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