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