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