lustre/mgs/mgs_llog.c

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