lustre/mgs/mgs_llog.c

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