Whamcloud - gitweb
LU-8922 lod: check master stripes properly
[fs/lustre-release.git] / lustre / mgs / mgs_llog.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2011, 2016, Intel Corporation.
27  */
28 /*
29  * This file is part of Lustre, http://www.lustre.org/
30  * Lustre is a trademark of Sun Microsystems, Inc.
31  *
32  * lustre/mgs/mgs_llog.c
33  *
34  * Lustre Management Server (mgs) config llog creation
35  *
36  * Author: Nathan Rutman <nathan@clusterfs.com>
37  * Author: Alex Zhuravlev <bzzz@whamcloud.com>
38  * Author: Mikhail Pershin <tappro@whamcloud.com>
39  */
40
41 #define DEBUG_SUBSYSTEM S_MGS
42 #define D_MGS D_CONFIG
43
44 #include <obd.h>
45 #include <lustre_ioctl.h>
46 #include <lustre_param.h>
47 #include <lustre_sec.h>
48 #include <lustre_quota.h>
49
50 #include "mgs_internal.h"
51
52 /********************** Class functions ********************/
53
54 /* Find all logs in CONFIG directory and link then into list */
55 int class_dentry_readdir(const struct lu_env *env,
56                          struct mgs_device *mgs, struct list_head *log_list)
57 {
58         struct dt_object    *dir = mgs->mgs_configs_dir;
59         const struct dt_it_ops *iops;
60         struct dt_it        *it;
61         struct mgs_direntry *de;
62         char                *key;
63         int                  rc, key_sz;
64
65         INIT_LIST_HEAD(log_list);
66
67         LASSERT(dir);
68         LASSERT(dir->do_index_ops);
69
70         iops = &dir->do_index_ops->dio_it;
71         it = iops->init(env, dir, LUDA_64BITHASH);
72         if (IS_ERR(it))
73                 RETURN(PTR_ERR(it));
74
75         rc = iops->load(env, it, 0);
76         if (rc <= 0)
77                 GOTO(fini, rc = 0);
78
79         /* main cycle */
80         do {
81                 key = (void *)iops->key(env, it);
82                 if (IS_ERR(key)) {
83                         CERROR("%s: key failed when listing %s: rc = %d\n",
84                                mgs->mgs_obd->obd_name, MOUNT_CONFIGS_DIR,
85                                (int) PTR_ERR(key));
86                         goto next;
87                 }
88                 key_sz = iops->key_size(env, it);
89                 LASSERT(key_sz > 0);
90
91                 /* filter out "." and ".." entries */
92                 if (key[0] == '.') {
93                         if (key_sz == 1)
94                                 goto next;
95                         if (key_sz == 2 && key[1] == '.')
96                                 goto next;
97                 }
98
99                 /* filter out ".bak" files */
100                 /* sizeof(".bak") - 1 == 3 */
101                 if (key_sz >= 3 &&
102                     !memcmp(".bak", key + key_sz - 3, 3)) {
103                         CDEBUG(D_MGS, "Skipping backup file %.*s\n",
104                                key_sz, key);
105                         goto next;
106                 }
107
108                 de = mgs_direntry_alloc(key_sz + 1);
109                 if (de == NULL) {
110                         rc = -ENOMEM;
111                         break;
112                 }
113
114                 memcpy(de->mde_name, key, key_sz);
115                 de->mde_name[key_sz] = 0;
116
117                 list_add(&de->mde_list, log_list);
118
119 next:
120                 rc = iops->next(env, it);
121         } while (rc == 0);
122         if (rc > 0)
123                 rc = 0;
124
125         iops->put(env, it);
126
127 fini:
128         iops->fini(env, it);
129         if (rc)
130                 CERROR("%s: key failed when listing %s: rc = %d\n",
131                        mgs->mgs_obd->obd_name, MOUNT_CONFIGS_DIR, rc);
132         RETURN(rc);
133 }
134
135 /******************** DB functions *********************/
136
137 static inline int name_create(char **newname, char *prefix, char *suffix)
138 {
139         LASSERT(newname);
140         OBD_ALLOC(*newname, strlen(prefix) + strlen(suffix) + 1);
141         if (!*newname)
142                 return -ENOMEM;
143         sprintf(*newname, "%s%s", prefix, suffix);
144         return 0;
145 }
146
147 static inline void name_destroy(char **name)
148 {
149         if (*name)
150                 OBD_FREE(*name, strlen(*name) + 1);
151         *name = NULL;
152 }
153
154 struct mgs_fsdb_handler_data
155 {
156         struct fs_db   *fsdb;
157         __u32           ver;
158 };
159
160 /* from the (client) config log, figure out:
161         1. which ost's/mdt's are configured (by index)
162         2. what the last config step is
163         3. COMPAT_18 osc name
164 */
165 /* It might be better to have a separate db file, instead of parsing the info
166    out of the client log.  This is slow and potentially error-prone. */
167 static int mgs_fsdb_handler(const struct lu_env *env, struct llog_handle *llh,
168                             struct llog_rec_hdr *rec, void *data)
169 {
170         struct mgs_fsdb_handler_data *d = data;
171         struct fs_db *fsdb = d->fsdb;
172         int cfg_len = rec->lrh_len;
173         char *cfg_buf = (char*) (rec + 1);
174         struct lustre_cfg *lcfg;
175         __u32 index;
176         int rc = 0;
177         ENTRY;
178
179         if (rec->lrh_type != OBD_CFG_REC) {
180                 CERROR("unhandled lrh_type: %#x\n", rec->lrh_type);
181                 RETURN(-EINVAL);
182         }
183
184         rc = lustre_cfg_sanity_check(cfg_buf, cfg_len);
185         if (rc) {
186                 CERROR("Insane cfg\n");
187                 RETURN(rc);
188         }
189
190         lcfg = (struct lustre_cfg *)cfg_buf;
191
192         CDEBUG(D_INFO, "cmd %x %s %s\n", lcfg->lcfg_command,
193                lustre_cfg_string(lcfg, 0), lustre_cfg_string(lcfg, 1));
194
195         /* Figure out ost indicies */
196         /* lov_modify_tgts add 0:lov1  1:ost1_UUID  2(index):0  3(gen):1 */
197         if (lcfg->lcfg_command == LCFG_LOV_ADD_OBD ||
198             lcfg->lcfg_command == LCFG_LOV_DEL_OBD) {
199                 index = simple_strtoul(lustre_cfg_string(lcfg, 2),
200                                        NULL, 10);
201                 CDEBUG(D_MGS, "OST index for %s is %u (%s)\n",
202                        lustre_cfg_string(lcfg, 1), index,
203                        lustre_cfg_string(lcfg, 2));
204                 set_bit(index, fsdb->fsdb_ost_index_map);
205         }
206
207         /* Figure out mdt indicies */
208         /* attach   0:MDC_uml1_mdsA_MNT_client  1:mdc  2:1d834_MNT_client_03f */
209         if ((lcfg->lcfg_command == LCFG_ATTACH) &&
210             (strcmp(lustre_cfg_string(lcfg, 1), LUSTRE_MDC_NAME) == 0)) {
211                 rc = server_name2index(lustre_cfg_string(lcfg, 0),
212                                        &index, NULL);
213                 if (rc != LDD_F_SV_TYPE_MDT) {
214                         CWARN("Unparsable MDC name %s, assuming index 0\n",
215                               lustre_cfg_string(lcfg, 0));
216                         index = 0;
217                 }
218                 rc = 0;
219                 CDEBUG(D_MGS, "MDT index is %u\n", index);
220                 set_bit(index, fsdb->fsdb_mdt_index_map);
221                 fsdb->fsdb_mdt_count ++;
222         }
223
224         /**
225          * figure out the old config. fsdb_gen = 0 means old log
226          * It is obsoleted and not supported anymore
227          */
228         if (fsdb->fsdb_gen == 0) {
229                 CERROR("Old config format is not supported\n");
230                 RETURN(-EINVAL);
231         }
232
233         /*
234          * compat to 1.8, check osc name used by MDT0 to OSTs, bz18548.
235          */
236         if (!test_bit(FSDB_OSCNAME18, &fsdb->fsdb_flags) &&
237             lcfg->lcfg_command == LCFG_ATTACH &&
238             strcmp(lustre_cfg_string(lcfg, 1), LUSTRE_OSC_NAME) == 0) {
239                 if (OBD_OCD_VERSION_MAJOR(d->ver) == 1 &&
240                     OBD_OCD_VERSION_MINOR(d->ver) <= 8) {
241                         CWARN("MDT using 1.8 OSC name scheme\n");
242                         set_bit(FSDB_OSCNAME18, &fsdb->fsdb_flags);
243                 }
244         }
245
246         if (lcfg->lcfg_command == LCFG_MARKER) {
247                 struct cfg_marker *marker;
248                 marker = lustre_cfg_buf(lcfg, 1);
249
250                 d->ver = marker->cm_vers;
251
252                 /* Keep track of the latest marker step */
253                 fsdb->fsdb_gen = max(fsdb->fsdb_gen, marker->cm_step);
254         }
255
256         RETURN(rc);
257 }
258
259 /* fsdb->fsdb_mutex is already held  in mgs_find_or_make_fsdb*/
260 static int mgs_get_fsdb_from_llog(const struct lu_env *env,
261                                   struct mgs_device *mgs,
262                                   struct fs_db *fsdb)
263 {
264         char                            *logname;
265         struct llog_handle              *loghandle;
266         struct llog_ctxt                *ctxt;
267         struct mgs_fsdb_handler_data     d = { fsdb, 0 };
268         int rc;
269
270         ENTRY;
271
272         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
273         LASSERT(ctxt != NULL);
274         rc = name_create(&logname, fsdb->fsdb_name, "-client");
275         if (rc)
276                 GOTO(out_put, rc);
277         rc = llog_open_create(env, ctxt, &loghandle, NULL, logname);
278         if (rc)
279                 GOTO(out_pop, rc);
280
281         rc = llog_init_handle(env, loghandle, LLOG_F_IS_PLAIN, NULL);
282         if (rc)
283                 GOTO(out_close, rc);
284
285         if (llog_get_size(loghandle) <= 1)
286                 set_bit(FSDB_LOG_EMPTY, &fsdb->fsdb_flags);
287
288         rc = llog_process(env, loghandle, mgs_fsdb_handler, (void *)&d, NULL);
289         CDEBUG(D_INFO, "get_db = %d\n", rc);
290 out_close:
291         llog_close(env, loghandle);
292 out_pop:
293         name_destroy(&logname);
294 out_put:
295         llog_ctxt_put(ctxt);
296
297         RETURN(rc);
298 }
299
300 static void mgs_free_fsdb_srpc(struct fs_db *fsdb)
301 {
302         struct mgs_tgt_srpc_conf *tgtconf;
303
304         /* free target-specific rules */
305         while (fsdb->fsdb_srpc_tgt) {
306                 tgtconf = fsdb->fsdb_srpc_tgt;
307                 fsdb->fsdb_srpc_tgt = tgtconf->mtsc_next;
308
309                 LASSERT(tgtconf->mtsc_tgt);
310
311                 sptlrpc_rule_set_free(&tgtconf->mtsc_rset);
312                 OBD_FREE(tgtconf->mtsc_tgt, strlen(tgtconf->mtsc_tgt) + 1);
313                 OBD_FREE_PTR(tgtconf);
314         }
315
316         /* free general rules */
317         sptlrpc_rule_set_free(&fsdb->fsdb_srpc_gen);
318 }
319
320 struct fs_db *mgs_find_fsdb(struct mgs_device *mgs, char *fsname)
321 {
322         struct fs_db *fsdb;
323         struct list_head *tmp;
324
325         list_for_each(tmp, &mgs->mgs_fs_db_list) {
326                 fsdb = list_entry(tmp, struct fs_db, fsdb_list);
327                 if (strcmp(fsdb->fsdb_name, fsname) == 0)
328                         return fsdb;
329         }
330         return NULL;
331 }
332
333 /* caller must hold the mgs->mgs_fs_db_lock */
334 static struct fs_db *mgs_new_fsdb(const struct lu_env *env,
335                                   struct mgs_device *mgs, char *fsname)
336 {
337         struct fs_db *fsdb;
338         int rc;
339         ENTRY;
340
341         if (strlen(fsname) >= sizeof(fsdb->fsdb_name)) {
342                 CERROR("fsname %s is too long\n", fsname);
343                 RETURN(ERR_PTR(-EINVAL));
344         }
345
346         OBD_ALLOC_PTR(fsdb);
347         if (!fsdb)
348                 RETURN(ERR_PTR(-ENOMEM));
349
350         strcpy(fsdb->fsdb_name, fsname);
351         mutex_init(&fsdb->fsdb_mutex);
352         set_bit(FSDB_UDESC, &fsdb->fsdb_flags);
353         fsdb->fsdb_gen = 1;
354
355         if (strcmp(fsname, MGSSELF_NAME) == 0) {
356                 set_bit(FSDB_MGS_SELF, &fsdb->fsdb_flags);
357                 fsdb->fsdb_mgs = mgs;
358         } else {
359                 OBD_ALLOC(fsdb->fsdb_ost_index_map, INDEX_MAP_SIZE);
360                 OBD_ALLOC(fsdb->fsdb_mdt_index_map, INDEX_MAP_SIZE);
361                 if (!fsdb->fsdb_ost_index_map || !fsdb->fsdb_mdt_index_map) {
362                         CERROR("No memory for index maps\n");
363                         GOTO(err, rc = -ENOMEM);
364                 }
365
366                 rc = name_create(&fsdb->fsdb_clilov, fsname, "-clilov");
367                 if (rc)
368                         GOTO(err, rc);
369                 rc = name_create(&fsdb->fsdb_clilmv, fsname, "-clilmv");
370                 if (rc)
371                         GOTO(err, rc);
372
373                 /* initialise data for NID table */
374                 mgs_ir_init_fs(env, mgs, fsdb);
375
376                 lproc_mgs_add_live(mgs, fsdb);
377         }
378
379         list_add(&fsdb->fsdb_list, &mgs->mgs_fs_db_list);
380
381         RETURN(fsdb);
382 err:
383         if (fsdb->fsdb_ost_index_map)
384                 OBD_FREE(fsdb->fsdb_ost_index_map, INDEX_MAP_SIZE);
385         if (fsdb->fsdb_mdt_index_map)
386                 OBD_FREE(fsdb->fsdb_mdt_index_map, INDEX_MAP_SIZE);
387         name_destroy(&fsdb->fsdb_clilov);
388         name_destroy(&fsdb->fsdb_clilmv);
389         OBD_FREE_PTR(fsdb);
390         RETURN(ERR_PTR(rc));
391 }
392
393 static void mgs_free_fsdb(struct mgs_device *mgs, struct fs_db *fsdb)
394 {
395         /* wait for anyone with the sem */
396         mutex_lock(&fsdb->fsdb_mutex);
397         lproc_mgs_del_live(mgs, fsdb);
398         list_del(&fsdb->fsdb_list);
399
400         /* deinitialize fsr */
401         mgs_ir_fini_fs(mgs, fsdb);
402
403         if (fsdb->fsdb_ost_index_map)
404                 OBD_FREE(fsdb->fsdb_ost_index_map, INDEX_MAP_SIZE);
405         if (fsdb->fsdb_mdt_index_map)
406                 OBD_FREE(fsdb->fsdb_mdt_index_map, INDEX_MAP_SIZE);
407         name_destroy(&fsdb->fsdb_clilov);
408         name_destroy(&fsdb->fsdb_clilmv);
409         mgs_free_fsdb_srpc(fsdb);
410         mutex_unlock(&fsdb->fsdb_mutex);
411         OBD_FREE_PTR(fsdb);
412 }
413
414 int mgs_init_fsdb_list(struct mgs_device *mgs)
415 {
416         INIT_LIST_HEAD(&mgs->mgs_fs_db_list);
417         return 0;
418 }
419
420 int mgs_cleanup_fsdb_list(struct mgs_device *mgs)
421 {
422         struct fs_db *fsdb;
423         struct list_head *tmp, *tmp2;
424
425         mutex_lock(&mgs->mgs_mutex);
426         list_for_each_safe(tmp, tmp2, &mgs->mgs_fs_db_list) {
427                 fsdb = list_entry(tmp, struct fs_db, fsdb_list);
428                 mgs_free_fsdb(mgs, fsdb);
429         }
430         mutex_unlock(&mgs->mgs_mutex);
431         return 0;
432 }
433
434 int mgs_find_or_make_fsdb(const struct lu_env *env,
435                           struct mgs_device *mgs, char *name,
436                           struct fs_db **dbh)
437 {
438         struct fs_db *fsdb;
439         int rc = 0;
440
441         ENTRY;
442         mutex_lock(&mgs->mgs_mutex);
443         fsdb = mgs_find_fsdb(mgs, name);
444         if (fsdb) {
445                 mutex_unlock(&mgs->mgs_mutex);
446                 *dbh = fsdb;
447                 RETURN(0);
448         }
449
450         CDEBUG(D_MGS, "Creating new db\n");
451         fsdb = mgs_new_fsdb(env, mgs, name);
452         /* lock fsdb_mutex until the db is loaded from llogs */
453         if (!IS_ERR(fsdb))
454                 mutex_lock(&fsdb->fsdb_mutex);
455         mutex_unlock(&mgs->mgs_mutex);
456         if (IS_ERR(fsdb))
457                 RETURN(PTR_ERR(fsdb));
458
459         if (!test_bit(FSDB_MGS_SELF, &fsdb->fsdb_flags)) {
460                 /* populate the db from the client llog */
461                 rc = mgs_get_fsdb_from_llog(env, mgs, fsdb);
462                 if (rc) {
463                         CERROR("Can't get db from client log %d\n", rc);
464                         GOTO(out_free, rc);
465                 }
466         }
467
468         /* populate srpc rules from params llog */
469         rc = mgs_get_fsdb_srpc_from_llog(env, mgs, fsdb);
470         if (rc) {
471                 CERROR("Can't get db from params log %d\n", rc);
472                 GOTO(out_free, rc);
473         }
474
475         mutex_unlock(&fsdb->fsdb_mutex);
476         *dbh = fsdb;
477
478         RETURN(0);
479
480 out_free:
481         mutex_unlock(&fsdb->fsdb_mutex);
482         mgs_free_fsdb(mgs, fsdb);
483         return rc;
484 }
485
486 /* 1 = index in use
487    0 = index unused
488    -1= empty client log */
489 int mgs_check_index(const struct lu_env *env,
490                     struct mgs_device *mgs,
491                     struct mgs_target_info *mti)
492 {
493         struct fs_db *fsdb;
494         void *imap;
495         int rc = 0;
496         ENTRY;
497
498         LASSERT(!(mti->mti_flags & LDD_F_NEED_INDEX));
499
500         rc = mgs_find_or_make_fsdb(env, mgs, mti->mti_fsname, &fsdb);
501         if (rc) {
502                 CERROR("Can't get db for %s\n", mti->mti_fsname);
503                 RETURN(rc);
504         }
505
506         if (test_bit(FSDB_LOG_EMPTY, &fsdb->fsdb_flags))
507                 RETURN(-1);
508
509         if (mti->mti_flags & LDD_F_SV_TYPE_OST)
510                 imap = fsdb->fsdb_ost_index_map;
511         else if (mti->mti_flags & LDD_F_SV_TYPE_MDT)
512                 imap = fsdb->fsdb_mdt_index_map;
513         else
514                 RETURN(-EINVAL);
515
516         if (test_bit(mti->mti_stripe_index, imap))
517                 RETURN(1);
518         RETURN(0);
519 }
520
521 static __inline__ int next_index(void *index_map, int map_len)
522 {
523         int i;
524         for (i = 0; i < map_len * 8; i++)
525                 if (!test_bit(i, index_map)) {
526                          return i;
527                  }
528         CERROR("max index %d exceeded.\n", i);
529         return -1;
530 }
531
532 /* Return codes:
533         0  newly marked as in use
534         <0 err
535         +EALREADY for update of an old index */
536 static int mgs_set_index(const struct lu_env *env,
537                          struct mgs_device *mgs,
538                          struct mgs_target_info *mti)
539 {
540         struct fs_db *fsdb;
541         void *imap;
542         int rc = 0;
543         ENTRY;
544
545         rc = mgs_find_or_make_fsdb(env, mgs, mti->mti_fsname, &fsdb);
546         if (rc) {
547                 CERROR("Can't get db for %s\n", mti->mti_fsname);
548                 RETURN(rc);
549         }
550
551         mutex_lock(&fsdb->fsdb_mutex);
552         if (mti->mti_flags & LDD_F_SV_TYPE_OST) {
553                 imap = fsdb->fsdb_ost_index_map;
554         } else if (mti->mti_flags & LDD_F_SV_TYPE_MDT) {
555                 imap = fsdb->fsdb_mdt_index_map;
556         } else {
557                 GOTO(out_up, rc = -EINVAL);
558         }
559
560         if (mti->mti_flags & LDD_F_NEED_INDEX) {
561                 rc = next_index(imap, INDEX_MAP_SIZE);
562                 if (rc == -1)
563                         GOTO(out_up, rc = -ERANGE);
564                 mti->mti_stripe_index = rc;
565                 if (mti->mti_flags & LDD_F_SV_TYPE_MDT)
566                         fsdb->fsdb_mdt_count ++;
567         }
568
569         /* the last index(0xffff) is reserved for default value. */
570         if (mti->mti_stripe_index >= INDEX_MAP_SIZE * 8 - 1) {
571                 LCONSOLE_ERROR_MSG(0x13f, "Server %s requested index %u, "
572                                    "but index must be less than %u.\n",
573                                    mti->mti_svname, mti->mti_stripe_index,
574                                    INDEX_MAP_SIZE * 8 - 1);
575                 GOTO(out_up, rc = -ERANGE);
576         }
577
578         if (test_bit(mti->mti_stripe_index, imap)) {
579                 if ((mti->mti_flags & LDD_F_VIRGIN) &&
580                     !(mti->mti_flags & LDD_F_WRITECONF)) {
581                         LCONSOLE_ERROR_MSG(0x140, "Server %s requested index "
582                                            "%d, but that index is already in "
583                                            "use. Use --writeconf to force\n",
584                                            mti->mti_svname,
585                                            mti->mti_stripe_index);
586                         GOTO(out_up, rc = -EADDRINUSE);
587                 } else {
588                         CDEBUG(D_MGS, "Server %s updating index %d\n",
589                                mti->mti_svname, mti->mti_stripe_index);
590                         GOTO(out_up, rc = EALREADY);
591                 }
592         }
593
594         set_bit(mti->mti_stripe_index, imap);
595         clear_bit(FSDB_LOG_EMPTY, &fsdb->fsdb_flags);
596         mutex_unlock(&fsdb->fsdb_mutex);
597         if (server_make_name(mti->mti_flags & ~(LDD_F_VIRGIN | LDD_F_WRITECONF),
598                              mti->mti_stripe_index, mti->mti_fsname,
599                              mti->mti_svname)) {
600                 CERROR("unknown server type %#x\n", mti->mti_flags);
601                 return -EINVAL;
602         }
603
604         CDEBUG(D_MGS, "Set index for %s to %d\n", mti->mti_svname,
605                mti->mti_stripe_index);
606
607         RETURN(0);
608 out_up:
609         mutex_unlock(&fsdb->fsdb_mutex);
610         return rc;
611 }
612
613 struct mgs_modify_lookup {
614         struct cfg_marker mml_marker;
615         int               mml_modified;
616 };
617
618 static int mgs_check_record_match(const struct lu_env *env,
619                                 struct llog_handle *llh,
620                                 struct llog_rec_hdr *rec, void *data)
621 {
622         struct cfg_marker *mc_marker = data;
623         struct cfg_marker *marker;
624         struct lustre_cfg *lcfg = REC_DATA(rec);
625         int cfg_len = REC_DATA_LEN(rec);
626         int rc;
627         ENTRY;
628
629
630         if (rec->lrh_type != OBD_CFG_REC) {
631                 CDEBUG(D_ERROR, "Unhandled lrh_type: %#x\n", rec->lrh_type);
632                 RETURN(-EINVAL);
633         }
634
635         rc = lustre_cfg_sanity_check(lcfg, cfg_len);
636         if (rc) {
637                 CDEBUG(D_ERROR, "Insane cfg\n");
638                 RETURN(rc);
639         }
640
641         /* We only care about markers */
642         if (lcfg->lcfg_command != LCFG_MARKER)
643                 RETURN(0);
644
645         marker = lustre_cfg_buf(lcfg, 1);
646
647         if (marker->cm_flags & CM_SKIP)
648                 RETURN(0);
649
650         if ((strcmp(mc_marker->cm_comment, marker->cm_comment) == 0) &&
651                 (strcmp(mc_marker->cm_tgtname, marker->cm_tgtname) == 0)) {
652                 /* Found a non-skipped marker match */
653                 CDEBUG(D_MGS, "Matched rec %u marker %d flag %x %s %s\n",
654                         rec->lrh_index, marker->cm_step,
655                         marker->cm_flags, marker->cm_tgtname,
656                         marker->cm_comment);
657                 rc = LLOG_PROC_BREAK;
658         }
659
660         RETURN(rc);
661 }
662
663 /**
664  * Check an existing config log record with matching comment and device
665  * Return code:
666  * 0 - checked successfully,
667  * LLOG_PROC_BREAK - record matches
668  * negative - error
669  */
670 static int mgs_check_marker(const struct lu_env *env, struct mgs_device *mgs,
671                 struct fs_db *fsdb, struct mgs_target_info *mti,
672                 char *logname, char *devname, char *comment)
673 {
674         struct llog_handle *loghandle;
675         struct llog_ctxt *ctxt;
676         struct cfg_marker *mc_marker;
677         int rc;
678
679         ENTRY;
680
681         LASSERT(mutex_is_locked(&fsdb->fsdb_mutex));
682         CDEBUG(D_MGS, "mgs check %s/%s/%s\n", logname, devname, comment);
683
684         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
685         LASSERT(ctxt != NULL);
686         rc = llog_open(env, ctxt, &loghandle, NULL, logname, LLOG_OPEN_EXISTS);
687         if (rc < 0) {
688                 if (rc == -ENOENT)
689                         rc = 0;
690                 GOTO(out_pop, rc);
691         }
692
693         rc = llog_init_handle(env, loghandle, LLOG_F_IS_PLAIN, NULL);
694         if (rc)
695                 GOTO(out_close, rc);
696
697         if (llog_get_size(loghandle) <= 1)
698                 GOTO(out_close, rc = 0);
699
700         OBD_ALLOC_PTR(mc_marker);
701         if (!mc_marker)
702                 GOTO(out_close, rc = -ENOMEM);
703         if (strlcpy(mc_marker->cm_comment, comment,
704                 sizeof(mc_marker->cm_comment)) >=
705                 sizeof(mc_marker->cm_comment))
706                 GOTO(out_free, rc = -E2BIG);
707         if (strlcpy(mc_marker->cm_tgtname, devname,
708                 sizeof(mc_marker->cm_tgtname)) >=
709                 sizeof(mc_marker->cm_tgtname))
710                 GOTO(out_free, rc = -E2BIG);
711
712         rc = llog_process(env, loghandle, mgs_check_record_match,
713                         (void *)mc_marker, NULL);
714
715 out_free:
716         OBD_FREE_PTR(mc_marker);
717
718 out_close:
719         llog_close(env, loghandle);
720 out_pop:
721         if (rc && rc != LLOG_PROC_BREAK)
722                 CDEBUG(D_ERROR, "%s: mgs check %s/%s failed: rc = %d\n",
723                         mgs->mgs_obd->obd_name, mti->mti_svname, comment, rc);
724         llog_ctxt_put(ctxt);
725         RETURN(rc);
726 }
727
728 static int mgs_modify_handler(const struct lu_env *env,
729                               struct llog_handle *llh,
730                               struct llog_rec_hdr *rec, void *data)
731 {
732         struct mgs_modify_lookup *mml = data;
733         struct cfg_marker *marker;
734         struct lustre_cfg *lcfg = REC_DATA(rec);
735         int cfg_len = REC_DATA_LEN(rec);
736         int rc;
737         ENTRY;
738
739         if (rec->lrh_type != OBD_CFG_REC) {
740                 CERROR("unhandled lrh_type: %#x\n", rec->lrh_type);
741                 RETURN(-EINVAL);
742         }
743
744         rc = lustre_cfg_sanity_check(lcfg, cfg_len);
745         if (rc) {
746                 CERROR("Insane cfg\n");
747                 RETURN(rc);
748         }
749
750         /* We only care about markers */
751         if (lcfg->lcfg_command != LCFG_MARKER)
752                 RETURN(0);
753
754         marker = lustre_cfg_buf(lcfg, 1);
755         if ((strcmp(mml->mml_marker.cm_comment, marker->cm_comment) == 0) &&
756             (strcmp(mml->mml_marker.cm_tgtname, marker->cm_tgtname) == 0) &&
757             !(marker->cm_flags & CM_SKIP)) {
758                 /* Found a non-skipped marker match */
759                 CDEBUG(D_MGS, "Changing rec %u marker %d %x->%x: %s %s\n",
760                        rec->lrh_index, marker->cm_step,
761                        marker->cm_flags, mml->mml_marker.cm_flags,
762                        marker->cm_tgtname, marker->cm_comment);
763                 /* Overwrite the old marker llog entry */
764                 marker->cm_flags &= ~CM_EXCLUDE; /* in case we're unexcluding */
765                 marker->cm_flags |= mml->mml_marker.cm_flags;
766                 marker->cm_canceltime = mml->mml_marker.cm_canceltime;
767                 rc = llog_write(env, llh, rec, rec->lrh_index);
768                 if (!rc)
769                          mml->mml_modified++;
770         }
771
772         RETURN(rc);
773 }
774
775 /**
776  * Modify an existing config log record (for CM_SKIP or CM_EXCLUDE)
777  * Return code:
778  * 0 - modified successfully,
779  * 1 - no modification was done
780  * negative - error
781  */
782 static int mgs_modify(const struct lu_env *env, struct mgs_device *mgs,
783                       struct fs_db *fsdb, struct mgs_target_info *mti,
784                       char *logname, char *devname, char *comment, int flags)
785 {
786         struct llog_handle *loghandle;
787         struct llog_ctxt *ctxt;
788         struct mgs_modify_lookup *mml;
789         int rc;
790
791         ENTRY;
792
793         LASSERT(mutex_is_locked(&fsdb->fsdb_mutex));
794         CDEBUG(D_MGS, "modify %s/%s/%s fl=%x\n", logname, devname, comment,
795                flags);
796
797         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
798         LASSERT(ctxt != NULL);
799         rc = llog_open(env, ctxt, &loghandle, NULL, logname, LLOG_OPEN_EXISTS);
800         if (rc < 0) {
801                 if (rc == -ENOENT)
802                         rc = 0;
803                 GOTO(out_pop, rc);
804         }
805
806         rc = llog_init_handle(env, loghandle, LLOG_F_IS_PLAIN, NULL);
807         if (rc)
808                 GOTO(out_close, rc);
809
810         if (llog_get_size(loghandle) <= 1)
811                 GOTO(out_close, rc = 0);
812
813         OBD_ALLOC_PTR(mml);
814         if (!mml)
815                 GOTO(out_close, rc = -ENOMEM);
816         if (strlcpy(mml->mml_marker.cm_comment, comment,
817                     sizeof(mml->mml_marker.cm_comment)) >=
818             sizeof(mml->mml_marker.cm_comment))
819                 GOTO(out_free, rc = -E2BIG);
820         if (strlcpy(mml->mml_marker.cm_tgtname, devname,
821                     sizeof(mml->mml_marker.cm_tgtname)) >=
822             sizeof(mml->mml_marker.cm_tgtname))
823                 GOTO(out_free, rc = -E2BIG);
824         /* Modify mostly means cancel */
825         mml->mml_marker.cm_flags = flags;
826         mml->mml_marker.cm_canceltime = flags ? cfs_time_current_sec() : 0;
827         mml->mml_modified = 0;
828         rc = llog_process(env, loghandle, mgs_modify_handler, (void *)mml,
829                           NULL);
830         if (!rc && !mml->mml_modified)
831                 rc = 1;
832
833 out_free:
834         OBD_FREE_PTR(mml);
835
836 out_close:
837         llog_close(env, loghandle);
838 out_pop:
839         if (rc < 0)
840                 CERROR("%s: modify %s/%s failed: rc = %d\n",
841                        mgs->mgs_obd->obd_name, mti->mti_svname, comment, rc);
842         llog_ctxt_put(ctxt);
843         RETURN(rc);
844 }
845
846 /** This structure is passed to mgs_replace_handler */
847 struct mgs_replace_uuid_lookup {
848         /* Nids are replaced for this target device */
849         struct mgs_target_info target;
850         /* Temporary modified llog */
851         struct llog_handle *temp_llh;
852         /* Flag is set if in target block*/
853         int in_target_device;
854         /* Nids already added. Just skip (multiple nids) */
855         int device_nids_added;
856         /* Flag is set if this block should not be copied */
857         int skip_it;
858 };
859
860 /**
861  * Check: a) if block should be skipped
862  * b) is it target block
863  *
864  * \param[in] lcfg
865  * \param[in] mrul
866  *
867  * \retval 0 should not to be skipped
868  * \retval 1 should to be skipped
869  */
870 static int check_markers(struct lustre_cfg *lcfg,
871                          struct mgs_replace_uuid_lookup *mrul)
872 {
873          struct cfg_marker *marker;
874
875         /* Track markers. Find given device */
876         if (lcfg->lcfg_command == LCFG_MARKER) {
877                 marker = lustre_cfg_buf(lcfg, 1);
878                 /* Clean llog from records marked as CM_EXCLUDE.
879                    CM_SKIP records are used for "active" command
880                    and can be restored if needed */
881                 if ((marker->cm_flags & (CM_EXCLUDE | CM_START)) ==
882                     (CM_EXCLUDE | CM_START)) {
883                         mrul->skip_it = 1;
884                         return 1;
885                 }
886
887                 if ((marker->cm_flags & (CM_EXCLUDE | CM_END)) ==
888                     (CM_EXCLUDE | CM_END)) {
889                         mrul->skip_it = 0;
890                         return 1;
891                 }
892
893                 if (strcmp(mrul->target.mti_svname, marker->cm_tgtname) == 0) {
894                         LASSERT(!(marker->cm_flags & CM_START) ||
895                                 !(marker->cm_flags & CM_END));
896                         if (marker->cm_flags & CM_START) {
897                                 mrul->in_target_device = 1;
898                                 mrul->device_nids_added = 0;
899                         } else if (marker->cm_flags & CM_END)
900                                 mrul->in_target_device = 0;
901                 }
902         }
903
904         return 0;
905 }
906
907 static int record_base(const struct lu_env *env, struct llog_handle *llh,
908                      char *cfgname, lnet_nid_t nid, int cmd,
909                      char *s1, char *s2, char *s3, char *s4)
910 {
911         struct mgs_thread_info  *mgi = mgs_env_info(env);
912         struct llog_cfg_rec     *lcr;
913         int rc;
914
915         CDEBUG(D_MGS, "lcfg %s %#x %s %s %s %s\n", cfgname,
916                cmd, s1, s2, s3, s4);
917
918         lustre_cfg_bufs_reset(&mgi->mgi_bufs, cfgname);
919         if (s1)
920                 lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 1, s1);
921         if (s2)
922                 lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 2, s2);
923         if (s3)
924                 lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 3, s3);
925         if (s4)
926                 lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 4, s4);
927
928         lcr = lustre_cfg_rec_new(cmd, &mgi->mgi_bufs);
929         if (lcr == NULL)
930                 return -ENOMEM;
931
932         lcr->lcr_cfg.lcfg_nid = nid;
933         rc = llog_write(env, llh, &lcr->lcr_hdr, LLOG_NEXT_IDX);
934
935         lustre_cfg_rec_free(lcr);
936
937         if (rc < 0)
938                 CDEBUG(D_MGS,
939                        "failed to write lcfg %s %#x %s %s %s %s: rc = %d\n",
940                        cfgname, cmd, s1, s2, s3, s4, rc);
941         return rc;
942 }
943
944 static inline int record_add_uuid(const struct lu_env *env,
945                                   struct llog_handle *llh,
946                                   uint64_t nid, char *uuid)
947 {
948         return record_base(env, llh, NULL, nid, LCFG_ADD_UUID, uuid,
949                            NULL, NULL, NULL);
950 }
951
952 static inline int record_add_conn(const struct lu_env *env,
953                                   struct llog_handle *llh,
954                                   char *devname, char *uuid)
955 {
956         return record_base(env, llh, devname, 0, LCFG_ADD_CONN, uuid,
957                            NULL, NULL, NULL);
958 }
959
960 static inline int record_attach(const struct lu_env *env,
961                                 struct llog_handle *llh, char *devname,
962                                 char *type, char *uuid)
963 {
964         return record_base(env, llh, devname, 0, LCFG_ATTACH, type, uuid,
965                            NULL, NULL);
966 }
967
968 static inline int record_setup(const struct lu_env *env,
969                                struct llog_handle *llh, char *devname,
970                                char *s1, char *s2, char *s3, char *s4)
971 {
972         return record_base(env, llh, devname, 0, LCFG_SETUP, s1, s2, s3, s4);
973 }
974
975 /**
976  * \retval <0 record processing error
977  * \retval n record is processed. No need copy original one.
978  * \retval 0 record is not processed.
979  */
980 static int process_command(const struct lu_env *env, struct lustre_cfg *lcfg,
981                            struct mgs_replace_uuid_lookup *mrul)
982 {
983         int nids_added = 0;
984         lnet_nid_t nid;
985         char *ptr;
986         int rc;
987
988         if (lcfg->lcfg_command == LCFG_ADD_UUID) {
989                 /* LCFG_ADD_UUID command found. Let's skip original command
990                    and add passed nids */
991                 ptr = mrul->target.mti_params;
992                 while (class_parse_nid(ptr, &nid, &ptr) == 0) {
993                         CDEBUG(D_MGS, "add nid %s with uuid %s, "
994                                "device %s\n", libcfs_nid2str(nid),
995                                 mrul->target.mti_params,
996                                 mrul->target.mti_svname);
997                         rc = record_add_uuid(env,
998                                              mrul->temp_llh, nid,
999                                              mrul->target.mti_params);
1000                         if (!rc)
1001                                 nids_added++;
1002                 }
1003
1004                 if (nids_added == 0) {
1005                         CERROR("No new nids were added, nid %s with uuid %s, "
1006                                "device %s\n", libcfs_nid2str(nid),
1007                                mrul->target.mti_params,
1008                                mrul->target.mti_svname);
1009                         RETURN(-ENXIO);
1010                 } else {
1011                         mrul->device_nids_added = 1;
1012                 }
1013
1014                 return nids_added;
1015         }
1016
1017         if (mrul->device_nids_added && lcfg->lcfg_command == LCFG_SETUP) {
1018                 /* LCFG_SETUP command found. UUID should be changed */
1019                 rc = record_setup(env,
1020                                   mrul->temp_llh,
1021                                   /* devname the same */
1022                                   lustre_cfg_string(lcfg, 0),
1023                                   /* s1 is not changed */
1024                                   lustre_cfg_string(lcfg, 1),
1025                                   /* new uuid should be
1026                                   the full nidlist */
1027                                   mrul->target.mti_params,
1028                                   /* s3 is not changed */
1029                                   lustre_cfg_string(lcfg, 3),
1030                                   /* s4 is not changed */
1031                                   lustre_cfg_string(lcfg, 4));
1032                 return rc ? rc : 1;
1033         }
1034
1035         /* Another commands in target device block */
1036         return 0;
1037 }
1038
1039 /**
1040  * Handler that called for every record in llog.
1041  * Records are processed in order they placed in llog.
1042  *
1043  * \param[in] llh       log to be processed
1044  * \param[in] rec       current record
1045  * \param[in] data      mgs_replace_uuid_lookup structure
1046  *
1047  * \retval 0    success
1048  */
1049 static int mgs_replace_handler(const struct lu_env *env,
1050                                struct llog_handle *llh,
1051                                struct llog_rec_hdr *rec,
1052                                void *data)
1053 {
1054         struct mgs_replace_uuid_lookup *mrul;
1055         struct lustre_cfg *lcfg = REC_DATA(rec);
1056         int cfg_len = REC_DATA_LEN(rec);
1057         int rc;
1058         ENTRY;
1059
1060         mrul = (struct mgs_replace_uuid_lookup *)data;
1061
1062         if (rec->lrh_type != OBD_CFG_REC) {
1063                 CERROR("unhandled lrh_type: %#x, cmd %x %s %s\n",
1064                        rec->lrh_type, lcfg->lcfg_command,
1065                        lustre_cfg_string(lcfg, 0),
1066                        lustre_cfg_string(lcfg, 1));
1067                 RETURN(-EINVAL);
1068         }
1069
1070         rc = lustre_cfg_sanity_check(lcfg, cfg_len);
1071         if (rc) {
1072                 /* Do not copy any invalidated records */
1073                 GOTO(skip_out, rc = 0);
1074         }
1075
1076         rc = check_markers(lcfg, mrul);
1077         if (rc || mrul->skip_it)
1078                 GOTO(skip_out, rc = 0);
1079
1080         /* Write to new log all commands outside target device block */
1081         if (!mrul->in_target_device)
1082                 GOTO(copy_out, rc = 0);
1083
1084         /* Skip all other LCFG_ADD_UUID and LCFG_ADD_CONN records
1085            (failover nids) for this target, assuming that if then
1086            primary is changing then so is the failover */
1087         if (mrul->device_nids_added &&
1088             (lcfg->lcfg_command == LCFG_ADD_UUID ||
1089              lcfg->lcfg_command == LCFG_ADD_CONN))
1090                 GOTO(skip_out, rc = 0);
1091
1092         rc = process_command(env, lcfg, mrul);
1093         if (rc < 0)
1094                 RETURN(rc);
1095
1096         if (rc)
1097                 RETURN(0);
1098 copy_out:
1099         /* Record is placed in temporary llog as is */
1100         rc = llog_write(env, mrul->temp_llh, rec, LLOG_NEXT_IDX);
1101
1102         CDEBUG(D_MGS, "Copied idx=%d, rc=%d, len=%d, cmd %x %s %s\n",
1103                rec->lrh_index, rc, rec->lrh_len, lcfg->lcfg_command,
1104                lustre_cfg_string(lcfg, 0), lustre_cfg_string(lcfg, 1));
1105         RETURN(rc);
1106
1107 skip_out:
1108         CDEBUG(D_MGS, "Skipped idx=%d, rc=%d, len=%d, cmd %x %s %s\n",
1109                rec->lrh_index, rc, rec->lrh_len, lcfg->lcfg_command,
1110                lustre_cfg_string(lcfg, 0), lustre_cfg_string(lcfg, 1));
1111         RETURN(rc);
1112 }
1113
1114 static int mgs_log_is_empty(const struct lu_env *env,
1115                             struct mgs_device *mgs, char *name)
1116 {
1117         struct llog_ctxt        *ctxt;
1118         int                      rc;
1119
1120         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
1121         LASSERT(ctxt != NULL);
1122
1123         rc = llog_is_empty(env, ctxt, name);
1124         llog_ctxt_put(ctxt);
1125         return rc;
1126 }
1127
1128 static int mgs_replace_nids_log(const struct lu_env *env,
1129                                 struct obd_device *mgs, struct fs_db *fsdb,
1130                                 char *logname, char *devname, char *nids)
1131 {
1132         struct llog_handle *orig_llh, *backup_llh;
1133         struct llog_ctxt *ctxt;
1134         struct mgs_replace_uuid_lookup *mrul;
1135         struct mgs_device *mgs_dev = lu2mgs_dev(mgs->obd_lu_dev);
1136         static struct obd_uuid   cfg_uuid = { .uuid = "config_uuid" };
1137         char *backup;
1138         int rc, rc2;
1139         ENTRY;
1140
1141         CDEBUG(D_MGS, "Replace nids for %s in %s\n", devname, logname);
1142
1143         ctxt = llog_get_context(mgs, LLOG_CONFIG_ORIG_CTXT);
1144         LASSERT(ctxt != NULL);
1145
1146         if (mgs_log_is_empty(env, mgs_dev, logname)) {
1147                 /* Log is empty. Nothing to replace */
1148                 GOTO(out_put, rc = 0);
1149         }
1150
1151         OBD_ALLOC(backup, strlen(logname) + strlen(".bak") + 1);
1152         if (backup == NULL)
1153                 GOTO(out_put, rc = -ENOMEM);
1154
1155         sprintf(backup, "%s.bak", logname);
1156
1157         rc = llog_backup(env, mgs, ctxt, ctxt, logname, backup);
1158         if (rc == 0) {
1159                 /* Now erase original log file. Connections are not allowed.
1160                    Backup is already saved */
1161                 rc = llog_erase(env, ctxt, NULL, logname);
1162                 if (rc < 0)
1163                         GOTO(out_free, rc);
1164         } else if (rc != -ENOENT) {
1165                 CERROR("%s: can't make backup for %s: rc = %d\n",
1166                        mgs->obd_name, logname, rc);
1167                 GOTO(out_free,rc);
1168         }
1169
1170         /* open local log */
1171         rc = llog_open_create(env, ctxt, &orig_llh, NULL, logname);
1172         if (rc)
1173                 GOTO(out_restore, rc);
1174
1175         rc = llog_init_handle(env, orig_llh, LLOG_F_IS_PLAIN, &cfg_uuid);
1176         if (rc)
1177                 GOTO(out_closel, rc);
1178
1179         /* open backup llog */
1180         rc = llog_open(env, ctxt, &backup_llh, NULL, backup,
1181                        LLOG_OPEN_EXISTS);
1182         if (rc)
1183                 GOTO(out_closel, rc);
1184
1185         rc = llog_init_handle(env, backup_llh, LLOG_F_IS_PLAIN, NULL);
1186         if (rc)
1187                 GOTO(out_close, rc);
1188
1189         if (llog_get_size(backup_llh) <= 1)
1190                 GOTO(out_close, rc = 0);
1191
1192         OBD_ALLOC_PTR(mrul);
1193         if (!mrul)
1194                 GOTO(out_close, rc = -ENOMEM);
1195         /* devname is only needed information to replace UUID records */
1196         strlcpy(mrul->target.mti_svname, devname,
1197                 sizeof(mrul->target.mti_svname));
1198         /* parse nids later */
1199         strlcpy(mrul->target.mti_params, nids, sizeof(mrul->target.mti_params));
1200         /* Copy records to this temporary llog */
1201         mrul->temp_llh = orig_llh;
1202
1203         rc = llog_process(env, backup_llh, mgs_replace_handler,
1204                           (void *)mrul, NULL);
1205         OBD_FREE_PTR(mrul);
1206 out_close:
1207         rc2 = llog_close(NULL, backup_llh);
1208         if (!rc)
1209                 rc = rc2;
1210 out_closel:
1211         rc2 = llog_close(NULL, orig_llh);
1212         if (!rc)
1213                 rc = rc2;
1214
1215 out_restore:
1216         if (rc) {
1217                 CERROR("%s: llog should be restored: rc = %d\n",
1218                        mgs->obd_name, rc);
1219                 rc2 = llog_backup(env, mgs, ctxt, ctxt, backup,
1220                                   logname);
1221                 if (rc2 < 0)
1222                         CERROR("%s: can't restore backup %s: rc = %d\n",
1223                                mgs->obd_name, logname, rc2);
1224         }
1225
1226 out_free:
1227         OBD_FREE(backup, strlen(backup) + 1);
1228
1229 out_put:
1230         llog_ctxt_put(ctxt);
1231
1232         if (rc)
1233                 CERROR("%s: failed to replace nids in log %s: rc = %d\n",
1234                        mgs->obd_name, logname, rc);
1235
1236         RETURN(rc);
1237 }
1238
1239 /**
1240  * Parse device name and get file system name and/or device index
1241  *
1242  * \param[in]   devname device name (ex. lustre-MDT0000)
1243  * \param[out]  fsname  file system name(optional)
1244  * \param[out]  index   device index(optional)
1245  *
1246  * \retval 0    success
1247  */
1248 static int mgs_parse_devname(char *devname, char *fsname, __u32 *index)
1249 {
1250         int rc;
1251         ENTRY;
1252
1253         /* Extract fsname */
1254         if (fsname) {
1255                 rc = server_name2fsname(devname, fsname, NULL);
1256                 if (rc < 0) {
1257                         CDEBUG(D_MGS, "Device name %s without fsname\n",
1258                                devname);
1259                         RETURN(-EINVAL);
1260                 }
1261         }
1262
1263         if (index) {
1264                 rc = server_name2index(devname, index, NULL);
1265                 if (rc < 0) {
1266                         CDEBUG(D_MGS, "Device name %s with wrong index\n",
1267                                devname);
1268                         RETURN(-EINVAL);
1269                 }
1270         }
1271
1272         RETURN(0);
1273 }
1274
1275 /* This is only called during replace_nids */
1276 static int only_mgs_is_running(struct obd_device *mgs_obd)
1277 {
1278         /* TDB: Is global variable with devices count exists? */
1279         int num_devices = get_devices_count();
1280         int num_exports = 0;
1281         struct obd_export *exp;
1282
1283         spin_lock(&mgs_obd->obd_dev_lock);
1284         list_for_each_entry(exp, &mgs_obd->obd_exports, exp_obd_chain) {
1285                 /* skip self export */
1286                 if (exp == mgs_obd->obd_self_export)
1287                         continue;
1288                 if (exp_connect_flags(exp) & OBD_CONNECT_MDS_MDS)
1289                         continue;
1290
1291                 ++num_exports;
1292
1293                 CERROR("%s: node %s still connected during replace_nids "
1294                        "connect_flags:%llx\n",
1295                        mgs_obd->obd_name,
1296                        libcfs_nid2str(exp->exp_nid_stats->nid),
1297                        exp_connect_flags(exp));
1298
1299         }
1300         spin_unlock(&mgs_obd->obd_dev_lock);
1301
1302         /* osd, MGS and MGC + self_export
1303            (wc -l /proc/fs/lustre/devices <= 2) && (non self exports == 0) */
1304         return (num_devices <= 3) && (num_exports == 0);
1305 }
1306
1307 static int name_create_mdt(char **logname, char *fsname, int i)
1308 {
1309         char mdt_index[9];
1310
1311         sprintf(mdt_index, "-MDT%04x", i);
1312         return name_create(logname, fsname, mdt_index);
1313 }
1314
1315 /**
1316  * Replace nids for \a device to \a nids values
1317  *
1318  * \param obd           MGS obd device
1319  * \param devname       nids need to be replaced for this device
1320  * (ex. lustre-OST0000)
1321  * \param nids          nids list (ex. nid1,nid2,nid3)
1322  *
1323  * \retval 0    success
1324  */
1325 int mgs_replace_nids(const struct lu_env *env,
1326                      struct mgs_device *mgs,
1327                      char *devname, char *nids)
1328 {
1329         /* Assume fsname is part of device name */
1330         char fsname[MTI_NAME_MAXLEN];
1331         int rc;
1332         __u32 index;
1333         char *logname;
1334         struct fs_db *fsdb;
1335         unsigned int i;
1336         int conn_state;
1337         struct obd_device *mgs_obd = mgs->mgs_obd;
1338         ENTRY;
1339
1340         /* We can only change NIDs if no other nodes are connected */
1341         spin_lock(&mgs_obd->obd_dev_lock);
1342         conn_state = mgs_obd->obd_no_conn;
1343         mgs_obd->obd_no_conn = 1;
1344         spin_unlock(&mgs_obd->obd_dev_lock);
1345
1346         /* We can not change nids if not only MGS is started */
1347         if (!only_mgs_is_running(mgs_obd)) {
1348                 CERROR("Only MGS is allowed to be started\n");
1349                 GOTO(out, rc = -EINPROGRESS);
1350         }
1351
1352         /* Get fsname and index*/
1353         rc = mgs_parse_devname(devname, fsname, &index);
1354         if (rc)
1355                 GOTO(out, rc);
1356
1357         rc = mgs_find_or_make_fsdb(env, mgs, fsname, &fsdb);
1358         if (rc) {
1359                 CERROR("%s: can't find fsdb: rc = %d\n", fsname, rc);
1360                 GOTO(out, rc);
1361         }
1362
1363         /* Process client llogs */
1364         name_create(&logname, fsname, "-client");
1365         rc = mgs_replace_nids_log(env, mgs_obd, fsdb, logname, devname, nids);
1366         name_destroy(&logname);
1367         if (rc) {
1368                 CERROR("%s: error while replacing NIDs for %s: rc = %d\n",
1369                        fsname, devname, rc);
1370                 GOTO(out, rc);
1371         }
1372
1373         /* Process MDT llogs */
1374         for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
1375                 if (!test_bit(i, fsdb->fsdb_mdt_index_map))
1376                         continue;
1377                 name_create_mdt(&logname, fsname, i);
1378                 rc = mgs_replace_nids_log(env, mgs_obd, fsdb, logname, devname, nids);
1379                 name_destroy(&logname);
1380                 if (rc)
1381                         GOTO(out, rc);
1382         }
1383
1384 out:
1385         spin_lock(&mgs_obd->obd_dev_lock);
1386         mgs_obd->obd_no_conn = conn_state;
1387         spin_unlock(&mgs_obd->obd_dev_lock);
1388
1389         RETURN(rc);
1390 }
1391
1392 static int record_lov_setup(const struct lu_env *env, struct llog_handle *llh,
1393                             char *devname, struct lov_desc *desc)
1394 {
1395         struct mgs_thread_info  *mgi = mgs_env_info(env);
1396         struct llog_cfg_rec     *lcr;
1397         int rc;
1398
1399         lustre_cfg_bufs_reset(&mgi->mgi_bufs, devname);
1400         lustre_cfg_bufs_set(&mgi->mgi_bufs, 1, desc, sizeof(*desc));
1401         lcr = lustre_cfg_rec_new(LCFG_SETUP, &mgi->mgi_bufs);
1402         if (lcr == NULL)
1403                 return -ENOMEM;
1404
1405         rc = llog_write(env, llh, &lcr->lcr_hdr, LLOG_NEXT_IDX);
1406         lustre_cfg_rec_free(lcr);
1407         return rc;
1408 }
1409
1410 static int record_lmv_setup(const struct lu_env *env, struct llog_handle *llh,
1411                             char *devname, struct lmv_desc *desc)
1412 {
1413         struct mgs_thread_info  *mgi = mgs_env_info(env);
1414         struct llog_cfg_rec     *lcr;
1415         int rc;
1416
1417         lustre_cfg_bufs_reset(&mgi->mgi_bufs, devname);
1418         lustre_cfg_bufs_set(&mgi->mgi_bufs, 1, desc, sizeof(*desc));
1419         lcr = lustre_cfg_rec_new(LCFG_SETUP, &mgi->mgi_bufs);
1420         if (lcr == NULL)
1421                 return -ENOMEM;
1422
1423         rc = llog_write(env, llh, &lcr->lcr_hdr, LLOG_NEXT_IDX);
1424         lustre_cfg_rec_free(lcr);
1425         return rc;
1426 }
1427
1428 static inline int record_mdc_add(const struct lu_env *env,
1429                                  struct llog_handle *llh,
1430                                  char *logname, char *mdcuuid,
1431                                  char *mdtuuid, char *index,
1432                                  char *gen)
1433 {
1434         return record_base(env,llh,logname,0,LCFG_ADD_MDC,
1435                            mdtuuid,index,gen,mdcuuid);
1436 }
1437
1438 static inline int record_lov_add(const struct lu_env *env,
1439                                  struct llog_handle *llh,
1440                                  char *lov_name, char *ost_uuid,
1441                                  char *index, char *gen)
1442 {
1443         return record_base(env, llh, lov_name, 0, LCFG_LOV_ADD_OBD,
1444                            ost_uuid, index, gen, NULL);
1445 }
1446
1447 static inline int record_mount_opt(const struct lu_env *env,
1448                                    struct llog_handle *llh,
1449                                    char *profile, char *lov_name,
1450                                    char *mdc_name)
1451 {
1452         return record_base(env, llh, NULL, 0, LCFG_MOUNTOPT,
1453                            profile, lov_name, mdc_name, NULL);
1454 }
1455
1456 static int record_marker(const struct lu_env *env,
1457                          struct llog_handle *llh,
1458                          struct fs_db *fsdb, __u32 flags,
1459                          char *tgtname, char *comment)
1460 {
1461         struct mgs_thread_info *mgi = mgs_env_info(env);
1462         struct llog_cfg_rec *lcr;
1463         int rc;
1464         int cplen = 0;
1465
1466         if (flags & CM_START)
1467                 fsdb->fsdb_gen++;
1468         mgi->mgi_marker.cm_step = fsdb->fsdb_gen;
1469         mgi->mgi_marker.cm_flags = flags;
1470         mgi->mgi_marker.cm_vers = LUSTRE_VERSION_CODE;
1471         cplen = strlcpy(mgi->mgi_marker.cm_tgtname, tgtname,
1472                         sizeof(mgi->mgi_marker.cm_tgtname));
1473         if (cplen >= sizeof(mgi->mgi_marker.cm_tgtname))
1474                 return -E2BIG;
1475         cplen = strlcpy(mgi->mgi_marker.cm_comment, comment,
1476                         sizeof(mgi->mgi_marker.cm_comment));
1477         if (cplen >= sizeof(mgi->mgi_marker.cm_comment))
1478                 return -E2BIG;
1479         mgi->mgi_marker.cm_createtime = cfs_time_current_sec();
1480         mgi->mgi_marker.cm_canceltime = 0;
1481         lustre_cfg_bufs_reset(&mgi->mgi_bufs, NULL);
1482         lustre_cfg_bufs_set(&mgi->mgi_bufs, 1, &mgi->mgi_marker,
1483                             sizeof(mgi->mgi_marker));
1484         lcr = lustre_cfg_rec_new(LCFG_MARKER, &mgi->mgi_bufs);
1485         if (lcr == NULL)
1486                 return -ENOMEM;
1487
1488         rc = llog_write(env, llh, &lcr->lcr_hdr, LLOG_NEXT_IDX);
1489         lustre_cfg_rec_free(lcr);
1490         return rc;
1491 }
1492
1493 static int record_start_log(const struct lu_env *env, struct mgs_device *mgs,
1494                             struct llog_handle **llh, char *name)
1495 {
1496         static struct obd_uuid   cfg_uuid = { .uuid = "config_uuid" };
1497         struct llog_ctxt        *ctxt;
1498         int                      rc = 0;
1499         ENTRY;
1500
1501         if (*llh)
1502                 GOTO(out, rc = -EBUSY);
1503
1504         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
1505         if (!ctxt)
1506                 GOTO(out, rc = -ENODEV);
1507         LASSERT(ctxt->loc_obd == mgs->mgs_obd);
1508
1509         rc = llog_open_create(env, ctxt, llh, NULL, name);
1510         if (rc)
1511                 GOTO(out_ctxt, rc);
1512         rc = llog_init_handle(env, *llh, LLOG_F_IS_PLAIN, &cfg_uuid);
1513         if (rc)
1514                 llog_close(env, *llh);
1515 out_ctxt:
1516         llog_ctxt_put(ctxt);
1517 out:
1518         if (rc) {
1519                 CERROR("%s: can't start log %s: rc = %d\n",
1520                        mgs->mgs_obd->obd_name, name, rc);
1521                 *llh = NULL;
1522         }
1523         RETURN(rc);
1524 }
1525
1526 static int record_end_log(const struct lu_env *env, struct llog_handle **llh)
1527 {
1528         int rc;
1529
1530         rc = llog_close(env, *llh);
1531         *llh = NULL;
1532
1533         return rc;
1534 }
1535
1536 /******************** config "macros" *********************/
1537
1538 /* write an lcfg directly into a log (with markers) */
1539 static int mgs_write_log_direct(const struct lu_env *env,
1540                                 struct mgs_device *mgs, struct fs_db *fsdb,
1541                                 char *logname, struct llog_cfg_rec *lcr,
1542                                 char *devname, char *comment)
1543 {
1544         struct llog_handle *llh = NULL;
1545         int rc;
1546
1547         ENTRY;
1548
1549         rc = record_start_log(env, mgs, &llh, logname);
1550         if (rc)
1551                 RETURN(rc);
1552
1553         /* FIXME These should be a single journal transaction */
1554         rc = record_marker(env, llh, fsdb, CM_START, devname, comment);
1555         if (rc)
1556                 GOTO(out_end, rc);
1557         rc = llog_write(env, llh, &lcr->lcr_hdr, LLOG_NEXT_IDX);
1558         if (rc)
1559                 GOTO(out_end, rc);
1560         rc = record_marker(env, llh, fsdb, CM_END, devname, comment);
1561         if (rc)
1562                 GOTO(out_end, rc);
1563 out_end:
1564         record_end_log(env, &llh);
1565         RETURN(rc);
1566 }
1567
1568 /* write the lcfg in all logs for the given fs */
1569 static int mgs_write_log_direct_all(const struct lu_env *env,
1570                                     struct mgs_device *mgs,
1571                                     struct fs_db *fsdb,
1572                                     struct mgs_target_info *mti,
1573                                     struct llog_cfg_rec *lcr, char *devname,
1574                                     char *comment, int server_only)
1575 {
1576         struct list_head         log_list;
1577         struct mgs_direntry     *dirent, *n;
1578         char                    *fsname = mti->mti_fsname;
1579         int                      rc = 0, len = strlen(fsname);
1580
1581         ENTRY;
1582         /* Find all the logs in the CONFIGS directory */
1583         rc = class_dentry_readdir(env, mgs, &log_list);
1584         if (rc)
1585                 RETURN(rc);
1586
1587         /* Could use fsdb index maps instead of directory listing */
1588         list_for_each_entry_safe(dirent, n, &log_list, mde_list) {
1589                 list_del_init(&dirent->mde_list);
1590                 /* don't write to sptlrpc rule log */
1591                 if (strstr(dirent->mde_name, "-sptlrpc") != NULL)
1592                         goto next;
1593
1594                 /* caller wants write server logs only */
1595                 if (server_only && strstr(dirent->mde_name, "-client") != NULL)
1596                         goto next;
1597
1598                 if (strlen(dirent->mde_name) <= len ||
1599                     strncmp(fsname, dirent->mde_name, len) != 0 ||
1600                     dirent->mde_name[len] != '-')
1601                         goto next;
1602
1603                 CDEBUG(D_MGS, "Changing log %s\n", dirent->mde_name);
1604                 /* Erase any old settings of this same parameter */
1605                 rc = mgs_modify(env, mgs, fsdb, mti, dirent->mde_name,
1606                                 devname, comment, CM_SKIP);
1607                 if (rc < 0)
1608                         CERROR("%s: Can't modify llog %s: rc = %d\n",
1609                                mgs->mgs_obd->obd_name, dirent->mde_name, rc);
1610                 if (lcr == NULL)
1611                         goto next;
1612                 /* Write the new one */
1613                 rc = mgs_write_log_direct(env, mgs, fsdb, dirent->mde_name,
1614                                           lcr, devname, comment);
1615                 if (rc != 0)
1616                         CERROR("%s: writing log %s: rc = %d\n",
1617                                mgs->mgs_obd->obd_name, dirent->mde_name, rc);
1618 next:
1619                 mgs_direntry_free(dirent);
1620         }
1621
1622         RETURN(rc);
1623 }
1624
1625 static int mgs_write_log_osp_to_mdt(const struct lu_env *env,
1626                                     struct mgs_device *mgs,
1627                                     struct fs_db *fsdb,
1628                                     struct mgs_target_info *mti,
1629                                     int index, char *logname);
1630 static int mgs_write_log_osc_to_lov(const struct lu_env *env,
1631                                     struct mgs_device *mgs,
1632                                     struct fs_db *fsdb,
1633                                     struct mgs_target_info *mti,
1634                                     char *logname, char *suffix, char *lovname,
1635                                     enum lustre_sec_part sec_part, int flags);
1636 static int name_create_mdt_and_lov(char **logname, char **lovname,
1637                                    struct fs_db *fsdb, int i);
1638
1639 static int add_param(char *params, char *key, char *val)
1640 {
1641         char *start = params + strlen(params);
1642         char *end = params + sizeof(((struct mgs_target_info *)0)->mti_params);
1643         int keylen = 0;
1644
1645         if (key != NULL)
1646                 keylen = strlen(key);
1647         if (start + 1 + keylen + strlen(val) >= end) {
1648                 CERROR("params are too long: %s %s%s\n",
1649                        params, key != NULL ? key : "", val);
1650                 return -EINVAL;
1651         }
1652
1653         sprintf(start, " %s%s", key != NULL ? key : "", val);
1654         return 0;
1655 }
1656
1657 /**
1658  * Walk through client config log record and convert the related records
1659  * into the target.
1660  **/
1661 static int mgs_steal_client_llog_handler(const struct lu_env *env,
1662                                          struct llog_handle *llh,
1663                                          struct llog_rec_hdr *rec, void *data)
1664 {
1665         struct mgs_device *mgs;
1666         struct obd_device *obd;
1667         struct mgs_target_info *mti, *tmti;
1668         struct fs_db *fsdb;
1669         int cfg_len = rec->lrh_len;
1670         char *cfg_buf = (char*) (rec + 1);
1671         struct lustre_cfg *lcfg;
1672         int rc = 0;
1673         struct llog_handle *mdt_llh = NULL;
1674         static int got_an_osc_or_mdc = 0;
1675         /* 0: not found any osc/mdc;
1676            1: found osc;
1677            2: found mdc;
1678         */
1679         static int last_step = -1;
1680         int cplen = 0;
1681
1682         ENTRY;
1683
1684         mti = ((struct temp_comp*)data)->comp_mti;
1685         tmti = ((struct temp_comp*)data)->comp_tmti;
1686         fsdb = ((struct temp_comp*)data)->comp_fsdb;
1687         obd = ((struct temp_comp *)data)->comp_obd;
1688         mgs = lu2mgs_dev(obd->obd_lu_dev);
1689         LASSERT(mgs);
1690
1691         if (rec->lrh_type != OBD_CFG_REC) {
1692                 CERROR("unhandled lrh_type: %#x\n", rec->lrh_type);
1693                 RETURN(-EINVAL);
1694         }
1695
1696         rc = lustre_cfg_sanity_check(cfg_buf, cfg_len);
1697         if (rc) {
1698                 CERROR("Insane cfg\n");
1699                 RETURN(rc);
1700         }
1701
1702         lcfg = (struct lustre_cfg *)cfg_buf;
1703
1704         if (lcfg->lcfg_command == LCFG_MARKER) {
1705                 struct cfg_marker *marker;
1706                 marker = lustre_cfg_buf(lcfg, 1);
1707                 if (!strncmp(marker->cm_comment, "add osc", 7) &&
1708                     (marker->cm_flags & CM_START) &&
1709                      !(marker->cm_flags & CM_SKIP)) {
1710                         got_an_osc_or_mdc = 1;
1711                         cplen = strlcpy(tmti->mti_svname, marker->cm_tgtname,
1712                                         sizeof(tmti->mti_svname));
1713                         if (cplen >= sizeof(tmti->mti_svname))
1714                                 RETURN(-E2BIG);
1715                         rc = record_start_log(env, mgs, &mdt_llh,
1716                                               mti->mti_svname);
1717                         if (rc)
1718                                 RETURN(rc);
1719                         rc = record_marker(env, mdt_llh, fsdb, CM_START,
1720                                            mti->mti_svname, "add osc(copied)");
1721                         record_end_log(env, &mdt_llh);
1722                         last_step = marker->cm_step;
1723                         RETURN(rc);
1724                 }
1725                 if (!strncmp(marker->cm_comment, "add osc", 7) &&
1726                     (marker->cm_flags & CM_END) &&
1727                      !(marker->cm_flags & CM_SKIP)) {
1728                         LASSERT(last_step == marker->cm_step);
1729                         last_step = -1;
1730                         got_an_osc_or_mdc = 0;
1731                         memset(tmti, 0, sizeof(*tmti));
1732                         rc = record_start_log(env, mgs, &mdt_llh,
1733                                               mti->mti_svname);
1734                         if (rc)
1735                                 RETURN(rc);
1736                         rc = record_marker(env, mdt_llh, fsdb, CM_END,
1737                                            mti->mti_svname, "add osc(copied)");
1738                         record_end_log(env, &mdt_llh);
1739                         RETURN(rc);
1740                 }
1741                 if (!strncmp(marker->cm_comment, "add mdc", 7) &&
1742                     (marker->cm_flags & CM_START) &&
1743                      !(marker->cm_flags & CM_SKIP)) {
1744                         got_an_osc_or_mdc = 2;
1745                         last_step = marker->cm_step;
1746                         memcpy(tmti->mti_svname, marker->cm_tgtname,
1747                                strlen(marker->cm_tgtname));
1748
1749                         RETURN(rc);
1750                 }
1751                 if (!strncmp(marker->cm_comment, "add mdc", 7) &&
1752                     (marker->cm_flags & CM_END) &&
1753                      !(marker->cm_flags & CM_SKIP)) {
1754                         LASSERT(last_step == marker->cm_step);
1755                         last_step = -1;
1756                         got_an_osc_or_mdc = 0;
1757                         memset(tmti, 0, sizeof(*tmti));
1758                         RETURN(rc);
1759                 }
1760         }
1761
1762         if (got_an_osc_or_mdc == 0 || last_step < 0)
1763                 RETURN(rc);
1764
1765         if (lcfg->lcfg_command == LCFG_ADD_UUID) {
1766                 __u64 nodenid = lcfg->lcfg_nid;
1767
1768                 if (strlen(tmti->mti_uuid) == 0) {
1769                         /* target uuid not set, this config record is before
1770                          * LCFG_SETUP, this nid is one of target node nid.
1771                          */
1772                         tmti->mti_nids[tmti->mti_nid_count] = nodenid;
1773                         tmti->mti_nid_count++;
1774                 } else {
1775                         char nidstr[LNET_NIDSTR_SIZE];
1776
1777                         /* failover node nid */
1778                         libcfs_nid2str_r(nodenid, nidstr, sizeof(nidstr));
1779                         rc = add_param(tmti->mti_params, PARAM_FAILNODE,
1780                                         nidstr);
1781                 }
1782
1783                 RETURN(rc);
1784         }
1785
1786         if (lcfg->lcfg_command == LCFG_SETUP) {
1787                 char *target;
1788
1789                 target = lustre_cfg_string(lcfg, 1);
1790                 memcpy(tmti->mti_uuid, target, strlen(target));
1791                 RETURN(rc);
1792         }
1793
1794         /* ignore client side sptlrpc_conf_log */
1795         if (lcfg->lcfg_command == LCFG_SPTLRPC_CONF)
1796                 RETURN(rc);
1797
1798         if (lcfg->lcfg_command == LCFG_ADD_MDC) {
1799                 int index;
1800
1801                 if (sscanf(lustre_cfg_buf(lcfg, 2), "%d", &index) != 1)
1802                         RETURN (-EINVAL);
1803
1804                 memcpy(tmti->mti_fsname, mti->mti_fsname,
1805                        strlen(mti->mti_fsname));
1806                 tmti->mti_stripe_index = index;
1807
1808                 rc = mgs_write_log_osp_to_mdt(env, mgs, fsdb, tmti,
1809                                               mti->mti_stripe_index,
1810                                               mti->mti_svname);
1811                 memset(tmti, 0, sizeof(*tmti));
1812                 RETURN(rc);
1813         }
1814
1815         if (lcfg->lcfg_command == LCFG_LOV_ADD_OBD) {
1816                 int index;
1817                 char mdt_index[9];
1818                 char *logname, *lovname;
1819
1820                 rc = name_create_mdt_and_lov(&logname, &lovname, fsdb,
1821                                              mti->mti_stripe_index);
1822                 if (rc)
1823                         RETURN(rc);
1824                 sprintf(mdt_index, "-MDT%04x", mti->mti_stripe_index);
1825
1826                 if (sscanf(lustre_cfg_buf(lcfg, 2), "%d", &index) != 1) {
1827                         name_destroy(&logname);
1828                         name_destroy(&lovname);
1829                         RETURN(-EINVAL);
1830                 }
1831
1832                 tmti->mti_stripe_index = index;
1833                 rc = mgs_write_log_osc_to_lov(env, mgs, fsdb, tmti, logname,
1834                                          mdt_index, lovname,
1835                                          LUSTRE_SP_MDT, 0);
1836                 name_destroy(&logname);
1837                 name_destroy(&lovname);
1838                 RETURN(rc);
1839         }
1840         RETURN(rc);
1841 }
1842
1843 /* fsdb->fsdb_mutex is already held  in mgs_write_log_target*/
1844 /* stealed from mgs_get_fsdb_from_llog*/
1845 static int mgs_steal_llog_for_mdt_from_client(const struct lu_env *env,
1846                                               struct mgs_device *mgs,
1847                                               char *client_name,
1848                                               struct temp_comp* comp)
1849 {
1850         struct llog_handle *loghandle;
1851         struct mgs_target_info *tmti;
1852         struct llog_ctxt *ctxt;
1853         int rc;
1854
1855         ENTRY;
1856
1857         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
1858         LASSERT(ctxt != NULL);
1859
1860         OBD_ALLOC_PTR(tmti);
1861         if (tmti == NULL)
1862                 GOTO(out_ctxt, rc = -ENOMEM);
1863
1864         comp->comp_tmti = tmti;
1865         comp->comp_obd = mgs->mgs_obd;
1866
1867         rc = llog_open(env, ctxt, &loghandle, NULL, client_name,
1868                        LLOG_OPEN_EXISTS);
1869         if (rc < 0) {
1870                 if (rc == -ENOENT)
1871                         rc = 0;
1872                 GOTO(out_pop, rc);
1873         }
1874
1875         rc = llog_init_handle(env, loghandle, LLOG_F_IS_PLAIN, NULL);
1876         if (rc)
1877                 GOTO(out_close, rc);
1878
1879         rc = llog_process_or_fork(env, loghandle, mgs_steal_client_llog_handler,
1880                                   (void *)comp, NULL, false);
1881         CDEBUG(D_MGS, "steal llog re = %d\n", rc);
1882 out_close:
1883         llog_close(env, loghandle);
1884 out_pop:
1885         OBD_FREE_PTR(tmti);
1886 out_ctxt:
1887         llog_ctxt_put(ctxt);
1888         RETURN(rc);
1889 }
1890
1891 /* lmv is the second thing for client logs */
1892 /* copied from mgs_write_log_lov. Please refer to that.  */
1893 static int mgs_write_log_lmv(const struct lu_env *env,
1894                              struct mgs_device *mgs,
1895                              struct fs_db *fsdb,
1896                              struct mgs_target_info *mti,
1897                              char *logname, char *lmvname)
1898 {
1899         struct llog_handle *llh = NULL;
1900         struct lmv_desc *lmvdesc;
1901         char *uuid;
1902         int rc = 0;
1903         ENTRY;
1904
1905         CDEBUG(D_MGS, "Writing lmv(%s) log for %s\n", lmvname,logname);
1906
1907         OBD_ALLOC_PTR(lmvdesc);
1908         if (lmvdesc == NULL)
1909                 RETURN(-ENOMEM);
1910         lmvdesc->ld_active_tgt_count = 0;
1911         lmvdesc->ld_tgt_count = 0;
1912         sprintf((char*)lmvdesc->ld_uuid.uuid, "%s_UUID", lmvname);
1913         uuid = (char *)lmvdesc->ld_uuid.uuid;
1914
1915         rc = record_start_log(env, mgs, &llh, logname);
1916         if (rc)
1917                 GOTO(out_free, rc);
1918         rc = record_marker(env, llh, fsdb, CM_START, lmvname, "lmv setup");
1919         if (rc)
1920                 GOTO(out_end, rc);
1921         rc = record_attach(env, llh, lmvname, "lmv", uuid);
1922         if (rc)
1923                 GOTO(out_end, rc);
1924         rc = record_lmv_setup(env, llh, lmvname, lmvdesc);
1925         if (rc)
1926                 GOTO(out_end, rc);
1927         rc = record_marker(env, llh, fsdb, CM_END, lmvname, "lmv setup");
1928         if (rc)
1929                 GOTO(out_end, rc);
1930 out_end:
1931         record_end_log(env, &llh);
1932 out_free:
1933         OBD_FREE_PTR(lmvdesc);
1934         RETURN(rc);
1935 }
1936
1937 /* lov is the first thing in the mdt and client logs */
1938 static int mgs_write_log_lov(const struct lu_env *env, struct mgs_device *mgs,
1939                              struct fs_db *fsdb, struct mgs_target_info *mti,
1940                              char *logname, char *lovname)
1941 {
1942         struct llog_handle *llh = NULL;
1943         struct lov_desc *lovdesc;
1944         char *uuid;
1945         int rc = 0;
1946         ENTRY;
1947
1948         CDEBUG(D_MGS, "Writing lov(%s) log for %s\n", lovname, logname);
1949
1950         /*
1951         #01 L attach   0:lov_mdsA  1:lov  2:71ccb_lov_mdsA_19f961a9e1
1952         #02 L lov_setup 0:lov_mdsA 1:(struct lov_desc)
1953               uuid=lov1_UUID, stripe count=1, size=1048576, offset=0, pattern=0
1954         */
1955
1956         /* FIXME just make lov_setup accept empty desc (put uuid in buf 2) */
1957         OBD_ALLOC_PTR(lovdesc);
1958         if (lovdesc == NULL)
1959                 RETURN(-ENOMEM);
1960         lovdesc->ld_magic = LOV_DESC_MAGIC;
1961         lovdesc->ld_tgt_count = 0;
1962         /* Defaults.  Can be changed later by lcfg config_param */
1963         lovdesc->ld_default_stripe_count = 1;
1964         lovdesc->ld_pattern = LOV_PATTERN_RAID0;
1965         lovdesc->ld_default_stripe_size = LOV_DESC_STRIPE_SIZE_DEFAULT;
1966         lovdesc->ld_default_stripe_offset = -1;
1967         lovdesc->ld_qos_maxage = LOV_DESC_QOS_MAXAGE_DEFAULT;
1968         sprintf((char*)lovdesc->ld_uuid.uuid, "%s_UUID", lovname);
1969         /* can these be the same? */
1970         uuid = (char *)lovdesc->ld_uuid.uuid;
1971
1972         /* This should always be the first entry in a log.
1973         rc = mgs_clear_log(obd, logname); */
1974         rc = record_start_log(env, mgs, &llh, logname);
1975         if (rc)
1976                 GOTO(out_free, rc);
1977         /* FIXME these should be a single journal transaction */
1978         rc = record_marker(env, llh, fsdb, CM_START, lovname, "lov setup");
1979         if (rc)
1980                 GOTO(out_end, rc);
1981         rc = record_attach(env, llh, lovname, "lov", uuid);
1982         if (rc)
1983                 GOTO(out_end, rc);
1984         rc = record_lov_setup(env, llh, lovname, lovdesc);
1985         if (rc)
1986                 GOTO(out_end, rc);
1987         rc = record_marker(env, llh, fsdb, CM_END, lovname, "lov setup");
1988         if (rc)
1989                 GOTO(out_end, rc);
1990         EXIT;
1991 out_end:
1992         record_end_log(env, &llh);
1993 out_free:
1994         OBD_FREE_PTR(lovdesc);
1995         return rc;
1996 }
1997
1998 /* add failnids to open log */
1999 static int mgs_write_log_failnids(const struct lu_env *env,
2000                                   struct mgs_target_info *mti,
2001                                   struct llog_handle *llh,
2002                                   char *cliname)
2003 {
2004         char *failnodeuuid = NULL;
2005         char *ptr = mti->mti_params;
2006         lnet_nid_t nid;
2007         int rc = 0;
2008
2009         /*
2010         #03 L add_uuid  nid=uml1@tcp(0x20000c0a80201) nal=90 0:  1:uml1_UUID
2011         #04 L add_uuid  nid=1@elan(0x1000000000001)   nal=90 0:  1:uml1_UUID
2012         #05 L setup    0:OSC_uml1_ost1_mdsA  1:ost1_UUID  2:uml1_UUID
2013         #06 L add_uuid  nid=uml2@tcp(0x20000c0a80202) nal=90 0:  1:uml2_UUID
2014         #0x L add_uuid  nid=2@elan(0x1000000000002)   nal=90 0:  1:uml2_UUID
2015         #07 L add_conn 0:OSC_uml1_ost1_mdsA  1:uml2_UUID
2016         */
2017
2018         /*
2019          * Pull failnid info out of params string, which may contain something
2020          * like "<nid1>,<nid2>:<nid3>,<nid4>".  class_parse_nid() does not
2021          * complain about abnormal inputs like ",:<nid1>", "<nid1>:,<nid2>",
2022          * etc.  However, convert_hostnames() should have caught those.
2023          */
2024         while (class_find_param(ptr, PARAM_FAILNODE, &ptr) == 0) {
2025                 while (class_parse_nid(ptr, &nid, &ptr) == 0) {
2026                         char nidstr[LNET_NIDSTR_SIZE];
2027
2028                         if (failnodeuuid == NULL) {
2029                                 /* We don't know the failover node name,
2030                                  * so just use the first nid as the uuid */
2031                                 libcfs_nid2str_r(nid, nidstr, sizeof(nidstr));
2032                                 rc = name_create(&failnodeuuid, nidstr, "");
2033                                 if (rc != 0)
2034                                         return rc;
2035                         }
2036                         CDEBUG(D_MGS, "add nid %s for failover uuid %s, "
2037                                 "client %s\n",
2038                                 libcfs_nid2str_r(nid, nidstr, sizeof(nidstr)),
2039                                 failnodeuuid, cliname);
2040                         rc = record_add_uuid(env, llh, nid, failnodeuuid);
2041                         /*
2042                          * If *ptr is ':', we have added all NIDs for
2043                          * failnodeuuid.
2044                          */
2045                         if (*ptr == ':') {
2046                                 rc = record_add_conn(env, llh, cliname,
2047                                                      failnodeuuid);
2048                                 name_destroy(&failnodeuuid);
2049                                 failnodeuuid = NULL;
2050                         }
2051                 }
2052                 if (failnodeuuid) {
2053                         rc = record_add_conn(env, llh, cliname, failnodeuuid);
2054                         name_destroy(&failnodeuuid);
2055                         failnodeuuid = NULL;
2056                 }
2057         }
2058
2059         return rc;
2060 }
2061
2062 static int mgs_write_log_mdc_to_lmv(const struct lu_env *env,
2063                                     struct mgs_device *mgs,
2064                                     struct fs_db *fsdb,
2065                                     struct mgs_target_info *mti,
2066                                     char *logname, char *lmvname)
2067 {
2068         struct llog_handle *llh = NULL;
2069         char *mdcname = NULL;
2070         char *nodeuuid = NULL;
2071         char *mdcuuid = NULL;
2072         char *lmvuuid = NULL;
2073         char index[6];
2074         char nidstr[LNET_NIDSTR_SIZE];
2075         int i, rc;
2076         ENTRY;
2077
2078         if (mgs_log_is_empty(env, mgs, logname)) {
2079                 CERROR("log is empty! Logical error\n");
2080                 RETURN(-EINVAL);
2081         }
2082
2083         CDEBUG(D_MGS, "adding mdc for %s to log %s:lmv(%s)\n",
2084                mti->mti_svname, logname, lmvname);
2085
2086         libcfs_nid2str_r(mti->mti_nids[0], nidstr, sizeof(nidstr));
2087         rc = name_create(&nodeuuid, nidstr, "");
2088         if (rc)
2089                 RETURN(rc);
2090         rc = name_create(&mdcname, mti->mti_svname, "-mdc");
2091         if (rc)
2092                 GOTO(out_free, rc);
2093         rc = name_create(&mdcuuid, mdcname, "_UUID");
2094         if (rc)
2095                 GOTO(out_free, rc);
2096         rc = name_create(&lmvuuid, lmvname, "_UUID");
2097         if (rc)
2098                 GOTO(out_free, rc);
2099
2100         rc = record_start_log(env, mgs, &llh, logname);
2101         if (rc)
2102                 GOTO(out_free, rc);
2103         rc = record_marker(env, llh, fsdb, CM_START, mti->mti_svname,
2104                            "add mdc");
2105         if (rc)
2106                 GOTO(out_end, rc);
2107         for (i = 0; i < mti->mti_nid_count; i++) {
2108                 CDEBUG(D_MGS, "add nid %s for mdt\n",
2109                         libcfs_nid2str_r(mti->mti_nids[i],
2110                                          nidstr, sizeof(nidstr)));
2111
2112                 rc = record_add_uuid(env, llh, mti->mti_nids[i], nodeuuid);
2113                 if (rc)
2114                         GOTO(out_end, rc);
2115         }
2116
2117         rc = record_attach(env, llh, mdcname, LUSTRE_MDC_NAME, lmvuuid);
2118         if (rc)
2119                 GOTO(out_end, rc);
2120         rc = record_setup(env, llh, mdcname, mti->mti_uuid, nodeuuid,
2121                           NULL, NULL);
2122         if (rc)
2123                 GOTO(out_end, rc);
2124         rc = mgs_write_log_failnids(env, mti, llh, mdcname);
2125         if (rc)
2126                 GOTO(out_end, rc);
2127         snprintf(index, sizeof(index), "%d", mti->mti_stripe_index);
2128         rc = record_mdc_add(env, llh, lmvname, mdcuuid, mti->mti_uuid,
2129                             index, "1");
2130         if (rc)
2131                 GOTO(out_end, rc);
2132         rc = record_marker(env, llh, fsdb, CM_END, mti->mti_svname,
2133                            "add mdc");
2134         if (rc)
2135                 GOTO(out_end, rc);
2136 out_end:
2137         record_end_log(env, &llh);
2138 out_free:
2139         name_destroy(&lmvuuid);
2140         name_destroy(&mdcuuid);
2141         name_destroy(&mdcname);
2142         name_destroy(&nodeuuid);
2143         RETURN(rc);
2144 }
2145
2146 static inline int name_create_lov(char **lovname, char *mdtname,
2147                                   struct fs_db *fsdb, int index)
2148 {
2149         /* COMPAT_180 */
2150         if (index == 0 && test_bit(FSDB_OSCNAME18, &fsdb->fsdb_flags))
2151                 return name_create(lovname, fsdb->fsdb_name, "-mdtlov");
2152         else
2153                 return name_create(lovname, mdtname, "-mdtlov");
2154 }
2155
2156 static int name_create_mdt_and_lov(char **logname, char **lovname,
2157                                    struct fs_db *fsdb, int i)
2158 {
2159         int rc;
2160
2161         rc = name_create_mdt(logname, fsdb->fsdb_name, i);
2162         if (rc)
2163                 return rc;
2164         /* COMPAT_180 */
2165         if (i == 0 && test_bit(FSDB_OSCNAME18, &fsdb->fsdb_flags))
2166                 rc = name_create(lovname, fsdb->fsdb_name, "-mdtlov");
2167         else
2168                 rc = name_create(lovname, *logname, "-mdtlov");
2169         if (rc) {
2170                 name_destroy(logname);
2171                 *logname = NULL;
2172         }
2173         return rc;
2174 }
2175
2176 static inline int name_create_mdt_osc(char **oscname, char *ostname,
2177                                       struct fs_db *fsdb, int i)
2178 {
2179         char suffix[16];
2180
2181         if (i == 0 && test_bit(FSDB_OSCNAME18, &fsdb->fsdb_flags))
2182                 sprintf(suffix, "-osc");
2183         else
2184                 sprintf(suffix, "-osc-MDT%04x", i);
2185         return name_create(oscname, ostname, suffix);
2186 }
2187
2188 /* add new mdc to already existent MDS */
2189 static int mgs_write_log_osp_to_mdt(const struct lu_env *env,
2190                                     struct mgs_device *mgs,
2191                                     struct fs_db *fsdb,
2192                                     struct mgs_target_info *mti,
2193                                     int mdt_index, char *logname)
2194 {
2195         struct llog_handle      *llh = NULL;
2196         char    *nodeuuid = NULL;
2197         char    *ospname = NULL;
2198         char    *lovuuid = NULL;
2199         char    *mdtuuid = NULL;
2200         char    *svname = NULL;
2201         char    *mdtname = NULL;
2202         char    *lovname = NULL;
2203         char    index_str[16];
2204         char    nidstr[LNET_NIDSTR_SIZE];
2205         int     i, rc;
2206
2207         ENTRY;
2208         if (mgs_log_is_empty(env, mgs, mti->mti_svname)) {
2209                 CERROR("log is empty! Logical error\n");
2210                 RETURN (-EINVAL);
2211         }
2212
2213         CDEBUG(D_MGS, "adding osp index %d to %s\n", mti->mti_stripe_index,
2214                logname);
2215
2216         rc = name_create_mdt(&mdtname, fsdb->fsdb_name, mti->mti_stripe_index);
2217         if (rc)
2218                 RETURN(rc);
2219
2220         libcfs_nid2str_r(mti->mti_nids[0], nidstr, sizeof(nidstr));
2221         rc = name_create(&nodeuuid, nidstr, "");
2222         if (rc)
2223                 GOTO(out_destory, rc);
2224
2225         rc = name_create(&svname, mdtname, "-osp");
2226         if (rc)
2227                 GOTO(out_destory, rc);
2228
2229         sprintf(index_str, "-MDT%04x", mdt_index);
2230         rc = name_create(&ospname, svname, index_str);
2231         if (rc)
2232                 GOTO(out_destory, rc);
2233
2234         rc = name_create_lov(&lovname, logname, fsdb, mdt_index);
2235         if (rc)
2236                 GOTO(out_destory, rc);
2237
2238         rc = name_create(&lovuuid, lovname, "_UUID");
2239         if (rc)
2240                 GOTO(out_destory, rc);
2241
2242         rc = name_create(&mdtuuid, mdtname, "_UUID");
2243         if (rc)
2244                 GOTO(out_destory, rc);
2245
2246         rc = record_start_log(env, mgs, &llh, logname);
2247         if (rc)
2248                 GOTO(out_destory, rc);
2249
2250         rc = record_marker(env, llh, fsdb, CM_START, mti->mti_svname,
2251                            "add osp");
2252         if (rc)
2253                 GOTO(out_destory, rc);
2254
2255         for (i = 0; i < mti->mti_nid_count; i++) {
2256                 CDEBUG(D_MGS, "add nid %s for mdt\n",
2257                         libcfs_nid2str_r(mti->mti_nids[i],
2258                                          nidstr, sizeof(nidstr)));
2259                 rc = record_add_uuid(env, llh, mti->mti_nids[i], nodeuuid);
2260                 if (rc)
2261                         GOTO(out_end, rc);
2262         }
2263
2264         rc = record_attach(env, llh, ospname, LUSTRE_OSP_NAME, lovuuid);
2265         if (rc)
2266                 GOTO(out_end, rc);
2267
2268         rc = record_setup(env, llh, ospname, mti->mti_uuid, nodeuuid,
2269                           NULL, NULL);
2270         if (rc)
2271                 GOTO(out_end, rc);
2272
2273         rc = mgs_write_log_failnids(env, mti, llh, ospname);
2274         if (rc)
2275                 GOTO(out_end, rc);
2276
2277         /* Add mdc(osp) to lod */
2278         snprintf(index_str, sizeof(index_str), "%d", mti->mti_stripe_index);
2279         rc = record_base(env, llh, lovname, 0, LCFG_ADD_MDC, mti->mti_uuid,
2280                          index_str, "1", NULL);
2281         if (rc)
2282                 GOTO(out_end, rc);
2283
2284         rc = record_marker(env, llh, fsdb, CM_END, mti->mti_svname, "add osp");
2285         if (rc)
2286                 GOTO(out_end, rc);
2287
2288 out_end:
2289         record_end_log(env, &llh);
2290
2291 out_destory:
2292         name_destroy(&mdtuuid);
2293         name_destroy(&lovuuid);
2294         name_destroy(&lovname);
2295         name_destroy(&ospname);
2296         name_destroy(&svname);
2297         name_destroy(&nodeuuid);
2298         name_destroy(&mdtname);
2299         RETURN(rc);
2300 }
2301
2302 static int mgs_write_log_mdt0(const struct lu_env *env,
2303                               struct mgs_device *mgs,
2304                               struct fs_db *fsdb,
2305                               struct mgs_target_info *mti)
2306 {
2307         char *log = mti->mti_svname;
2308         struct llog_handle *llh = NULL;
2309         char *uuid, *lovname;
2310         char mdt_index[6];
2311         char *ptr = mti->mti_params;
2312         int rc = 0, failout = 0;
2313         ENTRY;
2314
2315         OBD_ALLOC(uuid, sizeof(struct obd_uuid));
2316         if (uuid == NULL)
2317                 RETURN(-ENOMEM);
2318
2319         if (class_find_param(ptr, PARAM_FAILMODE, &ptr) == 0)
2320                 failout = (strncmp(ptr, "failout", 7) == 0);
2321
2322         rc = name_create(&lovname, log, "-mdtlov");
2323         if (rc)
2324                 GOTO(out_free, rc);
2325         if (mgs_log_is_empty(env, mgs, log)) {
2326                 rc = mgs_write_log_lov(env, mgs, fsdb, mti, log, lovname);
2327                 if (rc)
2328                         GOTO(out_lod, rc);
2329         }
2330
2331         sprintf(mdt_index, "%d", mti->mti_stripe_index);
2332
2333         rc = record_start_log(env, mgs, &llh, log);
2334         if (rc)
2335                 GOTO(out_lod, rc);
2336
2337         /* add MDT itself */
2338
2339         /* FIXME this whole fn should be a single journal transaction */
2340         sprintf(uuid, "%s_UUID", log);
2341         rc = record_marker(env, llh, fsdb, CM_START, log, "add mdt");
2342         if (rc)
2343                 GOTO(out_lod, rc);
2344         rc = record_attach(env, llh, log, LUSTRE_MDT_NAME, uuid);
2345         if (rc)
2346                 GOTO(out_end, rc);
2347         rc = record_mount_opt(env, llh, log, lovname, NULL);
2348         if (rc)
2349                 GOTO(out_end, rc);
2350         rc = record_setup(env, llh, log, uuid, mdt_index, lovname,
2351                         failout ? "n" : "f");
2352         if (rc)
2353                 GOTO(out_end, rc);
2354         rc = record_marker(env, llh, fsdb, CM_END, log, "add mdt");
2355         if (rc)
2356                 GOTO(out_end, rc);
2357 out_end:
2358         record_end_log(env, &llh);
2359 out_lod:
2360         name_destroy(&lovname);
2361 out_free:
2362         OBD_FREE(uuid, sizeof(struct obd_uuid));
2363         RETURN(rc);
2364 }
2365
2366 /* envelope method for all layers log */
2367 static int mgs_write_log_mdt(const struct lu_env *env,
2368                              struct mgs_device *mgs,
2369                              struct fs_db *fsdb,
2370                              struct mgs_target_info *mti)
2371 {
2372         struct mgs_thread_info *mgi = mgs_env_info(env);
2373         struct llog_handle *llh = NULL;
2374         char *cliname;
2375         int rc, i = 0;
2376         ENTRY;
2377
2378         CDEBUG(D_MGS, "writing new mdt %s\n", mti->mti_svname);
2379
2380         if (mti->mti_uuid[0] == '\0') {
2381                 /* Make up our own uuid */
2382                 snprintf(mti->mti_uuid, sizeof(mti->mti_uuid),
2383                          "%s_UUID", mti->mti_svname);
2384         }
2385
2386         /* add mdt */
2387         rc = mgs_write_log_mdt0(env, mgs, fsdb, mti);
2388         if (rc)
2389                 RETURN(rc);
2390         /* Append the mdt info to the client log */
2391         rc = name_create(&cliname, mti->mti_fsname, "-client");
2392         if (rc)
2393                 RETURN(rc);
2394
2395         if (mgs_log_is_empty(env, mgs, cliname)) {
2396                 /* Start client log */
2397                 rc = mgs_write_log_lov(env, mgs, fsdb, mti, cliname,
2398                                        fsdb->fsdb_clilov);
2399                 if (rc)
2400                         GOTO(out_free, rc);
2401                 rc = mgs_write_log_lmv(env, mgs, fsdb, mti, cliname,
2402                                        fsdb->fsdb_clilmv);
2403                 if (rc)
2404                         GOTO(out_free, rc);
2405         }
2406
2407         /*
2408         #09 L add_uuid nid=uml1@tcp(0x20000c0a80201) 0:  1:uml1_UUID
2409         #10 L attach   0:MDC_uml1_mdsA_MNT_client  1:mdc  2:1d834_MNT_client_03f
2410         #11 L setup    0:MDC_uml1_mdsA_MNT_client  1:mdsA_UUID  2:uml1_UUID
2411         #12 L add_uuid nid=uml2@tcp(0x20000c0a80202) 0:  1:uml2_UUID
2412         #13 L add_conn 0:MDC_uml1_mdsA_MNT_client  1:uml2_UUID
2413         #14 L mount_option 0:  1:client  2:lov1  3:MDC_uml1_mdsA_MNT_client
2414         */
2415
2416         /* copy client info about lov/lmv */
2417         mgi->mgi_comp.comp_mti = mti;
2418         mgi->mgi_comp.comp_fsdb = fsdb;
2419
2420         rc = mgs_steal_llog_for_mdt_from_client(env, mgs, cliname,
2421                                                 &mgi->mgi_comp);
2422         if (rc)
2423                 GOTO(out_free, rc);
2424         rc = mgs_write_log_mdc_to_lmv(env, mgs, fsdb, mti, cliname,
2425                                       fsdb->fsdb_clilmv);
2426         if (rc)
2427                 GOTO(out_free, rc);
2428
2429         /* add mountopts */
2430         rc = record_start_log(env, mgs, &llh, cliname);
2431         if (rc)
2432                 GOTO(out_free, rc);
2433
2434         rc = record_marker(env, llh, fsdb, CM_START, cliname,
2435                            "mount opts");
2436         if (rc)
2437                 GOTO(out_end, rc);
2438         rc = record_mount_opt(env, llh, cliname, fsdb->fsdb_clilov,
2439                               fsdb->fsdb_clilmv);
2440         if (rc)
2441                 GOTO(out_end, rc);
2442         rc = record_marker(env, llh, fsdb, CM_END, cliname,
2443                            "mount opts");
2444
2445         if (rc)
2446                 GOTO(out_end, rc);
2447
2448         /* for_all_existing_mdt except current one */
2449         for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
2450                 if (i !=  mti->mti_stripe_index &&
2451                     test_bit(i, fsdb->fsdb_mdt_index_map)) {
2452                         char *logname;
2453
2454                         rc = name_create_mdt(&logname, fsdb->fsdb_name, i);
2455                         if (rc)
2456                                 GOTO(out_end, rc);
2457
2458                         /* NB: If the log for the MDT is empty, it means
2459                          * the MDT is only added to the index
2460                          * map, and not being process yet, i.e. this
2461                          * is an unregistered MDT, see mgs_write_log_target().
2462                          * so we should skip it. Otherwise
2463                          *
2464                          * 1. MGS get register request for MDT1 and MDT2.
2465                          *
2466                          * 2. Then both MDT1 and MDT2 are added into
2467                          * fsdb_mdt_index_map. (see mgs_set_index()).
2468                          *
2469                          * 3. Then MDT1 get the lock of fsdb_mutex, then
2470                          * generate the config log, here, it will regard MDT2
2471                          * as an existent MDT, and generate "add osp" for
2472                          * lustre-MDT0001-osp-MDT0002. Note: at the moment
2473                          * MDT0002 config log is still empty, so it will
2474                          * add "add osp" even before "lov setup", which
2475                          * will definitly cause trouble.
2476                          *
2477                          * 4. MDT1 registeration finished, fsdb_mutex is
2478                          * released, then MDT2 get in, then in above
2479                          * mgs_steal_llog_for_mdt_from_client(), it will
2480                          * add another osp log for lustre-MDT0001-osp-MDT0002,
2481                          * which will cause another trouble.*/
2482                         if (!mgs_log_is_empty(env, mgs, logname))
2483                                 rc = mgs_write_log_osp_to_mdt(env, mgs, fsdb,
2484                                                               mti, i, logname);
2485
2486                         name_destroy(&logname);
2487                         if (rc)
2488                                 GOTO(out_end, rc);
2489                 }
2490         }
2491 out_end:
2492         record_end_log(env, &llh);
2493 out_free:
2494         name_destroy(&cliname);
2495         RETURN(rc);
2496 }
2497
2498 /* Add the ost info to the client/mdt lov */
2499 static int mgs_write_log_osc_to_lov(const struct lu_env *env,
2500                                     struct mgs_device *mgs, struct fs_db *fsdb,
2501                                     struct mgs_target_info *mti,
2502                                     char *logname, char *suffix, char *lovname,
2503                                     enum lustre_sec_part sec_part, int flags)
2504 {
2505         struct llog_handle *llh = NULL;
2506         char *nodeuuid = NULL;
2507         char *oscname = NULL;
2508         char *oscuuid = NULL;
2509         char *lovuuid = NULL;
2510         char *svname = NULL;
2511         char index[6];
2512         char nidstr[LNET_NIDSTR_SIZE];
2513         int i, rc;
2514         ENTRY;
2515
2516         CDEBUG(D_INFO, "adding osc for %s to log %s\n",
2517                 mti->mti_svname, logname);
2518
2519         if (mgs_log_is_empty(env, mgs, logname)) {
2520                 CERROR("log is empty! Logical error\n");
2521                 RETURN(-EINVAL);
2522         }
2523
2524         libcfs_nid2str_r(mti->mti_nids[0], nidstr, sizeof(nidstr));
2525         rc = name_create(&nodeuuid, nidstr, "");
2526         if (rc)
2527                 RETURN(rc);
2528         rc = name_create(&svname, mti->mti_svname, "-osc");
2529         if (rc)
2530                 GOTO(out_free, rc);
2531
2532         /* for the system upgraded from old 1.8, keep using the old osc naming
2533          * style for mdt, see name_create_mdt_osc(). LU-1257 */
2534         if (test_bit(FSDB_OSCNAME18, &fsdb->fsdb_flags))
2535                 rc = name_create(&oscname, svname, "");
2536         else
2537                 rc = name_create(&oscname, svname, suffix);
2538         if (rc)
2539                 GOTO(out_free, rc);
2540
2541         rc = name_create(&oscuuid, oscname, "_UUID");
2542         if (rc)
2543                 GOTO(out_free, rc);
2544         rc = name_create(&lovuuid, lovname, "_UUID");
2545         if (rc)
2546                 GOTO(out_free, rc);
2547
2548
2549         /*
2550         #03 L add_uuid nid=uml1@tcp(0x20000c0a80201) 0:  1:uml1_UUID
2551         multihomed (#4)
2552         #04 L add_uuid  nid=1@elan(0x1000000000001)  nal=90 0:  1:uml1_UUID
2553         #04 L attach   0:OSC_uml1_ost1_MNT_client  1:osc  2:89070_lov1_a41dff51a
2554         #05 L setup    0:OSC_uml1_ost1_MNT_client  1:ost1_UUID  2:uml1_UUID
2555         failover (#6,7)
2556         #06 L add_uuid nid=uml2@tcp(0x20000c0a80202) 0:  1:uml2_UUID
2557         #07 L add_conn 0:OSC_uml1_ost1_MNT_client  1:uml2_UUID
2558         #08 L lov_modify_tgts add 0:lov1  1:ost1_UUID  2(index):0  3(gen):1
2559         */
2560
2561         rc = record_start_log(env, mgs, &llh, logname);
2562         if (rc)
2563                 GOTO(out_free, rc);
2564
2565         /* FIXME these should be a single journal transaction */
2566         rc = record_marker(env, llh, fsdb, CM_START | flags, mti->mti_svname,
2567                            "add osc");
2568         if (rc)
2569                 GOTO(out_end, rc);
2570
2571         /* NB: don't change record order, because upon MDT steal OSC config
2572          * from client, it treats all nids before LCFG_SETUP as target nids
2573          * (multiple interfaces), while nids after as failover node nids.
2574          * See mgs_steal_client_llog_handler() LCFG_ADD_UUID.
2575          */
2576         for (i = 0; i < mti->mti_nid_count; i++) {
2577                 CDEBUG(D_MGS, "add nid %s\n",
2578                         libcfs_nid2str_r(mti->mti_nids[i],
2579                                          nidstr, sizeof(nidstr)));
2580                 rc = record_add_uuid(env, llh, mti->mti_nids[i], nodeuuid);
2581                 if (rc)
2582                         GOTO(out_end, rc);
2583         }
2584         rc = record_attach(env, llh, oscname, LUSTRE_OSC_NAME, lovuuid);
2585         if (rc)
2586                 GOTO(out_end, rc);
2587         rc = record_setup(env, llh, oscname, mti->mti_uuid, nodeuuid,
2588                           NULL, NULL);
2589         if (rc)
2590                 GOTO(out_end, rc);
2591         rc = mgs_write_log_failnids(env, mti, llh, oscname);
2592         if (rc)
2593                 GOTO(out_end, rc);
2594
2595         snprintf(index, sizeof(index), "%d", mti->mti_stripe_index);
2596
2597         rc = record_lov_add(env, llh, lovname, mti->mti_uuid, index, "1");
2598         if (rc)
2599                 GOTO(out_end, rc);
2600         rc = record_marker(env, llh, fsdb, CM_END | flags, mti->mti_svname,
2601                            "add osc");
2602         if (rc)
2603                 GOTO(out_end, rc);
2604 out_end:
2605         record_end_log(env, &llh);
2606 out_free:
2607         name_destroy(&lovuuid);
2608         name_destroy(&oscuuid);
2609         name_destroy(&oscname);
2610         name_destroy(&svname);
2611         name_destroy(&nodeuuid);
2612         RETURN(rc);
2613 }
2614
2615 static int mgs_write_log_ost(const struct lu_env *env,
2616                              struct mgs_device *mgs, struct fs_db *fsdb,
2617                              struct mgs_target_info *mti)
2618 {
2619         struct llog_handle *llh = NULL;
2620         char *logname, *lovname;
2621         char *ptr = mti->mti_params;
2622         int rc, flags = 0, failout = 0, i;
2623         ENTRY;
2624
2625         CDEBUG(D_MGS, "writing new ost %s\n", mti->mti_svname);
2626
2627         /* The ost startup log */
2628
2629         /* If the ost log already exists, that means that someone reformatted
2630            the ost and it called target_add again. */
2631         if (!mgs_log_is_empty(env, mgs, mti->mti_svname)) {
2632                 LCONSOLE_ERROR_MSG(0x141, "The config log for %s already "
2633                                    "exists, yet the server claims it never "
2634                                    "registered. It may have been reformatted, "
2635                                    "or the index changed. writeconf the MDT to "
2636                                    "regenerate all logs.\n", mti->mti_svname);
2637                 RETURN(-EALREADY);
2638         }
2639
2640         /*
2641         attach obdfilter ost1 ost1_UUID
2642         setup /dev/loop2 ldiskfs f|n errors=remount-ro,user_xattr
2643         */
2644         if (class_find_param(ptr, PARAM_FAILMODE, &ptr) == 0)
2645                 failout = (strncmp(ptr, "failout", 7) == 0);
2646         rc = record_start_log(env, mgs, &llh, mti->mti_svname);
2647         if (rc)
2648                 RETURN(rc);
2649         /* FIXME these should be a single journal transaction */
2650         rc = record_marker(env, llh, fsdb, CM_START, mti->mti_svname,"add ost");
2651         if (rc)
2652                 GOTO(out_end, rc);
2653         if (*mti->mti_uuid == '\0')
2654                 snprintf(mti->mti_uuid, sizeof(mti->mti_uuid),
2655                          "%s_UUID", mti->mti_svname);
2656         rc = record_attach(env, llh, mti->mti_svname,
2657                            "obdfilter"/*LUSTRE_OST_NAME*/, mti->mti_uuid);
2658         if (rc)
2659                 GOTO(out_end, rc);
2660         rc = record_setup(env, llh, mti->mti_svname,
2661                           "dev"/*ignored*/, "type"/*ignored*/,
2662                           failout ? "n" : "f", NULL/*options*/);
2663         if (rc)
2664                 GOTO(out_end, rc);
2665         rc = record_marker(env, llh, fsdb, CM_END, mti->mti_svname, "add ost");
2666         if (rc)
2667                 GOTO(out_end, rc);
2668 out_end:
2669         record_end_log(env, &llh);
2670         if (rc)
2671                 RETURN(rc);
2672         /* We also have to update the other logs where this osc is part of
2673            the lov */
2674
2675         if (test_bit(FSDB_OLDLOG14, &fsdb->fsdb_flags)) {
2676                 /* If we're upgrading, the old mdt log already has our
2677                    entry. Let's do a fake one for fun. */
2678                 /* Note that we can't add any new failnids, since we don't
2679                    know the old osc names. */
2680                 flags = CM_SKIP | CM_UPGRADE146;
2681
2682         } else if ((mti->mti_flags & LDD_F_UPDATE) != LDD_F_UPDATE) {
2683                 /* If the update flag isn't set, don't update client/mdt
2684                    logs. */
2685                 flags |= CM_SKIP;
2686                 LCONSOLE_WARN("Client log for %s was not updated; writeconf "
2687                               "the MDT first to regenerate it.\n",
2688                               mti->mti_svname);
2689         }
2690
2691         /* Add ost to all MDT lov defs */
2692         for (i = 0; i < INDEX_MAP_SIZE * 8; i++){
2693                 if (test_bit(i, fsdb->fsdb_mdt_index_map)) {
2694                         char mdt_index[9];
2695
2696                         rc = name_create_mdt_and_lov(&logname, &lovname, fsdb,
2697                                                      i);
2698                         if (rc)
2699                                 RETURN(rc);
2700                         sprintf(mdt_index, "-MDT%04x", i);
2701                         rc = mgs_write_log_osc_to_lov(env, mgs, fsdb, mti,
2702                                                       logname, mdt_index,
2703                                                       lovname, LUSTRE_SP_MDT,
2704                                                       flags);
2705                         name_destroy(&logname);
2706                         name_destroy(&lovname);
2707                         if (rc)
2708                                 RETURN(rc);
2709                 }
2710         }
2711
2712         /* Append ost info to the client log */
2713         rc = name_create(&logname, mti->mti_fsname, "-client");
2714         if (rc)
2715                 RETURN(rc);
2716         if (mgs_log_is_empty(env, mgs, logname)) {
2717                 /* Start client log */
2718                 rc = mgs_write_log_lov(env, mgs, fsdb, mti, logname,
2719                                        fsdb->fsdb_clilov);
2720                 if (rc)
2721                         GOTO(out_free, rc);
2722                 rc = mgs_write_log_lmv(env, mgs, fsdb, mti, logname,
2723                                        fsdb->fsdb_clilmv);
2724                 if (rc)
2725                         GOTO(out_free, rc);
2726         }
2727         rc = mgs_write_log_osc_to_lov(env, mgs, fsdb, mti, logname, "",
2728                                       fsdb->fsdb_clilov, LUSTRE_SP_CLI, flags);
2729 out_free:
2730         name_destroy(&logname);
2731         RETURN(rc);
2732 }
2733
2734 static __inline__ int mgs_param_empty(char *ptr)
2735 {
2736         char *tmp;
2737
2738         if ((tmp = strchr(ptr, '=')) && (*(++tmp) == '\0'))
2739                 return 1;
2740         return 0;
2741 }
2742
2743 static int mgs_write_log_failnid_internal(const struct lu_env *env,
2744                                           struct mgs_device *mgs,
2745                                           struct fs_db *fsdb,
2746                                           struct mgs_target_info *mti,
2747                                           char *logname, char *cliname)
2748 {
2749         int rc;
2750         struct llog_handle *llh = NULL;
2751
2752         if (mgs_param_empty(mti->mti_params)) {
2753                 /* Remove _all_ failnids */
2754                 rc = mgs_modify(env, mgs, fsdb, mti, logname,
2755                                 mti->mti_svname, "add failnid", CM_SKIP);
2756                 return rc < 0 ? rc : 0;
2757         }
2758
2759         /* Otherwise failover nids are additive */
2760         rc = record_start_log(env, mgs, &llh, logname);
2761         if (rc)
2762                 return rc;
2763                 /* FIXME this should be a single journal transaction */
2764         rc = record_marker(env, llh, fsdb, CM_START, mti->mti_svname,
2765                            "add failnid");
2766         if (rc)
2767                 goto out_end;
2768         rc = mgs_write_log_failnids(env, mti, llh, cliname);
2769         if (rc)
2770                 goto out_end;
2771         rc = record_marker(env, llh, fsdb, CM_END,
2772                            mti->mti_svname, "add failnid");
2773 out_end:
2774         record_end_log(env, &llh);
2775         return rc;
2776 }
2777
2778
2779 /* Add additional failnids to an existing log.
2780    The mdc/osc must have been added to logs first */
2781 /* tcp nids must be in dotted-quad ascii -
2782    we can't resolve hostnames from the kernel. */
2783 static int mgs_write_log_add_failnid(const struct lu_env *env,
2784                                      struct mgs_device *mgs,
2785                                      struct fs_db *fsdb,
2786                                      struct mgs_target_info *mti)
2787 {
2788         char *logname, *cliname;
2789         int rc;
2790         ENTRY;
2791
2792         /* FIXME we currently can't erase the failnids
2793          * given when a target first registers, since they aren't part of
2794          * an "add uuid" stanza */
2795
2796         /* Verify that we know about this target */
2797         if (mgs_log_is_empty(env, mgs, mti->mti_svname)) {
2798                 LCONSOLE_ERROR_MSG(0x142, "The target %s has not registered "
2799                                    "yet. It must be started before failnids "
2800                                    "can be added.\n", mti->mti_svname);
2801                 RETURN(-ENOENT);
2802         }
2803
2804         /* Create mdc/osc client name (e.g. lustre-OST0001-osc) */
2805         if (mti->mti_flags & LDD_F_SV_TYPE_MDT) {
2806                 rc = name_create(&cliname, mti->mti_svname, "-mdc");
2807         } else if (mti->mti_flags & LDD_F_SV_TYPE_OST) {
2808                 rc = name_create(&cliname, mti->mti_svname, "-osc");
2809         } else {
2810                 RETURN(-EINVAL);
2811         }
2812         if (rc)
2813                 RETURN(rc);
2814         /* Add failover nids to the client log */
2815         rc = name_create(&logname, mti->mti_fsname, "-client");
2816         if (rc) {
2817                 name_destroy(&cliname);
2818                 RETURN(rc);
2819         }
2820         rc = mgs_write_log_failnid_internal(env, mgs, fsdb,mti,logname,cliname);
2821         name_destroy(&logname);
2822         name_destroy(&cliname);
2823         if (rc)
2824                 RETURN(rc);
2825
2826         if (mti->mti_flags & LDD_F_SV_TYPE_OST) {
2827                 /* Add OST failover nids to the MDT logs as well */
2828                 int i;
2829
2830                 for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
2831                         if (!test_bit(i, fsdb->fsdb_mdt_index_map))
2832                                 continue;
2833                         rc = name_create_mdt(&logname, mti->mti_fsname, i);
2834                         if (rc)
2835                                 RETURN(rc);
2836                         rc = name_create_mdt_osc(&cliname, mti->mti_svname,
2837                                                  fsdb, i);
2838                         if (rc) {
2839                                 name_destroy(&logname);
2840                                 RETURN(rc);
2841                         }
2842                         rc = mgs_write_log_failnid_internal(env, mgs, fsdb,
2843                                                             mti, logname,
2844                                                             cliname);
2845                         name_destroy(&cliname);
2846                         name_destroy(&logname);
2847                         if (rc)
2848                                 RETURN(rc);
2849                 }
2850         }
2851
2852         RETURN(rc);
2853 }
2854
2855 static int mgs_wlp_lcfg(const struct lu_env *env,
2856                         struct mgs_device *mgs, struct fs_db *fsdb,
2857                         struct mgs_target_info *mti,
2858                         char *logname, struct lustre_cfg_bufs *bufs,
2859                         char *tgtname, char *ptr)
2860 {
2861         char comment[MTI_NAME_MAXLEN];
2862         char *tmp;
2863         struct llog_cfg_rec *lcr;
2864         int rc, del;
2865
2866         /* Erase any old settings of this same parameter */
2867         memcpy(comment, ptr, MTI_NAME_MAXLEN);
2868         comment[MTI_NAME_MAXLEN - 1] = 0;
2869         /* But don't try to match the value. */
2870         tmp = strchr(comment, '=');
2871         if (tmp != NULL)
2872                 *tmp = 0;
2873         /* FIXME we should skip settings that are the same as old values */
2874         rc = mgs_modify(env, mgs, fsdb, mti, logname, tgtname, comment,CM_SKIP);
2875         if (rc < 0)
2876                 return rc;
2877         del = mgs_param_empty(ptr);
2878
2879         LCONSOLE_INFO("%s parameter %s.%s in log %s\n", del ? "Disabling" : rc ?
2880                       "Setting" : "Modifying", tgtname, comment, logname);
2881         if (del) {
2882                 /* mgs_modify() will return 1 if nothing had to be done */
2883                 if (rc == 1)
2884                         rc = 0;
2885                 return rc;
2886         }
2887
2888         lustre_cfg_bufs_reset(bufs, tgtname);
2889         lustre_cfg_bufs_set_string(bufs, 1, ptr);
2890         if (mti->mti_flags & LDD_F_PARAM2)
2891                 lustre_cfg_bufs_set_string(bufs, 2, LCTL_UPCALL);
2892
2893         lcr = lustre_cfg_rec_new((mti->mti_flags & LDD_F_PARAM2) ?
2894                                  LCFG_SET_PARAM : LCFG_PARAM, bufs);
2895         if (lcr == NULL)
2896                 return -ENOMEM;
2897
2898         rc = mgs_write_log_direct(env, mgs, fsdb, logname, lcr, tgtname,
2899                                   comment);
2900         lustre_cfg_rec_free(lcr);
2901         return rc;
2902 }
2903
2904 static int mgs_write_log_param2(const struct lu_env *env,
2905                                 struct mgs_device *mgs,
2906                                 struct fs_db *fsdb,
2907                                 struct mgs_target_info *mti, char *ptr)
2908 {
2909         struct lustre_cfg_bufs  bufs;
2910         int                     rc = 0;
2911         ENTRY;
2912
2913         CDEBUG(D_MGS, "next param '%s'\n", ptr);
2914         rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, PARAMS_FILENAME, &bufs,
2915                           mti->mti_svname, ptr);
2916
2917         RETURN(rc);
2918 }
2919
2920 /* write global variable settings into log */
2921 static int mgs_write_log_sys(const struct lu_env *env,
2922                              struct mgs_device *mgs, struct fs_db *fsdb,
2923                              struct mgs_target_info *mti, char *sys, char *ptr)
2924 {
2925         struct mgs_thread_info  *mgi = mgs_env_info(env);
2926         struct lustre_cfg       *lcfg;
2927         struct llog_cfg_rec     *lcr;
2928         char *tmp, sep;
2929         int rc, cmd, convert = 1;
2930
2931         if (class_match_param(ptr, PARAM_TIMEOUT, &tmp) == 0) {
2932                 cmd = LCFG_SET_TIMEOUT;
2933         } else if (class_match_param(ptr, PARAM_LDLM_TIMEOUT, &tmp) == 0) {
2934                 cmd = LCFG_SET_LDLM_TIMEOUT;
2935         /* Check for known params here so we can return error to lctl */
2936         } else if ((class_match_param(ptr, PARAM_AT_MIN, &tmp) == 0) ||
2937                 (class_match_param(ptr, PARAM_AT_MAX, &tmp) == 0) ||
2938                 (class_match_param(ptr, PARAM_AT_EXTRA, &tmp) == 0) ||
2939                 (class_match_param(ptr, PARAM_AT_EARLY_MARGIN, &tmp) == 0) ||
2940                 (class_match_param(ptr, PARAM_AT_HISTORY, &tmp) == 0)) {
2941                 cmd = LCFG_PARAM;
2942         } else if (class_match_param(ptr, PARAM_JOBID_VAR, &tmp) == 0) {
2943                 convert = 0; /* Don't convert string value to integer */
2944                 cmd = LCFG_PARAM;
2945         } else {
2946                 return -EINVAL;
2947         }
2948
2949         if (mgs_param_empty(ptr))
2950                 CDEBUG(D_MGS, "global '%s' removed\n", sys);
2951         else
2952                 CDEBUG(D_MGS, "global '%s' val=%s\n", sys, tmp);
2953
2954         lustre_cfg_bufs_reset(&mgi->mgi_bufs, NULL);
2955         lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 1, sys);
2956         if (!convert && *tmp != '\0')
2957                 lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 2, tmp);
2958         lcr = lustre_cfg_rec_new(cmd, &mgi->mgi_bufs);
2959         if (lcr == NULL)
2960                 return -ENOMEM;
2961
2962         lcfg = &lcr->lcr_cfg;
2963         lcfg->lcfg_num = convert ? simple_strtoul(tmp, NULL, 0) : 0;
2964         /* truncate the comment to the parameter name */
2965         ptr = tmp - 1;
2966         sep = *ptr;
2967         *ptr = '\0';
2968         /* modify all servers and clients */
2969         rc = mgs_write_log_direct_all(env, mgs, fsdb, mti,
2970                                       *tmp == '\0' ? NULL : lcr,
2971                                       mti->mti_fsname, sys, 0);
2972         if (rc == 0 && *tmp != '\0') {
2973                 switch (cmd) {
2974                 case LCFG_SET_TIMEOUT:
2975                         if (!obd_timeout_set || lcfg->lcfg_num > obd_timeout)
2976                                 class_process_config(lcfg);
2977                         break;
2978                 case LCFG_SET_LDLM_TIMEOUT:
2979                         if (!ldlm_timeout_set || lcfg->lcfg_num > ldlm_timeout)
2980                                 class_process_config(lcfg);
2981                         break;
2982                 default:
2983                         break;
2984                 }
2985         }
2986         *ptr = sep;
2987         lustre_cfg_rec_free(lcr);
2988         return rc;
2989 }
2990
2991 /* write quota settings into log */
2992 static int mgs_write_log_quota(const struct lu_env *env, struct mgs_device *mgs,
2993                                struct fs_db *fsdb, struct mgs_target_info *mti,
2994                                char *quota, char *ptr)
2995 {
2996         struct mgs_thread_info  *mgi = mgs_env_info(env);
2997         struct llog_cfg_rec     *lcr;
2998         char                    *tmp;
2999         char                     sep;
3000         int                      rc, cmd = LCFG_PARAM;
3001
3002         /* support only 'meta' and 'data' pools so far */
3003         if (class_match_param(ptr, QUOTA_METAPOOL_NAME, &tmp) != 0 &&
3004             class_match_param(ptr, QUOTA_DATAPOOL_NAME, &tmp) != 0) {
3005                 CERROR("parameter quota.%s isn't supported (only quota.mdt "
3006                        "& quota.ost are)\n", ptr);
3007                 return -EINVAL;
3008         }
3009
3010         if (*tmp == '\0') {
3011                 CDEBUG(D_MGS, "global '%s' removed\n", quota);
3012         } else {
3013                 CDEBUG(D_MGS, "global '%s'\n", quota);
3014
3015                 if (strchr(tmp, 'u') == NULL && strchr(tmp, 'g') == NULL &&
3016                     strcmp(tmp, "none") != 0) {
3017                         CERROR("enable option(%s) isn't supported\n", tmp);
3018                         return -EINVAL;
3019                 }
3020         }
3021
3022         lustre_cfg_bufs_reset(&mgi->mgi_bufs, mti->mti_fsname);
3023         lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 1, quota);
3024         lcr = lustre_cfg_rec_new(cmd, &mgi->mgi_bufs);
3025         if (lcr == NULL)
3026                 return -ENOMEM;
3027
3028         /* truncate the comment to the parameter name */
3029         ptr = tmp - 1;
3030         sep = *ptr;
3031         *ptr = '\0';
3032
3033         /* XXX we duplicated quota enable information in all server
3034          *     config logs, it should be moved to a separate config
3035          *     log once we cleanup the config log for global param. */
3036         /* modify all servers */
3037         rc = mgs_write_log_direct_all(env, mgs, fsdb, mti,
3038                                       *tmp == '\0' ? NULL : lcr,
3039                                       mti->mti_fsname, quota, 1);
3040         *ptr = sep;
3041         lustre_cfg_rec_free(lcr);
3042         return rc < 0 ? rc : 0;
3043 }
3044
3045 static int mgs_srpc_set_param_disk(const struct lu_env *env,
3046                                    struct mgs_device *mgs,
3047                                    struct fs_db *fsdb,
3048                                    struct mgs_target_info *mti,
3049                                    char *param)
3050 {
3051         struct mgs_thread_info  *mgi = mgs_env_info(env);
3052         struct llog_cfg_rec     *lcr;
3053         struct llog_handle      *llh = NULL;
3054         char                    *logname;
3055         char                    *comment, *ptr;
3056         int                      rc, len;
3057
3058         ENTRY;
3059
3060         /* get comment */
3061         ptr = strchr(param, '=');
3062         LASSERT(ptr != NULL);
3063         len = ptr - param;
3064
3065         OBD_ALLOC(comment, len + 1);
3066         if (comment == NULL)
3067                 RETURN(-ENOMEM);
3068         strncpy(comment, param, len);
3069         comment[len] = '\0';
3070
3071         /* prepare lcfg */
3072         lustre_cfg_bufs_reset(&mgi->mgi_bufs, mti->mti_svname);
3073         lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 1, param);
3074         lcr = lustre_cfg_rec_new(LCFG_SPTLRPC_CONF, &mgi->mgi_bufs);
3075         if (lcr == NULL)
3076                 GOTO(out_comment, rc = -ENOMEM);
3077
3078         /* construct log name */
3079         rc = name_create(&logname, mti->mti_fsname, "-sptlrpc");
3080         if (rc < 0)
3081                 GOTO(out_lcfg, rc);
3082
3083         if (mgs_log_is_empty(env, mgs, logname)) {
3084                 rc = record_start_log(env, mgs, &llh, logname);
3085                 if (rc < 0)
3086                         GOTO(out, rc);
3087                 record_end_log(env, &llh);
3088         }
3089
3090         /* obsolete old one */
3091         rc = mgs_modify(env, mgs, fsdb, mti, logname, mti->mti_svname,
3092                         comment, CM_SKIP);
3093         if (rc < 0)
3094                 GOTO(out, rc);
3095         /* write the new one */
3096         rc = mgs_write_log_direct(env, mgs, fsdb, logname, lcr,
3097                                   mti->mti_svname, comment);
3098         if (rc)
3099                 CERROR("%s: error writing log %s: rc = %d\n",
3100                        mgs->mgs_obd->obd_name, logname, rc);
3101 out:
3102         name_destroy(&logname);
3103 out_lcfg:
3104         lustre_cfg_rec_free(lcr);
3105 out_comment:
3106         OBD_FREE(comment, len + 1);
3107         RETURN(rc);
3108 }
3109
3110 static int mgs_srpc_set_param_udesc_mem(struct fs_db *fsdb,
3111                                         char *param)
3112 {
3113         char    *ptr;
3114
3115         /* disable the adjustable udesc parameter for now, i.e. use default
3116          * setting that client always ship udesc to MDT if possible. to enable
3117          * it simply remove the following line */
3118         goto error_out;
3119
3120         ptr = strchr(param, '=');
3121         if (ptr == NULL)
3122                 goto error_out;
3123         *ptr++ = '\0';
3124
3125         if (strcmp(param, PARAM_SRPC_UDESC))
3126                 goto error_out;
3127
3128         if (strcmp(ptr, "yes") == 0) {
3129                 set_bit(FSDB_UDESC, &fsdb->fsdb_flags);
3130                 CWARN("Enable user descriptor shipping from client to MDT\n");
3131         } else if (strcmp(ptr, "no") == 0) {
3132                 clear_bit(FSDB_UDESC, &fsdb->fsdb_flags);
3133                 CWARN("Disable user descriptor shipping from client to MDT\n");
3134         } else {
3135                 *(ptr - 1) = '=';
3136                 goto error_out;
3137         }
3138         return 0;
3139
3140 error_out:
3141         CERROR("Invalid param: %s\n", param);
3142         return -EINVAL;
3143 }
3144
3145 static int mgs_srpc_set_param_mem(struct fs_db *fsdb,
3146                                   const char *svname,
3147                                   char *param)
3148 {
3149         struct sptlrpc_rule      rule;
3150         struct sptlrpc_rule_set *rset;
3151         int                      rc;
3152         ENTRY;
3153
3154         if (strncmp(param, PARAM_SRPC, sizeof(PARAM_SRPC) - 1) != 0) {
3155                 CERROR("Invalid sptlrpc parameter: %s\n", param);
3156                 RETURN(-EINVAL);
3157         }
3158
3159         if (strncmp(param, PARAM_SRPC_UDESC,
3160                     sizeof(PARAM_SRPC_UDESC) - 1) == 0) {
3161                 RETURN(mgs_srpc_set_param_udesc_mem(fsdb, param));
3162         }
3163
3164         if (strncmp(param, PARAM_SRPC_FLVR, sizeof(PARAM_SRPC_FLVR) - 1) != 0) {
3165                 CERROR("Invalid sptlrpc flavor parameter: %s\n", param);
3166                 RETURN(-EINVAL);
3167         }
3168
3169         param += sizeof(PARAM_SRPC_FLVR) - 1;
3170
3171         rc = sptlrpc_parse_rule(param, &rule);
3172         if (rc)
3173                 RETURN(rc);
3174
3175         /* mgs rules implies must be mgc->mgs */
3176         if (test_bit(FSDB_MGS_SELF, &fsdb->fsdb_flags)) {
3177                 if ((rule.sr_from != LUSTRE_SP_MGC &&
3178                      rule.sr_from != LUSTRE_SP_ANY) ||
3179                     (rule.sr_to != LUSTRE_SP_MGS &&
3180                      rule.sr_to != LUSTRE_SP_ANY))
3181                         RETURN(-EINVAL);
3182         }
3183
3184         /* preapre room for this coming rule. svcname format should be:
3185          * - fsname: general rule
3186          * - fsname-tgtname: target-specific rule
3187          */
3188         if (strchr(svname, '-')) {
3189                 struct mgs_tgt_srpc_conf *tgtconf;
3190                 int                       found = 0;
3191
3192                 for (tgtconf = fsdb->fsdb_srpc_tgt; tgtconf != NULL;
3193                      tgtconf = tgtconf->mtsc_next) {
3194                         if (!strcmp(tgtconf->mtsc_tgt, svname)) {
3195                                 found = 1;
3196                                 break;
3197                         }
3198                 }
3199
3200                 if (!found) {
3201                         int name_len;
3202
3203                         OBD_ALLOC_PTR(tgtconf);
3204                         if (tgtconf == NULL)
3205                                 RETURN(-ENOMEM);
3206
3207                         name_len = strlen(svname);
3208
3209                         OBD_ALLOC(tgtconf->mtsc_tgt, name_len + 1);
3210                         if (tgtconf->mtsc_tgt == NULL) {
3211                                 OBD_FREE_PTR(tgtconf);
3212                                 RETURN(-ENOMEM);
3213                         }
3214                         memcpy(tgtconf->mtsc_tgt, svname, name_len);
3215
3216                         tgtconf->mtsc_next = fsdb->fsdb_srpc_tgt;
3217                         fsdb->fsdb_srpc_tgt = tgtconf;
3218                 }
3219
3220                 rset = &tgtconf->mtsc_rset;
3221         } else if (strcmp(svname, MGSSELF_NAME) == 0) {
3222                 /* put _mgs related srpc rule directly in mgs ruleset */
3223                 rset = &fsdb->fsdb_mgs->mgs_lut.lut_sptlrpc_rset;
3224         } else {
3225                 rset = &fsdb->fsdb_srpc_gen;
3226         }
3227
3228         rc = sptlrpc_rule_set_merge(rset, &rule);
3229
3230         RETURN(rc);
3231 }
3232
3233 static int mgs_srpc_set_param(const struct lu_env *env,
3234                               struct mgs_device *mgs,
3235                               struct fs_db *fsdb,
3236                               struct mgs_target_info *mti,
3237                               char *param)
3238 {
3239         char                   *copy;
3240         int                     rc, copy_size;
3241         ENTRY;
3242
3243 #ifndef HAVE_GSS
3244         RETURN(-EINVAL);
3245 #endif
3246         /* keep a copy of original param, which could be destroied
3247          * during parsing */
3248         copy_size = strlen(param) + 1;
3249         OBD_ALLOC(copy, copy_size);
3250         if (copy == NULL)
3251                 return -ENOMEM;
3252         memcpy(copy, param, copy_size);
3253
3254         rc = mgs_srpc_set_param_mem(fsdb, mti->mti_svname, param);
3255         if (rc)
3256                 goto out_free;
3257
3258         /* previous steps guaranteed the syntax is correct */
3259         rc = mgs_srpc_set_param_disk(env, mgs, fsdb, mti, copy);
3260         if (rc)
3261                 goto out_free;
3262
3263         if (test_bit(FSDB_MGS_SELF, &fsdb->fsdb_flags)) {
3264                 /*
3265                  * for mgs rules, make them effective immediately.
3266                  */
3267                 LASSERT(fsdb->fsdb_srpc_tgt == NULL);
3268                 sptlrpc_target_update_exp_flavor(mgs->mgs_obd,
3269                                                  &fsdb->fsdb_srpc_gen);
3270         }
3271
3272 out_free:
3273         OBD_FREE(copy, copy_size);
3274         RETURN(rc);
3275 }
3276
3277 struct mgs_srpc_read_data {
3278         struct fs_db   *msrd_fsdb;
3279         int             msrd_skip;
3280 };
3281
3282 static int mgs_srpc_read_handler(const struct lu_env *env,
3283                                  struct llog_handle *llh,
3284                                  struct llog_rec_hdr *rec, void *data)
3285 {
3286         struct mgs_srpc_read_data *msrd = data;
3287         struct cfg_marker         *marker;
3288         struct lustre_cfg         *lcfg = REC_DATA(rec);
3289         char                      *svname, *param;
3290         int                        cfg_len, rc;
3291         ENTRY;
3292
3293         if (rec->lrh_type != OBD_CFG_REC) {
3294                 CERROR("unhandled lrh_type: %#x\n", rec->lrh_type);
3295                 RETURN(-EINVAL);
3296         }
3297
3298         cfg_len = REC_DATA_LEN(rec);
3299
3300         rc = lustre_cfg_sanity_check(lcfg, cfg_len);
3301         if (rc) {
3302                 CERROR("Insane cfg\n");
3303                 RETURN(rc);
3304         }
3305
3306         if (lcfg->lcfg_command == LCFG_MARKER) {
3307                 marker = lustre_cfg_buf(lcfg, 1);
3308
3309                 if (marker->cm_flags & CM_START &&
3310                     marker->cm_flags & CM_SKIP)
3311                         msrd->msrd_skip = 1;
3312                 if (marker->cm_flags & CM_END)
3313                         msrd->msrd_skip = 0;
3314
3315                 RETURN(0);
3316         }
3317
3318         if (msrd->msrd_skip)
3319                 RETURN(0);
3320
3321         if (lcfg->lcfg_command != LCFG_SPTLRPC_CONF) {
3322                 CERROR("invalid command (%x)\n", lcfg->lcfg_command);
3323                 RETURN(0);
3324         }
3325
3326         svname = lustre_cfg_string(lcfg, 0);
3327         if (svname == NULL) {
3328                 CERROR("svname is empty\n");
3329                 RETURN(0);
3330         }
3331
3332         param = lustre_cfg_string(lcfg, 1);
3333         if (param == NULL) {
3334                 CERROR("param is empty\n");
3335                 RETURN(0);
3336         }
3337
3338         rc = mgs_srpc_set_param_mem(msrd->msrd_fsdb, svname, param);
3339         if (rc)
3340                 CERROR("read sptlrpc record error (%d): %s\n", rc, param);
3341
3342         RETURN(0);
3343 }
3344
3345 int mgs_get_fsdb_srpc_from_llog(const struct lu_env *env,
3346                                 struct mgs_device *mgs,
3347                                 struct fs_db *fsdb)
3348 {
3349         struct llog_handle        *llh = NULL;
3350         struct llog_ctxt          *ctxt;
3351         char                      *logname;
3352         struct mgs_srpc_read_data  msrd;
3353         int                        rc;
3354         ENTRY;
3355
3356         /* construct log name */
3357         rc = name_create(&logname, fsdb->fsdb_name, "-sptlrpc");
3358         if (rc)
3359                 RETURN(rc);
3360
3361         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
3362         LASSERT(ctxt != NULL);
3363
3364         if (mgs_log_is_empty(env, mgs, logname))
3365                 GOTO(out, rc = 0);
3366
3367         rc = llog_open(env, ctxt, &llh, NULL, logname,
3368                        LLOG_OPEN_EXISTS);
3369         if (rc < 0) {
3370                 if (rc == -ENOENT)
3371                         rc = 0;
3372                 GOTO(out, rc);
3373         }
3374
3375         rc = llog_init_handle(env, llh, LLOG_F_IS_PLAIN, NULL);
3376         if (rc)
3377                 GOTO(out_close, rc);
3378
3379         if (llog_get_size(llh) <= 1)
3380                 GOTO(out_close, rc = 0);
3381
3382         msrd.msrd_fsdb = fsdb;
3383         msrd.msrd_skip = 0;
3384
3385         rc = llog_process(env, llh, mgs_srpc_read_handler, (void *)&msrd,
3386                           NULL);
3387
3388 out_close:
3389         llog_close(env, llh);
3390 out:
3391         llog_ctxt_put(ctxt);
3392         name_destroy(&logname);
3393
3394         if (rc)
3395                 CERROR("failed to read sptlrpc config database: %d\n", rc);
3396         RETURN(rc);
3397 }
3398
3399 /* Permanent settings of all parameters by writing into the appropriate
3400  * configuration logs.
3401  * A parameter with null value ("<param>='\0'") means to erase it out of
3402  * the logs.
3403  */
3404 static int mgs_write_log_param(const struct lu_env *env,
3405                                struct mgs_device *mgs, struct fs_db *fsdb,
3406                                struct mgs_target_info *mti, char *ptr)
3407 {
3408         struct mgs_thread_info *mgi = mgs_env_info(env);
3409         char *logname;
3410         char *tmp;
3411         int rc = 0;
3412         ENTRY;
3413
3414         /* For various parameter settings, we have to figure out which logs
3415            care about them (e.g. both mdt and client for lov settings) */
3416         CDEBUG(D_MGS, "next param '%s'\n", ptr);
3417
3418         /* The params are stored in MOUNT_DATA_FILE and modified via
3419            tunefs.lustre, or set using lctl conf_param */
3420
3421         /* Processed in lustre_start_mgc */
3422         if (class_match_param(ptr, PARAM_MGSNODE, NULL) == 0)
3423                 GOTO(end, rc);
3424
3425         /* Processed in ost/mdt */
3426         if (class_match_param(ptr, PARAM_NETWORK, NULL) == 0)
3427                 GOTO(end, rc);
3428
3429         /* Processed in mgs_write_log_ost */
3430         if (class_match_param(ptr, PARAM_FAILMODE, NULL) == 0) {
3431                 if (mti->mti_flags & LDD_F_PARAM) {
3432                         LCONSOLE_ERROR_MSG(0x169, "%s can only be "
3433                                            "changed with tunefs.lustre"
3434                                            "and --writeconf\n", ptr);
3435                         rc = -EPERM;
3436                 }
3437                 GOTO(end, rc);
3438         }
3439
3440         if (class_match_param(ptr, PARAM_SRPC, NULL) == 0) {
3441                 rc = mgs_srpc_set_param(env, mgs, fsdb, mti, ptr);
3442                 GOTO(end, rc);
3443         }
3444
3445         if (class_match_param(ptr, PARAM_FAILNODE, NULL) == 0) {
3446                 /* Add a failover nidlist */
3447                 rc = 0;
3448                 /* We already processed failovers params for new
3449                    targets in mgs_write_log_target */
3450                 if (mti->mti_flags & LDD_F_PARAM) {
3451                         CDEBUG(D_MGS, "Adding failnode\n");
3452                         rc = mgs_write_log_add_failnid(env, mgs, fsdb, mti);
3453                 }
3454                 GOTO(end, rc);
3455         }
3456
3457         if (class_match_param(ptr, PARAM_SYS, &tmp) == 0) {
3458                 rc = mgs_write_log_sys(env, mgs, fsdb, mti, ptr, tmp);
3459                 GOTO(end, rc);
3460         }
3461
3462         if (class_match_param(ptr, PARAM_QUOTA, &tmp) == 0) {
3463                 rc = mgs_write_log_quota(env, mgs, fsdb, mti, ptr, tmp);
3464                 GOTO(end, rc);
3465         }
3466
3467         if (class_match_param(ptr, PARAM_OSC PARAM_ACTIVE, &tmp) == 0 ||
3468             class_match_param(ptr, PARAM_MDC PARAM_ACTIVE, &tmp) == 0) {
3469                 /* active=0 means off, anything else means on */
3470                 int flag = (*tmp == '0') ? CM_EXCLUDE : 0;
3471                 bool deactive_osc = memcmp(ptr, PARAM_OSC PARAM_ACTIVE,
3472                                           strlen(PARAM_OSC PARAM_ACTIVE)) == 0;
3473                 int i;
3474
3475                 if (!deactive_osc) {
3476                         __u32   index;
3477
3478                         rc = server_name2index(mti->mti_svname, &index, NULL);
3479                         if (rc < 0)
3480                                 GOTO(end, rc);
3481
3482                         if (index == 0) {
3483                                 LCONSOLE_ERROR_MSG(0x144, "%s: MDC0 can not be"
3484                                                    " (de)activated.\n",
3485                                                    mti->mti_svname);
3486                                 GOTO(end, rc = -EPERM);
3487                         }
3488                 }
3489
3490                 LCONSOLE_WARN("Permanently %sactivating %s\n",
3491                               flag ? "de" : "re", mti->mti_svname);
3492                 /* Modify clilov */
3493                 rc = name_create(&logname, mti->mti_fsname, "-client");
3494                 if (rc < 0)
3495                         GOTO(end, rc);
3496                 rc = mgs_modify(env, mgs, fsdb, mti, logname,
3497                                 mti->mti_svname,
3498                                 deactive_osc ? "add osc" : "add mdc", flag);
3499                 name_destroy(&logname);
3500                 if (rc < 0)
3501                         goto active_err;
3502
3503                 /* Modify mdtlov */
3504                 /* Add to all MDT logs for DNE */
3505                 for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
3506                         if (!test_bit(i, fsdb->fsdb_mdt_index_map))
3507                                 continue;
3508                         rc = name_create_mdt(&logname, mti->mti_fsname, i);
3509                         if (rc < 0)
3510                                 GOTO(end, rc);
3511                         rc = mgs_modify(env, mgs, fsdb, mti, logname,
3512                                         mti->mti_svname,
3513                                         deactive_osc ? "add osc" : "add osp",
3514                                         flag);
3515                         name_destroy(&logname);
3516                         if (rc < 0)
3517                                 goto active_err;
3518                 }
3519 active_err:
3520                 if (rc < 0) {
3521                         LCONSOLE_ERROR_MSG(0x145, "Couldn't find %s in"
3522                                            "log (%d). No permanent "
3523                                            "changes were made to the "
3524                                            "config log.\n",
3525                                            mti->mti_svname, rc);
3526                         if (test_bit(FSDB_OLDLOG14, &fsdb->fsdb_flags))
3527                                 LCONSOLE_ERROR_MSG(0x146, "This may be"
3528                                                    " because the log"
3529                                                    "is in the old 1.4"
3530                                                    "style. Consider "
3531                                                    " --writeconf to "
3532                                                    "update the logs.\n");
3533                         GOTO(end, rc);
3534                 }
3535                 /* Fall through to osc/mdc proc for deactivating live
3536                    OSC/OSP on running MDT / clients. */
3537         }
3538         /* Below here, let obd's XXX_process_config methods handle it */
3539
3540         /* All lov. in proc */
3541         if (class_match_param(ptr, PARAM_LOV, NULL) == 0) {
3542                 char *mdtlovname;
3543
3544                 CDEBUG(D_MGS, "lov param %s\n", ptr);
3545                 if (!(mti->mti_flags & LDD_F_SV_TYPE_MDT)) {
3546                         LCONSOLE_ERROR_MSG(0x147, "LOV params must be "
3547                                            "set on the MDT, not %s. "
3548                                            "Ignoring.\n",
3549                                            mti->mti_svname);
3550                         GOTO(end, rc = 0);
3551                 }
3552
3553                 /* Modify mdtlov */
3554                 if (mgs_log_is_empty(env, mgs, mti->mti_svname))
3555                         GOTO(end, rc = -ENODEV);
3556
3557                 rc = name_create_mdt_and_lov(&logname, &mdtlovname, fsdb,
3558                                              mti->mti_stripe_index);
3559                 if (rc)
3560                         GOTO(end, rc);
3561                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, mti->mti_svname,
3562                                   &mgi->mgi_bufs, mdtlovname, ptr);
3563                 name_destroy(&logname);
3564                 name_destroy(&mdtlovname);
3565                 if (rc)
3566                         GOTO(end, rc);
3567
3568                 /* Modify clilov */
3569                 rc = name_create(&logname, mti->mti_fsname, "-client");
3570                 if (rc)
3571                         GOTO(end, rc);
3572                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, logname, &mgi->mgi_bufs,
3573                                   fsdb->fsdb_clilov, ptr);
3574                 name_destroy(&logname);
3575                 GOTO(end, rc);
3576         }
3577
3578         /* All osc., mdc., llite. params in proc */
3579         if ((class_match_param(ptr, PARAM_OSC, NULL) == 0) ||
3580             (class_match_param(ptr, PARAM_MDC, NULL) == 0) ||
3581             (class_match_param(ptr, PARAM_LLITE, NULL) == 0)) {
3582                 char *cname;
3583
3584                 if (test_bit(FSDB_OLDLOG14, &fsdb->fsdb_flags)) {
3585                         LCONSOLE_ERROR_MSG(0x148, "Upgraded client logs for %s"
3586                                            " cannot be modified. Consider"
3587                                            " updating the configuration with"
3588                                            " --writeconf\n",
3589                                            mti->mti_svname);
3590                         GOTO(end, rc = -EINVAL);
3591                 }
3592                 if (memcmp(ptr, PARAM_LLITE, strlen(PARAM_LLITE)) == 0) {
3593                         rc = name_create(&cname, mti->mti_fsname, "-client");
3594                         /* Add the client type to match the obdname in
3595                            class_config_llog_handler */
3596                 } else if (mti->mti_flags & LDD_F_SV_TYPE_MDT) {
3597                         rc = name_create(&cname, mti->mti_svname, "-mdc");
3598                 } else if (mti->mti_flags & LDD_F_SV_TYPE_OST) {
3599                         rc = name_create(&cname, mti->mti_svname, "-osc");
3600                 } else {
3601                         GOTO(end, rc = -EINVAL);
3602                 }
3603                 if (rc)
3604                         GOTO(end, rc);
3605
3606                 /* Forbid direct update of llite root squash parameters.
3607                  * These parameters are indirectly set via the MDT settings.
3608                  * See (LU-1778) */
3609                 if ((class_match_param(ptr, PARAM_LLITE, &tmp) == 0) &&
3610                     ((memcmp(tmp, "root_squash=", 12) == 0) ||
3611                      (memcmp(tmp, "nosquash_nids=", 14) == 0))) {
3612                         LCONSOLE_ERROR("%s: root squash parameters can only "
3613                                 "be updated through MDT component\n",
3614                                 mti->mti_fsname);
3615                         name_destroy(&cname);
3616                         GOTO(end, rc = -EINVAL);
3617                 }
3618
3619                 CDEBUG(D_MGS, "%.3s param %s\n", ptr, ptr + 4);
3620
3621                 /* Modify client */
3622                 rc = name_create(&logname, mti->mti_fsname, "-client");
3623                 if (rc) {
3624                         name_destroy(&cname);
3625                         GOTO(end, rc);
3626                 }
3627                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, logname, &mgi->mgi_bufs,
3628                                   cname, ptr);
3629
3630                 /* osc params affect the MDT as well */
3631                 if (!rc && (mti->mti_flags & LDD_F_SV_TYPE_OST)) {
3632                         int i;
3633
3634                         for (i = 0; i < INDEX_MAP_SIZE * 8; i++){
3635                                 if (!test_bit(i, fsdb->fsdb_mdt_index_map))
3636                                         continue;
3637                                 name_destroy(&cname);
3638                                 rc = name_create_mdt_osc(&cname, mti->mti_svname,
3639                                                          fsdb, i);
3640                                 name_destroy(&logname);
3641                                 if (rc)
3642                                         break;
3643                                 rc = name_create_mdt(&logname,
3644                                                      mti->mti_fsname, i);
3645                                 if (rc)
3646                                         break;
3647                                 if (!mgs_log_is_empty(env, mgs, logname)) {
3648                                         rc = mgs_wlp_lcfg(env, mgs, fsdb,
3649                                                           mti, logname,
3650                                                           &mgi->mgi_bufs,
3651                                                           cname, ptr);
3652                                         if (rc)
3653                                                 break;
3654                                 }
3655                         }
3656                 }
3657
3658                 /* For mdc activate/deactivate, it affects OSP on MDT as well */
3659                 if (class_match_param(ptr, PARAM_MDC PARAM_ACTIVE, &tmp) == 0 &&
3660                     rc == 0) {
3661                         char suffix[16];
3662                         char *lodname = NULL;
3663                         char *param_str = NULL;
3664                         int i;
3665                         int index;
3666
3667                         /* replace mdc with osp */
3668                         memcpy(ptr, PARAM_OSP, strlen(PARAM_OSP));
3669                         rc = server_name2index(mti->mti_svname, &index, NULL);
3670                         if (rc < 0) {
3671                                 memcpy(ptr, PARAM_MDC, strlen(PARAM_MDC));
3672                                 GOTO(end, rc);
3673                         }
3674
3675                         for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
3676                                 if (!test_bit(i, fsdb->fsdb_mdt_index_map))
3677                                         continue;
3678
3679                                 if (i == index)
3680                                         continue;
3681
3682                                 name_destroy(&logname);
3683                                 rc = name_create_mdt(&logname, mti->mti_fsname,
3684                                                      i);
3685                                 if (rc < 0)
3686                                         break;