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
  50 #include "mgs_internal.h"
  51
  52 /********************** Class functions ********************/
  53
  54 /**
  55  * Find all logs in CONFIG directory and link then into list.
  56  *
  57  * \param[in] env       pointer to the thread context
  58  * \param[in] mgs       pointer to the mgs device
  59  * \param[out] log_list the list to hold the found llog name entry
  60  *
  61  * \retval              0 for success
  62  * \retval              negative error number on failure
  63  **/
  64 int class_dentry_readdir(const struct lu_env *env, struct mgs_device *mgs,
  65                          struct list_head *log_list)
  66 {
  67         struct dt_object *dir = mgs->mgs_configs_dir;
  68         const struct dt_it_ops *iops;
  69         struct dt_it *it;
  70         struct mgs_direntry *de;
  71         char *key;
  72         int rc, key_sz;
  73
  74         INIT_LIST_HEAD(log_list);
  75
  76         LASSERT(dir);
  77         LASSERT(dir->do_index_ops);
  78
  79         iops = &dir->do_index_ops->dio_it;
  80         it = iops->init(env, dir, LUDA_64BITHASH);
  81         if (IS_ERR(it))
  82                 RETURN(PTR_ERR(it));
  83
  84         rc = iops->load(env, it, 0);
  85         if (rc <= 0)
  86                 GOTO(fini, rc = 0);
  87
  88         /* main cycle */
  89         do {
  90                 key = (void *)iops->key(env, it);
  91                 if (IS_ERR(key)) {
  92                         CERROR("%s: key failed when listing %s: rc = %d\n",
  93                                mgs->mgs_obd->obd_name, MOUNT_CONFIGS_DIR,
  94                                (int) PTR_ERR(key));
  95                         goto next;
  96                 }
  97                 key_sz = iops->key_size(env, it);
  98                 LASSERT(key_sz > 0);
  99
 100                 /* filter out "." and ".." entries */
 101                 if (key[0] == '.') {
 102                         if (key_sz == 1)
 103                                 goto next;
 104                         if (key_sz == 2 && key[1] == '.')
 105                                 goto next;
 106                 }
 107
 108                 /* filter out ".bak" files */
 109                 /* sizeof(".bak") - 1 == 3 */
 110                 if (key_sz >= 3 &&
 111                     !memcmp(".bak", key + key_sz - 3, 3)) {
 112                         CDEBUG(D_MGS, "Skipping backup file %.*s\n",
 113                                key_sz, key);
 114                         goto next;
 115                 }
 116
 117                 de = mgs_direntry_alloc(key_sz + 1);
 118                 if (de == NULL) {
 119                         rc = -ENOMEM;
 120                         break;
 121                 }
 122
 123                 memcpy(de->mde_name, key, key_sz);
 124                 de->mde_name[key_sz] = 0;
 125
 126                 list_add(&de->mde_list, log_list);
 127
 128 next:
 129                 rc = iops->next(env, it);
 130         } while (rc == 0);
 131         if (rc > 0)
 132                 rc = 0;
 133
 134         iops->put(env, it);
 135
 136 fini:
 137         iops->fini(env, it);
 138         if (rc) {
 139                 struct mgs_direntry *n;
 140
 141                 CERROR("%s: key failed when listing %s: rc = %d\n",
 142                        mgs->mgs_obd->obd_name, MOUNT_CONFIGS_DIR, rc);
 143
 144                 list_for_each_entry_safe(de, n, log_list, mde_list) {
 145                         list_del_init(&de->mde_list);
 146                         mgs_direntry_free(de);
 147                 }
 148         }
 149
 150         RETURN(rc);
 151 }
 152
 153 /******************** DB functions *********************/
 154
 155 static inline int name_create(char **newname, char *prefix, char *suffix)
 156 {
 157         LASSERT(newname);
 158         OBD_ALLOC(*newname, strlen(prefix) + strlen(suffix) + 1);
 159         if (!*newname)
 160                 return -ENOMEM;
 161         sprintf(*newname, "%s%s", prefix, suffix);
 162         return 0;
 163 }
 164
 165 static inline void name_destroy(char **name)
 166 {
 167         if (*name)
 168                 OBD_FREE(*name, strlen(*name) + 1);
 169         *name = NULL;
 170 }
 171
 172 struct mgs_fsdb_handler_data
 173 {
 174         struct fs_db   *fsdb;
 175         __u32           ver;
 176 };
 177
 178 /* from the (client) config log, figure out:
 179         1. which ost's/mdt's are configured (by index)
 180         2. what the last config step is
 181         3. COMPAT_18 osc name
 182 */
 183 /* It might be better to have a separate db file, instead of parsing the info
 184    out of the client log.  This is slow and potentially error-prone. */
 185 static int mgs_fsdb_handler(const struct lu_env *env, struct llog_handle *llh,
 186                             struct llog_rec_hdr *rec, void *data)
 187 {
 188         struct mgs_fsdb_handler_data *d = data;
 189         struct fs_db *fsdb = d->fsdb;
 190         int cfg_len = rec->lrh_len;
 191         char *cfg_buf = (char*) (rec + 1);
 192         struct lustre_cfg *lcfg;
 193         __u32 index;
 194         int rc = 0;
 195         ENTRY;
 196
 197         if (rec->lrh_type != OBD_CFG_REC) {
 198                 CERROR("unhandled lrh_type: %#x\n", rec->lrh_type);
 199                 RETURN(-EINVAL);
 200         }
 201
 202         rc = lustre_cfg_sanity_check(cfg_buf, cfg_len);
 203         if (rc) {
 204                 CERROR("Insane cfg\n");
 205                 RETURN(rc);
 206         }
 207
 208         lcfg = (struct lustre_cfg *)cfg_buf;
 209
 210         CDEBUG(D_INFO, "cmd %x %s %s\n", lcfg->lcfg_command,
 211                lustre_cfg_string(lcfg, 0), lustre_cfg_string(lcfg, 1));
 212
 213         /* Figure out ost indicies */
 214         /* lov_modify_tgts add 0:lov1  1:ost1_UUID  2(index):0  3(gen):1 */
 215         if (lcfg->lcfg_command == LCFG_LOV_ADD_OBD ||
 216             lcfg->lcfg_command == LCFG_LOV_DEL_OBD) {
 217                 index = simple_strtoul(lustre_cfg_string(lcfg, 2),
 218                                        NULL, 10);
 219                 CDEBUG(D_MGS, "OST index for %s is %u (%s)\n",
 220                        lustre_cfg_string(lcfg, 1), index,
 221                        lustre_cfg_string(lcfg, 2));
 222                 set_bit(index, fsdb->fsdb_ost_index_map);
 223         }
 224
 225         /* Figure out mdt indicies */
 226         /* attach   0:MDC_uml1_mdsA_MNT_client  1:mdc  2:1d834_MNT_client_03f */
 227         if ((lcfg->lcfg_command == LCFG_ATTACH) &&
 228             (strcmp(lustre_cfg_string(lcfg, 1), LUSTRE_MDC_NAME) == 0)) {
 229                 rc = server_name2index(lustre_cfg_string(lcfg, 0),
 230                                        &index, NULL);
 231                 if (rc != LDD_F_SV_TYPE_MDT) {
 232                         CWARN("Unparsable MDC name %s, assuming index 0\n",
 233                               lustre_cfg_string(lcfg, 0));
 234                         index = 0;
 235                 }
 236                 rc = 0;
 237                 CDEBUG(D_MGS, "MDT index is %u\n", index);
 238                 if (!test_bit(index, fsdb->fsdb_mdt_index_map)) {
 239                         set_bit(index, fsdb->fsdb_mdt_index_map);
 240                         fsdb->fsdb_mdt_count++;
 241                 }
 242         }
 243
 244         /**
 245          * figure out the old config. fsdb_gen = 0 means old log
 246          * It is obsoleted and not supported anymore
 247          */
 248         if (fsdb->fsdb_gen == 0) {
 249                 CERROR("Old config format is not supported\n");
 250                 RETURN(-EINVAL);
 251         }
 252
 253         /*
 254          * compat to 1.8, check osc name used by MDT0 to OSTs, bz18548.
 255          */
 256         if (!test_bit(FSDB_OSCNAME18, &fsdb->fsdb_flags) &&
 257             lcfg->lcfg_command == LCFG_ATTACH &&
 258             strcmp(lustre_cfg_string(lcfg, 1), LUSTRE_OSC_NAME) == 0) {
 259                 if (OBD_OCD_VERSION_MAJOR(d->ver) == 1 &&
 260                     OBD_OCD_VERSION_MINOR(d->ver) <= 8) {
 261                         CWARN("MDT using 1.8 OSC name scheme\n");
 262                         set_bit(FSDB_OSCNAME18, &fsdb->fsdb_flags);
 263                 }
 264         }
 265
 266         if (lcfg->lcfg_command == LCFG_MARKER) {
 267                 struct cfg_marker *marker;
 268                 marker = lustre_cfg_buf(lcfg, 1);
 269
 270                 d->ver = marker->cm_vers;
 271
 272                 /* Keep track of the latest marker step */
 273                 fsdb->fsdb_gen = max(fsdb->fsdb_gen, marker->cm_step);
 274         }
 275
 276         RETURN(rc);
 277 }
 278
 279 /* fsdb->fsdb_mutex is already held  in mgs_find_or_make_fsdb*/
 280 static int mgs_get_fsdb_from_llog(const struct lu_env *env,
 281                                   struct mgs_device *mgs,
 282                                   struct fs_db *fsdb)
 283 {
 284         char *logname;
 285         struct llog_handle *loghandle;
 286         struct llog_ctxt *ctxt;
 287         struct mgs_fsdb_handler_data d = {
 288                 .fsdb = fsdb,
 289         };
 290         int rc;
 291
 292         ENTRY;
 293
 294         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
 295         LASSERT(ctxt != NULL);
 296         rc = name_create(&logname, fsdb->fsdb_name, "-client");
 297         if (rc)
 298                 GOTO(out_put, rc);
 299         rc = llog_open_create(env, ctxt, &loghandle, NULL, logname);
 300         if (rc)
 301                 GOTO(out_pop, rc);
 302
 303         rc = llog_init_handle(env, loghandle, LLOG_F_IS_PLAIN, NULL);
 304         if (rc)
 305                 GOTO(out_close, rc);
 306
 307         if (llog_get_size(loghandle) <= 1)
 308                 set_bit(FSDB_LOG_EMPTY, &fsdb->fsdb_flags);
 309
 310         rc = llog_process(env, loghandle, mgs_fsdb_handler, (void *)&d, NULL);
 311         CDEBUG(D_INFO, "get_db = %d\n", rc);
 312 out_close:
 313         llog_close(env, loghandle);
 314 out_pop:
 315         name_destroy(&logname);
 316 out_put:
 317         llog_ctxt_put(ctxt);
 318
 319         RETURN(rc);
 320 }
 321
 322 static void mgs_free_fsdb_srpc(struct fs_db *fsdb)
 323 {
 324         struct mgs_tgt_srpc_conf *tgtconf;
 325
 326         /* free target-specific rules */
 327         while (fsdb->fsdb_srpc_tgt) {
 328                 tgtconf = fsdb->fsdb_srpc_tgt;
 329                 fsdb->fsdb_srpc_tgt = tgtconf->mtsc_next;
 330
 331                 LASSERT(tgtconf->mtsc_tgt);
 332
 333                 sptlrpc_rule_set_free(&tgtconf->mtsc_rset);
 334                 OBD_FREE(tgtconf->mtsc_tgt, strlen(tgtconf->mtsc_tgt) + 1);
 335                 OBD_FREE_PTR(tgtconf);
 336         }
 337
 338         /* free general rules */
 339         sptlrpc_rule_set_free(&fsdb->fsdb_srpc_gen);
 340 }
 341
 342 static void mgs_unlink_fsdb(struct mgs_device *mgs, struct fs_db *fsdb)
 343 {
 344         mutex_lock(&mgs->mgs_mutex);
 345         if (likely(!list_empty(&fsdb->fsdb_list))) {
 346                 LASSERTF(atomic_read(&fsdb->fsdb_ref) >= 2,
 347                          "Invalid ref %d on %s\n",
 348                          atomic_read(&fsdb->fsdb_ref),
 349                          fsdb->fsdb_name);
 350
 351                 list_del_init(&fsdb->fsdb_list);
 352                 /* Drop the reference on the list.*/
 353                 mgs_put_fsdb(mgs, fsdb);
 354         }
 355         mutex_unlock(&mgs->mgs_mutex);
 356 }
 357
 358 /* The caller must hold mgs->mgs_mutex. */
 359 static inline struct fs_db *
 360 mgs_find_fsdb_noref(struct mgs_device *mgs, const char *fsname)
 361 {
 362         struct fs_db *fsdb;
 363         struct list_head *tmp;
 364
 365         list_for_each(tmp, &mgs->mgs_fs_db_list) {
 366                 fsdb = list_entry(tmp, struct fs_db, fsdb_list);
 367                 if (strcmp(fsdb->fsdb_name, fsname) == 0)
 368                         return fsdb;
 369         }
 370
 371         return NULL;
 372 }
 373
 374 /* The caller must hold mgs->mgs_mutex. */
 375 static void mgs_remove_fsdb_by_name(struct mgs_device *mgs, const char *name)
 376 {
 377         struct fs_db *fsdb;
 378
 379         fsdb = mgs_find_fsdb_noref(mgs, name);
 380         if (fsdb) {
 381                 list_del_init(&fsdb->fsdb_list);
 382                 /* Drop the reference on the list.*/
 383                 mgs_put_fsdb(mgs, fsdb);
 384         }
 385 }
 386
 387 /* The caller must hold mgs->mgs_mutex. */
 388 struct fs_db *mgs_find_fsdb(struct mgs_device *mgs, const char *fsname)
 389 {
 390         struct fs_db *fsdb;
 391
 392         fsdb = mgs_find_fsdb_noref(mgs, fsname);
 393         if (fsdb)
 394                 atomic_inc(&fsdb->fsdb_ref);
 395
 396         return fsdb;
 397 }
 398
 399 /* The caller must hold mgs->mgs_mutex. */
 400 static struct fs_db *mgs_new_fsdb(const struct lu_env *env,
 401                                   struct mgs_device *mgs, char *fsname)
 402 {
 403         struct fs_db *fsdb;
 404         int rc;
 405         ENTRY;
 406
 407         if (strlen(fsname) >= sizeof(fsdb->fsdb_name)) {
 408                 CERROR("fsname %s is too long\n", fsname);
 409
 410                 RETURN(ERR_PTR(-EINVAL));
 411         }
 412
 413         OBD_ALLOC_PTR(fsdb);
 414         if (!fsdb)
 415                 RETURN(ERR_PTR(-ENOMEM));
 416
 417         strncpy(fsdb->fsdb_name, fsname, sizeof(fsdb->fsdb_name));
 418         mutex_init(&fsdb->fsdb_mutex);
 419         INIT_LIST_HEAD(&fsdb->fsdb_list);
 420         set_bit(FSDB_UDESC, &fsdb->fsdb_flags);
 421         fsdb->fsdb_gen = 1;
 422         INIT_LIST_HEAD(&fsdb->fsdb_clients);
 423         atomic_set(&fsdb->fsdb_notify_phase, 0);
 424         init_waitqueue_head(&fsdb->fsdb_notify_waitq);
 425         init_completion(&fsdb->fsdb_notify_comp);
 426
 427         if (strcmp(fsname, MGSSELF_NAME) == 0) {
 428                 set_bit(FSDB_MGS_SELF, &fsdb->fsdb_flags);
 429                 fsdb->fsdb_mgs = mgs;
 430                 if (logname_is_barrier(fsname))
 431                         goto add;
 432         } else {
 433                 OBD_ALLOC(fsdb->fsdb_mdt_index_map, INDEX_MAP_SIZE);
 434                 if (!fsdb->fsdb_mdt_index_map) {
 435                         CERROR("No memory for MDT index maps\n");
 436
 437                         GOTO(err, rc = -ENOMEM);
 438                 }
 439
 440                 OBD_ALLOC(fsdb->fsdb_ost_index_map, INDEX_MAP_SIZE);
 441                 if (!fsdb->fsdb_ost_index_map) {
 442                         CERROR("No memory for OST index maps\n");
 443
 444                         GOTO(err, rc = -ENOMEM);
 445                 }
 446
 447                 if (logname_is_barrier(fsname))
 448                         goto add;
 449
 450                 rc = name_create(&fsdb->fsdb_clilov, fsname, "-clilov");
 451                 if (rc)
 452                         GOTO(err, rc);
 453
 454                 rc = name_create(&fsdb->fsdb_clilmv, fsname, "-clilmv");
 455                 if (rc)
 456                         GOTO(err, rc);
 457
 458                 /* initialise data for NID table */
 459                 mgs_ir_init_fs(env, mgs, fsdb);
 460                 lproc_mgs_add_live(mgs, fsdb);
 461         }
 462
 463         if (!test_bit(FSDB_MGS_SELF, &fsdb->fsdb_flags) &&
 464             strcmp(PARAMS_FILENAME, fsname) != 0) {
 465                 /* populate the db from the client llog */
 466                 rc = mgs_get_fsdb_from_llog(env, mgs, fsdb);
 467                 if (rc) {
 468                         CERROR("Can't get db from client log %d\n", rc);
 469
 470                         GOTO(err, rc);
 471                 }
 472         }
 473
 474         /* populate srpc rules from params llog */
 475         rc = mgs_get_fsdb_srpc_from_llog(env, mgs, fsdb);
 476         if (rc) {
 477                 CERROR("Can't get db from params log %d\n", rc);
 478
 479                 GOTO(err, rc);
 480         }
 481
 482 add:
 483         /* One ref is for the fsdb on the list.
 484          * The other ref is for the caller. */
 485         atomic_set(&fsdb->fsdb_ref, 2);
 486         list_add(&fsdb->fsdb_list, &mgs->mgs_fs_db_list);
 487
 488         RETURN(fsdb);
 489
 490 err:
 491         atomic_set(&fsdb->fsdb_ref, 1);
 492         mgs_put_fsdb(mgs, fsdb);
 493
 494         RETURN(ERR_PTR(rc));
 495 }
 496
 497 static void mgs_free_fsdb(struct mgs_device *mgs, struct fs_db *fsdb)
 498 {
 499         LASSERT(list_empty(&fsdb->fsdb_list));
 500
 501         lproc_mgs_del_live(mgs, fsdb);
 502
 503         /* deinitialize fsr */
 504         if (fsdb->fsdb_mgs)
 505                 mgs_ir_fini_fs(mgs, fsdb);
 506
 507         if (fsdb->fsdb_ost_index_map)
 508                 OBD_FREE(fsdb->fsdb_ost_index_map, INDEX_MAP_SIZE);
 509         if (fsdb->fsdb_mdt_index_map)
 510                 OBD_FREE(fsdb->fsdb_mdt_index_map, INDEX_MAP_SIZE);
 511         name_destroy(&fsdb->fsdb_clilov);
 512         name_destroy(&fsdb->fsdb_clilmv);
 513         mgs_free_fsdb_srpc(fsdb);
 514         OBD_FREE_PTR(fsdb);
 515 }
 516
 517 void mgs_put_fsdb(struct mgs_device *mgs, struct fs_db *fsdb)
 518 {
 519         if (atomic_dec_and_test(&fsdb->fsdb_ref))
 520                 mgs_free_fsdb(mgs, fsdb);
 521 }
 522
 523 int mgs_init_fsdb_list(struct mgs_device *mgs)
 524 {
 525         INIT_LIST_HEAD(&mgs->mgs_fs_db_list);
 526         return 0;
 527 }
 528
 529 int mgs_cleanup_fsdb_list(struct mgs_device *mgs)
 530 {
 531         struct fs_db *fsdb;
 532         struct list_head *tmp, *tmp2;
 533
 534         mutex_lock(&mgs->mgs_mutex);
 535         list_for_each_safe(tmp, tmp2, &mgs->mgs_fs_db_list) {
 536                 fsdb = list_entry(tmp, struct fs_db, fsdb_list);
 537                 list_del_init(&fsdb->fsdb_list);
 538                 mgs_put_fsdb(mgs, fsdb);
 539         }
 540         mutex_unlock(&mgs->mgs_mutex);
 541         return 0;
 542 }
 543
 544 int mgs_find_or_make_fsdb(const struct lu_env *env, struct mgs_device *mgs,
 545                           char *name, struct fs_db **dbh)
 546 {
 547         struct fs_db *fsdb;
 548         int rc = 0;
 549         ENTRY;
 550
 551         mutex_lock(&mgs->mgs_mutex);
 552         fsdb = mgs_find_fsdb(mgs, name);
 553         if (!fsdb) {
 554                 fsdb = mgs_new_fsdb(env, mgs, name);
 555                 if (IS_ERR(fsdb))
 556                         rc = PTR_ERR(fsdb);
 557
 558                 CDEBUG(D_MGS, "Created new db: rc = %d\n", rc);
 559         }
 560         mutex_unlock(&mgs->mgs_mutex);
 561
 562         if (!rc)
 563                 *dbh = fsdb;
 564
 565         RETURN(rc);
 566 }
 567
 568 /* 1 = index in use
 569    0 = index unused
 570    -1= empty client log */
 571 int mgs_check_index(const struct lu_env *env,
 572                     struct mgs_device *mgs,
 573                     struct mgs_target_info *mti)
 574 {
 575         struct fs_db *fsdb;
 576         void *imap;
 577         int rc = 0;
 578         ENTRY;
 579
 580         LASSERT(!(mti->mti_flags & LDD_F_NEED_INDEX));
 581
 582         rc = mgs_find_or_make_fsdb(env, mgs, mti->mti_fsname, &fsdb);
 583         if (rc) {
 584                 CERROR("Can't get db for %s\n", mti->mti_fsname);
 585                 RETURN(rc);
 586         }
 587
 588         if (test_bit(FSDB_LOG_EMPTY, &fsdb->fsdb_flags))
 589                 GOTO(out, rc = -1);
 590
 591         if (mti->mti_flags & LDD_F_SV_TYPE_OST)
 592                 imap = fsdb->fsdb_ost_index_map;
 593         else if (mti->mti_flags & LDD_F_SV_TYPE_MDT)
 594                 imap = fsdb->fsdb_mdt_index_map;
 595         else
 596                 GOTO(out, rc = -EINVAL);
 597
 598         if (test_bit(mti->mti_stripe_index, imap))
 599                 GOTO(out, rc = 1);
 600
 601         GOTO(out, rc = 0);
 602
 603 out:
 604         mgs_put_fsdb(mgs, fsdb);
 605         return rc;
 606 }
 607
 608 static __inline__ int next_index(void *index_map, int map_len)
 609 {
 610         int i;
 611         for (i = 0; i < map_len * 8; i++)
 612                 if (!test_bit(i, index_map)) {
 613                          return i;
 614                  }
 615         CERROR("max index %d exceeded.\n", i);
 616         return -1;
 617 }
 618
 619 /* Make the mdt/ost server obd name based on the filesystem name */
 620 static bool server_make_name(u32 flags, u16 index, const char *fs,
 621                              char *name_buf, size_t name_buf_size)
 622 {
 623         bool invalid_flag = false;
 624
 625         if (flags & (LDD_F_SV_TYPE_MDT | LDD_F_SV_TYPE_OST)) {
 626                 if (!(flags & LDD_F_SV_ALL))
 627                         snprintf(name_buf, name_buf_size, "%.8s%c%s%04x", fs,
 628                                 (flags & LDD_F_VIRGIN) ? ':' :
 629                                         ((flags & LDD_F_WRITECONF) ? '=' : '-'),
 630                                 (flags & LDD_F_SV_TYPE_MDT) ? "MDT" : "OST",
 631                                 index);
 632         } else if (flags & LDD_F_SV_TYPE_MGS) {
 633                 snprintf(name_buf, name_buf_size, "MGS");
 634         } else {
 635                 CERROR("unknown server type %#x\n", flags);
 636                 invalid_flag = true;
 637         }
 638         return invalid_flag;
 639 }
 640
 641 /* Return codes:
 642         0  newly marked as in use
 643         <0 err
 644         +EALREADY for update of an old index */
 645 static int mgs_set_index(const struct lu_env *env,
 646                          struct mgs_device *mgs,
 647                          struct mgs_target_info *mti)
 648 {
 649         struct fs_db *fsdb;
 650         void *imap;
 651         int rc = 0;
 652         ENTRY;
 653
 654         rc = mgs_find_or_make_fsdb(env, mgs, mti->mti_fsname, &fsdb);
 655         if (rc) {
 656                 CERROR("Can't get db for %s\n", mti->mti_fsname);
 657                 RETURN(rc);
 658         }
 659
 660         mutex_lock(&fsdb->fsdb_mutex);
 661         if (mti->mti_flags & LDD_F_SV_TYPE_OST) {
 662                 imap = fsdb->fsdb_ost_index_map;
 663         } else if (mti->mti_flags & LDD_F_SV_TYPE_MDT) {
 664                 imap = fsdb->fsdb_mdt_index_map;
 665         } else {
 666                 GOTO(out_up, rc = -EINVAL);
 667         }
 668
 669         if (mti->mti_flags & LDD_F_NEED_INDEX) {
 670                 rc = next_index(imap, INDEX_MAP_SIZE);
 671                 if (rc == -1)
 672                         GOTO(out_up, rc = -ERANGE);
 673                 mti->mti_stripe_index = rc;
 674         }
 675
 676         /* the last index(0xffff) is reserved for default value. */
 677         if (mti->mti_stripe_index >= INDEX_MAP_SIZE * 8 - 1) {
 678                 LCONSOLE_ERROR_MSG(0x13f, "Server %s requested index %u, "
 679                                    "but index must be less than %u.\n",
 680                                    mti->mti_svname, mti->mti_stripe_index,
 681                                    INDEX_MAP_SIZE * 8 - 1);
 682                 GOTO(out_up, rc = -ERANGE);
 683         }
 684
 685         if (test_bit(mti->mti_stripe_index, imap)) {
 686                 if ((mti->mti_flags & LDD_F_VIRGIN) &&
 687                     !(mti->mti_flags & LDD_F_WRITECONF)) {
 688                         LCONSOLE_ERROR_MSG(0x140, "Server %s requested index "
 689                                            "%d, but that index is already in "
 690                                            "use. Use --writeconf to force\n",
 691                                            mti->mti_svname,
 692                                            mti->mti_stripe_index);
 693                         GOTO(out_up, rc = -EADDRINUSE);
 694                 } else {
 695                         CDEBUG(D_MGS, "Server %s updating index %d\n",
 696                                mti->mti_svname, mti->mti_stripe_index);
 697                         GOTO(out_up, rc = EALREADY);
 698                 }
 699         } else {
 700                 set_bit(mti->mti_stripe_index, imap);
 701                 if (mti->mti_flags & LDD_F_SV_TYPE_MDT)
 702                         fsdb->fsdb_mdt_count++;
 703         }
 704
 705         set_bit(mti->mti_stripe_index, imap);
 706         clear_bit(FSDB_LOG_EMPTY, &fsdb->fsdb_flags);
 707         if (server_make_name(mti->mti_flags & ~(LDD_F_VIRGIN | LDD_F_WRITECONF),
 708                              mti->mti_stripe_index, mti->mti_fsname,
 709                              mti->mti_svname, sizeof(mti->mti_svname))) {
 710                 CERROR("unknown server type %#x\n", mti->mti_flags);
 711                 GOTO(out_up, rc = -EINVAL);
 712         }
 713
 714         CDEBUG(D_MGS, "Set index for %s to %d\n", mti->mti_svname,
 715                mti->mti_stripe_index);
 716
 717         GOTO(out_up, rc = 0);
 718
 719 out_up:
 720         mutex_unlock(&fsdb->fsdb_mutex);
 721         mgs_put_fsdb(mgs, fsdb);
 722         return rc;
 723 }
 724
 725 struct mgs_modify_lookup {
 726         struct cfg_marker mml_marker;
 727         int               mml_modified;
 728 };
 729
 730 static int mgs_check_record_match(const struct lu_env *env,
 731                                 struct llog_handle *llh,
 732                                 struct llog_rec_hdr *rec, void *data)
 733 {
 734         struct cfg_marker *mc_marker = data;
 735         struct cfg_marker *marker;
 736         struct lustre_cfg *lcfg = REC_DATA(rec);
 737         int cfg_len = REC_DATA_LEN(rec);
 738         int rc;
 739         ENTRY;
 740
 741
 742         if (rec->lrh_type != OBD_CFG_REC) {
 743                 CDEBUG(D_ERROR, "Unhandled lrh_type: %#x\n", rec->lrh_type);
 744                 RETURN(-EINVAL);
 745         }
 746
 747         rc = lustre_cfg_sanity_check(lcfg, cfg_len);
 748         if (rc) {
 749                 CDEBUG(D_ERROR, "Insane cfg\n");
 750                 RETURN(rc);
 751         }
 752
 753         /* We only care about markers */
 754         if (lcfg->lcfg_command != LCFG_MARKER)
 755                 RETURN(0);
 756
 757         marker = lustre_cfg_buf(lcfg, 1);
 758
 759         if (marker->cm_flags & CM_SKIP)
 760                 RETURN(0);
 761
 762         if ((strcmp(mc_marker->cm_comment, marker->cm_comment) == 0) &&
 763                 (strcmp(mc_marker->cm_tgtname, marker->cm_tgtname) == 0)) {
 764                 /* Found a non-skipped marker match */
 765                 CDEBUG(D_MGS, "Matched rec %u marker %d flag %x %s %s\n",
 766                         rec->lrh_index, marker->cm_step,
 767                         marker->cm_flags, marker->cm_tgtname,
 768                         marker->cm_comment);
 769                 rc = LLOG_PROC_BREAK;
 770         }
 771
 772         RETURN(rc);
 773 }
 774
 775 /**
 776  * Check an existing config log record with matching comment and device
 777  * Return code:
 778  * 0 - checked successfully,
 779  * LLOG_PROC_BREAK - record matches
 780  * negative - error
 781  */
 782 static int mgs_check_marker(const struct lu_env *env, struct mgs_device *mgs,
 783                 struct fs_db *fsdb, struct mgs_target_info *mti,
 784                 char *logname, char *devname, char *comment)
 785 {
 786         struct llog_handle *loghandle;
 787         struct llog_ctxt *ctxt;
 788         struct cfg_marker *mc_marker;
 789         int rc;
 790
 791         ENTRY;
 792
 793         LASSERT(mutex_is_locked(&fsdb->fsdb_mutex));
 794         CDEBUG(D_MGS, "mgs check %s/%s/%s\n", logname, devname, comment);
 795
 796         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
 797         LASSERT(ctxt != NULL);
 798         rc = llog_open(env, ctxt, &loghandle, NULL, logname, LLOG_OPEN_EXISTS);
 799         if (rc < 0) {
 800                 if (rc == -ENOENT)
 801                         rc = 0;
 802                 GOTO(out_pop, rc);
 803         }
 804
 805         rc = llog_init_handle(env, loghandle, LLOG_F_IS_PLAIN, NULL);
 806         if (rc)
 807                 GOTO(out_close, rc);
 808
 809         if (llog_get_size(loghandle) <= 1)
 810                 GOTO(out_close, rc = 0);
 811
 812         OBD_ALLOC_PTR(mc_marker);
 813         if (!mc_marker)
 814                 GOTO(out_close, rc = -ENOMEM);
 815         if (strlcpy(mc_marker->cm_comment, comment,
 816                 sizeof(mc_marker->cm_comment)) >=
 817                 sizeof(mc_marker->cm_comment))
 818                 GOTO(out_free, rc = -E2BIG);
 819         if (strlcpy(mc_marker->cm_tgtname, devname,
 820                 sizeof(mc_marker->cm_tgtname)) >=
 821                 sizeof(mc_marker->cm_tgtname))
 822                 GOTO(out_free, rc = -E2BIG);
 823
 824         rc = llog_process(env, loghandle, mgs_check_record_match,
 825                         (void *)mc_marker, NULL);
 826
 827 out_free:
 828         OBD_FREE_PTR(mc_marker);
 829
 830 out_close:
 831         llog_close(env, loghandle);
 832 out_pop:
 833         if (rc && rc != LLOG_PROC_BREAK)
 834                 CDEBUG(D_ERROR, "%s: mgs check %s/%s failed: rc = %d\n",
 835                         mgs->mgs_obd->obd_name, mti->mti_svname, comment, rc);
 836         llog_ctxt_put(ctxt);
 837         RETURN(rc);
 838 }
 839
 840 static int mgs_modify_handler(const struct lu_env *env,
 841                               struct llog_handle *llh,
 842                               struct llog_rec_hdr *rec, void *data)
 843 {
 844         struct mgs_modify_lookup *mml = data;
 845         struct cfg_marker *marker;
 846         struct lustre_cfg *lcfg = REC_DATA(rec);
 847         int cfg_len = REC_DATA_LEN(rec);
 848         int rc;
 849         ENTRY;
 850
 851         if (rec->lrh_type != OBD_CFG_REC) {
 852                 CERROR("unhandled lrh_type: %#x\n", rec->lrh_type);
 853                 RETURN(-EINVAL);
 854         }
 855
 856         rc = lustre_cfg_sanity_check(lcfg, cfg_len);
 857         if (rc) {
 858                 CERROR("Insane cfg\n");
 859                 RETURN(rc);
 860         }
 861
 862         /* We only care about markers */
 863         if (lcfg->lcfg_command != LCFG_MARKER)
 864                 RETURN(0);
 865
 866         marker = lustre_cfg_buf(lcfg, 1);
 867         if ((strcmp(mml->mml_marker.cm_comment, marker->cm_comment) == 0) &&
 868             (strcmp(mml->mml_marker.cm_tgtname, marker->cm_tgtname) == 0) &&
 869             !(marker->cm_flags & CM_SKIP)) {
 870                 /* Found a non-skipped marker match */
 871                 CDEBUG(D_MGS, "Changing rec %u marker %d %x->%x: %s %s\n",
 872                        rec->lrh_index, marker->cm_step,
 873                        marker->cm_flags, mml->mml_marker.cm_flags,
 874                        marker->cm_tgtname, marker->cm_comment);
 875                 /* Overwrite the old marker llog entry */
 876                 marker->cm_flags &= ~CM_EXCLUDE; /* in case we're unexcluding */
 877                 marker->cm_flags |= mml->mml_marker.cm_flags;
 878                 marker->cm_canceltime = mml->mml_marker.cm_canceltime;
 879                 rc = llog_write(env, llh, rec, rec->lrh_index);
 880                 if (!rc)
 881                          mml->mml_modified++;
 882         }
 883
 884         RETURN(rc);
 885 }
 886
 887 /**
 888  * Modify an existing config log record (for CM_SKIP or CM_EXCLUDE)
 889  * Return code:
 890  * 0 - modified successfully,
 891  * 1 - no modification was done
 892  * negative - error
 893  */
 894 static int mgs_modify(const struct lu_env *env, struct mgs_device *mgs,
 895                       struct fs_db *fsdb, struct mgs_target_info *mti,
 896                       char *logname, char *devname, char *comment, int flags)
 897 {
 898         struct llog_handle *loghandle;
 899         struct llog_ctxt *ctxt;
 900         struct mgs_modify_lookup *mml;
 901         int rc;
 902
 903         ENTRY;
 904
 905         LASSERT(mutex_is_locked(&fsdb->fsdb_mutex));
 906         CDEBUG(D_MGS, "modify %s/%s/%s fl=%x\n", logname, devname, comment,
 907                flags);
 908
 909         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
 910         LASSERT(ctxt != NULL);
 911         rc = llog_open(env, ctxt, &loghandle, NULL, logname, LLOG_OPEN_EXISTS);
 912         if (rc < 0) {
 913                 if (rc == -ENOENT)
 914                         rc = 0;
 915                 GOTO(out_pop, rc);
 916         }
 917
 918         rc = llog_init_handle(env, loghandle, LLOG_F_IS_PLAIN, NULL);
 919         if (rc)
 920                 GOTO(out_close, rc);
 921
 922         if (llog_get_size(loghandle) <= 1)
 923                 GOTO(out_close, rc = 0);
 924
 925         OBD_ALLOC_PTR(mml);
 926         if (!mml)
 927                 GOTO(out_close, rc = -ENOMEM);
 928         if (strlcpy(mml->mml_marker.cm_comment, comment,
 929                     sizeof(mml->mml_marker.cm_comment)) >=
 930             sizeof(mml->mml_marker.cm_comment))
 931                 GOTO(out_free, rc = -E2BIG);
 932         if (strlcpy(mml->mml_marker.cm_tgtname, devname,
 933                     sizeof(mml->mml_marker.cm_tgtname)) >=
 934             sizeof(mml->mml_marker.cm_tgtname))
 935                 GOTO(out_free, rc = -E2BIG);
 936         /* Modify mostly means cancel */
 937         mml->mml_marker.cm_flags = flags;
 938         mml->mml_marker.cm_canceltime = flags ? ktime_get_real_seconds() : 0;
 939         mml->mml_modified = 0;
 940         rc = llog_process(env, loghandle, mgs_modify_handler, (void *)mml,
 941                           NULL);
 942         if (!rc && !mml->mml_modified)
 943                 rc = 1;
 944
 945 out_free:
 946         OBD_FREE_PTR(mml);
 947
 948 out_close:
 949         llog_close(env, loghandle);
 950 out_pop:
 951         if (rc < 0)
 952                 CERROR("%s: modify %s/%s failed: rc = %d\n",
 953                        mgs->mgs_obd->obd_name, mti->mti_svname, comment, rc);
 954         llog_ctxt_put(ctxt);
 955         RETURN(rc);
 956 }
 957
 958 /** This structure is passed to mgs_replace_handler */
 959 struct mgs_replace_uuid_lookup {
 960         /* Nids are replaced for this target device */
 961         struct mgs_target_info target;
 962         /* Temporary modified llog */
 963         struct llog_handle *temp_llh;
 964         /* Flag is set if in target block*/
 965         int in_target_device;
 966         /* Nids already added. Just skip (multiple nids) */
 967         int device_nids_added;
 968         /* Flag is set if this block should not be copied */
 969         int skip_it;
 970 };
 971
 972 /**
 973  * Check: a) if block should be skipped
 974  * b) is it target block
 975  *
 976  * \param[in] lcfg
 977  * \param[in] mrul
 978  *
 979  * \retval 0 should not to be skipped
 980  * \retval 1 should to be skipped
 981  */
 982 static int check_markers(struct lustre_cfg *lcfg,
 983                          struct mgs_replace_uuid_lookup *mrul)
 984 {
 985          struct cfg_marker *marker;
 986
 987         /* Track markers. Find given device */
 988         if (lcfg->lcfg_command == LCFG_MARKER) {
 989                 marker = lustre_cfg_buf(lcfg, 1);
 990                 /* Clean llog from records marked as CM_EXCLUDE.
 991                    CM_SKIP records are used for "active" command
 992                    and can be restored if needed */
 993                 if ((marker->cm_flags & (CM_EXCLUDE | CM_START)) ==
 994                     (CM_EXCLUDE | CM_START)) {
 995                         mrul->skip_it = 1;
 996                         return 1;
 997                 }
 998
 999                 if ((marker->cm_flags & (CM_EXCLUDE | CM_END)) ==
1000                     (CM_EXCLUDE | CM_END)) {
1001                         mrul->skip_it = 0;
1002                         return 1;
1003                 }
1004
1005                 if (strcmp(mrul->target.mti_svname, marker->cm_tgtname) == 0) {
1006                         LASSERT(!(marker->cm_flags & CM_START) ||
1007                                 !(marker->cm_flags & CM_END));
1008                         if (marker->cm_flags & CM_START) {
1009                                 mrul->in_target_device = 1;
1010                                 mrul->device_nids_added = 0;
1011                         } else if (marker->cm_flags & CM_END)
1012                                 mrul->in_target_device = 0;
1013                 }
1014         }
1015
1016         return 0;
1017 }
1018
1019 static int record_base(const struct lu_env *env, struct llog_handle *llh,
1020                      char *cfgname, lnet_nid_t nid, int cmd,
1021                      char *s1, char *s2, char *s3, char *s4)
1022 {
1023         struct mgs_thread_info  *mgi = mgs_env_info(env);
1024         struct llog_cfg_rec     *lcr;
1025         int rc;
1026
1027         CDEBUG(D_MGS, "lcfg %s %#x %s %s %s %s\n", cfgname,
1028                cmd, s1, s2, s3, s4);
1029
1030         lustre_cfg_bufs_reset(&mgi->mgi_bufs, cfgname);
1031         if (s1)
1032                 lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 1, s1);
1033         if (s2)
1034                 lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 2, s2);
1035         if (s3)
1036                 lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 3, s3);
1037         if (s4)
1038                 lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 4, s4);
1039
1040         lcr = lustre_cfg_rec_new(cmd, &mgi->mgi_bufs);
1041         if (lcr == NULL)
1042                 return -ENOMEM;
1043
1044         lcr->lcr_cfg.lcfg_nid = nid;
1045         rc = llog_write(env, llh, &lcr->lcr_hdr, LLOG_NEXT_IDX);
1046
1047         lustre_cfg_rec_free(lcr);
1048
1049         if (rc < 0)
1050                 CDEBUG(D_MGS,
1051                        "failed to write lcfg %s %#x %s %s %s %s: rc = %d\n",
1052                        cfgname, cmd, s1, s2, s3, s4, rc);
1053         return rc;
1054 }
1055
1056 static inline int record_add_uuid(const struct lu_env *env,
1057                                   struct llog_handle *llh,
1058                                   uint64_t nid, char *uuid)
1059 {
1060         return record_base(env, llh, NULL, nid, LCFG_ADD_UUID, uuid,
1061                            NULL, NULL, NULL);
1062 }
1063
1064 static inline int record_add_conn(const struct lu_env *env,
1065                                   struct llog_handle *llh,
1066                                   char *devname, char *uuid)
1067 {
1068         return record_base(env, llh, devname, 0, LCFG_ADD_CONN, uuid,
1069                            NULL, NULL, NULL);
1070 }
1071
1072 static inline int record_attach(const struct lu_env *env,
1073                                 struct llog_handle *llh, char *devname,
1074                                 char *type, char *uuid)
1075 {
1076         return record_base(env, llh, devname, 0, LCFG_ATTACH, type, uuid,
1077                            NULL, NULL);
1078 }
1079
1080 static inline int record_setup(const struct lu_env *env,
1081                                struct llog_handle *llh, char *devname,
1082                                char *s1, char *s2, char *s3, char *s4)
1083 {
1084         return record_base(env, llh, devname, 0, LCFG_SETUP, s1, s2, s3, s4);
1085 }
1086
1087 /**
1088  * \retval <0 record processing error
1089  * \retval n record is processed. No need copy original one.
1090  * \retval 0 record is not processed.
1091  */
1092 static int process_command(const struct lu_env *env, struct lustre_cfg *lcfg,
1093                            struct mgs_replace_uuid_lookup *mrul)
1094 {
1095         int nids_added = 0;
1096         lnet_nid_t nid;
1097         char *ptr;
1098         int rc;
1099
1100         if (lcfg->lcfg_command == LCFG_ADD_UUID) {
1101                 /* LCFG_ADD_UUID command found. Let's skip original command
1102                    and add passed nids */
1103                 ptr = mrul->target.mti_params;
1104                 while (class_parse_nid(ptr, &nid, &ptr) == 0) {
1105                         CDEBUG(D_MGS, "add nid %s with uuid %s, "
1106                                "device %s\n", libcfs_nid2str(nid),
1107                                 mrul->target.mti_params,
1108                                 mrul->target.mti_svname);
1109                         rc = record_add_uuid(env,
1110                                              mrul->temp_llh, nid,
1111                                              mrul->target.mti_params);
1112                         if (!rc)
1113                                 nids_added++;
1114                 }
1115
1116                 if (nids_added == 0) {
1117                         CERROR("No new nids were added, nid %s with uuid %s, "
1118                                "device %s\n", libcfs_nid2str(nid),
1119                                mrul->target.mti_params,
1120                                mrul->target.mti_svname);
1121                         RETURN(-ENXIO);
1122                 } else {
1123                         mrul->device_nids_added = 1;
1124                 }
1125
1126                 return nids_added;
1127         }
1128
1129         if (mrul->device_nids_added && lcfg->lcfg_command == LCFG_SETUP) {
1130                 /* LCFG_SETUP command found. UUID should be changed */
1131                 rc = record_setup(env,
1132                                   mrul->temp_llh,
1133                                   /* devname the same */
1134                                   lustre_cfg_string(lcfg, 0),
1135                                   /* s1 is not changed */
1136                                   lustre_cfg_string(lcfg, 1),
1137                                   /* new uuid should be
1138                                   the full nidlist */
1139                                   mrul->target.mti_params,
1140                                   /* s3 is not changed */
1141                                   lustre_cfg_string(lcfg, 3),
1142                                   /* s4 is not changed */
1143                                   lustre_cfg_string(lcfg, 4));
1144                 return rc ? rc : 1;
1145         }
1146
1147         /* Another commands in target device block */
1148         return 0;
1149 }
1150
1151 /**
1152  * Handler that called for every record in llog.
1153  * Records are processed in order they placed in llog.
1154  *
1155  * \param[in] llh       log to be processed
1156  * \param[in] rec       current record
1157  * \param[in] data      mgs_replace_uuid_lookup structure
1158  *
1159  * \retval 0    success
1160  */
1161 static int mgs_replace_handler(const struct lu_env *env,
1162                                struct llog_handle *llh,
1163                                struct llog_rec_hdr *rec,
1164                                void *data)
1165 {
1166         struct mgs_replace_uuid_lookup *mrul;
1167         struct lustre_cfg *lcfg = REC_DATA(rec);
1168         int cfg_len = REC_DATA_LEN(rec);
1169         int rc;
1170         ENTRY;
1171
1172         mrul = (struct mgs_replace_uuid_lookup *)data;
1173
1174         if (rec->lrh_type != OBD_CFG_REC) {
1175                 CERROR("unhandled lrh_type: %#x, cmd %x %s %s\n",
1176                        rec->lrh_type, lcfg->lcfg_command,
1177                        lustre_cfg_string(lcfg, 0),
1178                        lustre_cfg_string(lcfg, 1));
1179                 RETURN(-EINVAL);
1180         }
1181
1182         rc = lustre_cfg_sanity_check(lcfg, cfg_len);
1183         if (rc) {
1184                 /* Do not copy any invalidated records */
1185                 GOTO(skip_out, rc = 0);
1186         }
1187
1188         rc = check_markers(lcfg, mrul);
1189         if (rc || mrul->skip_it)
1190                 GOTO(skip_out, rc = 0);
1191
1192         /* Write to new log all commands outside target device block */
1193         if (!mrul->in_target_device)
1194                 GOTO(copy_out, rc = 0);
1195
1196         /* Skip all other LCFG_ADD_UUID and LCFG_ADD_CONN records
1197            (failover nids) for this target, assuming that if then
1198            primary is changing then so is the failover */
1199         if (mrul->device_nids_added &&
1200             (lcfg->lcfg_command == LCFG_ADD_UUID ||
1201              lcfg->lcfg_command == LCFG_ADD_CONN))
1202                 GOTO(skip_out, rc = 0);
1203
1204         rc = process_command(env, lcfg, mrul);
1205         if (rc < 0)
1206                 RETURN(rc);
1207
1208         if (rc)
1209                 RETURN(0);
1210 copy_out:
1211         /* Record is placed in temporary llog as is */
1212         rc = llog_write(env, mrul->temp_llh, rec, LLOG_NEXT_IDX);
1213
1214         CDEBUG(D_MGS, "Copied idx=%d, rc=%d, len=%d, cmd %x %s %s\n",
1215                rec->lrh_index, rc, rec->lrh_len, lcfg->lcfg_command,
1216                lustre_cfg_string(lcfg, 0), lustre_cfg_string(lcfg, 1));
1217         RETURN(rc);
1218
1219 skip_out:
1220         CDEBUG(D_MGS, "Skipped idx=%d, rc=%d, len=%d, cmd %x %s %s\n",
1221                rec->lrh_index, rc, rec->lrh_len, lcfg->lcfg_command,
1222                lustre_cfg_string(lcfg, 0), lustre_cfg_string(lcfg, 1));
1223         RETURN(rc);
1224 }
1225
1226 static int mgs_log_is_empty(const struct lu_env *env,
1227                             struct mgs_device *mgs, char *name)
1228 {
1229         struct llog_ctxt        *ctxt;
1230         int                      rc;
1231
1232         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
1233         LASSERT(ctxt != NULL);
1234
1235         rc = llog_is_empty(env, ctxt, name);
1236         llog_ctxt_put(ctxt);
1237         return rc;
1238 }
1239
1240 static int mgs_replace_nids_log(const struct lu_env *env,
1241                                 struct obd_device *mgs, struct fs_db *fsdb,
1242                                 char *logname, char *devname, char *nids)
1243 {
1244         struct llog_handle *orig_llh, *backup_llh;
1245         struct llog_ctxt *ctxt;
1246         struct mgs_replace_uuid_lookup *mrul;
1247         struct mgs_device *mgs_dev = lu2mgs_dev(mgs->obd_lu_dev);
1248         static struct obd_uuid   cfg_uuid = { .uuid = "config_uuid" };
1249         char *backup;
1250         int rc, rc2;
1251         ENTRY;
1252
1253         CDEBUG(D_MGS, "Replace nids for %s in %s\n", devname, logname);
1254
1255         ctxt = llog_get_context(mgs, LLOG_CONFIG_ORIG_CTXT);
1256         LASSERT(ctxt != NULL);
1257
1258         if (mgs_log_is_empty(env, mgs_dev, logname)) {
1259                 /* Log is empty. Nothing to replace */
1260                 GOTO(out_put, rc = 0);
1261         }
1262
1263         OBD_ALLOC(backup, strlen(logname) + strlen(".bak") + 1);
1264         if (backup == NULL)
1265                 GOTO(out_put, rc = -ENOMEM);
1266
1267         sprintf(backup, "%s.bak", logname);
1268
1269         rc = llog_backup(env, mgs, ctxt, ctxt, logname, backup);
1270         if (rc == 0) {
1271                 /* Now erase original log file. Connections are not allowed.
1272                    Backup is already saved */
1273                 rc = llog_erase(env, ctxt, NULL, logname);
1274                 if (rc < 0)
1275                         GOTO(out_free, rc);
1276         } else if (rc != -ENOENT) {
1277                 CERROR("%s: can't make backup for %s: rc = %d\n",
1278                        mgs->obd_name, logname, rc);
1279                 GOTO(out_free,rc);
1280         }
1281
1282         /* open local log */
1283         rc = llog_open_create(env, ctxt, &orig_llh, NULL, logname);
1284         if (rc)
1285                 GOTO(out_restore, rc);
1286
1287         rc = llog_init_handle(env, orig_llh, LLOG_F_IS_PLAIN, &cfg_uuid);
1288         if (rc)
1289                 GOTO(out_closel, rc);
1290
1291         /* open backup llog */
1292         rc = llog_open(env, ctxt, &backup_llh, NULL, backup,
1293                        LLOG_OPEN_EXISTS);
1294         if (rc)
1295                 GOTO(out_closel, rc);
1296
1297         rc = llog_init_handle(env, backup_llh, LLOG_F_IS_PLAIN, NULL);
1298         if (rc)
1299                 GOTO(out_close, rc);
1300
1301         if (llog_get_size(backup_llh) <= 1)
1302                 GOTO(out_close, rc = 0);
1303
1304         OBD_ALLOC_PTR(mrul);
1305         if (!mrul)
1306                 GOTO(out_close, rc = -ENOMEM);
1307         /* devname is only needed information to replace UUID records */
1308         strlcpy(mrul->target.mti_svname, devname,
1309                 sizeof(mrul->target.mti_svname));
1310         /* parse nids later */
1311         strlcpy(mrul->target.mti_params, nids, sizeof(mrul->target.mti_params));
1312         /* Copy records to this temporary llog */
1313         mrul->temp_llh = orig_llh;
1314
1315         rc = llog_process(env, backup_llh, mgs_replace_handler,
1316                           (void *)mrul, NULL);
1317         OBD_FREE_PTR(mrul);
1318 out_close:
1319         rc2 = llog_close(NULL, backup_llh);
1320         if (!rc)
1321                 rc = rc2;
1322 out_closel:
1323         rc2 = llog_close(NULL, orig_llh);
1324         if (!rc)
1325                 rc = rc2;
1326
1327 out_restore:
1328         if (rc) {
1329                 CERROR("%s: llog should be restored: rc = %d\n",
1330                        mgs->obd_name, rc);
1331                 rc2 = llog_backup(env, mgs, ctxt, ctxt, backup,
1332                                   logname);
1333                 if (rc2 < 0)
1334                         CERROR("%s: can't restore backup %s: rc = %d\n",
1335                                mgs->obd_name, logname, rc2);
1336         }
1337
1338 out_free:
1339         OBD_FREE(backup, strlen(backup) + 1);
1340
1341 out_put:
1342         llog_ctxt_put(ctxt);
1343
1344         if (rc)
1345                 CERROR("%s: failed to replace nids in log %s: rc = %d\n",
1346                        mgs->obd_name, logname, rc);
1347
1348         RETURN(rc);
1349 }
1350
1351 /**
1352  * Parse device name and get file system name and/or device index
1353  *
1354  * @devname     device name (ex. lustre-MDT0000)
1355  * @fsname      file system name extracted from @devname and returned
1356  *              to the caller (optional)
1357  * @index       device index extracted from @devname and returned to
1358  *              the caller (optional)
1359  *
1360  * RETURN       0                       success if we are only interested in
1361  *                                      extracting fsname from devname.
1362  *                                      i.e index is NULL
1363  *
1364  *              LDD_F_SV_TYPE_*         Besides extracting the fsname the
1365  *                                      user also wants the index. Report to
1366  *                                      the user the type of obd device the
1367  *                                      returned index belongs too.
1368  *
1369  *              -EINVAL                 The obd device name is improper so
1370  *                                      fsname could not be extracted.
1371  *
1372  *              -ENXIO                  Failed to extract the index out of
1373  *                                      the obd device name. Most likely an
1374  *                                      invalid obd device name
1375  */
1376 static int mgs_parse_devname(char *devname, char *fsname, u32 *index)
1377 {
1378         int rc = 0;
1379         ENTRY;
1380
1381         /* Extract fsname */
1382         if (fsname) {
1383                 rc = server_name2fsname(devname, fsname, NULL);
1384                 if (rc < 0) {
1385                         CDEBUG(D_MGS, "Device name %s without fsname\n",
1386                                devname);
1387                         RETURN(-EINVAL);
1388                 }
1389         }
1390
1391         if (index) {
1392                 rc = server_name2index(devname, index, NULL);
1393                 if (rc < 0) {
1394                         CDEBUG(D_MGS, "Device name %s with wrong index\n",
1395                                devname);
1396                         RETURN(-ENXIO);
1397                 }
1398         }
1399
1400         /* server_name2index can return LDD_F_SV_TYPE_* so always return rc */
1401         RETURN(rc);
1402 }
1403
1404 /* This is only called during replace_nids */
1405 static int only_mgs_is_running(struct obd_device *mgs_obd)
1406 {
1407         /* TDB: Is global variable with devices count exists? */
1408         int num_devices = get_devices_count();
1409         int num_exports = 0;
1410         struct obd_export *exp;
1411
1412         spin_lock(&mgs_obd->obd_dev_lock);
1413         list_for_each_entry(exp, &mgs_obd->obd_exports, exp_obd_chain) {
1414                 /* skip self export */
1415                 if (exp == mgs_obd->obd_self_export)
1416                         continue;
1417                 if (exp_connect_flags(exp) & OBD_CONNECT_MDS_MDS)
1418                         continue;
1419
1420                 ++num_exports;
1421
1422                 CERROR("%s: node %s still connected during replace_nids "
1423                        "connect_flags:%llx\n",
1424                        mgs_obd->obd_name,
1425                        libcfs_nid2str(exp->exp_nid_stats->nid),
1426                        exp_connect_flags(exp));
1427
1428         }
1429         spin_unlock(&mgs_obd->obd_dev_lock);
1430
1431         /* osd, MGS and MGC + self_export
1432            (wc -l /proc/fs/lustre/devices <= 2) && (non self exports == 0) */
1433         return (num_devices <= 3) && (num_exports == 0);
1434 }
1435
1436 static int name_create_mdt(char **logname, char *fsname, int i)
1437 {
1438         char mdt_index[9];
1439
1440         sprintf(mdt_index, "-MDT%04x", i);
1441         return name_create(logname, fsname, mdt_index);
1442 }
1443
1444 /**
1445  * Replace nids for \a device to \a nids values
1446  *
1447  * \param obd           MGS obd device
1448  * \param devname       nids need to be replaced for this device
1449  * (ex. lustre-OST0000)
1450  * \param nids          nids list (ex. nid1,nid2,nid3)
1451  *
1452  * \retval 0    success
1453  */
1454 int mgs_replace_nids(const struct lu_env *env,
1455                      struct mgs_device *mgs,
1456                      char *devname, char *nids)
1457 {
1458         /* Assume fsname is part of device name */
1459         char fsname[MTI_NAME_MAXLEN];
1460         int rc;
1461         __u32 index;
1462         char *logname;
1463         struct fs_db *fsdb = NULL;
1464         unsigned int i;
1465         int conn_state;
1466         struct obd_device *mgs_obd = mgs->mgs_obd;
1467         ENTRY;
1468
1469         /* We can only change NIDs if no other nodes are connected */
1470         spin_lock(&mgs_obd->obd_dev_lock);
1471         conn_state = mgs_obd->obd_no_conn;
1472         mgs_obd->obd_no_conn = 1;
1473         spin_unlock(&mgs_obd->obd_dev_lock);
1474
1475         /* We can not change nids if not only MGS is started */
1476         if (!only_mgs_is_running(mgs_obd)) {
1477                 CERROR("Only MGS is allowed to be started\n");
1478                 GOTO(out, rc = -EINPROGRESS);
1479         }
1480
1481         /* Get fsname and index */
1482         rc = mgs_parse_devname(devname, fsname, &index);
1483         if (rc < 0)
1484                 GOTO(out, rc);
1485
1486         rc = mgs_find_or_make_fsdb(env, mgs, fsname, &fsdb);
1487         if (rc) {
1488                 CERROR("%s: can't find fsdb: rc = %d\n", fsname, rc);
1489                 GOTO(out, rc);
1490         }
1491
1492         /* Process client llogs */
1493         name_create(&logname, fsname, "-client");
1494         rc = mgs_replace_nids_log(env, mgs_obd, fsdb, logname, devname, nids);
1495         name_destroy(&logname);
1496         if (rc) {
1497                 CERROR("%s: error while replacing NIDs for %s: rc = %d\n",
1498                        fsname, devname, rc);
1499                 GOTO(out, rc);
1500         }
1501
1502         /* Process MDT llogs */
1503         for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
1504                 if (!test_bit(i, fsdb->fsdb_mdt_index_map))
1505                         continue;
1506                 name_create_mdt(&logname, fsname, i);
1507                 rc = mgs_replace_nids_log(env, mgs_obd, fsdb, logname, devname, nids);
1508                 name_destroy(&logname);
1509                 if (rc)
1510                         GOTO(out, rc);
1511         }
1512
1513 out:
1514         spin_lock(&mgs_obd->obd_dev_lock);
1515         mgs_obd->obd_no_conn = conn_state;
1516         spin_unlock(&mgs_obd->obd_dev_lock);
1517
1518         if (fsdb)
1519                 mgs_put_fsdb(mgs, fsdb);
1520
1521         RETURN(rc);
1522 }
1523
1524 static int record_lov_setup(const struct lu_env *env, struct llog_handle *llh,
1525                             char *devname, struct lov_desc *desc)
1526 {
1527         struct mgs_thread_info  *mgi = mgs_env_info(env);
1528         struct llog_cfg_rec     *lcr;
1529         int rc;
1530
1531         lustre_cfg_bufs_reset(&mgi->mgi_bufs, devname);
1532         lustre_cfg_bufs_set(&mgi->mgi_bufs, 1, desc, sizeof(*desc));
1533         lcr = lustre_cfg_rec_new(LCFG_SETUP, &mgi->mgi_bufs);
1534         if (lcr == NULL)
1535                 return -ENOMEM;
1536
1537         rc = llog_write(env, llh, &lcr->lcr_hdr, LLOG_NEXT_IDX);
1538         lustre_cfg_rec_free(lcr);
1539         return rc;
1540 }
1541
1542 static int record_lmv_setup(const struct lu_env *env, struct llog_handle *llh,
1543                             char *devname, struct lmv_desc *desc)
1544 {
1545         struct mgs_thread_info  *mgi = mgs_env_info(env);
1546         struct llog_cfg_rec     *lcr;
1547         int rc;
1548
1549         lustre_cfg_bufs_reset(&mgi->mgi_bufs, devname);
1550         lustre_cfg_bufs_set(&mgi->mgi_bufs, 1, desc, sizeof(*desc));
1551         lcr = lustre_cfg_rec_new(LCFG_SETUP, &mgi->mgi_bufs);
1552         if (lcr == NULL)
1553                 return -ENOMEM;
1554
1555         rc = llog_write(env, llh, &lcr->lcr_hdr, LLOG_NEXT_IDX);
1556         lustre_cfg_rec_free(lcr);
1557         return rc;
1558 }
1559
1560 static inline int record_mdc_add(const struct lu_env *env,
1561                                  struct llog_handle *llh,
1562                                  char *logname, char *mdcuuid,
1563                                  char *mdtuuid, char *index,
1564                                  char *gen)
1565 {
1566         return record_base(env,llh,logname,0,LCFG_ADD_MDC,
1567                            mdtuuid,index,gen,mdcuuid);
1568 }
1569
1570 static inline int record_lov_add(const struct lu_env *env,
1571                                  struct llog_handle *llh,
1572                                  char *lov_name, char *ost_uuid,
1573                                  char *index, char *gen)
1574 {
1575         return record_base(env, llh, lov_name, 0, LCFG_LOV_ADD_OBD,
1576                            ost_uuid, index, gen, NULL);
1577 }
1578
1579 static inline int record_mount_opt(const struct lu_env *env,
1580                                    struct llog_handle *llh,
1581                                    char *profile, char *lov_name,
1582                                    char *mdc_name)
1583 {
1584         return record_base(env, llh, NULL, 0, LCFG_MOUNTOPT,
1585                            profile, lov_name, mdc_name, NULL);
1586 }
1587
1588 static int record_marker(const struct lu_env *env,
1589                          struct llog_handle *llh,
1590                          struct fs_db *fsdb, __u32 flags,
1591                          char *tgtname, char *comment)
1592 {
1593         struct mgs_thread_info *mgi = mgs_env_info(env);
1594         struct llog_cfg_rec *lcr;
1595         int rc;
1596         int cplen = 0;
1597
1598         if (flags & CM_START)
1599                 fsdb->fsdb_gen++;
1600         mgi->mgi_marker.cm_step = fsdb->fsdb_gen;
1601         mgi->mgi_marker.cm_flags = flags;
1602         mgi->mgi_marker.cm_vers = LUSTRE_VERSION_CODE;
1603         cplen = strlcpy(mgi->mgi_marker.cm_tgtname, tgtname,
1604                         sizeof(mgi->mgi_marker.cm_tgtname));
1605         if (cplen >= sizeof(mgi->mgi_marker.cm_tgtname))
1606                 return -E2BIG;
1607         cplen = strlcpy(mgi->mgi_marker.cm_comment, comment,
1608                         sizeof(mgi->mgi_marker.cm_comment));
1609         if (cplen >= sizeof(mgi->mgi_marker.cm_comment))
1610                 return -E2BIG;
1611         mgi->mgi_marker.cm_createtime = ktime_get_real_seconds();
1612         mgi->mgi_marker.cm_canceltime = 0;
1613         lustre_cfg_bufs_reset(&mgi->mgi_bufs, NULL);
1614         lustre_cfg_bufs_set(&mgi->mgi_bufs, 1, &mgi->mgi_marker,
1615                             sizeof(mgi->mgi_marker));
1616         lcr = lustre_cfg_rec_new(LCFG_MARKER, &mgi->mgi_bufs);
1617         if (lcr == NULL)
1618                 return -ENOMEM;
1619
1620         rc = llog_write(env, llh, &lcr->lcr_hdr, LLOG_NEXT_IDX);
1621         lustre_cfg_rec_free(lcr);
1622         return rc;
1623 }
1624
1625 static int record_start_log(const struct lu_env *env, struct mgs_device *mgs,
1626                             struct llog_handle **llh, char *name)
1627 {
1628         static struct obd_uuid   cfg_uuid = { .uuid = "config_uuid" };
1629         struct llog_ctxt        *ctxt;
1630         int                      rc = 0;
1631         ENTRY;
1632
1633         if (*llh)
1634                 GOTO(out, rc = -EBUSY);
1635
1636         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
1637         if (!ctxt)
1638                 GOTO(out, rc = -ENODEV);
1639         LASSERT(ctxt->loc_obd == mgs->mgs_obd);
1640
1641         rc = llog_open_create(env, ctxt, llh, NULL, name);
1642         if (rc)
1643                 GOTO(out_ctxt, rc);
1644         rc = llog_init_handle(env, *llh, LLOG_F_IS_PLAIN, &cfg_uuid);
1645         if (rc)
1646                 llog_close(env, *llh);
1647 out_ctxt:
1648         llog_ctxt_put(ctxt);
1649 out:
1650         if (rc) {
1651                 CERROR("%s: can't start log %s: rc = %d\n",
1652                        mgs->mgs_obd->obd_name, name, rc);
1653                 *llh = NULL;
1654         }
1655         RETURN(rc);
1656 }
1657
1658 static int record_end_log(const struct lu_env *env, struct llog_handle **llh)
1659 {
1660         int rc;
1661
1662         rc = llog_close(env, *llh);
1663         *llh = NULL;
1664
1665         return rc;
1666 }
1667
1668 /******************** config "macros" *********************/
1669
1670 /* write an lcfg directly into a log (with markers) */
1671 static int mgs_write_log_direct(const struct lu_env *env,
1672                                 struct mgs_device *mgs, struct fs_db *fsdb,
1673                                 char *logname, struct llog_cfg_rec *lcr,
1674                                 char *devname, char *comment)
1675 {
1676         struct llog_handle *llh = NULL;
1677         int rc;
1678
1679         ENTRY;
1680
1681         rc = record_start_log(env, mgs, &llh, logname);
1682         if (rc)
1683                 RETURN(rc);
1684
1685         /* FIXME These should be a single journal transaction */
1686         rc = record_marker(env, llh, fsdb, CM_START, devname, comment);
1687         if (rc)
1688                 GOTO(out_end, rc);
1689         rc = llog_write(env, llh, &lcr->lcr_hdr, LLOG_NEXT_IDX);
1690         if (rc)
1691                 GOTO(out_end, rc);
1692         rc = record_marker(env, llh, fsdb, CM_END, devname, comment);
1693         if (rc)
1694                 GOTO(out_end, rc);
1695 out_end:
1696         record_end_log(env, &llh);
1697         RETURN(rc);
1698 }
1699
1700 /* write the lcfg in all logs for the given fs */
1701 static int mgs_write_log_direct_all(const struct lu_env *env,
1702                                     struct mgs_device *mgs,
1703                                     struct fs_db *fsdb,
1704                                     struct mgs_target_info *mti,
1705                                     struct llog_cfg_rec *lcr, char *devname,
1706                                     char *comment, int server_only)
1707 {
1708         struct list_head         log_list;
1709         struct mgs_direntry     *dirent, *n;
1710         char                    *fsname = mti->mti_fsname;
1711         int                      rc = 0, len = strlen(fsname);
1712
1713         ENTRY;
1714         /* Find all the logs in the CONFIGS directory */
1715         rc = class_dentry_readdir(env, mgs, &log_list);
1716         if (rc)
1717                 RETURN(rc);
1718
1719         /* Could use fsdb index maps instead of directory listing */
1720         list_for_each_entry_safe(dirent, n, &log_list, mde_list) {
1721                 list_del_init(&dirent->mde_list);
1722                 /* don't write to sptlrpc rule log */
1723                 if (strstr(dirent->mde_name, "-sptlrpc") != NULL)
1724                         goto next;
1725
1726                 /* caller wants write server logs only */
1727                 if (server_only && strstr(dirent->mde_name, "-client") != NULL)
1728                         goto next;
1729
1730                 if (strlen(dirent->mde_name) <= len ||
1731                     strncmp(fsname, dirent->mde_name, len) != 0 ||
1732                     dirent->mde_name[len] != '-')
1733                         goto next;
1734
1735                 CDEBUG(D_MGS, "Changing log %s\n", dirent->mde_name);
1736                 /* Erase any old settings of this same parameter */
1737                 rc = mgs_modify(env, mgs, fsdb, mti, dirent->mde_name,
1738                                 devname, comment, CM_SKIP);
1739                 if (rc < 0)
1740                         CERROR("%s: Can't modify llog %s: rc = %d\n",
1741                                mgs->mgs_obd->obd_name, dirent->mde_name, rc);
1742                 if (lcr == NULL)
1743                         goto next;
1744                 /* Write the new one */
1745                 rc = mgs_write_log_direct(env, mgs, fsdb, dirent->mde_name,
1746                                           lcr, devname, comment);
1747                 if (rc != 0)
1748                         CERROR("%s: writing log %s: rc = %d\n",
1749                                mgs->mgs_obd->obd_name, dirent->mde_name, rc);
1750 next:
1751                 mgs_direntry_free(dirent);
1752         }
1753
1754         RETURN(rc);
1755 }
1756
1757 static int mgs_write_log_osp_to_mdt(const struct lu_env *env,
1758                                     struct mgs_device *mgs,
1759                                     struct fs_db *fsdb,
1760                                     struct mgs_target_info *mti,
1761                                     int index, char *logname);
1762 static int mgs_write_log_osc_to_lov(const struct lu_env *env,
1763                                     struct mgs_device *mgs,
1764                                     struct fs_db *fsdb,
1765                                     struct mgs_target_info *mti,
1766                                     char *logname, char *suffix, char *lovname,
1767                                     enum lustre_sec_part sec_part, int flags);
1768 static int name_create_mdt_and_lov(char **logname, char **lovname,
1769                                    struct fs_db *fsdb, int i);
1770
1771 static int add_param(char *params, char *key, char *val)
1772 {
1773         char *start = params + strlen(params);
1774         char *end = params + sizeof(((struct mgs_target_info *)0)->mti_params);
1775         int keylen = 0;
1776
1777         if (key != NULL)
1778                 keylen = strlen(key);
1779         if (start + 1 + keylen + strlen(val) >= end) {
1780                 CERROR("params are too long: %s %s%s\n",
1781                        params, key != NULL ? key : "", val);
1782                 return -EINVAL;
1783         }
1784
1785         sprintf(start, " %s%s", key != NULL ? key : "", val);
1786         return 0;
1787 }
1788
1789 /**
1790  * Walk through client config log record and convert the related records
1791  * into the target.
1792  **/
1793 static int mgs_steal_client_llog_handler(const struct lu_env *env,
1794                                          struct llog_handle *llh,
1795                                          struct llog_rec_hdr *rec, void *data)
1796 {
1797         struct mgs_device *mgs;
1798         struct obd_device *obd;
1799         struct mgs_target_info *mti, *tmti;
1800         struct fs_db *fsdb;
1801         int cfg_len = rec->lrh_len;
1802         char *cfg_buf = (char*) (rec + 1);
1803         struct lustre_cfg *lcfg;
1804         int rc = 0;
1805         struct llog_handle *mdt_llh = NULL;
1806         static int got_an_osc_or_mdc = 0;
1807         /* 0: not found any osc/mdc;
1808            1: found osc;
1809            2: found mdc;
1810         */
1811         static int last_step = -1;
1812         int cplen = 0;
1813
1814         ENTRY;
1815
1816         mti = ((struct temp_comp*)data)->comp_mti;
1817         tmti = ((struct temp_comp*)data)->comp_tmti;
1818         fsdb = ((struct temp_comp*)data)->comp_fsdb;
1819         obd = ((struct temp_comp *)data)->comp_obd;
1820         mgs = lu2mgs_dev(obd->obd_lu_dev);
1821         LASSERT(mgs);
1822
1823         if (rec->lrh_type != OBD_CFG_REC) {
1824                 CERROR("unhandled lrh_type: %#x\n", rec->lrh_type);
1825                 RETURN(-EINVAL);
1826         }
1827
1828         rc = lustre_cfg_sanity_check(cfg_buf, cfg_len);
1829         if (rc) {
1830                 CERROR("Insane cfg\n");
1831                 RETURN(rc);
1832         }
1833
1834         lcfg = (struct lustre_cfg *)cfg_buf;
1835
1836         if (lcfg->lcfg_command == LCFG_MARKER) {
1837                 struct cfg_marker *marker;
1838                 marker = lustre_cfg_buf(lcfg, 1);
1839                 if (!strncmp(marker->cm_comment, "add osc", 7) &&
1840                     (marker->cm_flags & CM_START) &&
1841                      !(marker->cm_flags & CM_SKIP)) {
1842                         got_an_osc_or_mdc = 1;
1843                         cplen = strlcpy(tmti->mti_svname, marker->cm_tgtname,
1844                                         sizeof(tmti->mti_svname));
1845                         if (cplen >= sizeof(tmti->mti_svname))
1846                                 RETURN(-E2BIG);
1847                         rc = record_start_log(env, mgs, &mdt_llh,
1848                                               mti->mti_svname);
1849                         if (rc)
1850                                 RETURN(rc);
1851                         rc = record_marker(env, mdt_llh, fsdb, CM_START,
1852                                            mti->mti_svname, "add osc(copied)");
1853                         record_end_log(env, &mdt_llh);
1854                         last_step = marker->cm_step;
1855                         RETURN(rc);
1856                 }
1857                 if (!strncmp(marker->cm_comment, "add osc", 7) &&
1858                     (marker->cm_flags & CM_END) &&
1859                      !(marker->cm_flags & CM_SKIP)) {
1860                         LASSERT(last_step == marker->cm_step);
1861                         last_step = -1;
1862                         got_an_osc_or_mdc = 0;
1863                         memset(tmti, 0, sizeof(*tmti));
1864                         rc = record_start_log(env, mgs, &mdt_llh,
1865                                               mti->mti_svname);
1866                         if (rc)
1867                                 RETURN(rc);
1868                         rc = record_marker(env, mdt_llh, fsdb, CM_END,
1869                                            mti->mti_svname, "add osc(copied)");
1870                         record_end_log(env, &mdt_llh);
1871                         RETURN(rc);
1872                 }
1873                 if (!strncmp(marker->cm_comment, "add mdc", 7) &&
1874                     (marker->cm_flags & CM_START) &&
1875                      !(marker->cm_flags & CM_SKIP)) {
1876                         got_an_osc_or_mdc = 2;
1877                         last_step = marker->cm_step;
1878                         memcpy(tmti->mti_svname, marker->cm_tgtname,
1879                                strlen(marker->cm_tgtname));
1880
1881                         RETURN(rc);
1882                 }
1883                 if (!strncmp(marker->cm_comment, "add mdc", 7) &&
1884                     (marker->cm_flags & CM_END) &&
1885                      !(marker->cm_flags & CM_SKIP)) {
1886                         LASSERT(last_step == marker->cm_step);
1887                         last_step = -1;
1888                         got_an_osc_or_mdc = 0;
1889                         memset(tmti, 0, sizeof(*tmti));
1890                         RETURN(rc);
1891                 }
1892         }
1893
1894         if (got_an_osc_or_mdc == 0 || last_step < 0)
1895                 RETURN(rc);
1896
1897         if (lcfg->lcfg_command == LCFG_ADD_UUID) {
1898                 __u64 nodenid = lcfg->lcfg_nid;
1899
1900                 if (strlen(tmti->mti_uuid) == 0) {
1901                         /* target uuid not set, this config record is before
1902                          * LCFG_SETUP, this nid is one of target node nid.
1903                          */
1904                         tmti->mti_nids[tmti->mti_nid_count] = nodenid;
1905                         tmti->mti_nid_count++;
1906                 } else {
1907                         char nidstr[LNET_NIDSTR_SIZE];
1908
1909                         /* failover node nid */
1910                         libcfs_nid2str_r(nodenid, nidstr, sizeof(nidstr));
1911                         rc = add_param(tmti->mti_params, PARAM_FAILNODE,
1912                                         nidstr);
1913                 }
1914
1915                 RETURN(rc);
1916         }
1917
1918         if (lcfg->lcfg_command == LCFG_SETUP) {
1919                 char *target;
1920
1921                 target = lustre_cfg_string(lcfg, 1);
1922                 memcpy(tmti->mti_uuid, target, strlen(target));
1923                 RETURN(rc);
1924         }
1925
1926         /* ignore client side sptlrpc_conf_log */
1927         if (lcfg->lcfg_command == LCFG_SPTLRPC_CONF)
1928                 RETURN(rc);
1929
1930         if (lcfg->lcfg_command == LCFG_ADD_MDC &&
1931             strstr(lustre_cfg_string(lcfg, 0), "-clilmv") != NULL) {
1932                 int index;
1933
1934                 if (sscanf(lustre_cfg_buf(lcfg, 2), "%d", &index) != 1)
1935                         RETURN (-EINVAL);
1936
1937                 memcpy(tmti->mti_fsname, mti->mti_fsname,
1938                        strlen(mti->mti_fsname));
1939                 tmti->mti_stripe_index = index;
1940
1941                 rc = mgs_write_log_osp_to_mdt(env, mgs, fsdb, tmti,
1942                                               mti->mti_stripe_index,
1943                                               mti->mti_svname);
1944                 memset(tmti, 0, sizeof(*tmti));
1945                 RETURN(rc);
1946         }
1947
1948         if (lcfg->lcfg_command == LCFG_LOV_ADD_OBD) {
1949                 int index;
1950                 char mdt_index[9];
1951                 char *logname, *lovname;
1952
1953                 rc = name_create_mdt_and_lov(&logname, &lovname, fsdb,
1954                                              mti->mti_stripe_index);
1955                 if (rc)
1956                         RETURN(rc);
1957                 sprintf(mdt_index, "-MDT%04x", mti->mti_stripe_index);
1958
1959                 if (sscanf(lustre_cfg_buf(lcfg, 2), "%d", &index) != 1) {
1960                         name_destroy(&logname);
1961                         name_destroy(&lovname);
1962                         RETURN(-EINVAL);
1963                 }
1964
1965                 tmti->mti_stripe_index = index;
1966                 rc = mgs_write_log_osc_to_lov(env, mgs, fsdb, tmti, logname,
1967                                          mdt_index, lovname,
1968                                          LUSTRE_SP_MDT, 0);
1969                 name_destroy(&logname);
1970                 name_destroy(&lovname);
1971                 RETURN(rc);
1972         }
1973         RETURN(rc);
1974 }
1975
1976 /* fsdb->fsdb_mutex is already held  in mgs_write_log_target*/
1977 /* stealed from mgs_get_fsdb_from_llog*/
1978 static int mgs_steal_llog_for_mdt_from_client(const struct lu_env *env,
1979                                               struct mgs_device *mgs,
1980                                               char *client_name,
1981                                               struct temp_comp* comp)
1982 {
1983         struct llog_handle *loghandle;
1984         struct mgs_target_info *tmti;
1985         struct llog_ctxt *ctxt;
1986         int rc;
1987
1988         ENTRY;
1989
1990         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
1991         LASSERT(ctxt != NULL);
1992
1993         OBD_ALLOC_PTR(tmti);
1994         if (tmti == NULL)
1995                 GOTO(out_ctxt, rc = -ENOMEM);
1996
1997         comp->comp_tmti = tmti;
1998         comp->comp_obd = mgs->mgs_obd;
1999
2000         rc = llog_open(env, ctxt, &loghandle, NULL, client_name,
2001                        LLOG_OPEN_EXISTS);
2002         if (rc < 0) {
2003                 if (rc == -ENOENT)
2004                         rc = 0;
2005                 GOTO(out_pop, rc);
2006         }
2007
2008         rc = llog_init_handle(env, loghandle, LLOG_F_IS_PLAIN, NULL);
2009         if (rc)
2010                 GOTO(out_close, rc);
2011
2012         rc = llog_process_or_fork(env, loghandle, mgs_steal_client_llog_handler,
2013                                   (void *)comp, NULL, false);
2014         CDEBUG(D_MGS, "steal llog re = %d\n", rc);
2015 out_close:
2016         llog_close(env, loghandle);
2017 out_pop:
2018         OBD_FREE_PTR(tmti);
2019 out_ctxt:
2020         llog_ctxt_put(ctxt);
2021         RETURN(rc);
2022 }
2023
2024 /* lmv is the second thing for client logs */
2025 /* copied from mgs_write_log_lov. Please refer to that.  */
2026 static int mgs_write_log_lmv(const struct lu_env *env,
2027                              struct mgs_device *mgs,
2028                              struct fs_db *fsdb,
2029                              struct mgs_target_info *mti,
2030                              char *logname, char *lmvname)
2031 {
2032         struct llog_handle *llh = NULL;
2033         struct lmv_desc *lmvdesc;
2034         char *uuid;
2035         int rc = 0;
2036         ENTRY;
2037
2038         CDEBUG(D_MGS, "Writing lmv(%s) log for %s\n", lmvname,logname);
2039
2040         OBD_ALLOC_PTR(lmvdesc);
2041         if (lmvdesc == NULL)
2042                 RETURN(-ENOMEM);
2043         lmvdesc->ld_active_tgt_count = 0;
2044         lmvdesc->ld_tgt_count = 0;
2045         sprintf((char*)lmvdesc->ld_uuid.uuid, "%s_UUID", lmvname);
2046         uuid = (char *)lmvdesc->ld_uuid.uuid;
2047
2048         rc = record_start_log(env, mgs, &llh, logname);
2049         if (rc)
2050                 GOTO(out_free, rc);
2051         rc = record_marker(env, llh, fsdb, CM_START, lmvname, "lmv setup");
2052         if (rc)
2053                 GOTO(out_end, rc);
2054         rc = record_attach(env, llh, lmvname, "lmv", uuid);
2055         if (rc)
2056                 GOTO(out_end, rc);
2057         rc = record_lmv_setup(env, llh, lmvname, lmvdesc);
2058         if (rc)
2059                 GOTO(out_end, rc);
2060         rc = record_marker(env, llh, fsdb, CM_END, lmvname, "lmv setup");
2061         if (rc)
2062                 GOTO(out_end, rc);
2063 out_end:
2064         record_end_log(env, &llh);
2065 out_free:
2066         OBD_FREE_PTR(lmvdesc);
2067         RETURN(rc);
2068 }
2069
2070 /* lov is the first thing in the mdt and client logs */
2071 static int mgs_write_log_lov(const struct lu_env *env, struct mgs_device *mgs,
2072                              struct fs_db *fsdb, struct mgs_target_info *mti,
2073                              char *logname, char *lovname)
2074 {
2075         struct llog_handle *llh = NULL;
2076         struct lov_desc *lovdesc;
2077         char *uuid;
2078         int rc = 0;
2079         ENTRY;
2080
2081         CDEBUG(D_MGS, "Writing lov(%s) log for %s\n", lovname, logname);
2082
2083         /*
2084         #01 L attach   0:lov_mdsA  1:lov  2:71ccb_lov_mdsA_19f961a9e1
2085         #02 L lov_setup 0:lov_mdsA 1:(struct lov_desc)
2086               uuid=lov1_UUID, stripe count=1, size=1048576, offset=0, pattern=0
2087         */
2088
2089         /* FIXME just make lov_setup accept empty desc (put uuid in buf 2) */
2090         OBD_ALLOC_PTR(lovdesc);
2091         if (lovdesc == NULL)
2092                 RETURN(-ENOMEM);
2093         lovdesc->ld_magic = LOV_DESC_MAGIC;
2094         lovdesc->ld_tgt_count = 0;
2095         /* Defaults.  Can be changed later by lcfg config_param */
2096         lovdesc->ld_default_stripe_count = 1;
2097         lovdesc->ld_pattern = LOV_PATTERN_RAID0;
2098         lovdesc->ld_default_stripe_size = LOV_DESC_STRIPE_SIZE_DEFAULT;
2099         lovdesc->ld_default_stripe_offset = -1;
2100         lovdesc->ld_qos_maxage = LOV_DESC_QOS_MAXAGE_DEFAULT;
2101         sprintf((char*)lovdesc->ld_uuid.uuid, "%s_UUID", lovname);
2102         /* can these be the same? */
2103         uuid = (char *)lovdesc->ld_uuid.uuid;
2104
2105         /* This should always be the first entry in a log.
2106         rc = mgs_clear_log(obd, logname); */
2107         rc = record_start_log(env, mgs, &llh, logname);
2108         if (rc)
2109                 GOTO(out_free, rc);
2110         /* FIXME these should be a single journal transaction */
2111         rc = record_marker(env, llh, fsdb, CM_START, lovname, "lov setup");
2112         if (rc)
2113                 GOTO(out_end, rc);
2114         rc = record_attach(env, llh, lovname, "lov", uuid);
2115         if (rc)
2116                 GOTO(out_end, rc);
2117         rc = record_lov_setup(env, llh, lovname, lovdesc);
2118         if (rc)
2119                 GOTO(out_end, rc);
2120         rc = record_marker(env, llh, fsdb, CM_END, lovname, "lov setup");
2121         if (rc)
2122                 GOTO(out_end, rc);
2123         EXIT;
2124 out_end:
2125         record_end_log(env, &llh);
2126 out_free:
2127         OBD_FREE_PTR(lovdesc);
2128         return rc;
2129 }
2130
2131 /* add failnids to open log */
2132 static int mgs_write_log_failnids(const struct lu_env *env,
2133                                   struct mgs_target_info *mti,
2134                                   struct llog_handle *llh,
2135                                   char *cliname)
2136 {
2137         char *failnodeuuid = NULL;
2138         char *ptr = mti->mti_params;
2139         lnet_nid_t nid;
2140         int rc = 0;
2141
2142         /*
2143         #03 L add_uuid  nid=uml1@tcp(0x20000c0a80201) nal=90 0:  1:uml1_UUID
2144         #04 L add_uuid  nid=1@elan(0x1000000000001)   nal=90 0:  1:uml1_UUID
2145         #05 L setup    0:OSC_uml1_ost1_mdsA  1:ost1_UUID  2:uml1_UUID
2146         #06 L add_uuid  nid=uml2@tcp(0x20000c0a80202) nal=90 0:  1:uml2_UUID
2147         #0x L add_uuid  nid=2@elan(0x1000000000002)   nal=90 0:  1:uml2_UUID
2148         #07 L add_conn 0:OSC_uml1_ost1_mdsA  1:uml2_UUID
2149         */
2150
2151         /*
2152          * Pull failnid info out of params string, which may contain something
2153          * like "<nid1>,<nid2>:<nid3>,<nid4>".  class_parse_nid() does not
2154          * complain about abnormal inputs like ",:<nid1>", "<nid1>:,<nid2>",
2155          * etc.  However, convert_hostnames() should have caught those.
2156          */
2157         while (class_find_param(ptr, PARAM_FAILNODE, &ptr) == 0) {
2158                 while (class_parse_nid(ptr, &nid, &ptr) == 0) {
2159                         char nidstr[LNET_NIDSTR_SIZE];
2160
2161                         if (failnodeuuid == NULL) {
2162                                 /* We don't know the failover node name,
2163                                  * so just use the first nid as the uuid */
2164                                 libcfs_nid2str_r(nid, nidstr, sizeof(nidstr));
2165                                 rc = name_create(&failnodeuuid, nidstr, "");
2166                                 if (rc != 0)
2167                                         return rc;
2168                         }
2169                         CDEBUG(D_MGS, "add nid %s for failover uuid %s, "
2170                                 "client %s\n",
2171                                 libcfs_nid2str_r(nid, nidstr, sizeof(nidstr)),
2172                                 failnodeuuid, cliname);
2173                         rc = record_add_uuid(env, llh, nid, failnodeuuid);
2174                         /*
2175                          * If *ptr is ':', we have added all NIDs for
2176                          * failnodeuuid.
2177                          */
2178                         if (*ptr == ':') {
2179                                 rc = record_add_conn(env, llh, cliname,
2180                                                      failnodeuuid);
2181                                 name_destroy(&failnodeuuid);
2182                                 failnodeuuid = NULL;
2183                         }
2184                 }
2185                 if (failnodeuuid) {
2186                         rc = record_add_conn(env, llh, cliname, failnodeuuid);
2187                         name_destroy(&failnodeuuid);
2188                         failnodeuuid = NULL;
2189                 }
2190         }
2191
2192         return rc;
2193 }
2194
2195 static int mgs_write_log_mdc_to_lmv(const struct lu_env *env,
2196                                     struct mgs_device *mgs,
2197                                     struct fs_db *fsdb,
2198                                     struct mgs_target_info *mti,
2199                                     char *logname, char *lmvname)
2200 {
2201         struct llog_handle *llh = NULL;
2202         char *mdcname = NULL;
2203         char *nodeuuid = NULL;
2204         char *mdcuuid = NULL;
2205         char *lmvuuid = NULL;
2206         char index[6];
2207         char nidstr[LNET_NIDSTR_SIZE];
2208         int i, rc;
2209         ENTRY;
2210
2211         if (mgs_log_is_empty(env, mgs, logname)) {
2212                 CERROR("log is empty! Logical error\n");
2213                 RETURN(-EINVAL);
2214         }
2215
2216         CDEBUG(D_MGS, "adding mdc for %s to log %s:lmv(%s)\n",
2217                mti->mti_svname, logname, lmvname);
2218
2219         libcfs_nid2str_r(mti->mti_nids[0], nidstr, sizeof(nidstr));
2220         rc = name_create(&nodeuuid, nidstr, "");
2221         if (rc)
2222                 RETURN(rc);
2223         rc = name_create(&mdcname, mti->mti_svname, "-mdc");
2224         if (rc)
2225                 GOTO(out_free, rc);
2226         rc = name_create(&mdcuuid, mdcname, "_UUID");
2227         if (rc)
2228                 GOTO(out_free, rc);
2229         rc = name_create(&lmvuuid, lmvname, "_UUID");
2230         if (rc)
2231                 GOTO(out_free, rc);
2232
2233         rc = record_start_log(env, mgs, &llh, logname);
2234         if (rc)
2235                 GOTO(out_free, rc);
2236         rc = record_marker(env, llh, fsdb, CM_START, mti->mti_svname,
2237                            "add mdc");
2238         if (rc)
2239                 GOTO(out_end, rc);
2240         for (i = 0; i < mti->mti_nid_count; i++) {
2241                 CDEBUG(D_MGS, "add nid %s for mdt\n",
2242                         libcfs_nid2str_r(mti->mti_nids[i],
2243                                          nidstr, sizeof(nidstr)));
2244
2245                 rc = record_add_uuid(env, llh, mti->mti_nids[i], nodeuuid);
2246                 if (rc)
2247                         GOTO(out_end, rc);
2248         }
2249
2250         rc = record_attach(env, llh, mdcname, LUSTRE_MDC_NAME, lmvuuid);
2251         if (rc)
2252                 GOTO(out_end, rc);
2253         rc = record_setup(env, llh, mdcname, mti->mti_uuid, nodeuuid,
2254                           NULL, NULL);
2255         if (rc)
2256                 GOTO(out_end, rc);
2257         rc = mgs_write_log_failnids(env, mti, llh, mdcname);
2258         if (rc)
2259                 GOTO(out_end, rc);
2260         snprintf(index, sizeof(index), "%d", mti->mti_stripe_index);
2261         rc = record_mdc_add(env, llh, lmvname, mdcuuid, mti->mti_uuid,
2262                             index, "1");
2263         if (rc)
2264                 GOTO(out_end, rc);
2265         rc = record_marker(env, llh, fsdb, CM_END, mti->mti_svname,
2266                            "add mdc");
2267         if (rc)
2268                 GOTO(out_end, rc);
2269 out_end:
2270         record_end_log(env, &llh);
2271 out_free:
2272         name_destroy(&lmvuuid);
2273         name_destroy(&mdcuuid);
2274         name_destroy(&mdcname);
2275         name_destroy(&nodeuuid);
2276         RETURN(rc);
2277 }
2278
2279 static inline int name_create_lov(char **lovname, char *mdtname,
2280                                   struct fs_db *fsdb, int index)
2281 {
2282         /* COMPAT_180 */
2283         if (index == 0 && test_bit(FSDB_OSCNAME18, &fsdb->fsdb_flags))
2284                 return name_create(lovname, fsdb->fsdb_name, "-mdtlov");
2285         else
2286                 return name_create(lovname, mdtname, "-mdtlov");
2287 }
2288
2289 static int name_create_mdt_and_lov(char **logname, char **lovname,
2290                                    struct fs_db *fsdb, int i)
2291 {
2292         int rc;
2293
2294         rc = name_create_mdt(logname, fsdb->fsdb_name, i);
2295         if (rc)
2296                 return rc;
2297         /* COMPAT_180 */
2298         if (i == 0 && test_bit(FSDB_OSCNAME18, &fsdb->fsdb_flags))
2299                 rc = name_create(lovname, fsdb->fsdb_name, "-mdtlov");
2300         else
2301                 rc = name_create(lovname, *logname, "-mdtlov");
2302         if (rc) {
2303                 name_destroy(logname);
2304                 *logname = NULL;
2305         }
2306         return rc;
2307 }
2308
2309 static inline int name_create_mdt_osc(char **oscname, char *ostname,
2310                                       struct fs_db *fsdb, int i)
2311 {
2312         char suffix[16];
2313
2314         if (i == 0 && test_bit(FSDB_OSCNAME18, &fsdb->fsdb_flags))
2315                 sprintf(suffix, "-osc");
2316         else
2317                 sprintf(suffix, "-osc-MDT%04x", i);
2318         return name_create(oscname, ostname, suffix);
2319 }
2320
2321 /* add new mdc to already existent MDS */
2322 static int mgs_write_log_osp_to_mdt(const struct lu_env *env,
2323                                     struct mgs_device *mgs,
2324                                     struct fs_db *fsdb,
2325                                     struct mgs_target_info *mti,
2326                                     int mdt_index, char *logname)
2327 {
2328         struct llog_handle      *llh = NULL;
2329         char    *nodeuuid = NULL;
2330         char    *ospname = NULL;
2331         char    *lovuuid = NULL;
2332         char    *mdtuuid = NULL;
2333         char    *svname = NULL;
2334         char    *mdtname = NULL;
2335         char    *lovname = NULL;
2336         char    index_str[16];
2337         char    nidstr[LNET_NIDSTR_SIZE];
2338         int     i, rc;
2339
2340         ENTRY;
2341         if (mgs_log_is_empty(env, mgs, mti->mti_svname)) {
2342                 CERROR("log is empty! Logical error\n");
2343                 RETURN (-EINVAL);
2344         }
2345
2346         CDEBUG(D_MGS, "adding osp index %d to %s\n", mti->mti_stripe_index,
2347                logname);
2348
2349         rc = name_create_mdt(&mdtname, fsdb->fsdb_name, mti->mti_stripe_index);
2350         if (rc)
2351                 RETURN(rc);
2352
2353         libcfs_nid2str_r(mti->mti_nids[0], nidstr, sizeof(nidstr));
2354         rc = name_create(&nodeuuid, nidstr, "");
2355         if (rc)
2356                 GOTO(out_destory, rc);
2357
2358         rc = name_create(&svname, mdtname, "-osp");
2359         if (rc)
2360                 GOTO(out_destory, rc);
2361
2362         sprintf(index_str, "-MDT%04x", mdt_index);
2363         rc = name_create(&ospname, svname, index_str);
2364         if (rc)
2365                 GOTO(out_destory, rc);
2366
2367         rc = name_create_lov(&lovname, logname, fsdb, mdt_index);
2368         if (rc)
2369                 GOTO(out_destory, rc);
2370
2371         rc = name_create(&lovuuid, lovname, "_UUID");
2372         if (rc)
2373                 GOTO(out_destory, rc);
2374
2375         rc = name_create(&mdtuuid, mdtname, "_UUID");
2376         if (rc)
2377                 GOTO(out_destory, rc);
2378
2379         rc = record_start_log(env, mgs, &llh, logname);
2380         if (rc)
2381                 GOTO(out_destory, rc);
2382
2383         rc = record_marker(env, llh, fsdb, CM_START, mti->mti_svname,
2384                            "add osp");
2385         if (rc)
2386                 GOTO(out_destory, rc);
2387
2388         for (i = 0; i < mti->mti_nid_count; i++) {
2389                 CDEBUG(D_MGS, "add nid %s for mdt\n",
2390                         libcfs_nid2str_r(mti->mti_nids[i],
2391                                          nidstr, sizeof(nidstr)));
2392                 rc = record_add_uuid(env, llh, mti->mti_nids[i], nodeuuid);
2393                 if (rc)
2394                         GOTO(out_end, rc);
2395         }
2396
2397         rc = record_attach(env, llh, ospname, LUSTRE_OSP_NAME, lovuuid);
2398         if (rc)
2399                 GOTO(out_end, rc);
2400
2401         rc = record_setup(env, llh, ospname, mti->mti_uuid, nodeuuid,
2402                           NULL, NULL);
2403         if (rc)
2404                 GOTO(out_end, rc);
2405
2406         rc = mgs_write_log_failnids(env, mti, llh, ospname);
2407         if (rc)
2408                 GOTO(out_end, rc);
2409
2410         /* Add mdc(osp) to lod */
2411         snprintf(index_str, sizeof(index_str), "%d", mti->mti_stripe_index);
2412         rc = record_base(env, llh, lovname, 0, LCFG_ADD_MDC, mti->mti_uuid,
2413                          index_str, "1", NULL);
2414         if (rc)
2415                 GOTO(out_end, rc);
2416
2417         rc = record_marker(env, llh, fsdb, CM_END, mti->mti_svname, "add osp");
2418         if (rc)
2419                 GOTO(out_end, rc);
2420
2421 out_end:
2422         record_end_log(env, &llh);
2423
2424 out_destory:
2425         name_destroy(&mdtuuid);
2426         name_destroy(&lovuuid);
2427         name_destroy(&lovname);
2428         name_destroy(&ospname);
2429         name_destroy(&svname);
2430         name_destroy(&nodeuuid);
2431         name_destroy(&mdtname);
2432         RETURN(rc);
2433 }
2434
2435 static int mgs_write_log_mdt0(const struct lu_env *env,
2436                               struct mgs_device *mgs,
2437                               struct fs_db *fsdb,
2438                               struct mgs_target_info *mti)
2439 {
2440         char *log = mti->mti_svname;
2441         struct llog_handle *llh = NULL;
2442         char *uuid, *lovname;
2443         char mdt_index[6];
2444         char *ptr = mti->mti_params;
2445         int rc = 0, failout = 0;
2446         ENTRY;
2447
2448         OBD_ALLOC(uuid, sizeof(struct obd_uuid));
2449         if (uuid == NULL)
2450                 RETURN(-ENOMEM);
2451
2452         if (class_find_param(ptr, PARAM_FAILMODE, &ptr) == 0)
2453                 failout = (strncmp(ptr, "failout", 7) == 0);
2454
2455         rc = name_create(&lovname, log, "-mdtlov");
2456         if (rc)
2457                 GOTO(out_free, rc);
2458         if (mgs_log_is_empty(env, mgs, log)) {
2459                 rc = mgs_write_log_lov(env, mgs, fsdb, mti, log, lovname);
2460                 if (rc)
2461                         GOTO(out_lod, rc);
2462         }
2463
2464         sprintf(mdt_index, "%d", mti->mti_stripe_index);
2465
2466         rc = record_start_log(env, mgs, &llh, log);
2467         if (rc)
2468                 GOTO(out_lod, rc);
2469
2470         /* add MDT itself */
2471
2472         /* FIXME this whole fn should be a single journal transaction */
2473         sprintf(uuid, "%s_UUID", log);
2474         rc = record_marker(env, llh, fsdb, CM_START, log, "add mdt");
2475         if (rc)
2476                 GOTO(out_lod, rc);
2477         rc = record_attach(env, llh, log, LUSTRE_MDT_NAME, uuid);
2478         if (rc)
2479                 GOTO(out_end, rc);
2480         rc = record_mount_opt(env, llh, log, lovname, NULL);
2481         if (rc)
2482                 GOTO(out_end, rc);
2483         rc = record_setup(env, llh, log, uuid, mdt_index, lovname,
2484                         failout ? "n" : "f");
2485         if (rc)
2486                 GOTO(out_end, rc);
2487         rc = record_marker(env, llh, fsdb, CM_END, log, "add mdt");
2488         if (rc)
2489                 GOTO(out_end, rc);
2490 out_end:
2491         record_end_log(env, &llh);
2492 out_lod:
2493         name_destroy(&lovname);
2494 out_free:
2495         OBD_FREE(uuid, sizeof(struct obd_uuid));
2496         RETURN(rc);
2497 }
2498
2499 /* envelope method for all layers log */
2500 static int mgs_write_log_mdt(const struct lu_env *env,
2501                              struct mgs_device *mgs,
2502                              struct fs_db *fsdb,
2503                              struct mgs_target_info *mti)
2504 {
2505         struct mgs_thread_info *mgi = mgs_env_info(env);
2506         struct llog_handle *llh = NULL;
2507         char *cliname;
2508         int rc, i = 0;
2509         ENTRY;
2510
2511         CDEBUG(D_MGS, "writing new mdt %s\n", mti->mti_svname);
2512
2513         if (mti->mti_uuid[0] == '\0') {
2514                 /* Make up our own uuid */
2515                 snprintf(mti->mti_uuid, sizeof(mti->mti_uuid),
2516                          "%s_UUID", mti->mti_svname);
2517         }
2518
2519         /* add mdt */
2520         rc = mgs_write_log_mdt0(env, mgs, fsdb, mti);
2521         if (rc)
2522                 RETURN(rc);
2523         /* Append the mdt info to the client log */
2524         rc = name_create(&cliname, mti->mti_fsname, "-client");
2525         if (rc)
2526                 RETURN(rc);
2527
2528         if (mgs_log_is_empty(env, mgs, cliname)) {
2529                 /* Start client log */
2530                 rc = mgs_write_log_lov(env, mgs, fsdb, mti, cliname,
2531                                        fsdb->fsdb_clilov);
2532                 if (rc)
2533                         GOTO(out_free, rc);
2534                 rc = mgs_write_log_lmv(env, mgs, fsdb, mti, cliname,
2535                                        fsdb->fsdb_clilmv);
2536                 if (rc)
2537                         GOTO(out_free, rc);
2538         }
2539
2540         /*
2541         #09 L add_uuid nid=uml1@tcp(0x20000c0a80201) 0:  1:uml1_UUID
2542         #10 L attach   0:MDC_uml1_mdsA_MNT_client  1:mdc  2:1d834_MNT_client_03f
2543         #11 L setup    0:MDC_uml1_mdsA_MNT_client  1:mdsA_UUID  2:uml1_UUID
2544         #12 L add_uuid nid=uml2@tcp(0x20000c0a80202) 0:  1:uml2_UUID
2545         #13 L add_conn 0:MDC_uml1_mdsA_MNT_client  1:uml2_UUID
2546         #14 L mount_option 0:  1:client  2:lov1  3:MDC_uml1_mdsA_MNT_client
2547         */
2548
2549         /* copy client info about lov/lmv */
2550         mgi->mgi_comp.comp_mti = mti;
2551         mgi->mgi_comp.comp_fsdb = fsdb;
2552
2553         rc = mgs_steal_llog_for_mdt_from_client(env, mgs, cliname,
2554                                                 &mgi->mgi_comp);
2555         if (rc)
2556                 GOTO(out_free, rc);
2557         rc = mgs_write_log_mdc_to_lmv(env, mgs, fsdb, mti, cliname,
2558                                       fsdb->fsdb_clilmv);
2559         if (rc)
2560                 GOTO(out_free, rc);
2561
2562         /* add mountopts */
2563         rc = record_start_log(env, mgs, &llh, cliname);
2564         if (rc)
2565                 GOTO(out_free, rc);
2566
2567         rc = record_marker(env, llh, fsdb, CM_START, cliname, "mount opts");
2568         if (rc)
2569                 GOTO(out_end, rc);
2570         rc = record_mount_opt(env, llh, cliname, fsdb->fsdb_clilov,
2571                               fsdb->fsdb_clilmv);
2572         if (rc)
2573                 GOTO(out_end, rc);
2574         rc = record_marker(env, llh, fsdb, CM_END, cliname, "mount opts");
2575         if (rc)
2576                 GOTO(out_end, rc);
2577
2578         /* for_all_existing_mdt except current one */
2579         for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
2580                 if (i !=  mti->mti_stripe_index &&
2581                     test_bit(i, fsdb->fsdb_mdt_index_map)) {
2582                         char *logname;
2583
2584                         rc = name_create_mdt(&logname, fsdb->fsdb_name, i);
2585                         if (rc)
2586                                 GOTO(out_end, rc);
2587
2588                         /* NB: If the log for the MDT is empty, it means
2589                          * the MDT is only added to the index
2590                          * map, and not being process yet, i.e. this
2591                          * is an unregistered MDT, see mgs_write_log_target().
2592                          * so we should skip it. Otherwise
2593                          *
2594                          * 1. MGS get register request for MDT1 and MDT2.
2595                          *
2596                          * 2. Then both MDT1 and MDT2 are added into
2597                          * fsdb_mdt_index_map. (see mgs_set_index()).
2598                          *
2599                          * 3. Then MDT1 get the lock of fsdb_mutex, then
2600                          * generate the config log, here, it will regard MDT2
2601                          * as an existent MDT, and generate "add osp" for
2602                          * lustre-MDT0001-osp-MDT0002. Note: at the moment
2603                          * MDT0002 config log is still empty, so it will
2604                          * add "add osp" even before "lov setup", which
2605                          * will definitly cause trouble.
2606                          *
2607                          * 4. MDT1 registeration finished, fsdb_mutex is
2608                          * released, then MDT2 get in, then in above
2609                          * mgs_steal_llog_for_mdt_from_client(), it will
2610                          * add another osp log for lustre-MDT0001-osp-MDT0002,
2611                          * which will cause another trouble.*/
2612                         if (!mgs_log_is_empty(env, mgs, logname))
2613                                 rc = mgs_write_log_osp_to_mdt(env, mgs, fsdb,
2614                                                               mti, i, logname);
2615
2616                         name_destroy(&logname);
2617                         if (rc)
2618                                 GOTO(out_end, rc);
2619                 }
2620         }
2621 out_end:
2622         record_end_log(env, &llh);
2623 out_free:
2624         name_destroy(&cliname);
2625         RETURN(rc);
2626 }
2627
2628 /* Add the ost info to the client/mdt lov */
2629 static int mgs_write_log_osc_to_lov(const struct lu_env *env,
2630                                     struct mgs_device *mgs, struct fs_db *fsdb,
2631                                     struct mgs_target_info *mti,
2632                                     char *logname, char *suffix, char *lovname,
2633                                     enum lustre_sec_part sec_part, int flags)
2634 {
2635         struct llog_handle *llh = NULL;
2636         char *nodeuuid = NULL;
2637         char *oscname = NULL;
2638         char *oscuuid = NULL;
2639         char *lovuuid = NULL;
2640         char *svname = NULL;
2641         char index[6];
2642         char nidstr[LNET_NIDSTR_SIZE];
2643         int i, rc;
2644         ENTRY;
2645
2646         CDEBUG(D_INFO, "adding osc for %s to log %s\n",
2647                 mti->mti_svname, logname);
2648
2649         if (mgs_log_is_empty(env, mgs, logname)) {
2650                 CERROR("log is empty! Logical error\n");
2651                 RETURN(-EINVAL);
2652         }
2653
2654         libcfs_nid2str_r(mti->mti_nids[0], nidstr, sizeof(nidstr));
2655         rc = name_create(&nodeuuid, nidstr, "");
2656         if (rc)
2657                 RETURN(rc);
2658         rc = name_create(&svname, mti->mti_svname, "-osc");
2659         if (rc)
2660                 GOTO(out_free, rc);
2661
2662         /* for the system upgraded from old 1.8, keep using the old osc naming
2663          * style for mdt, see name_create_mdt_osc(). LU-1257 */
2664         if (test_bit(FSDB_OSCNAME18, &fsdb->fsdb_flags))
2665                 rc = name_create(&oscname, svname, "");
2666         else
2667                 rc = name_create(&oscname, svname, suffix);
2668         if (rc)
2669                 GOTO(out_free, rc);
2670
2671         rc = name_create(&oscuuid, oscname, "_UUID");
2672         if (rc)
2673                 GOTO(out_free, rc);
2674         rc = name_create(&lovuuid, lovname, "_UUID");
2675         if (rc)
2676                 GOTO(out_free, rc);
2677
2678
2679         /*
2680         #03 L add_uuid nid=uml1@tcp(0x20000c0a80201) 0:  1:uml1_UUID
2681         multihomed (#4)
2682         #04 L add_uuid  nid=1@elan(0x1000000000001)  nal=90 0:  1:uml1_UUID
2683         #04 L attach   0:OSC_uml1_ost1_MNT_client  1:osc  2:89070_lov1_a41dff51a
2684         #05 L setup    0:OSC_uml1_ost1_MNT_client  1:ost1_UUID  2:uml1_UUID
2685         failover (#6,7)
2686         #06 L add_uuid nid=uml2@tcp(0x20000c0a80202) 0:  1:uml2_UUID
2687         #07 L add_conn 0:OSC_uml1_ost1_MNT_client  1:uml2_UUID
2688         #08 L lov_modify_tgts add 0:lov1  1:ost1_UUID  2(index):0  3(gen):1
2689         */
2690
2691         rc = record_start_log(env, mgs, &llh, logname);
2692         if (rc)
2693                 GOTO(out_free, rc);
2694
2695         /* FIXME these should be a single journal transaction */
2696         rc = record_marker(env, llh, fsdb, CM_START | flags, mti->mti_svname,
2697                            "add osc");
2698         if (rc)
2699                 GOTO(out_end, rc);
2700
2701         /* NB: don't change record order, because upon MDT steal OSC config
2702          * from client, it treats all nids before LCFG_SETUP as target nids
2703          * (multiple interfaces), while nids after as failover node nids.
2704          * See mgs_steal_client_llog_handler() LCFG_ADD_UUID.
2705          */
2706         for (i = 0; i < mti->mti_nid_count; i++) {
2707                 CDEBUG(D_MGS, "add nid %s\n",
2708                         libcfs_nid2str_r(mti->mti_nids[i],
2709                                          nidstr, sizeof(nidstr)));
2710                 rc = record_add_uuid(env, llh, mti->mti_nids[i], nodeuuid);
2711                 if (rc)
2712                         GOTO(out_end, rc);
2713         }
2714         rc = record_attach(env, llh, oscname, LUSTRE_OSC_NAME, lovuuid);
2715         if (rc)
2716                 GOTO(out_end, rc);
2717         rc = record_setup(env, llh, oscname, mti->mti_uuid, nodeuuid,
2718                           NULL, NULL);
2719         if (rc)
2720                 GOTO(out_end, rc);
2721         rc = mgs_write_log_failnids(env, mti, llh, oscname);
2722         if (rc)
2723                 GOTO(out_end, rc);
2724
2725         snprintf(index, sizeof(index), "%d", mti->mti_stripe_index);
2726
2727         rc = record_lov_add(env, llh, lovname, mti->mti_uuid, index, "1");
2728         if (rc)
2729                 GOTO(out_end, rc);
2730         rc = record_marker(env, llh, fsdb, CM_END | flags, mti->mti_svname,
2731                            "add osc");
2732         if (rc)
2733                 GOTO(out_end, rc);
2734 out_end:
2735         record_end_log(env, &llh);
2736 out_free:
2737         name_destroy(&lovuuid);
2738         name_destroy(&oscuuid);
2739         name_destroy(&oscname);
2740         name_destroy(&svname);
2741         name_destroy(&nodeuuid);
2742         RETURN(rc);
2743 }
2744
2745 static int mgs_write_log_ost(const struct lu_env *env,
2746                              struct mgs_device *mgs, struct fs_db *fsdb,
2747                              struct mgs_target_info *mti)
2748 {
2749         struct llog_handle *llh = NULL;
2750         char *logname, *lovname;
2751         char *ptr = mti->mti_params;
2752         int rc, flags = 0, failout = 0, i;
2753         ENTRY;
2754
2755         CDEBUG(D_MGS, "writing new ost %s\n", mti->mti_svname);
2756
2757         /* The ost startup log */
2758
2759         /* If the ost log already exists, that means that someone reformatted
2760            the ost and it called target_add again. */
2761         if (!mgs_log_is_empty(env, mgs, mti->mti_svname)) {
2762                 LCONSOLE_ERROR_MSG(0x141, "The config log for %s already "
2763                                    "exists, yet the server claims it never "
2764                                    "registered. It may have been reformatted, "
2765                                    "or the index changed. writeconf the MDT to "
2766                                    "regenerate all logs.\n", mti->mti_svname);
2767                 RETURN(-EALREADY);
2768         }
2769
2770         /*
2771         attach obdfilter ost1 ost1_UUID
2772         setup /dev/loop2 ldiskfs f|n errors=remount-ro,user_xattr
2773         */
2774         if (class_find_param(ptr, PARAM_FAILMODE, &ptr) == 0)
2775                 failout = (strncmp(ptr, "failout", 7) == 0);
2776         rc = record_start_log(env, mgs, &llh, mti->mti_svname);
2777         if (rc)
2778                 RETURN(rc);
2779         /* FIXME these should be a single journal transaction */
2780         rc = record_marker(env, llh, fsdb, CM_START, mti->mti_svname,"add ost");
2781         if (rc)
2782                 GOTO(out_end, rc);
2783         if (*mti->mti_uuid == '\0')
2784                 snprintf(mti->mti_uuid, sizeof(mti->mti_uuid),
2785                          "%s_UUID", mti->mti_svname);
2786         rc = record_attach(env, llh, mti->mti_svname,
2787                            "obdfilter"/*LUSTRE_OST_NAME*/, mti->mti_uuid);
2788         if (rc)
2789                 GOTO(out_end, rc);
2790         rc = record_setup(env, llh, mti->mti_svname,
2791                           "dev"/*ignored*/, "type"/*ignored*/,
2792                           failout ? "n" : "f", NULL/*options*/);
2793         if (rc)
2794                 GOTO(out_end, rc);
2795         rc = record_marker(env, llh, fsdb, CM_END, mti->mti_svname, "add ost");
2796         if (rc)
2797                 GOTO(out_end, rc);
2798 out_end:
2799         record_end_log(env, &llh);
2800         if (rc)
2801                 RETURN(rc);
2802         /* We also have to update the other logs where this osc is part of
2803            the lov */
2804
2805         if (test_bit(FSDB_OLDLOG14, &fsdb->fsdb_flags)) {
2806                 /* If we're upgrading, the old mdt log already has our
2807                    entry. Let's do a fake one for fun. */
2808                 /* Note that we can't add any new failnids, since we don't
2809                    know the old osc names. */
2810                 flags = CM_SKIP | CM_UPGRADE146;
2811
2812         } else if ((mti->mti_flags & LDD_F_UPDATE) != LDD_F_UPDATE) {
2813                 /* If the update flag isn't set, don't update client/mdt
2814                    logs. */
2815                 flags |= CM_SKIP;
2816                 LCONSOLE_WARN("Client log for %s was not updated; writeconf "
2817                               "the MDT first to regenerate it.\n",
2818                               mti->mti_svname);
2819         }
2820
2821         /* Add ost to all MDT lov defs */
2822         for (i = 0; i < INDEX_MAP_SIZE * 8; i++){
2823                 if (test_bit(i, fsdb->fsdb_mdt_index_map)) {
2824                         char mdt_index[9];
2825
2826                         rc = name_create_mdt_and_lov(&logname, &lovname, fsdb,
2827                                                      i);
2828                         if (rc)
2829                                 RETURN(rc);
2830                         sprintf(mdt_index, "-MDT%04x", i);
2831                         rc = mgs_write_log_osc_to_lov(env, mgs, fsdb, mti,
2832                                                       logname, mdt_index,
2833                                                       lovname, LUSTRE_SP_MDT,
2834                                                       flags);
2835                         name_destroy(&logname);
2836                         name_destroy(&lovname);
2837                         if (rc)
2838                                 RETURN(rc);
2839                 }
2840         }
2841
2842         /* Append ost info to the client log */
2843         rc = name_create(&logname, mti->mti_fsname, "-client");
2844         if (rc)
2845                 RETURN(rc);
2846         if (mgs_log_is_empty(env, mgs, logname)) {
2847                 /* Start client log */
2848                 rc = mgs_write_log_lov(env, mgs, fsdb, mti, logname,
2849                                        fsdb->fsdb_clilov);
2850                 if (rc)
2851                         GOTO(out_free, rc);
2852                 rc = mgs_write_log_lmv(env, mgs, fsdb, mti, logname,
2853                                        fsdb->fsdb_clilmv);
2854                 if (rc)
2855                         GOTO(out_free, rc);
2856         }
2857         rc = mgs_write_log_osc_to_lov(env, mgs, fsdb, mti, logname, "",
2858                                       fsdb->fsdb_clilov, LUSTRE_SP_CLI, flags);
2859 out_free:
2860         name_destroy(&logname);
2861         RETURN(rc);
2862 }
2863
2864 static __inline__ int mgs_param_empty(char *ptr)
2865 {
2866         char *tmp;
2867
2868         if ((tmp = strchr(ptr, '=')) && (*(++tmp) == '\0'))
2869                 return 1;
2870         return 0;
2871 }
2872
2873 static int mgs_write_log_failnid_internal(const struct lu_env *env,
2874                                           struct mgs_device *mgs,
2875                                           struct fs_db *fsdb,
2876                                           struct mgs_target_info *mti,
2877                                           char *logname, char *cliname)
2878 {
2879         int rc;
2880         struct llog_handle *llh = NULL;
2881
2882         if (mgs_param_empty(mti->mti_params)) {
2883                 /* Remove _all_ failnids */
2884                 rc = mgs_modify(env, mgs, fsdb, mti, logname,
2885                                 mti->mti_svname, "add failnid", CM_SKIP);
2886                 return rc < 0 ? rc : 0;
2887         }
2888
2889         /* Otherwise failover nids are additive */
2890         rc = record_start_log(env, mgs, &llh, logname);
2891         if (rc)
2892                 return rc;
2893                 /* FIXME this should be a single journal transaction */
2894         rc = record_marker(env, llh, fsdb, CM_START, mti->mti_svname,
2895                            "add failnid");
2896         if (rc)
2897                 goto out_end;
2898         rc = mgs_write_log_failnids(env, mti, llh, cliname);
2899         if (rc)
2900                 goto out_end;
2901         rc = record_marker(env, llh, fsdb, CM_END,
2902                            mti->mti_svname, "add failnid");
2903 out_end:
2904         record_end_log(env, &llh);
2905         return rc;
2906 }
2907
2908
2909 /* Add additional failnids to an existing log.
2910    The mdc/osc must have been added to logs first */
2911 /* tcp nids must be in dotted-quad ascii -
2912    we can't resolve hostnames from the kernel. */
2913 static int mgs_write_log_add_failnid(const struct lu_env *env,
2914                                      struct mgs_device *mgs,
2915                                      struct fs_db *fsdb,
2916                                      struct mgs_target_info *mti)
2917 {
2918         char *logname, *cliname;
2919         int rc;
2920         ENTRY;
2921
2922         /* FIXME we currently can't erase the failnids
2923          * given when a target first registers, since they aren't part of
2924          * an "add uuid" stanza
2925          */
2926
2927         /* Verify that we know about this target */
2928         if (mgs_log_is_empty(env, mgs, mti->mti_svname)) {
2929                 LCONSOLE_ERROR_MSG(0x142, "The target %s has not registered "
2930                                    "yet. It must be started before failnids "
2931                                    "can be added.\n", mti->mti_svname);
2932                 RETURN(-ENOENT);
2933         }
2934
2935         /* Create mdc/osc client name (e.g. lustre-OST0001-osc) */
2936         if (mti->mti_flags & LDD_F_SV_TYPE_MDT) {
2937                 rc = name_create(&cliname, mti->mti_svname, "-mdc");
2938         } else if (mti->mti_flags & LDD_F_SV_TYPE_OST) {
2939                 rc = name_create(&cliname, mti->mti_svname, "-osc");
2940         } else {
2941                 RETURN(-EINVAL);
2942         }
2943         if (rc)
2944                 RETURN(rc);
2945
2946         /* Add failover nids to the client log */
2947         rc = name_create(&logname, mti->mti_fsname, "-client");
2948         if (rc) {
2949                 name_destroy(&cliname);
2950                 RETURN(rc);
2951         }
2952
2953         rc = mgs_write_log_failnid_internal(env, mgs, fsdb,mti,logname,cliname);
2954         name_destroy(&logname);
2955         name_destroy(&cliname);
2956         if (rc)
2957                 RETURN(rc);
2958
2959         if (mti->mti_flags & LDD_F_SV_TYPE_OST) {
2960                 /* Add OST failover nids to the MDT logs as well */
2961                 int i;
2962
2963                 for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
2964                         if (!test_bit(i, fsdb->fsdb_mdt_index_map))
2965                                 continue;
2966                         rc = name_create_mdt(&logname, mti->mti_fsname, i);
2967                         if (rc)
2968                                 RETURN(rc);
2969                         rc = name_create_mdt_osc(&cliname, mti->mti_svname,
2970                                                  fsdb, i);
2971                         if (rc) {
2972                                 name_destroy(&logname);
2973                                 RETURN(rc);
2974                         }
2975                         rc = mgs_write_log_failnid_internal(env, mgs, fsdb,
2976                                                             mti, logname,
2977                                                             cliname);
2978                         name_destroy(&cliname);
2979                         name_destroy(&logname);
2980                         if (rc)
2981                                 RETURN(rc);
2982                 }
2983         }
2984
2985         RETURN(rc);
2986 }
2987
2988 static int mgs_wlp_lcfg(const struct lu_env *env,
2989                         struct mgs_device *mgs, struct fs_db *fsdb,
2990                         struct mgs_target_info *mti,
2991                         char *logname, struct lustre_cfg_bufs *bufs,
2992                         char *tgtname, char *ptr)
2993 {
2994         char comment[MTI_NAME_MAXLEN];
2995         char *tmp;
2996         struct llog_cfg_rec *lcr;
2997         int rc, del;
2998
2999         /* Erase any old settings of this same parameter */
3000         memcpy(comment, ptr, MTI_NAME_MAXLEN);
3001         comment[MTI_NAME_MAXLEN - 1] = 0;
3002         /* But don't try to match the value. */
3003         tmp = strchr(comment, '=');
3004         if (tmp != NULL)
3005                 *tmp = 0;
3006         /* FIXME we should skip settings that are the same as old values */
3007         rc = mgs_modify(env, mgs, fsdb, mti, logname, tgtname, comment,CM_SKIP);
3008         if (rc < 0)
3009                 return rc;
3010         del = mgs_param_empty(ptr);
3011
3012         LCONSOLE_INFO("%s parameter %s.%s in log %s\n", del ? "Disabling" : rc ?
3013                       "Setting" : "Modifying", tgtname, comment, logname);
3014         if (del) {
3015                 /* mgs_modify() will return 1 if nothing had to be done */
3016                 if (rc == 1)
3017                         rc = 0;
3018                 return rc;
3019         }
3020
3021         lustre_cfg_bufs_reset(bufs, tgtname);
3022         lustre_cfg_bufs_set_string(bufs, 1, ptr);
3023         if (mti->mti_flags & LDD_F_PARAM2)
3024                 lustre_cfg_bufs_set_string(bufs, 2, LCTL_UPCALL);
3025
3026         lcr = lustre_cfg_rec_new((mti->mti_flags & LDD_F_PARAM2) ?
3027                                  LCFG_SET_PARAM : LCFG_PARAM, bufs);
3028         if (lcr == NULL)
3029                 return -ENOMEM;
3030
3031         rc = mgs_write_log_direct(env, mgs, fsdb, logname, lcr, tgtname,
3032                                   comment);
3033         lustre_cfg_rec_free(lcr);
3034         return rc;
3035 }
3036
3037 /* write global variable settings into log */
3038 static int mgs_write_log_sys(const struct lu_env *env,
3039                              struct mgs_device *mgs, struct fs_db *fsdb,
3040                              struct mgs_target_info *mti, char *sys, char *ptr)
3041 {
3042         struct mgs_thread_info  *mgi = mgs_env_info(env);
3043         struct lustre_cfg       *lcfg;
3044         struct llog_cfg_rec     *lcr;
3045         char *tmp, sep;
3046         int rc, cmd, convert = 1;
3047
3048         if (class_match_param(ptr, PARAM_TIMEOUT, &tmp) == 0) {
3049                 cmd = LCFG_SET_TIMEOUT;
3050         } else if (class_match_param(ptr, PARAM_LDLM_TIMEOUT, &tmp) == 0) {
3051                 cmd = LCFG_SET_LDLM_TIMEOUT;
3052         /* Check for known params here so we can return error to lctl */
3053         } else if ((class_match_param(ptr, PARAM_AT_MIN, &tmp) == 0) ||
3054                 (class_match_param(ptr, PARAM_AT_MAX, &tmp) == 0) ||
3055                 (class_match_param(ptr, PARAM_AT_EXTRA, &tmp) == 0) ||
3056                 (class_match_param(ptr, PARAM_AT_EARLY_MARGIN, &tmp) == 0) ||
3057                 (class_match_param(ptr, PARAM_AT_HISTORY, &tmp) == 0)) {
3058                 cmd = LCFG_PARAM;
3059         } else if (class_match_param(ptr, PARAM_JOBID_VAR, &tmp) == 0) {
3060                 convert = 0; /* Don't convert string value to integer */
3061                 cmd = LCFG_PARAM;
3062         } else {
3063                 return -EINVAL;
3064         }
3065
3066         if (mgs_param_empty(ptr))
3067                 CDEBUG(D_MGS, "global '%s' removed\n", sys);
3068         else
3069                 CDEBUG(D_MGS, "global '%s' val=%s\n", sys, tmp);
3070
3071         lustre_cfg_bufs_reset(&mgi->mgi_bufs, NULL);
3072         lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 1, sys);
3073         if (!convert && *tmp != '\0')
3074                 lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 2, tmp);
3075         lcr = lustre_cfg_rec_new(cmd, &mgi->mgi_bufs);
3076         if (lcr == NULL)
3077                 return -ENOMEM;
3078
3079         lcfg = &lcr->lcr_cfg;
3080         lcfg->lcfg_num = convert ? simple_strtoul(tmp, NULL, 0) : 0;
3081         /* truncate the comment to the parameter name */
3082         ptr = tmp - 1;
3083         sep = *ptr;
3084         *ptr = '\0';
3085         /* modify all servers and clients */
3086         rc = mgs_write_log_direct_all(env, mgs, fsdb, mti,
3087                                       *tmp == '\0' ? NULL : lcr,
3088                                       mti->mti_fsname, sys, 0);
3089         if (rc == 0 && *tmp != '\0') {
3090                 switch (cmd) {
3091                 case LCFG_SET_TIMEOUT:
3092                         if (!obd_timeout_set || lcfg->lcfg_num > obd_timeout)
3093                                 class_process_config(lcfg);
3094                         break;
3095                 case LCFG_SET_LDLM_TIMEOUT:
3096                         if (!ldlm_timeout_set || lcfg->lcfg_num > ldlm_timeout)
3097                                 class_process_config(lcfg);
3098                         break;
3099                 default:
3100                         break;
3101                 }
3102         }
3103         *ptr = sep;
3104         lustre_cfg_rec_free(lcr);
3105         return rc;
3106 }
3107
3108 /* write quota settings into log */
3109 static int mgs_write_log_quota(const struct lu_env *env, struct mgs_device *mgs,
3110                                struct fs_db *fsdb, struct mgs_target_info *mti,
3111                                char *quota, char *ptr)
3112 {
3113         struct mgs_thread_info  *mgi = mgs_env_info(env);
3114         struct llog_cfg_rec     *lcr;
3115         char                    *tmp;
3116         char                     sep;
3117         int                      rc, cmd = LCFG_PARAM;
3118
3119         /* support only 'meta' and 'data' pools so far */
3120         if (class_match_param(ptr, QUOTA_METAPOOL_NAME, &tmp) != 0 &&
3121             class_match_param(ptr, QUOTA_DATAPOOL_NAME, &tmp) != 0) {
3122                 CERROR("parameter quota.%s isn't supported (only quota.mdt "
3123                        "& quota.ost are)\n", ptr);
3124                 return -EINVAL;
3125         }
3126
3127         if (*tmp == '\0') {
3128                 CDEBUG(D_MGS, "global '%s' removed\n", quota);
3129         } else {
3130                 CDEBUG(D_MGS, "global '%s'\n", quota);
3131
3132                 if (strchr(tmp, 'u') == NULL && strchr(tmp, 'g') == NULL &&
3133                     strchr(tmp, 'p') == NULL &&
3134                     strcmp(tmp, "none") != 0) {
3135                         CERROR("enable option(%s) isn't supported\n", tmp);
3136                         return -EINVAL;
3137                 }
3138         }
3139
3140         lustre_cfg_bufs_reset(&mgi->mgi_bufs, mti->mti_fsname);
3141         lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 1, quota);
3142         lcr = lustre_cfg_rec_new(cmd, &mgi->mgi_bufs);
3143         if (lcr == NULL)
3144                 return -ENOMEM;
3145
3146         /* truncate the comment to the parameter name */
3147         ptr = tmp - 1;
3148         sep = *ptr;
3149         *ptr = '\0';
3150
3151         /* XXX we duplicated quota enable information in all server
3152          *     config logs, it should be moved to a separate config
3153          *     log once we cleanup the config log for global param. */
3154         /* modify all servers */
3155         rc = mgs_write_log_direct_all(env, mgs, fsdb, mti,
3156                                       *tmp == '\0' ? NULL : lcr,
3157                                       mti->mti_fsname, quota, 1);
3158         *ptr = sep;
3159         lustre_cfg_rec_free(lcr);
3160         return rc < 0 ? rc : 0;
3161 }
3162
3163 static int mgs_srpc_set_param_disk(const struct lu_env *env,
3164                                    struct mgs_device *mgs,
3165                                    struct fs_db *fsdb,
3166                                    struct mgs_target_info *mti,
3167                                    char *param)
3168 {
3169         struct mgs_thread_info  *mgi = mgs_env_info(env);
3170         struct llog_cfg_rec     *lcr;
3171         struct llog_handle      *llh = NULL;
3172         char                    *logname;
3173         char                    *comment, *ptr;
3174         int                      rc, len;
3175
3176         ENTRY;
3177
3178         /* get comment */
3179         ptr = strchr(param, '=');
3180         LASSERT(ptr != NULL);
3181         len = ptr - param;
3182
3183         OBD_ALLOC(comment, len + 1);
3184         if (comment == NULL)
3185                 RETURN(-ENOMEM);
3186         strncpy(comment, param, len);
3187         comment[len] = '\0';
3188
3189         /* prepare lcfg */
3190         lustre_cfg_bufs_reset(&mgi->mgi_bufs, mti->mti_svname);
3191         lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 1, param);
3192         lcr = lustre_cfg_rec_new(LCFG_SPTLRPC_CONF, &mgi->mgi_bufs);
3193         if (lcr == NULL)
3194                 GOTO(out_comment, rc = -ENOMEM);
3195
3196         /* construct log name */
3197         rc = name_create(&logname, mti->mti_fsname, "-sptlrpc");
3198         if (rc < 0)
3199                 GOTO(out_lcfg, rc);
3200
3201         if (mgs_log_is_empty(env, mgs, logname)) {
3202                 rc = record_start_log(env, mgs, &llh, logname);
3203                 if (rc < 0)
3204                         GOTO(out, rc);
3205                 record_end_log(env, &llh);
3206         }
3207
3208         /* obsolete old one */
3209         rc = mgs_modify(env, mgs, fsdb, mti, logname, mti->mti_svname,
3210                         comment, CM_SKIP);
3211         if (rc < 0)
3212                 GOTO(out, rc);
3213         /* write the new one */
3214         rc = mgs_write_log_direct(env, mgs, fsdb, logname, lcr,
3215                                   mti->mti_svname, comment);
3216         if (rc)
3217                 CERROR("%s: error writing log %s: rc = %d\n",
3218                        mgs->mgs_obd->obd_name, logname, rc);
3219 out:
3220         name_destroy(&logname);
3221 out_lcfg:
3222         lustre_cfg_rec_free(lcr);
3223 out_comment:
3224         OBD_FREE(comment, len + 1);
3225         RETURN(rc);
3226 }
3227
3228 static int mgs_srpc_set_param_udesc_mem(struct fs_db *fsdb,
3229                                         char *param)
3230 {
3231         char    *ptr;
3232
3233         /* disable the adjustable udesc parameter for now, i.e. use default
3234          * setting that client always ship udesc to MDT if possible. to enable
3235          * it simply remove the following line */
3236         goto error_out;
3237
3238         ptr = strchr(param, '=');
3239         if (ptr == NULL)
3240                 goto error_out;
3241         *ptr++ = '\0';
3242
3243         if (strcmp(param, PARAM_SRPC_UDESC))
3244                 goto error_out;
3245
3246         if (strcmp(ptr, "yes") == 0) {
3247                 set_bit(FSDB_UDESC, &fsdb->fsdb_flags);
3248                 CWARN("Enable user descriptor shipping from client to MDT\n");
3249         } else if (strcmp(ptr, "no") == 0) {
3250                 clear_bit(FSDB_UDESC, &fsdb->fsdb_flags);
3251                 CWARN("Disable user descriptor shipping from client to MDT\n");
3252         } else {
3253                 *(ptr - 1) = '=';
3254                 goto error_out;
3255         }
3256         return 0;
3257
3258 error_out:
3259         CERROR("Invalid param: %s\n", param);
3260         return -EINVAL;
3261 }
3262
3263 static int mgs_srpc_set_param_mem(struct fs_db *fsdb,
3264                                   const char *svname,
3265                                   char *param)
3266 {
3267         struct sptlrpc_rule      rule;
3268         struct sptlrpc_rule_set *rset;
3269         int                      rc;
3270         ENTRY;
3271
3272         if (strncmp(param, PARAM_SRPC, sizeof(PARAM_SRPC) - 1) != 0) {
3273                 CERROR("Invalid sptlrpc parameter: %s\n", param);
3274                 RETURN(-EINVAL);
3275         }
3276
3277         if (strncmp(param, PARAM_SRPC_UDESC,
3278                     sizeof(PARAM_SRPC_UDESC) - 1) == 0) {
3279                 RETURN(mgs_srpc_set_param_udesc_mem(fsdb, param));
3280         }
3281
3282         if (strncmp(param, PARAM_SRPC_FLVR, sizeof(PARAM_SRPC_FLVR) - 1) != 0) {
3283                 CERROR("Invalid sptlrpc flavor parameter: %s\n", param);
3284                 RETURN(-EINVAL);
3285         }
3286
3287         param += sizeof(PARAM_SRPC_FLVR) - 1;
3288
3289         rc = sptlrpc_parse_rule(param, &rule);
3290         if (rc)
3291                 RETURN(rc);
3292
3293         /* mgs rules implies must be mgc->mgs */
3294         if (test_bit(FSDB_MGS_SELF, &fsdb->fsdb_flags)) {
3295                 if ((rule.sr_from != LUSTRE_SP_MGC &&
3296                      rule.sr_from != LUSTRE_SP_ANY) ||
3297                     (rule.sr_to != LUSTRE_SP_MGS &&
3298                      rule.sr_to != LUSTRE_SP_ANY))
3299                         RETURN(-EINVAL);
3300         }
3301
3302         /* preapre room for this coming rule. svcname format should be:
3303          * - fsname: general rule
3304          * - fsname-tgtname: target-specific rule
3305          */
3306         if (strchr(svname, '-')) {
3307                 struct mgs_tgt_srpc_conf *tgtconf;
3308                 int                       found = 0;
3309
3310                 for (tgtconf = fsdb->fsdb_srpc_tgt; tgtconf != NULL;
3311                      tgtconf = tgtconf->mtsc_next) {
3312                         if (!strcmp(tgtconf->mtsc_tgt, svname)) {
3313                                 found = 1;
3314                                 break;
3315                         }
3316                 }
3317
3318                 if (!found) {
3319                         int name_len;
3320
3321                         OBD_ALLOC_PTR(tgtconf);
3322                         if (tgtconf == NULL)
3323                                 RETURN(-ENOMEM);
3324
3325                         name_len = strlen(svname);
3326
3327                         OBD_ALLOC(tgtconf->mtsc_tgt, name_len + 1);
3328                         if (tgtconf->mtsc_tgt == NULL) {
3329                                 OBD_FREE_PTR(tgtconf);
3330                                 RETURN(-ENOMEM);
3331                         }
3332                         memcpy(tgtconf->mtsc_tgt, svname, name_len);
3333
3334                         tgtconf->mtsc_next = fsdb->fsdb_srpc_tgt;
3335                         fsdb->fsdb_srpc_tgt = tgtconf;
3336                 }
3337
3338                 rset = &tgtconf->mtsc_rset;
3339         } else if (strcmp(svname, MGSSELF_NAME) == 0) {
3340                 /* put _mgs related srpc rule directly in mgs ruleset */
3341                 rset = &fsdb->fsdb_mgs->mgs_lut.lut_sptlrpc_rset;
3342         } else {
3343                 rset = &fsdb->fsdb_srpc_gen;
3344         }
3345
3346         rc = sptlrpc_rule_set_merge(rset, &rule);
3347
3348         RETURN(rc);
3349 }
3350
3351 static int mgs_srpc_set_param(const struct lu_env *env,
3352                               struct mgs_device *mgs,
3353                               struct fs_db *fsdb,
3354                               struct mgs_target_info *mti,
3355                               char *param)
3356 {
3357         char                   *copy;
3358         int                     rc, copy_size;
3359         ENTRY;
3360
3361 #ifndef HAVE_GSS
3362         RETURN(-EINVAL);
3363 #endif
3364         /* keep a copy of original param, which could be destroied
3365          * during parsing */
3366         copy_size = strlen(param) + 1;
3367         OBD_ALLOC(copy, copy_size);
3368         if (copy == NULL)
3369                 return -ENOMEM;
3370         memcpy(copy, param, copy_size);
3371
3372         rc = mgs_srpc_set_param_mem(fsdb, mti->mti_svname, param);
3373         if (rc)
3374                 goto out_free;
3375
3376         /* previous steps guaranteed the syntax is correct */
3377         rc = mgs_srpc_set_param_disk(env, mgs, fsdb, mti, copy);
3378         if (rc)
3379                 goto out_free;
3380
3381         if (test_bit(FSDB_MGS_SELF, &fsdb->fsdb_flags)) {
3382                 /*
3383                  * for mgs rules, make them effective immediately.
3384                  */
3385                 LASSERT(fsdb->fsdb_srpc_tgt == NULL);
3386                 sptlrpc_target_update_exp_flavor(mgs->mgs_obd,
3387                                                  &fsdb->fsdb_srpc_gen);
3388         }
3389
3390 out_free:
3391         OBD_FREE(copy, copy_size);
3392         RETURN(rc);
3393 }
3394
3395 struct mgs_srpc_read_data {
3396         struct fs_db   *msrd_fsdb;
3397         int             msrd_skip;
3398 };
3399
3400 static int mgs_srpc_read_handler(const struct lu_env *env,
3401                                  struct llog_handle *llh,
3402                                  struct llog_rec_hdr *rec, void *data)
3403 {
3404         struct mgs_srpc_read_data *msrd = data;
3405         struct cfg_marker         *marker;
3406         struct lustre_cfg         *lcfg = REC_DATA(rec);
3407         char                      *svname, *param;
3408         int                        cfg_len, rc;
3409         ENTRY;
3410
3411         if (rec->lrh_type != OBD_CFG_REC) {
3412                 CERROR("unhandled lrh_type: %#x\n", rec->lrh_type);
3413                 RETURN(-EINVAL);
3414         }
3415
3416         cfg_len = REC_DATA_LEN(rec);
3417
3418         rc = lustre_cfg_sanity_check(lcfg, cfg_len);
3419         if (rc) {
3420                 CERROR("Insane cfg\n");
3421                 RETURN(rc);
3422         }
3423
3424         if (lcfg->lcfg_command == LCFG_MARKER) {
3425                 marker = lustre_cfg_buf(lcfg, 1);
3426
3427                 if (marker->cm_flags & CM_START &&
3428                     marker->cm_flags & CM_SKIP)
3429                         msrd->msrd_skip = 1;
3430                 if (marker->cm_flags & CM_END)
3431                         msrd->msrd_skip = 0;
3432
3433                 RETURN(0);
3434         }
3435
3436         if (msrd->msrd_skip)
3437                 RETURN(0);
3438
3439         if (lcfg->lcfg_command != LCFG_SPTLRPC_CONF) {
3440                 CERROR("invalid command (%x)\n", lcfg->lcfg_command);
3441                 RETURN(0);
3442         }
3443
3444         svname = lustre_cfg_string(lcfg, 0);
3445         if (svname == NULL) {
3446                 CERROR("svname is empty\n");
3447                 RETURN(0);
3448         }
3449
3450         param = lustre_cfg_string(lcfg, 1);
3451         if (param == NULL) {
3452                 CERROR("param is empty\n");
3453                 RETURN(0);
3454         }
3455
3456         rc = mgs_srpc_set_param_mem(msrd->msrd_fsdb, svname, param);
3457         if (rc)
3458                 CERROR("read sptlrpc record error (%d): %s\n", rc, param);
3459
3460         RETURN(0);
3461 }
3462
3463 int mgs_get_fsdb_srpc_from_llog(const struct lu_env *env,
3464                                 struct mgs_device *mgs,
3465                                 struct fs_db *fsdb)
3466 {
3467         struct llog_handle        *llh = NULL;
3468         struct llog_ctxt          *ctxt;
3469         char                      *logname;
3470         struct mgs_srpc_read_data  msrd;
3471         int                        rc;
3472         ENTRY;
3473
3474         /* construct log name */
3475         rc = name_create(&logname, fsdb->fsdb_name, "-sptlrpc");
3476         if (rc)
3477                 RETURN(rc);
3478
3479         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
3480         LASSERT(ctxt != NULL);
3481
3482         if (mgs_log_is_empty(env, mgs, logname))
3483                 GOTO(out, rc = 0);
3484
3485         rc = llog_open(env, ctxt, &llh, NULL, logname,
3486                        LLOG_OPEN_EXISTS);
3487         if (rc < 0) {
3488                 if (rc == -ENOENT)
3489                         rc = 0;
3490                 GOTO(out, rc);
3491         }
3492
3493         rc = llog_init_handle(env, llh, LLOG_F_IS_PLAIN, NULL);
3494         if (rc)
3495                 GOTO(out_close, rc);
3496
3497         if (llog_get_size(llh) <= 1)
3498                 GOTO(out_close, rc = 0);
3499
3500         msrd.msrd_fsdb = fsdb;
3501         msrd.msrd_skip = 0;
3502
3503         rc = llog_process(env, llh, mgs_srpc_read_handler, (void *)&msrd,
3504                           NULL);
3505
3506 out_close:
3507         llog_close(env, llh);
3508 out:
3509         llog_ctxt_put(ctxt);
3510         name_destroy(&logname);
3511
3512         if (rc)
3513                 CERROR("failed to read sptlrpc config database: %d\n", rc);
3514         RETURN(rc);
3515 }
3516
3517 static int mgs_write_log_param2(const struct lu_env *env,
3518                                 struct mgs_device *mgs,
3519                                 struct fs_db *fsdb,
3520                                 struct mgs_target_info *mti, char *ptr)
3521 {
3522         struct lustre_cfg_bufs bufs;
3523         int rc;
3524
3525         ENTRY;
3526         CDEBUG(D_MGS, "next param '%s'\n", ptr);
3527
3528         /* PARAM_MGSNODE and PARAM_NETWORK are set only when formating
3529          * or during the inital mount. It can never change after that.
3530          */
3531         if (!class_match_param(ptr, PARAM_MGSNODE, NULL) ||
3532             !class_match_param(ptr, PARAM_NETWORK, NULL)) {
3533                 rc = 0;
3534                 goto end;
3535         }
3536
3537         /* Processed in mgs_write_log_ost. Another value that can't
3538          * be changed by lctl set_param -P.
3539          */
3540         if (!class_match_param(ptr, PARAM_FAILMODE, NULL)) {
3541                 LCONSOLE_ERROR_MSG(0x169,
3542                                    "%s can only be changed with tunefs.lustre and --writeconf\n",
3543                                    ptr);
3544                 rc = -EPERM;
3545                 goto end;
3546         }
3547
3548         /* FIXME !!! Support for sptlrpc is incomplete. Currently the change
3549          * doesn't transmit to the client. See LU-7183.
3550          */
3551         if (!class_match_param(ptr, PARAM_SRPC, NULL)) {
3552                 rc = mgs_srpc_set_param(env, mgs, fsdb, mti, ptr);
3553                 goto end;
3554         }
3555
3556         /* Can't use class_match_param since ptr doesn't start with
3557          * PARAM_FAILNODE. So we look for PARAM_FAILNODE contained in ptr.
3558          */
3559         if (strstr(ptr, PARAM_FAILNODE)) {
3560                 /* Add a failover nidlist. We already processed failovers
3561                  * params for new targets in mgs_write_log_target.
3562                  */
3563                 const char *param;
3564
3565                 /* can't use wildcards with failover.node */
3566                 if (strchr(ptr, '*')) {
3567                         rc = -ENODEV;
3568                         goto end;
3569                 }
3570
3571                 param = strstr(ptr, PARAM_FAILNODE);
3572                 if (strlcpy(mti->mti_params, param, sizeof(mti->mti_params)) >=
3573                     sizeof(mti->mti_params)) {
3574                         rc = -E2BIG;
3575                         goto end;
3576                 }
3577
3578                 CDEBUG(D_MGS, "Adding failnode with param %s\n",
3579                        mti->mti_params);
3580                 rc = mgs_write_log_add_failnid(env, mgs, fsdb, mti);
3581                 goto end;
3582         }
3583
3584         rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, PARAMS_FILENAME, &bufs,
3585                           mti->mti_svname, ptr);
3586 end:
3587         RETURN(rc);
3588 }
3589
3590 /* Permanent settings of all parameters by writing into the appropriate
3591  * configuration logs.
3592  * A parameter with null value ("<param>='\0'") means to erase it out of
3593  * the logs.
3594  */
3595 static int mgs_write_log_param(const struct lu_env *env,
3596                                struct mgs_device *mgs, struct fs_db *fsdb,
3597                                struct mgs_target_info *mti, char *ptr)
3598 {
3599         struct mgs_thread_info *mgi = mgs_env_info(env);
3600         char *logname;
3601         char *tmp;
3602         int rc = 0;
3603         ENTRY;
3604
3605         /* For various parameter settings, we have to figure out which logs
3606            care about them (e.g. both mdt and client for lov settings) */
3607         CDEBUG(D_MGS, "next param '%s'\n", ptr);
3608
3609         /* The params are stored in MOUNT_DATA_FILE and modified via
3610            tunefs.lustre, or set using lctl conf_param */
3611
3612         /* Processed in lustre_start_mgc */
3613         if (class_match_param(ptr, PARAM_MGSNODE, NULL) == 0)
3614                 GOTO(end, rc);
3615
3616         /* Processed in ost/mdt */
3617         if (class_match_param(ptr, PARAM_NETWORK, NULL) == 0)
3618                 GOTO(end, rc);
3619
3620         /* Processed in mgs_write_log_ost */
3621         if (class_match_param(ptr, PARAM_FAILMODE, NULL) == 0) {
3622                 if (mti->mti_flags & LDD_F_PARAM) {
3623                         LCONSOLE_ERROR_MSG(0x169, "%s can only be "
3624                                            "changed with tunefs.lustre"
3625                                            "and --writeconf\n", ptr);
3626                         rc = -EPERM;
3627                 }
3628                 GOTO(end, rc);
3629         }
3630
3631         if (class_match_param(ptr, PARAM_SRPC, NULL) == 0) {
3632                 rc = mgs_srpc_set_param(env, mgs, fsdb, mti, ptr);
3633                 GOTO(end, rc);
3634         }
3635
3636         if (class_match_param(ptr, PARAM_FAILNODE, NULL) == 0) {
3637                 /* Add a failover nidlist */
3638                 rc = 0;
3639                 /* We already processed failovers params for new
3640                    targets in mgs_write_log_target */
3641                 if (mti->mti_flags & LDD_F_PARAM) {
3642                         CDEBUG(D_MGS, "Adding failnode\n");
3643                         rc = mgs_write_log_add_failnid(env, mgs, fsdb, mti);
3644                 }
3645                 GOTO(end, rc);
3646         }
3647
3648         if (class_match_param(ptr, PARAM_SYS, &tmp) == 0) {
3649                 rc = mgs_write_log_sys(env, mgs, fsdb, mti, ptr, tmp);
3650                 GOTO(end, rc);
3651         }
3652
3653         if (class_match_param(ptr, PARAM_QUOTA, &tmp) == 0) {
3654                 rc = mgs_write_log_quota(env, mgs, fsdb, mti, ptr, tmp);
3655                 GOTO(end, rc);
3656         }
3657
3658         if (class_match_param(ptr, PARAM_OSC PARAM_ACTIVE, &tmp) == 0 ||
3659             class_match_param(ptr, PARAM_MDC PARAM_ACTIVE, &tmp) == 0) {
3660                 /* active=0 means off, anything else means on */
3661                 int flag = (*tmp == '0') ? CM_EXCLUDE : 0;
3662                 bool deactive_osc = memcmp(ptr, PARAM_OSC PARAM_ACTIVE,
3663                                           strlen(PARAM_OSC PARAM_ACTIVE)) == 0;
3664                 int i;
3665
3666                 if (!deactive_osc) {
3667                         __u32   index;
3668
3669                         rc = server_name2index(mti->mti_svname, &index, NULL);
3670                         if (rc < 0)
3671                                 GOTO(end, rc);
3672
3673                         if (index == 0) {
3674                                 LCONSOLE_ERROR_MSG(0x144, "%s: MDC0 can not be"
3675                                                    " (de)activated.\n",
3676                                                    mti->mti_svname);
3677                                 GOTO(end, rc = -EPERM);
3678                         }
3679                 }
3680
3681                 LCONSOLE_WARN("Permanently %sactivating %s\n",
3682                               flag ? "de" : "re", mti->mti_svname);
3683                 /* Modify clilov */
3684                 rc = name_create(&logname, mti->mti_fsname, "-client");
3685                 if (rc < 0)
3686                         GOTO(end, rc);
3687                 rc = mgs_modify(env, mgs, fsdb, mti, logname,
3688                                 mti->mti_svname,
3689                                 deactive_osc ? "add osc" : "add mdc", flag);
3690                 name_destroy(&logname);
3691                 if (rc < 0)
3692                         goto active_err;
3693
3694                 /* Modify mdtlov */
3695                 /* Add to all MDT logs for DNE */
3696                 for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
3697                         if (!test_bit(i, fsdb->fsdb_mdt_index_map))
3698                                 continue;
3699                         rc = name_create_mdt(&logname, mti->mti_fsname, i);
3700                         if (rc < 0)
3701                                 GOTO(end, rc);
3702                         rc = mgs_modify(env, mgs, fsdb, mti, logname,
3703                                         mti->mti_svname,
3704                                         deactive_osc ? "add osc" : "add osp",
3705                                         flag);
3706                         name_destroy(&logname);
3707                         if (rc < 0)
3708                                 goto active_err;
3709                 }
3710 active_err:
3711                 if (rc < 0) {
3712                         LCONSOLE_ERROR_MSG(0x145, "Couldn't find %s in"
3713                                            "log (%d). No permanent "
3714                                            "changes were made to the "
3715                                            "config log.\n",
3716                                            mti->mti_svname, rc);
3717                         if (test_bit(FSDB_OLDLOG14, &fsdb->fsdb_flags))
3718                                 LCONSOLE_ERROR_MSG(0x146, "This may be"
3719                                                    " because the log"
3720                                                    "is in the old 1.4"
3721                                                    "style. Consider "
3722                                                    " --writeconf to "
3723                                                    "update the logs.\n");
3724                         GOTO(end, rc);
3725                 }
3726                 /* Fall through to osc/mdc proc for deactivating live
3727                    OSC/OSP on running MDT / clients. */
3728         }
3729         /* Below here, let obd's XXX_process_config methods handle it */
3730
3731         /* All lov. in proc */
3732         if (class_match_param(ptr, PARAM_LOV, NULL) == 0) {
3733                 char *mdtlovname;
3734
3735                 CDEBUG(D_MGS, "lov param %s\n", ptr);
3736                 if (!(mti->mti_flags & LDD_F_SV_TYPE_MDT)) {
3737                         LCONSOLE_ERROR_MSG(0x147, "LOV params must be "
3738                                            "set on the MDT, not %s. "
3739                                            "Ignoring.\n",
3740                                            mti->mti_svname);
3741                         GOTO(end, rc = 0);
3742                 }
3743
3744                 /* Modify mdtlov */
3745                 if (mgs_log_is_empty(env, mgs, mti->mti_svname))
3746                         GOTO(end, rc = -ENODEV);
3747
3748                 rc = name_create_mdt_and_lov(&logname, &mdtlovname, fsdb,
3749                                              mti->mti_stripe_index);
3750                 if (rc)
3751                         GOTO(end, rc);
3752                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, mti->mti_svname,
3753                                   &mgi->mgi_bufs, mdtlovname, ptr);
3754                 name_destroy(&logname);
3755                 name_destroy(&mdtlovname);
3756                 if (rc)
3757                         GOTO(end, rc);
3758
3759                 /* Modify clilov */
3760                 rc = name_create(&logname, mti->mti_fsname, "-client");
3761                 if (rc)
3762                         GOTO(end, rc);
3763                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, logname, &mgi->mgi_bufs,
3764                                   fsdb->fsdb_clilov, ptr);
3765                 name_destroy(&logname);
3766                 GOTO(end, rc);
3767         }
3768
3769         /* All osc., mdc., llite. params in proc */
3770         if ((class_match_param(ptr, PARAM_OSC, NULL) == 0) ||
3771             (class_match_param(ptr, PARAM_MDC, NULL) == 0) ||
3772             (class_match_param(ptr, PARAM_LLITE, NULL) == 0)) {
3773                 char *cname;
3774
3775                 if (test_bit(FSDB_OLDLOG14, &fsdb->fsdb_flags)) {
3776                         LCONSOLE_ERROR_MSG(0x148, "Upgraded client logs for %s"
3777                                            " cannot be modified. Consider"
3778                                            " updating the configuration with"
3779                                            " --writeconf\n",
3780                                            mti->mti_svname);
3781                         GOTO(end, rc = -EINVAL);
3782                 }
3783                 if (memcmp(ptr, PARAM_LLITE, strlen(PARAM_LLITE)) == 0) {
3784                         rc = name_create(&cname, mti->mti_fsname, "-client");
3785                         /* Add the client type to match the obdname in
3786                            class_config_llog_handler */
3787                 } else if (mti->mti_flags & LDD_F_SV_TYPE_MDT) {
3788                         rc = name_create(&cname, mti->mti_svname, "-mdc");
3789                 } else if (mti->mti_flags & LDD_F_SV_TYPE_OST) {
3790                         rc = name_create(&cname, mti->mti_svname, "-osc");
3791                 } else {
3792                         GOTO(end, rc = -EINVAL);
3793                 }
3794                 if (rc)
3795                         GOTO(end, rc);
3796
3797                 /* Forbid direct update of llite root squash parameters.
3798                  * These parameters are indirectly set via the MDT settings.
3799                  * See (LU-1778) */
3800                 if ((class_match_param(ptr, PARAM_LLITE, &tmp) == 0) &&
3801                     ((memcmp(tmp, "root_squash=", 12) == 0) ||
3802                      (memcmp(tmp, "nosquash_nids=", 14) == 0))) {
3803                         LCONSOLE_ERROR("%s: root squash parameters can only "
3804                                 "be updated through MDT component\n",
3805                                 mti->mti_fsname);
3806                         name_destroy(&cname);
3807                         GOTO(end, rc = -EINVAL);
3808                 }
3809
3810                 CDEBUG(D_MGS, "%.3s param %s\n", ptr, ptr + 4);
3811
3812                 /* Modify client */
3813                 rc = name_create(&logname, mti->mti_fsname, "-client");
3814                 if (rc) {
3815                         name_destroy(&cname);
3816                         GOTO(end, rc);
3817                 }
3818                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, logname, &mgi->mgi_bufs,
3819                                   cname, ptr);
3820
3821                 /* osc params affect the MDT as well */
3822                 if (!rc && (mti->mti_flags & LDD_F_SV_TYPE_OST)) {
3823                         int i;
3824
3825                         for (i = 0; i < INDEX_MAP_SIZE * 8; i++){
3826                                 if (!test_bit(i, fsdb->fsdb_mdt_index_map))
3827                                         continue;
3828                                 name_destroy(&cname);
3829                                 rc = name_create_mdt_osc(&cname, mti->mti_svname,
3830                                                          fsdb, i);
3831                                 name_destroy(&logname);
3832                                 if (rc)
3833                                         break;
3834                                 rc = name_create_mdt(&logname,
3835                                                      mti->mti_fsname, i);
3836                                 if (rc)
3837                                         break;
3838                                 if (!mgs_log_is_empty(env, mgs, logname)) {
3839                                         rc = mgs_wlp_lcfg(env, mgs, fsdb,
3840                                                           mti, logname,
3841                                                           &mgi->mgi_bufs,
3842                                                           cname, ptr);
3843                                         if (rc)
3844                                                 break;
3845                                 }
3846                         }
3847                 }
3848
3849                 /* For mdc activate/deactivate, it affects OSP on MDT as well */
3850                 if (class_match_param(ptr, PARAM_MDC PARAM_ACTIVE, &tmp) == 0 &&
3851                     rc == 0) {
3852                         char suffix[16];
3853                         char *lodname = NULL;
3854                         char *param_str = NULL;
3855                         int i;
3856                         int index;
3857
3858                         /* replace mdc with osp */
3859                         memcpy(ptr, PARAM_OSP, strlen(PARAM_OSP));
3860                         rc = server_name2index(mti->mti_svname, &index, NULL);
3861                         if (rc < 0) {
3862                                 memcpy(ptr, PARAM_MDC, strlen(PARAM_MDC));
3863                                 GOTO(end, rc);
3864                         }
3865
3866                         for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
3867                                 if (!test_bit(i, fsdb->fsdb_mdt_index_map))
3868                                         continue;
3869
3870                                 if (i == index)
3871                                         continue;
3872
3873                                 name_destroy(&logname);
3874                                 rc = name_create_mdt(&logname, mti->mti_fsname,
3875                                                      i);
3876                                 if (rc < 0)
3877                                         break;
3878
3879                                 if (mgs_log_is_empty(env, mgs, logname))
3880                                         continue;
3881
3882                                 snprintf(suffix, sizeof(suffix), "-osp-MDT%04x",
3883                                          i);
3884                                 name_destroy(&cname);
3885                                 rc = name_create(&cname, mti->mti_svname,
3886                                                  suffix);
3887                                 if (rc < 0)
3888                                         break;
3889
3890                                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, logname,
3891                                                   &mgi->mgi_bufs, cname, ptr);
3892                                 if (rc < 0)
3893                                         break;
3894
3895                                 /* Add configuration log for noitfying LOD
3896                                  * to active/deactive the OSP. */
3897                                 name_destroy(&param_str);
3898                                 rc = name_create(&param_str, cname,
3899                                                  (*tmp == '0') ?  ".active=0" :
3900                                                  ".active=1");
3901                                 if (rc < 0)
3902                                         break;
3903
3904                                 name_destroy(&lodname);
3905                                 rc = name_create(&lodname, logname, "-mdtlov");
3906                                 if (rc < 0)
3907                                         break;
3908
3909                                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, logname,
3910                                                   &mgi->mgi_bufs, lodname,
3911                                                   param_str);
3912                                 if (rc < 0)
3913                                         break;
3914                         }
3915                         memcpy(ptr, PARAM_MDC, strlen(PARAM_MDC));
3916                         name_destroy(&lodname);
3917                         name_destroy(&param_str);
3918                 }
3919
3920                 name_destroy(&logname);
3921                 name_destroy(&cname);
3922                 GOTO(end, rc);
3923         }
3924
3925         /* All mdt. params in proc */
3926         if (class_match_param(ptr, PARAM_MDT, &tmp) == 0) {
3927                 int i;
3928                 __u32 idx;
3929
3930                 CDEBUG(D_MGS, "%.3s param %s\n", ptr, ptr + 4);
3931                 if (strncmp(mti->mti_svname, mti->mti_fsname,
3932                             MTI_NAME_MAXLEN) == 0)
3933                         /* device is unspecified completely? */
3934                         rc = LDD_F_SV_TYPE_MDT | LDD_F_SV_ALL;
3935                 else
3936                         rc = server_name2index(mti->mti_svname, &idx, NULL);
3937                 if (rc < 0)
3938                         goto active_err;
3939                 if ((rc & LDD_F_SV_TYPE_MDT) == 0)
3940                         goto active_err;
3941                 if (rc & LDD_F_SV_ALL) {
3942                         for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
3943                                 if (!test_bit(i,
3944                                                   fsdb->fsdb_mdt_index_map))
3945                                         continue;
3946                                 rc = name_create_mdt(&logname,
3947                                                 mti->mti_fsname, i);
3948                                 if (rc)
3949                                         goto active_err;
3950                                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti,
3951                                                   logname, &mgi->mgi_bufs,
3952                                                   logname, ptr);
3953                                 name_destroy(&logname);
3954                                 if (rc)
3955                                         goto active_err;
3956                         }
3957                 } else {
3958                         if ((memcmp(tmp, "root_squash=", 12) == 0) ||
3959                             (memcmp(tmp, "nosquash_nids=", 14) == 0)) {
3960                                 LCONSOLE_ERROR("%s: root squash parameters "
3961                                         "cannot be applied to a single MDT\n",
3962                                         mti->mti_fsname);
3963                                 GOTO(end, rc = -EINVAL);
3964                         }
3965                         rc = mgs_wlp_lcfg(env, mgs, fsdb, mti,
3966                                           mti->mti_svname, &mgi->mgi_bufs,
3967                                           mti->mti_svname, ptr);
3968                         if (rc)
3969                                 goto active_err;
3970                 }
3971
3972                 /* root squash settings are also applied to llite
3973                  * config log (see LU-1778) */
3974                 if (rc == 0 &&
3975                     ((memcmp(tmp, "root_squash=", 12) == 0) ||
3976                      (memcmp(tmp, "nosquash_nids=", 14) == 0))) {
3977                         char *cname;
3978                         char *ptr2;
3979
3980                         rc = name_create(&cname, mti->mti_fsname, "-client");
3981                         if (rc)
3982                                 GOTO(end, rc);
3983                         rc = name_create(&logname, mti->mti_fsname, "-client");
3984                         if (rc) {
3985                                 name_destroy(&cname);
3986                                 GOTO(end, rc);
3987                         }
3988                         rc = name_create(&ptr2, PARAM_LLITE, tmp);
3989                         if (rc) {
3990                                 name_destroy(&cname);
3991                                 name_destroy(&logname);
3992                                 GOTO(end, rc);
3993                         }
3994                         rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, logname,
3995                                           &mgi->mgi_bufs, cname, ptr2);
3996                         name_destroy(&ptr2);
3997                         name_destroy(&logname);
3998                         name_destroy(&cname);
3999                 }
4000                 GOTO(end, rc);
4001         }
4002
4003         /* All mdd., ost. and osd. params in proc */
4004         if ((class_match_param(ptr, PARAM_MDD, NULL) == 0) ||
4005             (class_match_param(ptr, PARAM_LOD, NULL) == 0) ||
4006             (class_match_param(ptr, PARAM_OST, NULL) == 0) ||
4007             (class_match_param(ptr, PARAM_OSD, NULL) == 0)) {
4008                 CDEBUG(D_MGS, "%.3s param %s\n", ptr, ptr + 4);
4009                 if (mgs_log_is_empty(env, mgs, mti->mti_svname))
4010                         GOTO(end, rc = -ENODEV);
4011
4012                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, mti->mti_svname,
4013                                   &mgi->mgi_bufs, mti->mti_svname, ptr);
4014                 GOTO(end, rc);
4015         }
4016
4017         LCONSOLE_WARN("Ignoring unrecognized param '%s'\n", ptr);
4018
4019 end:
4020         if (rc)
4021                 CERROR("err %d on param '%s'\n", rc, ptr);
4022
4023         RETURN(rc);
4024 }
4025
4026 int mgs_write_log_target(const struct lu_env *env, struct mgs_device *mgs,
4027                          struct mgs_target_info *mti, struct fs_db *fsdb)
4028 {
4029         char    *buf, *params;
4030         int      rc = -EINVAL;
4031
4032         ENTRY;
4033
4034         /* set/check the new target index */
4035         rc = mgs_set_index(env, mgs, mti);
4036         if (rc < 0)
4037                 RETURN(rc);
4038
4039         if (rc == EALREADY) {
4040                 LCONSOLE_WARN("Found index %d for %s, updating log\n",
4041                               mti->mti_stripe_index, mti->mti_svname);
4042                 /* We would like to mark old log sections as invalid
4043                    and add new log sections in the client and mdt logs.
4044                    But if we add new sections, then live clients will
4045                    get repeat setup instructions for already running
4046                    osc's. So don't update the client/mdt logs. */
4047                 mti->mti_flags &= ~LDD_F_UPDATE;
4048                 rc = 0;
4049         }
4050
4051         OBD_FAIL_TIMEOUT(OBD_FAIL_MGS_WRITE_TARGET_DELAY, cfs_fail_val > 0 ?
4052                          cfs_fail_val : 10);
4053
4054         mutex_lock(&fsdb->fsdb_mutex);
4055
4056         if (mti->mti_flags &
4057             (LDD_F_VIRGIN | LDD_F_UPGRADE14 | LDD_F_WRITECONF)) {
4058                 /* Generate a log from scratch */
4059                 if (mti->mti_flags & LDD_F_SV_TYPE_MDT) {
4060                         rc = mgs_write_log_mdt(env, mgs, fsdb, mti);
4061                 } else if (mti->mti_flags & LDD_F_SV_TYPE_OST) {
4062                         rc = mgs_write_log_ost(env, mgs, fsdb, mti);
4063                 } else {
4064                         CERROR("Unknown target type %#x, can't create log for "
4065                                "%s\n", mti->mti_flags, mti->mti_svname);
4066                 }
4067                 if (rc) {
4068                         CERROR("Can't write logs for %s (%d)\n",
4069                                mti->mti_svname, rc);
4070                         GOTO(out_up, rc);
4071                 }
4072         } else {
4073                 /* Just update the params from tunefs in mgs_write_log_params */
4074                 CDEBUG(D_MGS, "Update params for %s\n", mti->mti_svname);
4075                 mti->mti_flags |= LDD_F_PARAM;
4076         }
4077
4078         /* allocate temporary buffer, where class_get_next_param will
4079            make copy of a current  parameter */
4080         OBD_ALLOC(buf, strlen(mti->mti_params) + 1);
4081         if (buf == NULL)
4082                 GOTO(out_up, rc = -ENOMEM);
4083         params = mti->mti_params;
4084         while (params != NULL) {
4085                 rc = class_get_next_param(&params, buf);
4086                 if (rc) {
4087                         if (rc == 1)
4088                                 /* there is no next parameter, that is
4089                                    not an error */
4090                                 rc = 0;
4091                         break;
4092                 }
4093                 CDEBUG(D_MGS, "remaining string: '%s', param: '%s'\n",
4094                        params, buf);
4095                 rc = mgs_write_log_param(env, mgs, fsdb, mti, buf);
4096                 if (rc)
4097                         break;
4098         }
4099
4100         OBD_FREE(buf, strlen(mti->mti_params) + 1);
4101
4102 out_up:
4103         mutex_unlock(&fsdb->fsdb_mutex);
4104         RETURN(rc);
4105 }
4106
4107 int mgs_erase_log(const struct lu_env *env, struct mgs_device *mgs, char *name)
4108 {
4109         struct llog_ctxt        *ctxt;
4110         int                      rc = 0;
4111
4112         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
4113         if (ctxt == NULL) {
4114                 CERROR("%s: MGS config context doesn't exist\n",
4115                        mgs->mgs_obd->obd_name);
4116                 rc = -ENODEV;
4117         } else {
4118                 rc = llog_erase(env, ctxt, NULL, name);
4119                 /* llog may not exist */
4120                 if (rc == -ENOENT)
4121                         rc = 0;
4122                 llog_ctxt_put(ctxt);
4123         }
4124
4125         if (rc)
4126                 CERROR("%s: failed to clear log %s: %d\n",
4127                        mgs->mgs_obd->obd_name, name, rc);
4128
4129         return rc;
4130 }
4131
4132 /* erase all logs for the given fs */
4133 int mgs_erase_logs(const struct lu_env *env, struct mgs_device *mgs,
4134                    const char *fsname)
4135 {
4136         struct list_head log_list;
4137         struct mgs_direntry *dirent, *n;
4138         char barrier_name[20] = {};
4139         char *suffix;
4140         int count = 0;
4141         int rc, len = strlen(fsname);
4142         ENTRY;
4143
4144         mutex_lock(&mgs->mgs_mutex);
4145
4146         /* Find all the logs in the CONFIGS directory */
4147         rc = class_dentry_readdir(env, mgs, &log_list);
4148         if (rc) {
4149                 mutex_unlock(&mgs->mgs_mutex);
4150                 RETURN(rc);
4151         }
4152
4153         if (list_empty(&log_list)) {
4154                 mutex_unlock(&mgs->mgs_mutex);
4155                 RETURN(-ENOENT);
4156         }
4157
4158         snprintf(barrier_name, sizeof(barrier_name) - 1, "%s-%s",
4159                  fsname, BARRIER_FILENAME);
4160         /* Delete the barrier fsdb */
4161         mgs_remove_fsdb_by_name(mgs, barrier_name);
4162         /* Delete the fs db */
4163         mgs_remove_fsdb_by_name(mgs, fsname);
4164         mutex_unlock(&mgs->mgs_mutex);
4165
4166         list_for_each_entry_safe(dirent, n, &log_list, mde_list) {
4167                 list_del_init(&dirent->mde_list);
4168                 suffix = strrchr(dirent->mde_name, '-');
4169                 if (suffix != NULL) {
4170                         if ((len == suffix - dirent->mde_name) &&
4171                             (strncmp(fsname, dirent->mde_name, len) == 0)) {
4172                                 CDEBUG(D_MGS, "Removing log %s\n",
4173                                        dirent->mde_name);
4174                                 mgs_erase_log(env, mgs, dirent->mde_name);
4175                                 count++;
4176                         }
4177                 }
4178                 mgs_direntry_free(dirent);
4179         }
4180
4181         if (count == 0)
4182                 rc = -ENOENT;
4183
4184         RETURN(rc);
4185 }
4186
4187 /* list all logs for the given fs */
4188 int mgs_list_logs(const struct lu_env *env, struct mgs_device *mgs,
4189                   struct obd_ioctl_data *data)
4190 {
4191         struct list_head         log_list;
4192         struct mgs_direntry     *dirent, *n;
4193         char                    *out, *suffix;
4194         int                      l, remains, rc;
4195
4196         ENTRY;
4197
4198         /* Find all the logs in the CONFIGS directory */
4199         rc = class_dentry_readdir(env, mgs, &log_list);
4200         if (rc)
4201                 RETURN(rc);
4202
4203         out = data->ioc_bulk;
4204         remains = data->ioc_inllen1;
4205         list_for_each_entry_safe(dirent, n, &log_list, mde_list) {
4206                 list_del_init(&dirent->mde_list);
4207                 suffix = strrchr(dirent->mde_name, '-');
4208                 if (suffix != NULL) {
4209                         l = snprintf(out, remains, "config_log: %s\n",
4210                                      dirent->mde_name);
4211                         out += l;
4212                         remains -= l;
4213                 }
4214                 mgs_direntry_free(dirent);
4215                 if (remains <= 0)
4216                         break;
4217         }
4218         RETURN(rc);
4219 }
4220
4221 struct mgs_lcfg_fork_data {
4222         struct lustre_cfg_bufs   mlfd_bufs;
4223         struct mgs_device       *mlfd_mgs;
4224         struct llog_handle      *mlfd_llh;
4225         const char              *mlfd_oldname;
4226         const char              *mlfd_newname;
4227         char                     mlfd_data[0];
4228 };
4229
4230 static bool contain_valid_fsname(char *buf, const char *fsname,
4231                                  int buflen, int namelen)
4232 {
4233         if (buflen < namelen)
4234                 return false;
4235
4236         if (memcmp(buf, fsname, namelen) != 0)
4237                 return false;
4238
4239         if (buf[namelen] != '\0' && buf[namelen] != '-')
4240                 return false;
4241
4242         return true;
4243 }
4244
4245 static int mgs_lcfg_fork_handler(const struct lu_env *env,
4246                                  struct llog_handle *o_llh,
4247                                  struct llog_rec_hdr *o_rec, void *data)
4248 {
4249         struct mgs_lcfg_fork_data *mlfd = data;
4250         struct lustre_cfg_bufs *n_bufs = &mlfd->mlfd_bufs;
4251         struct lustre_cfg *o_lcfg = (struct lustre_cfg *)(o_rec + 1);
4252         struct llog_cfg_rec *lcr;
4253         char *o_buf;
4254         char *n_buf = mlfd->mlfd_data;
4255         int o_buflen;
4256         int o_namelen = strlen(mlfd->mlfd_oldname);
4257         int n_namelen = strlen(mlfd->mlfd_newname);
4258         int diff = n_namelen - o_namelen;
4259         __u32 cmd = o_lcfg->lcfg_command;
4260         __u32 cnt = o_lcfg->lcfg_bufcount;
4261         int rc;
4262         int i;
4263         ENTRY;
4264
4265         /* buf[0] */
4266         o_buf = lustre_cfg_buf(o_lcfg, 0);
4267         o_buflen = o_lcfg->lcfg_buflens[0];
4268         if (contain_valid_fsname(o_buf, mlfd->mlfd_oldname, o_buflen,
4269                                  o_namelen)) {
4270                 memcpy(n_buf, mlfd->mlfd_newname, n_namelen);
4271                 memcpy(n_buf + n_namelen, o_buf + o_namelen,
4272                        o_buflen - o_namelen);
4273                 lustre_cfg_bufs_reset(n_bufs, n_buf);
4274                 n_buf += cfs_size_round(o_buflen + diff);
4275         } else {
4276                 lustre_cfg_bufs_reset(n_bufs, o_buflen != 0 ? o_buf : NULL);
4277         }
4278
4279         switch (cmd) {
4280         case LCFG_MARKER: {
4281                 struct cfg_marker *o_marker;
4282                 struct cfg_marker *n_marker;
4283                 int tgt_namelen;
4284
4285                 if (cnt != 2) {
4286                         CDEBUG(D_MGS, "Unknown cfg marker entry with %d "
4287                                "buffers\n", cnt);
4288                         RETURN(-EINVAL);
4289                 }
4290
4291                 /* buf[1] is marker */
4292                 o_buf = lustre_cfg_buf(o_lcfg, 1);
4293                 o_buflen = o_lcfg->lcfg_buflens[1];
4294                 o_marker = (struct cfg_marker *)o_buf;
4295                 if (!contain_valid_fsname(o_marker->cm_tgtname,
4296                                           mlfd->mlfd_oldname,
4297                                           sizeof(o_marker->cm_tgtname),
4298                                           o_namelen)) {
4299                         lustre_cfg_bufs_set(n_bufs, 1, o_marker,
4300                                             sizeof(*o_marker));
4301                         break;
4302                 }
4303
4304                 n_marker = (struct cfg_marker *)n_buf;
4305                 *n_marker = *o_marker;
4306                 memcpy(n_marker->cm_tgtname, mlfd->mlfd_newname, n_namelen);
4307                 tgt_namelen = strlen(o_marker->cm_tgtname);
4308                 if (tgt_namelen > o_namelen)
4309                         memcpy(n_marker->cm_tgtname + n_namelen,
4310                                o_marker->cm_tgtname + o_namelen,
4311                                tgt_namelen - o_namelen);
4312                 n_marker->cm_tgtname[tgt_namelen + diff] = '\0';
4313                 lustre_cfg_bufs_set(n_bufs, 1, n_marker, sizeof(*n_marker));
4314                 break;
4315         }
4316         case LCFG_PARAM:
4317         case LCFG_SET_PARAM: {
4318                 for (i = 1; i < cnt; i++)
4319                         /* buf[i] is the param value, reuse it directly */
4320                         lustre_cfg_bufs_set(n_bufs, i,
4321                                             lustre_cfg_buf(o_lcfg, i),
4322                                             o_lcfg->lcfg_buflens[i]);
4323                 break;
4324         }
4325         case LCFG_POOL_NEW:
4326         case LCFG_POOL_ADD:
4327         case LCFG_POOL_REM:
4328         case LCFG_POOL_DEL: {
4329                 if (cnt < 3 || cnt > 4) {
4330                         CDEBUG(D_MGS, "Unknown cfg pool (%x) entry with %d "
4331                                "buffers\n", cmd, cnt);
4332                         RETURN(-EINVAL);
4333                 }
4334
4335                 /* buf[1] is fsname */
4336                 o_buf = lustre_cfg_buf(o_lcfg, 1);
4337                 o_buflen = o_lcfg->lcfg_buflens[1];
4338                 memcpy(n_buf, mlfd->mlfd_newname, n_namelen);
4339                 memcpy(n_buf + n_namelen, o_buf + o_namelen,
4340                        o_buflen - o_namelen);
4341                 lustre_cfg_bufs_set(n_bufs, 1, n_buf, o_buflen + diff);
4342                 n_buf += cfs_size_round(o_buflen + diff);
4343
4344                 /* buf[2] is the pool name, reuse it directly */
4345                 lustre_cfg_bufs_set(n_bufs, 2, lustre_cfg_buf(o_lcfg, 2),
4346                                     o_lcfg->lcfg_buflens[2]);
4347
4348                 if (cnt == 3)
4349                         break;
4350
4351                 /* buf[3] is ostname */
4352                 o_buf = lustre_cfg_buf(o_lcfg, 3);
4353                 o_buflen = o_lcfg->lcfg_buflens[3];
4354                 memcpy(n_buf, mlfd->mlfd_newname, n_namelen);
4355                 memcpy(n_buf + n_namelen, o_buf + o_namelen,
4356                        o_buflen - o_namelen);
4357                 lustre_cfg_bufs_set(n_bufs, 3, n_buf, o_buflen + diff);
4358                 break;
4359         }
4360         case LCFG_SETUP: {
4361                 if (cnt == 2) {
4362                         o_buflen = o_lcfg->lcfg_buflens[1];
4363                         if (o_buflen == sizeof(struct lov_desc) ||
4364                             o_buflen == sizeof(struct lmv_desc)) {
4365                                 char *o_uuid;
4366                                 char *n_uuid;
4367                                 int uuid_len;
4368
4369                                 /* buf[1] */
4370                                 o_buf = lustre_cfg_buf(o_lcfg, 1);
4371                                 if (o_buflen == sizeof(struct lov_desc)) {
4372                                         struct lov_desc *o_desc =
4373                                                 (struct lov_desc *)o_buf;
4374                                         struct lov_desc *n_desc =
4375                                                 (struct lov_desc *)n_buf;
4376
4377                                         *n_desc = *o_desc;
4378                                         o_uuid = o_desc->ld_uuid.uuid;
4379                                         n_uuid = n_desc->ld_uuid.uuid;
4380                                         uuid_len = sizeof(o_desc->ld_uuid.uuid);
4381                                 } else {
4382                                         struct lmv_desc *o_desc =
4383                                                 (struct lmv_desc *)o_buf;
4384                                         struct lmv_desc *n_desc =
4385                                                 (struct lmv_desc *)n_buf;
4386
4387                                         *n_desc = *o_desc;
4388                                         o_uuid = o_desc->ld_uuid.uuid;
4389                                         n_uuid = n_desc->ld_uuid.uuid;
4390                                         uuid_len = sizeof(o_desc->ld_uuid.uuid);
4391                                 }
4392
4393                                 if (unlikely(!contain_valid_fsname(o_uuid,
4394                                                 mlfd->mlfd_oldname, uuid_len,
4395                                                 o_namelen))) {
4396                                         lustre_cfg_bufs_set(n_bufs, 1, o_buf,
4397                                                             o_buflen);
4398                                         break;
4399                                 }
4400
4401                                 memcpy(n_uuid, mlfd->mlfd_newname, n_namelen);
4402                                 uuid_len = strlen(o_uuid);
4403                                 if (uuid_len > o_namelen)
4404                                         memcpy(n_uuid + n_namelen,
4405                                                o_uuid + o_namelen,
4406                                                uuid_len - o_namelen);
4407                                 n_uuid[uuid_len + diff] = '\0';
4408                                 lustre_cfg_bufs_set(n_bufs, 1, n_buf, o_buflen);
4409                                 break;
4410                         } /* else case fall through */
4411                 } /* else case fall through */
4412         }
4413         default: {
4414                 for (i = 1; i < cnt; i++) {
4415                         o_buflen = o_lcfg->lcfg_buflens[i];
4416                         if (o_buflen == 0)
4417                                 continue;
4418
4419                         o_buf = lustre_cfg_buf(o_lcfg, i);
4420                         if (!contain_valid_fsname(o_buf, mlfd->mlfd_oldname,
4421                                                   o_buflen, o_namelen)) {
4422                                 lustre_cfg_bufs_set(n_bufs, i, o_buf, o_buflen);
4423                                 continue;
4424                         }
4425
4426                         memcpy(n_buf, mlfd->mlfd_newname, n_namelen);
4427                         if (o_buflen == o_namelen) {
4428                                 lustre_cfg_bufs_set(n_bufs, i, n_buf,
4429                                                     n_namelen);
4430                                 n_buf += cfs_size_round(n_namelen);
4431                                 continue;
4432                         }
4433
4434                         memcpy(n_buf + n_namelen, o_buf + o_namelen,
4435                                o_buflen - o_namelen);
4436                         lustre_cfg_bufs_set(n_bufs, i, n_buf, o_buflen + diff);
4437                         n_buf += cfs_size_round(o_buflen + diff);
4438                 }
4439                 break;
4440         }
4441         }
4442
4443         lcr = lustre_cfg_rec_new(cmd, n_bufs);
4444         if (!lcr)
4445                 RETURN(-ENOMEM);
4446
4447         lcr->lcr_cfg = *o_lcfg;
4448         rc = llog_write(env, mlfd->mlfd_llh, &lcr->lcr_hdr, LLOG_NEXT_IDX);
4449         lustre_cfg_rec_free(lcr);
4450
4451         RETURN(rc);
4452 }
4453
4454 static int mgs_lcfg_fork_one(const struct lu_env *env, struct mgs_device *mgs,
4455                              struct mgs_direntry *mde, const char *oldname,
4456                              const char *newname)
4457 {
4458         struct llog_handle *old_llh = NULL;
4459         struct llog_handle *new_llh = NULL;
4460         struct llog_ctxt *ctxt = NULL;
4461         struct mgs_lcfg_fork_data *mlfd = NULL;
4462         char *name_buf = NULL;
4463         int name_buflen;
4464         int old_namelen = strlen(oldname);
4465         int new_namelen = strlen(newname);
4466         int rc;
4467         ENTRY;
4468
4469         name_buflen = mde->mde_len + new_namelen - old_namelen;
4470         OBD_ALLOC(name_buf, name_buflen);
4471         if (!name_buf)
4472                 RETURN(-ENOMEM);
4473
4474         memcpy(name_buf, newname, new_namelen);
4475         memcpy(name_buf + new_namelen, mde->mde_name + old_namelen,
4476                mde->mde_len - old_namelen);
4477
4478         CDEBUG(D_MGS, "Fork the config-log from %s to %s\n",
4479                mde->mde_name, name_buf);
4480
4481         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
4482         LASSERT(ctxt);
4483
4484         rc = llog_open_create(env, ctxt, &new_llh, NULL, name_buf);
4485         if (rc)
4486                 GOTO(out, rc);
4487
4488         rc = llog_init_handle(env, new_llh, LLOG_F_IS_PLAIN, NULL);
4489         if (rc)
4490                 GOTO(out, rc);
4491
4492         if (unlikely(mgs_log_is_empty(env, mgs, mde->mde_name)))
4493                 GOTO(out, rc = 0);
4494
4495         rc = llog_open(env, ctxt, &old_llh, NULL, mde->mde_name,
4496                        LLOG_OPEN_EXISTS);
4497         if (rc)
4498                 GOTO(out, rc);
4499
4500         rc = llog_init_handle(env, old_llh, LLOG_F_IS_PLAIN, NULL);
4501         if (rc)
4502                 GOTO(out, rc);
4503
4504         new_llh->lgh_hdr->llh_tgtuuid = old_llh->lgh_hdr->llh_tgtuuid;
4505
4506         OBD_ALLOC(mlfd, LLOG_MIN_CHUNK_SIZE);
4507         if (!mlfd)
4508                 GOTO(out, rc = -ENOMEM);
4509
4510         mlfd->mlfd_mgs = mgs;
4511         mlfd->mlfd_llh = new_llh;
4512         mlfd->mlfd_oldname = oldname;
4513         mlfd->mlfd_newname = newname;
4514
4515         rc = llog_process(env, old_llh, mgs_lcfg_fork_handler, mlfd, NULL);
4516         OBD_FREE(mlfd, LLOG_MIN_CHUNK_SIZE);
4517
4518         GOTO(out, rc);
4519
4520 out:
4521         if (old_llh)
4522                 llog_close(env, old_llh);
4523         if (new_llh)
4524                 llog_close(env, new_llh);
4525         if (name_buf)
4526                 OBD_FREE(name_buf, name_buflen);
4527         if (ctxt)
4528                 llog_ctxt_put(ctxt);
4529
4530         return rc;
4531 }
4532
4533 int mgs_lcfg_fork(const struct lu_env *env, struct mgs_device *mgs,
4534                   const char *oldname, const char *newname)
4535 {
4536         struct list_head log_list;
4537         struct mgs_direntry *dirent, *n;
4538         int olen = strlen(oldname);
4539         int nlen = strlen(newname);
4540         int count = 0;
4541         int rc = 0;
4542         ENTRY;
4543
4544         if (unlikely(!oldname || oldname[0] == '\0' ||
4545                      !newname || newname[0] == '\0'))
4546                 RETURN(-EINVAL);
4547
4548         if (strcmp(oldname, newname) == 0)
4549                 RETURN(-EINVAL);
4550
4551         /* lock it to prevent fork/erase/register in parallel. */
4552         mutex_lock(&mgs->mgs_mutex);
4553
4554         rc = class_dentry_readdir(env, mgs, &log_list);
4555         if (rc) {
4556                 mutex_unlock(&mgs->mgs_mutex);
4557                 RETURN(rc);
4558         }
4559
4560         if (list_empty(&log_list)) {
4561                 mutex_unlock(&mgs->mgs_mutex);
4562                 RETURN(-ENOENT);
4563         }
4564
4565         list_for_each_entry_safe(dirent, n, &log_list, mde_list) {
4566                 char *ptr;
4567
4568                 ptr = strrchr(dirent->mde_name, '-');
4569                 if (ptr) {
4570                         int tlen = ptr - dirent->mde_name;
4571
4572                         if (tlen == nlen &&
4573                             strncmp(newname, dirent->mde_name, tlen) == 0)
4574                                 GOTO(out, rc = -EEXIST);
4575
4576                         if (tlen == olen &&
4577                             strncmp(oldname, dirent->mde_name, tlen) == 0)
4578                                 continue;
4579                 }
4580
4581                 list_del_init(&dirent->mde_list);
4582                 mgs_direntry_free(dirent);
4583         }
4584
4585         if (list_empty(&log_list)) {
4586                 mutex_unlock(&mgs->mgs_mutex);
4587                 RETURN(-ENOENT);
4588         }
4589
4590         list_for_each_entry(dirent, &log_list, mde_list) {
4591                 rc = mgs_lcfg_fork_one(env, mgs, dirent, oldname, newname);
4592                 if (rc)
4593                         break;
4594
4595                 count++;
4596         }
4597
4598 out:
4599         mutex_unlock(&mgs->mgs_mutex);
4600
4601         list_for_each_entry_safe(dirent, n, &log_list, mde_list) {
4602                 list_del_init(&dirent->mde_list);
4603                 mgs_direntry_free(dirent);
4604         }
4605
4606         if (rc && count > 0)
4607                 mgs_erase_logs(env, mgs, newname);
4608
4609         RETURN(rc);
4610 }
4611
4612 int mgs_lcfg_erase(const struct lu_env *env, struct mgs_device *mgs,
4613                    const char *fsname)
4614 {
4615         int rc;
4616         ENTRY;
4617
4618         if (unlikely(!fsname || fsname[0] == '\0'))
4619                 RETURN(-EINVAL);
4620
4621         rc = mgs_erase_logs(env, mgs, fsname);
4622
4623         RETURN(rc);
4624 }
4625
4626 static int mgs_xattr_del(const struct lu_env *env, struct dt_object *obj)
4627 {
4628         struct dt_device *dev;
4629         struct thandle *th = NULL;
4630         int rc = 0;
4631
4632         ENTRY;
4633
4634         dev = container_of0(obj->do_lu.lo_dev, struct dt_device, dd_lu_dev);
4635         th = dt_trans_create(env, dev);
4636         if (IS_ERR(th))
4637                 RETURN(PTR_ERR(th));
4638
4639         rc = dt_declare_xattr_del(env, obj, XATTR_TARGET_RENAME, th);
4640         if (rc)
4641                 GOTO(stop, rc);
4642
4643         rc = dt_trans_start_local(env, dev, th);
4644         if (rc)
4645                 GOTO(stop, rc);
4646
4647         dt_write_lock(env, obj, 0);
4648         rc = dt_xattr_del(env, obj, XATTR_TARGET_RENAME, th);
4649
4650         GOTO(unlock, rc);
4651
4652 unlock:
4653         dt_write_unlock(env, obj);
4654
4655 stop:
4656         dt_trans_stop(env, dev, th);
4657
4658         return rc;
4659 }
4660
4661 int mgs_lcfg_rename(const struct lu_env *env, struct mgs_device *mgs)
4662 {
4663         struct list_head log_list;
4664         struct mgs_direntry *dirent, *n;
4665         char fsname[16];
4666         struct lu_buf buf = {
4667                 .lb_buf = fsname,
4668                 .lb_len = sizeof(fsname)
4669         };
4670         int rc = 0;
4671
4672         ENTRY;
4673
4674         rc = class_dentry_readdir(env, mgs, &log_list);
4675         if (rc)
4676                 RETURN(rc);
4677
4678         if (list_empty(&log_list))
4679                 RETURN(0);
4680
4681         list_for_each_entry_safe(dirent, n, &log_list, mde_list) {
4682                 struct dt_object *o = NULL;
4683                 char oldname[16];
4684                 char *ptr;
4685                 int len;
4686
4687                 list_del_init(&dirent->mde_list);
4688                 ptr = strrchr(dirent->mde_name, '-');
4689                 if (!ptr)
4690                         goto next;
4691
4692                 len = ptr - dirent->mde_name;
4693                 if (unlikely(len >= sizeof(oldname))) {
4694                         CDEBUG(D_MGS, "Skip invalid configuration file %s\n",
4695                                dirent->mde_name);
4696                         goto next;
4697                 }
4698
4699                 o = local_file_find(env, mgs->mgs_los, mgs->mgs_configs_dir,
4700                                     dirent->mde_name);
4701                 if (IS_ERR(o)) {
4702                         rc = PTR_ERR(o);
4703                         CDEBUG(D_MGS, "Fail to locate file %s: rc = %d\n",
4704                                dirent->mde_name, rc);
4705                         goto next;
4706                 }
4707
4708                 rc = dt_xattr_get(env, o, &buf, XATTR_TARGET_RENAME);
4709                 if (rc < 0) {
4710                         if (rc == -ENODATA)
4711                                 rc = 0;
4712                         else
4713                                 CDEBUG(D_MGS,
4714                                        "Fail to get EA for %s: rc = %d\n",
4715                                        dirent->mde_name, rc);
4716                         goto next;
4717                 }
4718
4719                 if (unlikely(rc == len &&
4720                              memcmp(fsname, dirent->mde_name, len) == 0)) {
4721                         /* The new fsname is the same as the old one. */
4722                         rc = mgs_xattr_del(env, o);
4723                         goto next;
4724                 }
4725
4726                 memcpy(oldname, dirent->mde_name, len);
4727                 oldname[len] = '\0';
4728                 fsname[rc] = '\0';
4729                 rc = mgs_lcfg_fork_one(env, mgs, dirent, oldname, fsname);
4730                 if (rc && rc != -EEXIST) {
4731                         CDEBUG(D_MGS, "Fail to fork %s: rc = %d\n",
4732                                dirent->mde_name, rc);
4733                         goto next;
4734                 }
4735
4736                 rc = mgs_erase_log(env, mgs, dirent->mde_name);
4737                 if (rc) {
4738                         CDEBUG(D_MGS, "Fail to erase old %s: rc = %d\n",
4739                                dirent->mde_name, rc);
4740                         /* keep it there if failed to remove it. */
4741                         rc = 0;
4742                 }
4743
4744 next:
4745                 if (o && !IS_ERR(o))
4746                         lu_object_put(env, &o->do_lu);
4747
4748                 mgs_direntry_free(dirent);
4749                 if (rc)
4750                         break;
4751         }
4752
4753         list_for_each_entry_safe(dirent, n, &log_list, mde_list) {
4754                 list_del_init(&dirent->mde_list);
4755                 mgs_direntry_free(dirent);
4756         }
4757
4758         RETURN(rc);
4759 }
4760
4761 /* Setup _mgs fsdb and log
4762  */
4763 int mgs__mgs_fsdb_setup(const struct lu_env *env, struct mgs_device *mgs)
4764 {
4765         struct fs_db *fsdb = NULL;
4766         int rc;
4767         ENTRY;
4768
4769         rc = mgs_find_or_make_fsdb(env, mgs, MGSSELF_NAME, &fsdb);
4770         if (!rc)
4771                 mgs_put_fsdb(mgs, fsdb);
4772
4773         RETURN(rc);
4774 }
4775
4776 /* Setup params fsdb and log
4777  */
4778 int mgs_params_fsdb_setup(const struct lu_env *env, struct mgs_device *mgs)
4779 {
4780         struct fs_db *fsdb = NULL;
4781         struct llog_handle *params_llh = NULL;
4782         int rc;
4783         ENTRY;
4784
4785         rc = mgs_find_or_make_fsdb(env, mgs, PARAMS_FILENAME, &fsdb);
4786         if (!rc) {
4787                 mutex_lock(&fsdb->fsdb_mutex);
4788                 rc = record_start_log(env, mgs, &params_llh, PARAMS_FILENAME);
4789                 if (!rc)
4790                         rc = record_end_log(env, &params_llh);
4791                 mutex_unlock(&fsdb->fsdb_mutex);
4792                 mgs_put_fsdb(mgs, fsdb);
4793         }
4794
4795         RETURN(rc);
4796 }
4797
4798 /* Cleanup params fsdb and log
4799  */
4800 int mgs_params_fsdb_cleanup(const struct lu_env *env, struct mgs_device *mgs)
4801 {
4802         int rc;
4803
4804         rc = mgs_erase_logs(env, mgs, PARAMS_FILENAME);
4805         return rc == -ENOENT ? 0 : rc;
4806 }
4807
4808 /**
4809  * Fill in the mgs_target_info based on data devname and param provide.
4810  *
4811  * @env         thread context
4812  * @mgs         mgs device
4813  * @mti         mgs target info. We want to set this based other paramters
4814  *              passed to this function. Once setup we write it to the config
4815  *              logs.
4816  * @devname     optional OBD device name
4817  * @param       string that contains both what tunable to set and the value to
4818  *              set it to.
4819  *
4820  * RETURN       0 for success
4821  *              negative error number on failure
4822  **/
4823 static int mgs_set_conf_param(const struct lu_env *env, struct mgs_device *mgs,
4824                               struct mgs_target_info *mti, const char *devname,
4825                               const char *param)
4826 {
4827         struct fs_db *fsdb = NULL;
4828         int dev_type;
4829         int rc = 0;
4830
4831         ENTRY;
4832         /* lustre, lustre-mdtlov, lustre-client, lustre-MDT0000 */
4833         if (!devname) {
4834                 size_t len;
4835
4836                 /* We have two possible cases here:
4837                  *
4838                  * 1) the device name embedded in the param:
4839                  *    lustre-OST0000.osc.max_dirty_mb=32
4840                  *
4841                  * 2) the file system name is embedded in
4842                  *    the param: lustre.sys.at.min=0
4843                  */
4844                 len = strcspn(param, ".=");
4845                 if (!len || param[len] == '=')
4846                         RETURN(-EINVAL);
4847
4848                 if (len >= sizeof(mti->mti_svname))
4849                         RETURN(-E2BIG);
4850
4851                 snprintf(mti->mti_svname, sizeof(mti->mti_svname),
4852                          "%.*s", (int)len, param);
4853                 param += len + 1;
4854         } else {
4855                 if (strlcpy(mti->mti_svname, devname, sizeof(mti->mti_svname)) >=
4856                     sizeof(mti->mti_svname))
4857                         RETURN(-E2BIG);
4858         }
4859
4860         if (!strlen(mti->mti_svname)) {
4861                 LCONSOLE_ERROR_MSG(0x14d, "No target specified: %s\n", param);
4862                 RETURN(-ENOSYS);
4863         }
4864
4865         dev_type = mgs_parse_devname(mti->mti_svname, mti->mti_fsname,
4866                                      &mti->mti_stripe_index);
4867         switch (dev_type) {
4868         /* For this case we have an invalid obd device name */
4869         case -ENXIO:
4870                 CDEBUG(D_MGS, "%s don't contain an index\n", mti->mti_svname);
4871                 strlcpy(mti->mti_fsname, mti->mti_svname, MTI_NAME_MAXLEN);
4872                 dev_type = 0;
4873                 break;
4874         /* Not an obd device, assume devname is the fsname.
4875          * User might of only provided fsname and not obd device
4876          */
4877         case -EINVAL:
4878                 CDEBUG(D_MGS, "%s is seen as a file system name\n", mti->mti_svname);
4879                 strlcpy(mti->mti_fsname, mti->mti_svname, MTI_NAME_MAXLEN);
4880                 dev_type = 0;
4881                 break;
4882         default:
4883                 if (dev_type < 0)
4884                         GOTO(out, rc = dev_type);
4885
4886                 /* param related to llite isn't allowed to set by OST or MDT */
4887                 if (dev_type & LDD_F_SV_TYPE_OST ||
4888                     dev_type & LDD_F_SV_TYPE_MDT) {
4889                         /* param related to llite isn't allowed to set by OST
4890                          * or MDT
4891                          */
4892                         if (!strncmp(param, PARAM_LLITE,
4893                                      sizeof(PARAM_LLITE) - 1))
4894                                 GOTO(out, rc = -EINVAL);
4895
4896                         /* Strip -osc or -mdc suffix from svname */
4897                         if (server_make_name(dev_type, mti->mti_stripe_index,
4898                                              mti->mti_fsname, mti->mti_svname,
4899                                              sizeof(mti->mti_svname)))
4900                                 GOTO(out, rc = -EINVAL);
4901                 }
4902                 break;
4903         }
4904
4905         if (strlcpy(mti->mti_params, param, sizeof(mti->mti_params)) >=
4906             sizeof(mti->mti_params))
4907                 GOTO(out, rc = -E2BIG);
4908
4909         CDEBUG(D_MGS, "set_conf_param fs='%s' device='%s' param='%s'\n",
4910                mti->mti_fsname, mti->mti_svname, mti->mti_params);
4911
4912         rc = mgs_find_or_make_fsdb(env, mgs, mti->mti_fsname, &fsdb);
4913         if (rc)
4914                 GOTO(out, rc);
4915
4916         if (!test_bit(FSDB_MGS_SELF, &fsdb->fsdb_flags) &&
4917             test_bit(FSDB_LOG_EMPTY, &fsdb->fsdb_flags)) {
4918                 CERROR("No filesystem targets for %s. cfg_device from lctl "
4919                        "is '%s'\n", mti->mti_fsname, mti->mti_svname);
4920                 mgs_unlink_fsdb(mgs, fsdb);
4921                 GOTO(out, rc = -EINVAL);
4922         }
4923
4924         /*
4925          * Revoke lock so everyone updates.  Should be alright if
4926          * someone was already reading while we were updating the logs,
4927          * so we don't really need to hold the lock while we're
4928          * writing (above).
4929          */
4930         mti->mti_flags = dev_type | LDD_F_PARAM;
4931         mutex_lock(&fsdb->fsdb_mutex);
4932         rc = mgs_write_log_param(env, mgs, fsdb, mti, mti->mti_params);
4933         mutex_unlock(&fsdb->fsdb_mutex);
4934         mgs_revoke_lock(mgs, fsdb, CONFIG_T_CONFIG);
4935
4936 out:
4937         if (fsdb)
4938                 mgs_put_fsdb(mgs, fsdb);
4939
4940         RETURN(rc);
4941 }
4942
4943 static int mgs_set_param2(const struct lu_env *env, struct mgs_device *mgs,
4944                           struct mgs_target_info *mti, const char *param)
4945 {
4946         struct fs_db *fsdb = NULL;
4947         int dev_type;
4948         size_t len;
4949         int rc;
4950
4951         if (strlcpy(mti->mti_params, param, sizeof(mti->mti_params)) >=
4952             sizeof(mti->mti_params))
4953                 GOTO(out, rc = -E2BIG);
4954
4955         /* obdname2fsname reports devname as an obd device */
4956         len = strcspn(param, ".=");
4957         if (len && param[len] != '=') {
4958                 char *ptr;
4959
4960                 param += len + 1;
4961                 ptr = strchr(param, '.');
4962
4963                 len = strlen(param);
4964                 if (ptr)
4965                         len -= strlen(ptr);
4966                 if (len >= sizeof(mti->mti_svname))
4967                         GOTO(out, rc = -E2BIG);
4968
4969                 snprintf(mti->mti_svname, sizeof(mti->mti_svname), "%.*s",
4970                         (int)len, param);
4971
4972                 obdname2fsname(mti->mti_svname, mti->mti_fsname,
4973                                sizeof(mti->mti_fsname));
4974         } else {
4975                 snprintf(mti->mti_svname, sizeof(mti->mti_svname), "general");
4976         }
4977
4978         CDEBUG(D_MGS, "set_param2 fs='%s' device='%s' param='%s'\n",
4979                mti->mti_fsname, mti->mti_svname, mti->mti_params);
4980
4981         /* The return value should be the device type i.e LDD_F_SV_TYPE_XXX.
4982          * A returned error tells us we don't have a target obd device.
4983          */
4984         dev_type = server_name2index(mti->mti_svname, &mti->mti_stripe_index,
4985                                      NULL);
4986         if (dev_type < 0)
4987                 dev_type = 0;
4988
4989         /* the return value should be the device type i.e LDD_F_SV_TYPE_XXX.
4990          * Strip -osc or -mdc suffix from svname
4991          */
4992         if ((dev_type & LDD_F_SV_TYPE_OST || dev_type & LDD_F_SV_TYPE_MDT) &&
4993             server_make_name(dev_type, mti->mti_stripe_index,
4994                              mti->mti_fsname, mti->mti_svname,
4995                              sizeof(mti->mti_svname)))
4996                 GOTO(out, rc = -EINVAL);
4997
4998         rc = mgs_find_or_make_fsdb(env, mgs, PARAMS_FILENAME, &fsdb);
4999         if (rc)
5000                 GOTO(out, rc);
5001         /*
5002          * Revoke lock so everyone updates.  Should be alright if
5003          * someone was already reading while we were updating the logs,
5004          * so we don't really need to hold the lock while we're
5005          * writing (above).
5006          */
5007         mti->mti_flags = dev_type | LDD_F_PARAM2;
5008         mutex_lock(&fsdb->fsdb_mutex);
5009         rc = mgs_write_log_param2(env, mgs, fsdb, mti, mti->mti_params);
5010         mutex_unlock(&fsdb->fsdb_mutex);
5011         mgs_revoke_lock(mgs, fsdb, CONFIG_T_PARAMS);
5012         mgs_put_fsdb(mgs, fsdb);
5013 out:
5014         RETURN(rc);
5015 }
5016
5017 /* Set a permanent (config log) param for a target or fs
5018  *
5019  * @lcfg buf0 may contain the device (testfs-MDT0000) name
5020  *       buf1 contains the single parameter
5021  */
5022 int mgs_set_param(const struct lu_env *env, struct mgs_device *mgs,
5023                   struct lustre_cfg *lcfg)
5024 {
5025         const char *param = lustre_cfg_string(lcfg, 1);
5026         struct mgs_target_info *mti;
5027         int rc;
5028
5029         /* Create a fake mti to hold everything */
5030         OBD_ALLOC_PTR(mti);
5031         if (!mti)
5032                 return -ENOMEM;
5033
5034         print_lustre_cfg(lcfg);
5035
5036         if (lcfg->lcfg_command == LCFG_PARAM) {
5037                 /* For the case of lctl conf_param devname can be
5038                  * lustre, lustre-mdtlov, lustre-client, lustre-MDT0000
5039                  */
5040                 const char *devname = lustre_cfg_string(lcfg, 0);
5041
5042                 rc = mgs_set_conf_param(env, mgs, mti, devname, param);
5043         } else {
5044                 /* In the case of lctl set_param -P lcfg[0] will always
5045                  * be 'general'. At least for now.
5046                  */
5047                 rc = mgs_set_param2(env, mgs, mti, param);
5048         }
5049
5050         OBD_FREE_PTR(mti);
5051
5052         return rc;
5053 }
5054
5055 static int mgs_write_log_pool(const struct lu_env *env,
5056                               struct mgs_device *mgs, char *logname,
5057                               struct fs_db *fsdb, char *tgtname,
5058                               enum lcfg_command_type cmd,
5059                               char *fsname, char *poolname,
5060                               char *ostname, char *comment)
5061 {
5062         struct llog_handle *llh = NULL;
5063         int rc;
5064
5065         rc = record_start_log(env, mgs, &llh, logname);
5066         if (rc)
5067                 return rc;
5068         rc = record_marker(env, llh, fsdb, CM_START, tgtname, comment);
5069         if (rc)
5070                 goto out;
5071         rc = record_base(env, llh, tgtname, 0, cmd,
5072                          fsname, poolname, ostname, NULL);
5073         if (rc)
5074                 goto out;
5075         rc = record_marker(env, llh, fsdb, CM_END, tgtname, comment);
5076 out:
5077         record_end_log(env, &llh);
5078         return rc;
5079 }
5080
5081 int mgs_nodemap_cmd(const struct lu_env *env, struct mgs_device *mgs,
5082                     enum lcfg_command_type cmd, const char *nodemap_name,
5083                     char *param)
5084 {
5085         lnet_nid_t      nid[2];
5086         __u32           idmap[2];
5087         bool            bool_switch;
5088         __u32           int_id;
5089         int             rc = 0;
5090         ENTRY;
5091
5092         switch (cmd) {
5093         case LCFG_NODEMAP_ADD:
5094                 rc = nodemap_add(nodemap_name);
5095                 break;
5096         case LCFG_NODEMAP_DEL:
5097                 rc = nodemap_del(nodemap_name);
5098                 break;
5099         case LCFG_NODEMAP_ADD_RANGE:
5100                 rc = nodemap_parse_range(param, nid);
5101                 if (rc != 0)
5102                         break;
5103                 rc = nodemap_add_range(nodemap_name, nid);
5104                 break;
5105         case LCFG_NODEMAP_DEL_RANGE:
5106                 rc = nodemap_parse_range(param, nid);
5107                 if (rc != 0)
5108                         break;
5109                 rc = nodemap_del_range(nodemap_name, nid);
5110                 break;
5111         case LCFG_NODEMAP_ADMIN:
5112                 bool_switch = simple_strtoul(param, NULL, 10);
5113                 rc = nodemap_set_allow_root(nodemap_name, bool_switch);
5114                 break;
5115         case LCFG_NODEMAP_DENY_UNKNOWN:
5116                 bool_switch = simple_strtoul(param, NULL, 10);
5117                 rc = nodemap_set_deny_unknown(nodemap_name, bool_switch);
5118                 break;
5119         case LCFG_NODEMAP_AUDIT_MODE:
5120                 rc = kstrtoul(param, 10, (unsigned long *)&bool_switch);
5121                 if (rc == 0)
5122                         rc = nodemap_set_audit_mode(nodemap_name, bool_switch);
5123                 break;
5124         case LCFG_NODEMAP_MAP_MODE:
5125                 if (strcmp("both", param) == 0)
5126                         rc = nodemap_set_mapping_mode(nodemap_name,
5127                                                       NODEMAP_MAP_BOTH);
5128                 else if (strcmp("uid_only", param) == 0)
5129                         rc = nodemap_set_mapping_mode(nodemap_name,
5130                                                       NODEMAP_MAP_UID_ONLY);
5131                 else if (strcmp("gid_only", param) == 0)
5132                         rc = nodemap_set_mapping_mode(nodemap_name,
5133                                                       NODEMAP_MAP_GID_ONLY);
5134                 else
5135                         rc = -EINVAL;
5136                 break;
5137         case LCFG_NODEMAP_TRUSTED:
5138                 bool_switch = simple_strtoul(param, NULL, 10);
5139                 rc = nodemap_set_trust_client_ids(nodemap_name, bool_switch);
5140                 break;
5141         case LCFG_NODEMAP_SQUASH_UID:
5142                 int_id = simple_strtoul(param, NULL, 10);
5143                 rc = nodemap_set_squash_uid(nodemap_name, int_id);
5144                 break;
5145         case LCFG_NODEMAP_SQUASH_GID:
5146                 int_id = simple_strtoul(param, NULL, 10);
5147                 rc = nodemap_set_squash_gid(nodemap_name, int_id);
5148                 break;
5149         case LCFG_NODEMAP_ADD_UIDMAP:
5150         case LCFG_NODEMAP_ADD_GIDMAP:
5151                 rc = nodemap_parse_idmap(param, idmap);
5152                 if (rc != 0)
5153                         break;
5154                 if (cmd == LCFG_NODEMAP_ADD_UIDMAP)
5155                         rc = nodemap_add_idmap(nodemap_name, NODEMAP_UID,
5156                                                idmap);
5157                 else
5158                         rc = nodemap_add_idmap(nodemap_name, NODEMAP_GID,
5159                                                idmap);
5160                 break;
5161         case LCFG_NODEMAP_DEL_UIDMAP:
5162         case LCFG_NODEMAP_DEL_GIDMAP:
5163                 rc = nodemap_parse_idmap(param, idmap);
5164                 if (rc != 0)
5165                         break;
5166                 if (cmd == LCFG_NODEMAP_DEL_UIDMAP)
5167                         rc = nodemap_del_idmap(nodemap_name, NODEMAP_UID,
5168                                                idmap);
5169                 else
5170                         rc = nodemap_del_idmap(nodemap_name, NODEMAP_GID,
5171                                                idmap);
5172                 break;
5173         case LCFG_NODEMAP_SET_FILESET:
5174                 rc = nodemap_set_fileset(nodemap_name, param);
5175                 break;
5176         default:
5177                 rc = -EINVAL;
5178         }
5179
5180         RETURN(rc);
5181 }
5182
5183 int mgs_pool_cmd(const struct lu_env *env, struct mgs_device *mgs,
5184                  enum lcfg_command_type cmd, char *fsname,
5185                  char *poolname, char *ostname)
5186 {
5187         struct fs_db *fsdb;
5188         char *lovname;
5189         char *logname;
5190         char *label = NULL, *canceled_label = NULL;
5191         int label_sz;
5192         struct mgs_target_info *mti = NULL;
5193         bool checked = false;
5194         bool locked = false;
5195         bool free = false;
5196         int rc, i;
5197         ENTRY;
5198
5199         rc = mgs_find_or_make_fsdb(env, mgs, fsname, &fsdb);
5200         if (rc) {
5201                 CERROR("Can't get db for %s\n", fsname);
5202                 RETURN(rc);
5203         }
5204         if (test_bit(FSDB_LOG_EMPTY, &fsdb->fsdb_flags)) {
5205                 CERROR("%s is not defined\n", fsname);
5206                 free = true;
5207                 GOTO(out_fsdb, rc = -EINVAL);
5208         }
5209
5210         label_sz = 10 + strlen(fsname) + strlen(poolname);
5211
5212         /* check if ostname match fsname */
5213         if (ostname != NULL) {
5214                 char *ptr;
5215
5216                 ptr = strrchr(ostname, '-');
5217                 if ((ptr == NULL) ||
5218                     (strncmp(fsname, ostname, ptr-ostname) != 0))
5219                         RETURN(-EINVAL);
5220                 label_sz += strlen(ostname);
5221         }
5222
5223         OBD_ALLOC(label, label_sz);
5224         if (!label)
5225                 GOTO(out_fsdb, rc = -ENOMEM);
5226
5227         switch(cmd) {
5228         case LCFG_POOL_NEW:
5229                 sprintf(label,
5230                         "new %s.%s", fsname, poolname);
5231                 break;
5232         case LCFG_POOL_ADD:
5233                 sprintf(label,
5234                         "add %s.%s.%s", fsname, poolname, ostname);
5235                 break;
5236         case LCFG_POOL_REM:
5237                 OBD_ALLOC(canceled_label, label_sz);
5238                 if (canceled_label == NULL)
5239                         GOTO(out_label, rc = -ENOMEM);
5240                 sprintf(label,
5241                         "rem %s.%s.%s", fsname, poolname, ostname);
5242                 sprintf(canceled_label,
5243                         "add %s.%s.%s", fsname, poolname, ostname);
5244                 break;
5245         case LCFG_POOL_DEL:
5246                 OBD_ALLOC(canceled_label, label_sz);
5247                 if (canceled_label == NULL)
5248                         GOTO(out_label, rc = -ENOMEM);
5249                 sprintf(label,
5250                         "del %s.%s", fsname, poolname);
5251                 sprintf(canceled_label,
5252                         "new %s.%s", fsname, poolname);
5253                 break;
5254         default:
5255                 break;
5256         }
5257
5258         OBD_ALLOC_PTR(mti);
5259         if (mti == NULL)
5260                 GOTO(out_cancel, rc = -ENOMEM);
5261         strncpy(mti->mti_svname, "lov pool", sizeof(mti->mti_svname));
5262
5263         mutex_lock(&fsdb->fsdb_mutex);
5264         locked = true;
5265         /* write pool def to all MDT logs */
5266         for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
5267                 if (test_bit(i,  fsdb->fsdb_mdt_index_map)) {
5268                         rc = name_create_mdt_and_lov(&logname, &lovname,
5269                                                      fsdb, i);
5270                         if (rc)
5271                                 GOTO(out_mti, rc);
5272
5273                         if (!checked && (canceled_label == NULL)) {
5274                                 rc = mgs_check_marker(env, mgs, fsdb, mti,
5275                                                 logname, lovname, label);
5276                                 if (rc) {
5277                                         name_destroy(&logname);
5278                                         name_destroy(&lovname);
5279                                         GOTO(out_mti,
5280                                                 rc = (rc == LLOG_PROC_BREAK ?
5281                                                         -EEXIST : rc));
5282                                 }
5283                                 checked = true;
5284                         }
5285                         if (canceled_label != NULL)
5286                                 rc = mgs_modify(env, mgs, fsdb, mti, logname,
5287                                                 lovname, canceled_label,
5288                                                 CM_SKIP);
5289
5290                         if (rc >= 0)
5291                                 rc = mgs_write_log_pool(env, mgs, logname,
5292                                                         fsdb, lovname, cmd,
5293                                                         fsname, poolname,
5294                                                         ostname, label);
5295                         name_destroy(&logname);
5296                         name_destroy(&lovname);
5297                         if (rc)
5298                                 GOTO(out_mti, rc);
5299                 }
5300         }
5301
5302         rc = name_create(&logname, fsname, "-client");
5303         if (rc)
5304                 GOTO(out_mti, rc);
5305
5306         if (!checked && (canceled_label == NULL)) {
5307                 rc = mgs_check_marker(env, mgs, fsdb, mti, logname,
5308                                 fsdb->fsdb_clilov, label);
5309                 if (rc) {
5310                         name_destroy(&logname);
5311                         GOTO(out_mti, rc = (rc == LLOG_PROC_BREAK ?
5312                                 -EEXIST : rc));
5313                 }
5314         }
5315         if (canceled_label != NULL) {
5316                 rc = mgs_modify(env, mgs, fsdb, mti, logname,
5317                                 fsdb->fsdb_clilov, canceled_label, CM_SKIP);
5318                 if (rc < 0) {
5319                         name_destroy(&logname);
5320                         GOTO(out_mti, rc);
5321                 }
5322         }
5323
5324         rc = mgs_write_log_pool(env, mgs, logname, fsdb, fsdb->fsdb_clilov,
5325                                 cmd, fsname, poolname, ostname, label);
5326         mutex_unlock(&fsdb->fsdb_mutex);
5327         locked = false;
5328         name_destroy(&logname);
5329         /* request for update */
5330         mgs_revoke_lock(mgs, fsdb, CONFIG_T_CONFIG);
5331
5332         GOTO(out_mti, rc);
5333
5334 out_mti:
5335         if (locked)
5336                 mutex_unlock(&fsdb->fsdb_mutex);
5337         if (mti != NULL)
5338                 OBD_FREE_PTR(mti);
5339 out_cancel:
5340         if (canceled_label != NULL)
5341                 OBD_FREE(canceled_label, label_sz);
5342 out_label:
5343         OBD_FREE(label, label_sz);
5344 out_fsdb:
5345         if (free)
5346                 mgs_unlink_fsdb(mgs, fsdb);
5347         mgs_put_fsdb(mgs, fsdb);
5348
5349         return rc;
5350 }