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