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, 2016, 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_ioctl.h>
  46 #include <uapi/linux/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 ? cfs_time_current_sec() : 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 = cfs_time_current_sec();
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                 int index;
2144
2145                 if (sscanf(lustre_cfg_buf(lcfg, 2), "%d", &index) != 1)
2146                         RETURN (-EINVAL);
2147
2148                 memcpy(tmti->mti_fsname, mti->mti_fsname,
2149                        strlen(mti->mti_fsname));
2150                 tmti->mti_stripe_index = index;
2151
2152                 rc = mgs_write_log_osp_to_mdt(env, mgs, fsdb, tmti,
2153                                               mti->mti_stripe_index,
2154                                               mti->mti_svname);
2155                 memset(tmti, 0, sizeof(*tmti));
2156                 RETURN(rc);
2157         }
2158
2159         if (lcfg->lcfg_command == LCFG_LOV_ADD_OBD) {
2160                 int index;
2161                 char mdt_index[9];
2162                 char *logname, *lovname;
2163
2164                 rc = name_create_mdt_and_lov(&logname, &lovname, fsdb,
2165                                              mti->mti_stripe_index);
2166                 if (rc)
2167                         RETURN(rc);
2168                 sprintf(mdt_index, "-MDT%04x", mti->mti_stripe_index);
2169
2170                 if (sscanf(lustre_cfg_buf(lcfg, 2), "%d", &index) != 1) {
2171                         name_destroy(&logname);
2172                         name_destroy(&lovname);
2173                         RETURN(-EINVAL);
2174                 }
2175
2176                 tmti->mti_stripe_index = index;
2177                 rc = mgs_write_log_osc_to_lov(env, mgs, fsdb, tmti, logname,
2178                                          mdt_index, lovname,
2179                                          LUSTRE_SP_MDT, 0);
2180                 name_destroy(&logname);
2181                 name_destroy(&lovname);
2182                 RETURN(rc);
2183         }
2184         RETURN(rc);
2185 }
2186
2187 /* fsdb->fsdb_mutex is already held  in mgs_write_log_target*/
2188 /* stealed from mgs_get_fsdb_from_llog*/
2189 static int mgs_steal_llog_for_mdt_from_client(const struct lu_env *env,
2190                                               struct mgs_device *mgs,
2191                                               char *client_name,
2192                                               struct temp_comp* comp)
2193 {
2194         struct llog_handle *loghandle;
2195         struct mgs_target_info *tmti;
2196         struct llog_ctxt *ctxt;
2197         int rc;
2198
2199         ENTRY;
2200
2201         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
2202         LASSERT(ctxt != NULL);
2203
2204         OBD_ALLOC_PTR(tmti);
2205         if (tmti == NULL)
2206                 GOTO(out_ctxt, rc = -ENOMEM);
2207
2208         comp->comp_tmti = tmti;
2209         comp->comp_obd = mgs->mgs_obd;
2210
2211         rc = llog_open(env, ctxt, &loghandle, NULL, client_name,
2212                        LLOG_OPEN_EXISTS);
2213         if (rc < 0) {
2214                 if (rc == -ENOENT)
2215                         rc = 0;
2216                 GOTO(out_pop, rc);
2217         }
2218
2219         rc = llog_init_handle(env, loghandle, LLOG_F_IS_PLAIN, NULL);
2220         if (rc)
2221                 GOTO(out_close, rc);
2222
2223         rc = llog_process_or_fork(env, loghandle, mgs_steal_client_llog_handler,
2224                                   (void *)comp, NULL, false);
2225         CDEBUG(D_MGS, "steal llog re = %d\n", rc);
2226 out_close:
2227         llog_close(env, loghandle);
2228 out_pop:
2229         OBD_FREE_PTR(tmti);
2230 out_ctxt:
2231         llog_ctxt_put(ctxt);
2232         RETURN(rc);
2233 }
2234
2235 /* lmv is the second thing for client logs */
2236 /* copied from mgs_write_log_lov. Please refer to that.  */
2237 static int mgs_write_log_lmv(const struct lu_env *env,
2238                              struct mgs_device *mgs,
2239                              struct fs_db *fsdb,
2240                              struct mgs_target_info *mti,
2241                              char *logname, char *lmvname)
2242 {
2243         struct llog_handle *llh = NULL;
2244         struct lmv_desc *lmvdesc;
2245         char *uuid;
2246         int rc = 0;
2247         ENTRY;
2248
2249         CDEBUG(D_MGS, "Writing lmv(%s) log for %s\n", lmvname,logname);
2250
2251         OBD_ALLOC_PTR(lmvdesc);
2252         if (lmvdesc == NULL)
2253                 RETURN(-ENOMEM);
2254         lmvdesc->ld_active_tgt_count = 0;
2255         lmvdesc->ld_tgt_count = 0;
2256         sprintf((char*)lmvdesc->ld_uuid.uuid, "%s_UUID", lmvname);
2257         uuid = (char *)lmvdesc->ld_uuid.uuid;
2258
2259         rc = record_start_log(env, mgs, &llh, logname);
2260         if (rc)
2261                 GOTO(out_free, rc);
2262         rc = record_marker(env, llh, fsdb, CM_START, lmvname, "lmv setup");
2263         if (rc)
2264                 GOTO(out_end, rc);
2265         rc = record_attach(env, llh, lmvname, "lmv", uuid);
2266         if (rc)
2267                 GOTO(out_end, rc);
2268         rc = record_lmv_setup(env, llh, lmvname, lmvdesc);
2269         if (rc)
2270                 GOTO(out_end, rc);
2271         rc = record_marker(env, llh, fsdb, CM_END, lmvname, "lmv setup");
2272         if (rc)
2273                 GOTO(out_end, rc);
2274 out_end:
2275         record_end_log(env, &llh);
2276 out_free:
2277         OBD_FREE_PTR(lmvdesc);
2278         RETURN(rc);
2279 }
2280
2281 /* lov is the first thing in the mdt and client logs */
2282 static int mgs_write_log_lov(const struct lu_env *env, struct mgs_device *mgs,
2283                              struct fs_db *fsdb, struct mgs_target_info *mti,
2284                              char *logname, char *lovname)
2285 {
2286         struct llog_handle *llh = NULL;
2287         struct lov_desc *lovdesc;
2288         char *uuid;
2289         int rc = 0;
2290         ENTRY;
2291
2292         CDEBUG(D_MGS, "Writing lov(%s) log for %s\n", lovname, logname);
2293
2294         /*
2295         #01 L attach   0:lov_mdsA  1:lov  2:71ccb_lov_mdsA_19f961a9e1
2296         #02 L lov_setup 0:lov_mdsA 1:(struct lov_desc)
2297               uuid=lov1_UUID, stripe count=1, size=1048576, offset=0, pattern=0
2298         */
2299
2300         /* FIXME just make lov_setup accept empty desc (put uuid in buf 2) */
2301         OBD_ALLOC_PTR(lovdesc);
2302         if (lovdesc == NULL)
2303                 RETURN(-ENOMEM);
2304         lovdesc->ld_magic = LOV_DESC_MAGIC;
2305         lovdesc->ld_tgt_count = 0;
2306         /* Defaults.  Can be changed later by lcfg config_param */
2307         lovdesc->ld_default_stripe_count = 1;
2308         lovdesc->ld_pattern = LOV_PATTERN_RAID0;
2309         lovdesc->ld_default_stripe_size = LOV_DESC_STRIPE_SIZE_DEFAULT;
2310         lovdesc->ld_default_stripe_offset = -1;
2311         lovdesc->ld_qos_maxage = LOV_DESC_QOS_MAXAGE_DEFAULT;
2312         sprintf((char*)lovdesc->ld_uuid.uuid, "%s_UUID", lovname);
2313         /* can these be the same? */
2314         uuid = (char *)lovdesc->ld_uuid.uuid;
2315
2316         /* This should always be the first entry in a log.
2317         rc = mgs_clear_log(obd, logname); */
2318         rc = record_start_log(env, mgs, &llh, logname);
2319         if (rc)
2320                 GOTO(out_free, rc);
2321         /* FIXME these should be a single journal transaction */
2322         rc = record_marker(env, llh, fsdb, CM_START, lovname, "lov setup");
2323         if (rc)
2324                 GOTO(out_end, rc);
2325         rc = record_attach(env, llh, lovname, "lov", uuid);
2326         if (rc)
2327                 GOTO(out_end, rc);
2328         rc = record_lov_setup(env, llh, lovname, lovdesc);
2329         if (rc)
2330                 GOTO(out_end, rc);
2331         rc = record_marker(env, llh, fsdb, CM_END, lovname, "lov setup");
2332         if (rc)
2333                 GOTO(out_end, rc);
2334         EXIT;
2335 out_end:
2336         record_end_log(env, &llh);
2337 out_free:
2338         OBD_FREE_PTR(lovdesc);
2339         return rc;
2340 }
2341
2342 /* add failnids to open log */
2343 static int mgs_write_log_failnids(const struct lu_env *env,
2344                                   struct mgs_target_info *mti,
2345                                   struct llog_handle *llh,
2346                                   char *cliname)
2347 {
2348         char *failnodeuuid = NULL;
2349         char *ptr = mti->mti_params;
2350         lnet_nid_t nid;
2351         int rc = 0;
2352
2353         /*
2354         #03 L add_uuid  nid=uml1@tcp(0x20000c0a80201) nal=90 0:  1:uml1_UUID
2355         #04 L add_uuid  nid=1@elan(0x1000000000001)   nal=90 0:  1:uml1_UUID
2356         #05 L setup    0:OSC_uml1_ost1_mdsA  1:ost1_UUID  2:uml1_UUID
2357         #06 L add_uuid  nid=uml2@tcp(0x20000c0a80202) nal=90 0:  1:uml2_UUID
2358         #0x L add_uuid  nid=2@elan(0x1000000000002)   nal=90 0:  1:uml2_UUID
2359         #07 L add_conn 0:OSC_uml1_ost1_mdsA  1:uml2_UUID
2360         */
2361
2362         /*
2363          * Pull failnid info out of params string, which may contain something
2364          * like "<nid1>,<nid2>:<nid3>,<nid4>".  class_parse_nid() does not
2365          * complain about abnormal inputs like ",:<nid1>", "<nid1>:,<nid2>",
2366          * etc.  However, convert_hostnames() should have caught those.
2367          */
2368         while (class_find_param(ptr, PARAM_FAILNODE, &ptr) == 0) {
2369                 while (class_parse_nid(ptr, &nid, &ptr) == 0) {
2370                         char nidstr[LNET_NIDSTR_SIZE];
2371
2372                         if (failnodeuuid == NULL) {
2373                                 /* We don't know the failover node name,
2374                                  * so just use the first nid as the uuid */
2375                                 libcfs_nid2str_r(nid, nidstr, sizeof(nidstr));
2376                                 rc = name_create(&failnodeuuid, nidstr, "");
2377                                 if (rc != 0)
2378                                         return rc;
2379                         }
2380                         CDEBUG(D_MGS, "add nid %s for failover uuid %s, "
2381                                 "client %s\n",
2382                                 libcfs_nid2str_r(nid, nidstr, sizeof(nidstr)),
2383                                 failnodeuuid, cliname);
2384                         rc = record_add_uuid(env, llh, nid, failnodeuuid);
2385                         /*
2386                          * If *ptr is ':', we have added all NIDs for
2387                          * failnodeuuid.
2388                          */
2389                         if (*ptr == ':') {
2390                                 rc = record_add_conn(env, llh, cliname,
2391                                                      failnodeuuid);
2392                                 name_destroy(&failnodeuuid);
2393                                 failnodeuuid = NULL;
2394                         }
2395                 }
2396                 if (failnodeuuid) {
2397                         rc = record_add_conn(env, llh, cliname, failnodeuuid);
2398                         name_destroy(&failnodeuuid);
2399                         failnodeuuid = NULL;
2400                 }
2401         }
2402
2403         return rc;
2404 }
2405
2406 static int mgs_write_log_mdc_to_lmv(const struct lu_env *env,
2407                                     struct mgs_device *mgs,
2408                                     struct fs_db *fsdb,
2409                                     struct mgs_target_info *mti,
2410                                     char *logname, char *lmvname)
2411 {
2412         struct llog_handle *llh = NULL;
2413         char *mdcname = NULL;
2414         char *nodeuuid = NULL;
2415         char *mdcuuid = NULL;
2416         char *lmvuuid = NULL;
2417         char index[6];
2418         char nidstr[LNET_NIDSTR_SIZE];
2419         int i, rc;
2420         ENTRY;
2421
2422         if (mgs_log_is_empty(env, mgs, logname)) {
2423                 CERROR("log is empty! Logical error\n");
2424                 RETURN(-EINVAL);
2425         }
2426
2427         CDEBUG(D_MGS, "adding mdc for %s to log %s:lmv(%s)\n",
2428                mti->mti_svname, logname, lmvname);
2429
2430         libcfs_nid2str_r(mti->mti_nids[0], nidstr, sizeof(nidstr));
2431         rc = name_create(&nodeuuid, nidstr, "");
2432         if (rc)
2433                 RETURN(rc);
2434         rc = name_create(&mdcname, mti->mti_svname, "-mdc");
2435         if (rc)
2436                 GOTO(out_free, rc);
2437         rc = name_create(&mdcuuid, mdcname, "_UUID");
2438         if (rc)
2439                 GOTO(out_free, rc);
2440         rc = name_create(&lmvuuid, lmvname, "_UUID");
2441         if (rc)
2442                 GOTO(out_free, rc);
2443
2444         rc = record_start_log(env, mgs, &llh, logname);
2445         if (rc)
2446                 GOTO(out_free, rc);
2447         rc = record_marker(env, llh, fsdb, CM_START, mti->mti_svname,
2448                            "add mdc");
2449         if (rc)
2450                 GOTO(out_end, rc);
2451         for (i = 0; i < mti->mti_nid_count; i++) {
2452                 CDEBUG(D_MGS, "add nid %s for mdt\n",
2453                         libcfs_nid2str_r(mti->mti_nids[i],
2454                                          nidstr, sizeof(nidstr)));
2455
2456                 rc = record_add_uuid(env, llh, mti->mti_nids[i], nodeuuid);
2457                 if (rc)
2458                         GOTO(out_end, rc);
2459         }
2460
2461         rc = record_attach(env, llh, mdcname, LUSTRE_MDC_NAME, lmvuuid);
2462         if (rc)
2463                 GOTO(out_end, rc);
2464         rc = record_setup(env, llh, mdcname, mti->mti_uuid, nodeuuid,
2465                           NULL, NULL);
2466         if (rc)
2467                 GOTO(out_end, rc);
2468         rc = mgs_write_log_failnids(env, mti, llh, mdcname);
2469         if (rc)
2470                 GOTO(out_end, rc);
2471         snprintf(index, sizeof(index), "%d", mti->mti_stripe_index);
2472         rc = record_mdc_add(env, llh, lmvname, mdcuuid, mti->mti_uuid,
2473                             index, "1");
2474         if (rc)
2475                 GOTO(out_end, rc);
2476         rc = record_marker(env, llh, fsdb, CM_END, mti->mti_svname,
2477                            "add mdc");
2478         if (rc)
2479                 GOTO(out_end, rc);
2480 out_end:
2481         record_end_log(env, &llh);
2482 out_free:
2483         name_destroy(&lmvuuid);
2484         name_destroy(&mdcuuid);
2485         name_destroy(&mdcname);
2486         name_destroy(&nodeuuid);
2487         RETURN(rc);
2488 }
2489
2490 static inline int name_create_lov(char **lovname, char *mdtname,
2491                                   struct fs_db *fsdb, int index)
2492 {
2493         /* COMPAT_180 */
2494         if (index == 0 && test_bit(FSDB_OSCNAME18, &fsdb->fsdb_flags))
2495                 return name_create(lovname, fsdb->fsdb_name, "-mdtlov");
2496         else
2497                 return name_create(lovname, mdtname, "-mdtlov");
2498 }
2499
2500 static int name_create_mdt_and_lov(char **logname, char **lovname,
2501                                    struct fs_db *fsdb, int i)
2502 {
2503         int rc;
2504
2505         rc = name_create_mdt(logname, fsdb->fsdb_name, i);
2506         if (rc)
2507                 return rc;
2508         /* COMPAT_180 */
2509         if (i == 0 && test_bit(FSDB_OSCNAME18, &fsdb->fsdb_flags))
2510                 rc = name_create(lovname, fsdb->fsdb_name, "-mdtlov");
2511         else
2512                 rc = name_create(lovname, *logname, "-mdtlov");
2513         if (rc) {
2514                 name_destroy(logname);
2515                 *logname = NULL;
2516         }
2517         return rc;
2518 }
2519
2520 static inline int name_create_mdt_osc(char **oscname, char *ostname,
2521                                       struct fs_db *fsdb, int i)
2522 {
2523         char suffix[16];
2524
2525         if (i == 0 && test_bit(FSDB_OSCNAME18, &fsdb->fsdb_flags))
2526                 sprintf(suffix, "-osc");
2527         else
2528                 sprintf(suffix, "-osc-MDT%04x", i);
2529         return name_create(oscname, ostname, suffix);
2530 }
2531
2532 /* add new mdc to already existent MDS */
2533 static int mgs_write_log_osp_to_mdt(const struct lu_env *env,
2534                                     struct mgs_device *mgs,
2535                                     struct fs_db *fsdb,
2536                                     struct mgs_target_info *mti,
2537                                     int mdt_index, char *logname)
2538 {
2539         struct llog_handle      *llh = NULL;
2540         char    *nodeuuid = NULL;
2541         char    *ospname = NULL;
2542         char    *lovuuid = NULL;
2543         char    *mdtuuid = NULL;
2544         char    *svname = NULL;
2545         char    *mdtname = NULL;
2546         char    *lovname = NULL;
2547         char    index_str[16];
2548         char    nidstr[LNET_NIDSTR_SIZE];
2549         int     i, rc;
2550
2551         ENTRY;
2552         if (mgs_log_is_empty(env, mgs, mti->mti_svname)) {
2553                 CERROR("log is empty! Logical error\n");
2554                 RETURN (-EINVAL);
2555         }
2556
2557         CDEBUG(D_MGS, "adding osp index %d to %s\n", mti->mti_stripe_index,
2558                logname);
2559
2560         rc = name_create_mdt(&mdtname, fsdb->fsdb_name, mti->mti_stripe_index);
2561         if (rc)
2562                 RETURN(rc);
2563
2564         libcfs_nid2str_r(mti->mti_nids[0], nidstr, sizeof(nidstr));
2565         rc = name_create(&nodeuuid, nidstr, "");
2566         if (rc)
2567                 GOTO(out_destory, rc);
2568
2569         rc = name_create(&svname, mdtname, "-osp");
2570         if (rc)
2571                 GOTO(out_destory, rc);
2572
2573         sprintf(index_str, "-MDT%04x", mdt_index);
2574         rc = name_create(&ospname, svname, index_str);
2575         if (rc)
2576                 GOTO(out_destory, rc);
2577
2578         rc = name_create_lov(&lovname, logname, fsdb, mdt_index);
2579         if (rc)
2580                 GOTO(out_destory, rc);
2581
2582         rc = name_create(&lovuuid, lovname, "_UUID");
2583         if (rc)
2584                 GOTO(out_destory, rc);
2585
2586         rc = name_create(&mdtuuid, mdtname, "_UUID");
2587         if (rc)
2588                 GOTO(out_destory, rc);
2589
2590         rc = record_start_log(env, mgs, &llh, logname);
2591         if (rc)
2592                 GOTO(out_destory, rc);
2593
2594         rc = record_marker(env, llh, fsdb, CM_START, mti->mti_svname,
2595                            "add osp");
2596         if (rc)
2597                 GOTO(out_destory, rc);
2598
2599         for (i = 0; i < mti->mti_nid_count; i++) {
2600                 CDEBUG(D_MGS, "add nid %s for mdt\n",
2601                         libcfs_nid2str_r(mti->mti_nids[i],
2602                                          nidstr, sizeof(nidstr)));
2603                 rc = record_add_uuid(env, llh, mti->mti_nids[i], nodeuuid);
2604                 if (rc)
2605                         GOTO(out_end, rc);
2606         }
2607
2608         rc = record_attach(env, llh, ospname, LUSTRE_OSP_NAME, lovuuid);
2609         if (rc)
2610                 GOTO(out_end, rc);
2611
2612         rc = record_setup(env, llh, ospname, mti->mti_uuid, nodeuuid,
2613                           NULL, NULL);
2614         if (rc)
2615                 GOTO(out_end, rc);
2616
2617         rc = mgs_write_log_failnids(env, mti, llh, ospname);
2618         if (rc)
2619                 GOTO(out_end, rc);
2620
2621         /* Add mdc(osp) to lod */
2622         snprintf(index_str, sizeof(index_str), "%d", mti->mti_stripe_index);
2623         rc = record_base(env, llh, lovname, 0, LCFG_ADD_MDC, mti->mti_uuid,
2624                          index_str, "1", NULL);
2625         if (rc)
2626                 GOTO(out_end, rc);
2627
2628         rc = record_marker(env, llh, fsdb, CM_END, mti->mti_svname, "add osp");
2629         if (rc)
2630                 GOTO(out_end, rc);
2631
2632 out_end:
2633         record_end_log(env, &llh);
2634
2635 out_destory:
2636         name_destroy(&mdtuuid);
2637         name_destroy(&lovuuid);
2638         name_destroy(&lovname);
2639         name_destroy(&ospname);
2640         name_destroy(&svname);
2641         name_destroy(&nodeuuid);
2642         name_destroy(&mdtname);
2643         RETURN(rc);
2644 }
2645
2646 static int mgs_write_log_mdt0(const struct lu_env *env,
2647                               struct mgs_device *mgs,
2648                               struct fs_db *fsdb,
2649                               struct mgs_target_info *mti)
2650 {
2651         char *log = mti->mti_svname;
2652         struct llog_handle *llh = NULL;
2653         char *uuid, *lovname;
2654         char mdt_index[6];
2655         char *ptr = mti->mti_params;
2656         int rc = 0, failout = 0;
2657         ENTRY;
2658
2659         OBD_ALLOC(uuid, sizeof(struct obd_uuid));
2660         if (uuid == NULL)
2661                 RETURN(-ENOMEM);
2662
2663         if (class_find_param(ptr, PARAM_FAILMODE, &ptr) == 0)
2664                 failout = (strncmp(ptr, "failout", 7) == 0);
2665
2666         rc = name_create(&lovname, log, "-mdtlov");
2667         if (rc)
2668                 GOTO(out_free, rc);
2669         if (mgs_log_is_empty(env, mgs, log)) {
2670                 rc = mgs_write_log_lov(env, mgs, fsdb, mti, log, lovname);
2671                 if (rc)
2672                         GOTO(out_lod, rc);
2673         }
2674
2675         sprintf(mdt_index, "%d", mti->mti_stripe_index);
2676
2677         rc = record_start_log(env, mgs, &llh, log);
2678         if (rc)
2679                 GOTO(out_lod, rc);
2680
2681         /* add MDT itself */
2682
2683         /* FIXME this whole fn should be a single journal transaction */
2684         sprintf(uuid, "%s_UUID", log);
2685         rc = record_marker(env, llh, fsdb, CM_START, log, "add mdt");
2686         if (rc)
2687                 GOTO(out_lod, rc);
2688         rc = record_attach(env, llh, log, LUSTRE_MDT_NAME, uuid);
2689         if (rc)
2690                 GOTO(out_end, rc);
2691         rc = record_mount_opt(env, llh, log, lovname, NULL);
2692         if (rc)
2693                 GOTO(out_end, rc);
2694         rc = record_setup(env, llh, log, uuid, mdt_index, lovname,
2695                         failout ? "n" : "f");
2696         if (rc)
2697                 GOTO(out_end, rc);
2698         rc = record_marker(env, llh, fsdb, CM_END, log, "add mdt");
2699         if (rc)
2700                 GOTO(out_end, rc);
2701 out_end:
2702         record_end_log(env, &llh);
2703 out_lod:
2704         name_destroy(&lovname);
2705 out_free:
2706         OBD_FREE(uuid, sizeof(struct obd_uuid));
2707         RETURN(rc);
2708 }
2709
2710 /* envelope method for all layers log */
2711 static int mgs_write_log_mdt(const struct lu_env *env,
2712                              struct mgs_device *mgs,
2713                              struct fs_db *fsdb,
2714                              struct mgs_target_info *mti)
2715 {
2716         struct mgs_thread_info *mgi = mgs_env_info(env);
2717         struct llog_handle *llh = NULL;
2718         char *cliname;
2719         int rc, i = 0;
2720         ENTRY;
2721
2722         CDEBUG(D_MGS, "writing new mdt %s\n", mti->mti_svname);
2723
2724         if (mti->mti_uuid[0] == '\0') {
2725                 /* Make up our own uuid */
2726                 snprintf(mti->mti_uuid, sizeof(mti->mti_uuid),
2727                          "%s_UUID", mti->mti_svname);
2728         }
2729
2730         /* add mdt */
2731         rc = mgs_write_log_mdt0(env, mgs, fsdb, mti);
2732         if (rc)
2733                 RETURN(rc);
2734         /* Append the mdt info to the client log */
2735         rc = name_create(&cliname, mti->mti_fsname, "-client");
2736         if (rc)
2737                 RETURN(rc);
2738
2739         if (mgs_log_is_empty(env, mgs, cliname)) {
2740                 /* Start client log */
2741                 rc = mgs_write_log_lov(env, mgs, fsdb, mti, cliname,
2742                                        fsdb->fsdb_clilov);
2743                 if (rc)
2744                         GOTO(out_free, rc);
2745                 rc = mgs_write_log_lmv(env, mgs, fsdb, mti, cliname,
2746                                        fsdb->fsdb_clilmv);
2747                 if (rc)
2748                         GOTO(out_free, rc);
2749         }
2750
2751         /*
2752         #09 L add_uuid nid=uml1@tcp(0x20000c0a80201) 0:  1:uml1_UUID
2753         #10 L attach   0:MDC_uml1_mdsA_MNT_client  1:mdc  2:1d834_MNT_client_03f
2754         #11 L setup    0:MDC_uml1_mdsA_MNT_client  1:mdsA_UUID  2:uml1_UUID
2755         #12 L add_uuid nid=uml2@tcp(0x20000c0a80202) 0:  1:uml2_UUID
2756         #13 L add_conn 0:MDC_uml1_mdsA_MNT_client  1:uml2_UUID
2757         #14 L mount_option 0:  1:client  2:lov1  3:MDC_uml1_mdsA_MNT_client
2758         */
2759
2760         /* copy client info about lov/lmv */
2761         mgi->mgi_comp.comp_mti = mti;
2762         mgi->mgi_comp.comp_fsdb = fsdb;
2763
2764         rc = mgs_steal_llog_for_mdt_from_client(env, mgs, cliname,
2765                                                 &mgi->mgi_comp);
2766         if (rc)
2767                 GOTO(out_free, rc);
2768         rc = mgs_write_log_mdc_to_lmv(env, mgs, fsdb, mti, cliname,
2769                                       fsdb->fsdb_clilmv);
2770         if (rc)
2771                 GOTO(out_free, rc);
2772
2773         /* add mountopts */
2774         rc = record_start_log(env, mgs, &llh, cliname);
2775         if (rc)
2776                 GOTO(out_free, rc);
2777
2778         rc = record_marker(env, llh, fsdb, CM_START, cliname,
2779                            "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,
2787                            "mount opts");
2788
2789         if (rc)
2790                 GOTO(out_end, rc);
2791
2792         /* for_all_existing_mdt except current one */
2793         for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
2794                 if (i !=  mti->mti_stripe_index &&
2795                     test_bit(i, fsdb->fsdb_mdt_index_map)) {
2796                         char *logname;
2797
2798                         rc = name_create_mdt(&logname, fsdb->fsdb_name, i);
2799                         if (rc)
2800                                 GOTO(out_end, rc);
2801
2802                         /* NB: If the log for the MDT is empty, it means
2803                          * the MDT is only added to the index
2804                          * map, and not being process yet, i.e. this
2805                          * is an unregistered MDT, see mgs_write_log_target().
2806                          * so we should skip it. Otherwise
2807                          *
2808                          * 1. MGS get register request for MDT1 and MDT2.
2809                          *
2810                          * 2. Then both MDT1 and MDT2 are added into
2811                          * fsdb_mdt_index_map. (see mgs_set_index()).
2812                          *
2813                          * 3. Then MDT1 get the lock of fsdb_mutex, then
2814                          * generate the config log, here, it will regard MDT2
2815                          * as an existent MDT, and generate "add osp" for
2816                          * lustre-MDT0001-osp-MDT0002. Note: at the moment
2817                          * MDT0002 config log is still empty, so it will
2818                          * add "add osp" even before "lov setup", which
2819                          * will definitly cause trouble.
2820                          *
2821                          * 4. MDT1 registeration finished, fsdb_mutex is
2822                          * released, then MDT2 get in, then in above
2823                          * mgs_steal_llog_for_mdt_from_client(), it will
2824                          * add another osp log for lustre-MDT0001-osp-MDT0002,
2825                          * which will cause another trouble.*/
2826                         if (!mgs_log_is_empty(env, mgs, logname))
2827                                 rc = mgs_write_log_osp_to_mdt(env, mgs, fsdb,
2828                                                               mti, i, logname);
2829
2830                         name_destroy(&logname);
2831                         if (rc)
2832                                 GOTO(out_end, rc);
2833                 }
2834         }
2835 out_end:
2836         record_end_log(env, &llh);
2837 out_free:
2838         name_destroy(&cliname);
2839         RETURN(rc);
2840 }
2841
2842 /* Add the ost info to the client/mdt lov */
2843 static int mgs_write_log_osc_to_lov(const struct lu_env *env,
2844                                     struct mgs_device *mgs, struct fs_db *fsdb,
2845                                     struct mgs_target_info *mti,
2846                                     char *logname, char *suffix, char *lovname,
2847                                     enum lustre_sec_part sec_part, int flags)
2848 {
2849         struct llog_handle *llh = NULL;
2850         char *nodeuuid = NULL;
2851         char *oscname = NULL;
2852         char *oscuuid = NULL;
2853         char *lovuuid = NULL;
2854         char *svname = NULL;
2855         char index[6];
2856         char nidstr[LNET_NIDSTR_SIZE];
2857         int i, rc;
2858         ENTRY;
2859
2860         CDEBUG(D_INFO, "adding osc for %s to log %s\n",
2861                 mti->mti_svname, logname);
2862
2863         if (mgs_log_is_empty(env, mgs, logname)) {
2864                 CERROR("log is empty! Logical error\n");
2865                 RETURN(-EINVAL);
2866         }
2867
2868         libcfs_nid2str_r(mti->mti_nids[0], nidstr, sizeof(nidstr));
2869         rc = name_create(&nodeuuid, nidstr, "");
2870         if (rc)
2871                 RETURN(rc);
2872         rc = name_create(&svname, mti->mti_svname, "-osc");
2873         if (rc)
2874                 GOTO(out_free, rc);
2875
2876         /* for the system upgraded from old 1.8, keep using the old osc naming
2877          * style for mdt, see name_create_mdt_osc(). LU-1257 */
2878         if (test_bit(FSDB_OSCNAME18, &fsdb->fsdb_flags))
2879                 rc = name_create(&oscname, svname, "");
2880         else
2881                 rc = name_create(&oscname, svname, suffix);
2882         if (rc)
2883                 GOTO(out_free, rc);
2884
2885         rc = name_create(&oscuuid, oscname, "_UUID");
2886         if (rc)
2887                 GOTO(out_free, rc);
2888         rc = name_create(&lovuuid, lovname, "_UUID");
2889         if (rc)
2890                 GOTO(out_free, rc);
2891
2892
2893         /*
2894         #03 L add_uuid nid=uml1@tcp(0x20000c0a80201) 0:  1:uml1_UUID
2895         multihomed (#4)
2896         #04 L add_uuid  nid=1@elan(0x1000000000001)  nal=90 0:  1:uml1_UUID
2897         #04 L attach   0:OSC_uml1_ost1_MNT_client  1:osc  2:89070_lov1_a41dff51a
2898         #05 L setup    0:OSC_uml1_ost1_MNT_client  1:ost1_UUID  2:uml1_UUID
2899         failover (#6,7)
2900         #06 L add_uuid nid=uml2@tcp(0x20000c0a80202) 0:  1:uml2_UUID
2901         #07 L add_conn 0:OSC_uml1_ost1_MNT_client  1:uml2_UUID
2902         #08 L lov_modify_tgts add 0:lov1  1:ost1_UUID  2(index):0  3(gen):1
2903         */
2904
2905         rc = record_start_log(env, mgs, &llh, logname);
2906         if (rc)
2907                 GOTO(out_free, rc);
2908
2909         /* FIXME these should be a single journal transaction */
2910         rc = record_marker(env, llh, fsdb, CM_START | flags, mti->mti_svname,
2911                            "add osc");
2912         if (rc)
2913                 GOTO(out_end, rc);
2914
2915         /* NB: don't change record order, because upon MDT steal OSC config
2916          * from client, it treats all nids before LCFG_SETUP as target nids
2917          * (multiple interfaces), while nids after as failover node nids.
2918          * See mgs_steal_client_llog_handler() LCFG_ADD_UUID.
2919          */
2920         for (i = 0; i < mti->mti_nid_count; i++) {
2921                 CDEBUG(D_MGS, "add nid %s\n",
2922                         libcfs_nid2str_r(mti->mti_nids[i],
2923                                          nidstr, sizeof(nidstr)));
2924                 rc = record_add_uuid(env, llh, mti->mti_nids[i], nodeuuid);
2925                 if (rc)
2926                         GOTO(out_end, rc);
2927         }
2928         rc = record_attach(env, llh, oscname, LUSTRE_OSC_NAME, lovuuid);
2929         if (rc)
2930                 GOTO(out_end, rc);
2931         rc = record_setup(env, llh, oscname, mti->mti_uuid, nodeuuid,
2932                           NULL, NULL);
2933         if (rc)
2934                 GOTO(out_end, rc);
2935         rc = mgs_write_log_failnids(env, mti, llh, oscname);
2936         if (rc)
2937                 GOTO(out_end, rc);
2938
2939         snprintf(index, sizeof(index), "%d", mti->mti_stripe_index);
2940
2941         rc = record_lov_add(env, llh, lovname, mti->mti_uuid, index, "1");
2942         if (rc)
2943                 GOTO(out_end, rc);
2944         rc = record_marker(env, llh, fsdb, CM_END | flags, mti->mti_svname,
2945                            "add osc");
2946         if (rc)
2947                 GOTO(out_end, rc);
2948 out_end:
2949         record_end_log(env, &llh);
2950 out_free:
2951         name_destroy(&lovuuid);
2952         name_destroy(&oscuuid);
2953         name_destroy(&oscname);
2954         name_destroy(&svname);
2955         name_destroy(&nodeuuid);
2956         RETURN(rc);
2957 }
2958
2959 static int mgs_write_log_ost(const struct lu_env *env,
2960                              struct mgs_device *mgs, struct fs_db *fsdb,
2961                              struct mgs_target_info *mti)
2962 {
2963         struct llog_handle *llh = NULL;
2964         char *logname, *lovname;
2965         char *ptr = mti->mti_params;
2966         int rc, flags = 0, failout = 0, i;
2967         ENTRY;
2968
2969         CDEBUG(D_MGS, "writing new ost %s\n", mti->mti_svname);
2970
2971         /* The ost startup log */
2972
2973         /* If the ost log already exists, that means that someone reformatted
2974            the ost and it called target_add again. */
2975         if (!mgs_log_is_empty(env, mgs, mti->mti_svname)) {
2976                 LCONSOLE_ERROR_MSG(0x141, "The config log for %s already "
2977                                    "exists, yet the server claims it never "
2978                                    "registered. It may have been reformatted, "
2979                                    "or the index changed. writeconf the MDT to "
2980                                    "regenerate all logs.\n", mti->mti_svname);
2981                 RETURN(-EALREADY);
2982         }
2983
2984         /*
2985         attach obdfilter ost1 ost1_UUID
2986         setup /dev/loop2 ldiskfs f|n errors=remount-ro,user_xattr
2987         */
2988         if (class_find_param(ptr, PARAM_FAILMODE, &ptr) == 0)
2989                 failout = (strncmp(ptr, "failout", 7) == 0);
2990         rc = record_start_log(env, mgs, &llh, mti->mti_svname);
2991         if (rc)
2992                 RETURN(rc);
2993         /* FIXME these should be a single journal transaction */
2994         rc = record_marker(env, llh, fsdb, CM_START, mti->mti_svname,"add ost");
2995         if (rc)
2996                 GOTO(out_end, rc);
2997         if (*mti->mti_uuid == '\0')
2998                 snprintf(mti->mti_uuid, sizeof(mti->mti_uuid),
2999                          "%s_UUID", mti->mti_svname);
3000         rc = record_attach(env, llh, mti->mti_svname,
3001                            "obdfilter"/*LUSTRE_OST_NAME*/, mti->mti_uuid);
3002         if (rc)
3003                 GOTO(out_end, rc);
3004         rc = record_setup(env, llh, mti->mti_svname,
3005                           "dev"/*ignored*/, "type"/*ignored*/,
3006                           failout ? "n" : "f", NULL/*options*/);
3007         if (rc)
3008                 GOTO(out_end, rc);
3009         rc = record_marker(env, llh, fsdb, CM_END, mti->mti_svname, "add ost");
3010         if (rc)
3011                 GOTO(out_end, rc);
3012 out_end:
3013         record_end_log(env, &llh);
3014         if (rc)
3015                 RETURN(rc);
3016         /* We also have to update the other logs where this osc is part of
3017            the lov */
3018
3019         if (test_bit(FSDB_OLDLOG14, &fsdb->fsdb_flags)) {
3020                 /* If we're upgrading, the old mdt log already has our
3021                    entry. Let's do a fake one for fun. */
3022                 /* Note that we can't add any new failnids, since we don't
3023                    know the old osc names. */
3024                 flags = CM_SKIP | CM_UPGRADE146;
3025
3026         } else if ((mti->mti_flags & LDD_F_UPDATE) != LDD_F_UPDATE) {
3027                 /* If the update flag isn't set, don't update client/mdt
3028                    logs. */
3029                 flags |= CM_SKIP;
3030                 LCONSOLE_WARN("Client log for %s was not updated; writeconf "
3031                               "the MDT first to regenerate it.\n",
3032                               mti->mti_svname);
3033         }
3034
3035         /* Add ost to all MDT lov defs */
3036         for (i = 0; i < INDEX_MAP_SIZE * 8; i++){
3037                 if (test_bit(i, fsdb->fsdb_mdt_index_map)) {
3038                         char mdt_index[9];
3039
3040                         rc = name_create_mdt_and_lov(&logname, &lovname, fsdb,
3041                                                      i);
3042                         if (rc)
3043                                 RETURN(rc);
3044                         sprintf(mdt_index, "-MDT%04x", i);
3045                         rc = mgs_write_log_osc_to_lov(env, mgs, fsdb, mti,
3046                                                       logname, mdt_index,
3047                                                       lovname, LUSTRE_SP_MDT,
3048                                                       flags);
3049                         name_destroy(&logname);
3050                         name_destroy(&lovname);
3051                         if (rc)
3052                                 RETURN(rc);
3053                 }
3054         }
3055
3056         /* Append ost info to the client log */
3057         rc = name_create(&logname, mti->mti_fsname, "-client");
3058         if (rc)
3059                 RETURN(rc);
3060         if (mgs_log_is_empty(env, mgs, logname)) {
3061                 /* Start client log */
3062                 rc = mgs_write_log_lov(env, mgs, fsdb, mti, logname,
3063                                        fsdb->fsdb_clilov);
3064                 if (rc)
3065                         GOTO(out_free, rc);
3066                 rc = mgs_write_log_lmv(env, mgs, fsdb, mti, logname,
3067                                        fsdb->fsdb_clilmv);
3068                 if (rc)
3069                         GOTO(out_free, rc);
3070         }
3071         rc = mgs_write_log_osc_to_lov(env, mgs, fsdb, mti, logname, "",
3072                                       fsdb->fsdb_clilov, LUSTRE_SP_CLI, flags);
3073 out_free:
3074         name_destroy(&logname);
3075         RETURN(rc);
3076 }
3077
3078 static __inline__ int mgs_param_empty(char *ptr)
3079 {
3080         char *tmp;
3081
3082         if ((tmp = strchr(ptr, '=')) && (*(++tmp) == '\0'))
3083                 return 1;
3084         return 0;
3085 }
3086
3087 static int mgs_write_log_failnid_internal(const struct lu_env *env,
3088                                           struct mgs_device *mgs,
3089                                           struct fs_db *fsdb,
3090                                           struct mgs_target_info *mti,
3091                                           char *logname, char *cliname)
3092 {
3093         int rc;
3094         struct llog_handle *llh = NULL;
3095
3096         if (mgs_param_empty(mti->mti_params)) {
3097                 /* Remove _all_ failnids */
3098                 rc = mgs_modify(env, mgs, fsdb, mti, logname,
3099                                 mti->mti_svname, "add failnid", CM_SKIP);
3100                 return rc < 0 ? rc : 0;
3101         }
3102
3103         /* Otherwise failover nids are additive */
3104         rc = record_start_log(env, mgs, &llh, logname);
3105         if (rc)
3106                 return rc;
3107                 /* FIXME this should be a single journal transaction */
3108         rc = record_marker(env, llh, fsdb, CM_START, mti->mti_svname,
3109                            "add failnid");
3110         if (rc)
3111                 goto out_end;
3112         rc = mgs_write_log_failnids(env, mti, llh, cliname);
3113         if (rc)
3114                 goto out_end;
3115         rc = record_marker(env, llh, fsdb, CM_END,
3116                            mti->mti_svname, "add failnid");
3117 out_end:
3118         record_end_log(env, &llh);
3119         return rc;
3120 }
3121
3122
3123 /* Add additional failnids to an existing log.
3124    The mdc/osc must have been added to logs first */
3125 /* tcp nids must be in dotted-quad ascii -
3126    we can't resolve hostnames from the kernel. */
3127 static int mgs_write_log_add_failnid(const struct lu_env *env,
3128                                      struct mgs_device *mgs,
3129                                      struct fs_db *fsdb,
3130                                      struct mgs_target_info *mti)
3131 {
3132         char *logname, *cliname;
3133         int rc;
3134         ENTRY;
3135
3136         /* FIXME we currently can't erase the failnids
3137          * given when a target first registers, since they aren't part of
3138          * an "add uuid" stanza
3139          */
3140
3141         /* Verify that we know about this target */
3142         if (mgs_log_is_empty(env, mgs, mti->mti_svname)) {
3143                 LCONSOLE_ERROR_MSG(0x142, "The target %s has not registered "
3144                                    "yet. It must be started before failnids "
3145                                    "can be added.\n", mti->mti_svname);
3146                 RETURN(-ENOENT);
3147         }
3148
3149         /* Create mdc/osc client name (e.g. lustre-OST0001-osc) */
3150         if (mti->mti_flags & LDD_F_SV_TYPE_MDT) {
3151                 rc = name_create(&cliname, mti->mti_svname, "-mdc");
3152         } else if (mti->mti_flags & LDD_F_SV_TYPE_OST) {
3153                 rc = name_create(&cliname, mti->mti_svname, "-osc");
3154         } else {
3155                 RETURN(-EINVAL);
3156         }
3157         if (rc)
3158                 RETURN(rc);
3159
3160         /* Add failover nids to the client log */
3161         rc = name_create(&logname, mti->mti_fsname, "-client");
3162         if (rc) {
3163                 name_destroy(&cliname);
3164                 RETURN(rc);
3165         }
3166
3167         rc = mgs_write_log_failnid_internal(env, mgs, fsdb,mti,logname,cliname);
3168         name_destroy(&logname);
3169         name_destroy(&cliname);
3170         if (rc)
3171                 RETURN(rc);
3172
3173         if (mti->mti_flags & LDD_F_SV_TYPE_OST) {
3174                 /* Add OST failover nids to the MDT logs as well */
3175                 int i;
3176
3177                 for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
3178                         if (!test_bit(i, fsdb->fsdb_mdt_index_map))
3179                                 continue;
3180                         rc = name_create_mdt(&logname, mti->mti_fsname, i);
3181                         if (rc)
3182                                 RETURN(rc);
3183                         rc = name_create_mdt_osc(&cliname, mti->mti_svname,
3184                                                  fsdb, i);
3185                         if (rc) {
3186                                 name_destroy(&logname);
3187                                 RETURN(rc);
3188                         }
3189                         rc = mgs_write_log_failnid_internal(env, mgs, fsdb,
3190                                                             mti, logname,
3191                                                             cliname);
3192                         name_destroy(&cliname);
3193                         name_destroy(&logname);
3194                         if (rc)
3195                                 RETURN(rc);
3196                 }
3197         }
3198
3199         RETURN(rc);
3200 }
3201
3202 static int mgs_wlp_lcfg(const struct lu_env *env,
3203                         struct mgs_device *mgs, struct fs_db *fsdb,
3204                         struct mgs_target_info *mti,
3205                         char *logname, struct lustre_cfg_bufs *bufs,
3206                         char *tgtname, char *ptr)
3207 {
3208         char comment[MTI_NAME_MAXLEN];
3209         char *tmp;
3210         struct llog_cfg_rec *lcr;
3211         int rc, del;
3212
3213         /* Erase any old settings of this same parameter */
3214         memcpy(comment, ptr, MTI_NAME_MAXLEN);
3215         comment[MTI_NAME_MAXLEN - 1] = 0;
3216         /* But don't try to match the value. */
3217         tmp = strchr(comment, '=');
3218         if (tmp != NULL)
3219                 *tmp = 0;
3220         /* FIXME we should skip settings that are the same as old values */
3221         rc = mgs_modify(env, mgs, fsdb, mti, logname, tgtname, comment,CM_SKIP);
3222         if (rc < 0)
3223                 return rc;
3224         del = mgs_param_empty(ptr);
3225
3226         LCONSOLE_INFO("%s parameter %s.%s in log %s\n", del ? "Disabling" : rc ?
3227                       "Setting" : "Modifying", tgtname, comment, logname);
3228         if (del) {
3229                 /* mgs_modify() will return 1 if nothing had to be done */
3230                 if (rc == 1)
3231                         rc = 0;
3232                 return rc;
3233         }
3234
3235         lustre_cfg_bufs_reset(bufs, tgtname);
3236         lustre_cfg_bufs_set_string(bufs, 1, ptr);
3237         if (mti->mti_flags & LDD_F_PARAM2)
3238                 lustre_cfg_bufs_set_string(bufs, 2, LCTL_UPCALL);
3239
3240         lcr = lustre_cfg_rec_new((mti->mti_flags & LDD_F_PARAM2) ?
3241                                  LCFG_SET_PARAM : LCFG_PARAM, bufs);
3242         if (lcr == NULL)
3243                 return -ENOMEM;
3244
3245         rc = mgs_write_log_direct(env, mgs, fsdb, logname, lcr, tgtname,
3246                                   comment);
3247         lustre_cfg_rec_free(lcr);
3248         return rc;
3249 }
3250
3251 /* write global variable settings into log */
3252 static int mgs_write_log_sys(const struct lu_env *env,
3253                              struct mgs_device *mgs, struct fs_db *fsdb,
3254                              struct mgs_target_info *mti, char *sys, char *ptr)
3255 {
3256         struct mgs_thread_info  *mgi = mgs_env_info(env);
3257         struct lustre_cfg       *lcfg;
3258         struct llog_cfg_rec     *lcr;
3259         char *tmp, sep;
3260         int rc, cmd, convert = 1;
3261
3262         if (class_match_param(ptr, PARAM_TIMEOUT, &tmp) == 0) {
3263                 cmd = LCFG_SET_TIMEOUT;
3264         } else if (class_match_param(ptr, PARAM_LDLM_TIMEOUT, &tmp) == 0) {
3265                 cmd = LCFG_SET_LDLM_TIMEOUT;
3266         /* Check for known params here so we can return error to lctl */
3267         } else if ((class_match_param(ptr, PARAM_AT_MIN, &tmp) == 0) ||
3268                 (class_match_param(ptr, PARAM_AT_MAX, &tmp) == 0) ||
3269                 (class_match_param(ptr, PARAM_AT_EXTRA, &tmp) == 0) ||
3270                 (class_match_param(ptr, PARAM_AT_EARLY_MARGIN, &tmp) == 0) ||
3271                 (class_match_param(ptr, PARAM_AT_HISTORY, &tmp) == 0)) {
3272                 cmd = LCFG_PARAM;
3273         } else if (class_match_param(ptr, PARAM_JOBID_VAR, &tmp) == 0) {
3274                 convert = 0; /* Don't convert string value to integer */
3275                 cmd = LCFG_PARAM;
3276         } else {
3277                 return -EINVAL;
3278         }
3279
3280         if (mgs_param_empty(ptr))
3281                 CDEBUG(D_MGS, "global '%s' removed\n", sys);
3282         else
3283                 CDEBUG(D_MGS, "global '%s' val=%s\n", sys, tmp);
3284
3285         lustre_cfg_bufs_reset(&mgi->mgi_bufs, NULL);
3286         lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 1, sys);
3287         if (!convert && *tmp != '\0')
3288                 lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 2, tmp);
3289         lcr = lustre_cfg_rec_new(cmd, &mgi->mgi_bufs);
3290         if (lcr == NULL)
3291                 return -ENOMEM;
3292
3293         lcfg = &lcr->lcr_cfg;
3294         lcfg->lcfg_num = convert ? simple_strtoul(tmp, NULL, 0) : 0;
3295         /* truncate the comment to the parameter name */
3296         ptr = tmp - 1;
3297         sep = *ptr;
3298         *ptr = '\0';
3299         /* modify all servers and clients */
3300         rc = mgs_write_log_direct_all(env, mgs, fsdb, mti,
3301                                       *tmp == '\0' ? NULL : lcr,
3302                                       mti->mti_fsname, sys, 0);
3303         if (rc == 0 && *tmp != '\0') {
3304                 switch (cmd) {
3305                 case LCFG_SET_TIMEOUT:
3306                         if (!obd_timeout_set || lcfg->lcfg_num > obd_timeout)
3307                                 class_process_config(lcfg);
3308                         break;
3309                 case LCFG_SET_LDLM_TIMEOUT:
3310                         if (!ldlm_timeout_set || lcfg->lcfg_num > ldlm_timeout)
3311                                 class_process_config(lcfg);
3312                         break;
3313                 default:
3314                         break;
3315                 }
3316         }
3317         *ptr = sep;
3318         lustre_cfg_rec_free(lcr);
3319         return rc;
3320 }
3321
3322 /* write quota settings into log */
3323 static int mgs_write_log_quota(const struct lu_env *env, struct mgs_device *mgs,
3324                                struct fs_db *fsdb, struct mgs_target_info *mti,
3325                                char *quota, char *ptr)
3326 {
3327         struct mgs_thread_info  *mgi = mgs_env_info(env);
3328         struct llog_cfg_rec     *lcr;
3329         char                    *tmp;
3330         char                     sep;
3331         int                      rc, cmd = LCFG_PARAM;
3332
3333         /* support only 'meta' and 'data' pools so far */
3334         if (class_match_param(ptr, QUOTA_METAPOOL_NAME, &tmp) != 0 &&
3335             class_match_param(ptr, QUOTA_DATAPOOL_NAME, &tmp) != 0) {
3336                 CERROR("parameter quota.%s isn't supported (only quota.mdt "
3337                        "& quota.ost are)\n", ptr);
3338                 return -EINVAL;
3339         }
3340
3341         if (*tmp == '\0') {
3342                 CDEBUG(D_MGS, "global '%s' removed\n", quota);
3343         } else {
3344                 CDEBUG(D_MGS, "global '%s'\n", quota);
3345
3346                 if (strchr(tmp, 'u') == NULL && strchr(tmp, 'g') == NULL &&
3347                     strchr(tmp, 'p') == NULL &&
3348                     strcmp(tmp, "none") != 0) {
3349                         CERROR("enable option(%s) isn't supported\n", tmp);
3350                         return -EINVAL;
3351                 }
3352         }
3353
3354         lustre_cfg_bufs_reset(&mgi->mgi_bufs, mti->mti_fsname);
3355         lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 1, quota);
3356         lcr = lustre_cfg_rec_new(cmd, &mgi->mgi_bufs);
3357         if (lcr == NULL)
3358                 return -ENOMEM;
3359
3360         /* truncate the comment to the parameter name */
3361         ptr = tmp - 1;
3362         sep = *ptr;
3363         *ptr = '\0';
3364
3365         /* XXX we duplicated quota enable information in all server
3366          *     config logs, it should be moved to a separate config
3367          *     log once we cleanup the config log for global param. */
3368         /* modify all servers */
3369         rc = mgs_write_log_direct_all(env, mgs, fsdb, mti,
3370                                       *tmp == '\0' ? NULL : lcr,
3371                                       mti->mti_fsname, quota, 1);
3372         *ptr = sep;
3373         lustre_cfg_rec_free(lcr);
3374         return rc < 0 ? rc : 0;
3375 }
3376
3377 static int mgs_srpc_set_param_disk(const struct lu_env *env,
3378                                    struct mgs_device *mgs,
3379                                    struct fs_db *fsdb,
3380                                    struct mgs_target_info *mti,
3381                                    char *param)
3382 {
3383         struct mgs_thread_info  *mgi = mgs_env_info(env);
3384         struct llog_cfg_rec     *lcr;
3385         struct llog_handle      *llh = NULL;
3386         char                    *logname;
3387         char                    *comment, *ptr;
3388         int                      rc, len;
3389
3390         ENTRY;
3391
3392         /* get comment */
3393         ptr = strchr(param, '=');
3394         LASSERT(ptr != NULL);
3395         len = ptr - param;
3396
3397         OBD_ALLOC(comment, len + 1);
3398         if (comment == NULL)
3399                 RETURN(-ENOMEM);
3400         strncpy(comment, param, len);
3401         comment[len] = '\0';
3402
3403         /* prepare lcfg */
3404         lustre_cfg_bufs_reset(&mgi->mgi_bufs, mti->mti_svname);
3405         lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 1, param);
3406         lcr = lustre_cfg_rec_new(LCFG_SPTLRPC_CONF, &mgi->mgi_bufs);
3407         if (lcr == NULL)
3408                 GOTO(out_comment, rc = -ENOMEM);
3409
3410         /* construct log name */
3411         rc = name_create(&logname, mti->mti_fsname, "-sptlrpc");
3412         if (rc < 0)
3413                 GOTO(out_lcfg, rc);
3414
3415         if (mgs_log_is_empty(env, mgs, logname)) {
3416                 rc = record_start_log(env, mgs, &llh, logname);
3417                 if (rc < 0)
3418                         GOTO(out, rc);
3419                 record_end_log(env, &llh);
3420         }
3421
3422         /* obsolete old one */
3423         rc = mgs_modify(env, mgs, fsdb, mti, logname, mti->mti_svname,
3424                         comment, CM_SKIP);
3425         if (rc < 0)
3426                 GOTO(out, rc);
3427         /* write the new one */
3428         rc = mgs_write_log_direct(env, mgs, fsdb, logname, lcr,
3429                                   mti->mti_svname, comment);
3430         if (rc)
3431                 CERROR("%s: error writing log %s: rc = %d\n",
3432                        mgs->mgs_obd->obd_name, logname, rc);
3433 out:
3434         name_destroy(&logname);
3435 out_lcfg:
3436         lustre_cfg_rec_free(lcr);
3437 out_comment:
3438         OBD_FREE(comment, len + 1);
3439         RETURN(rc);
3440 }
3441
3442 static int mgs_srpc_set_param_udesc_mem(struct fs_db *fsdb,
3443                                         char *param)
3444 {
3445         char    *ptr;
3446
3447         /* disable the adjustable udesc parameter for now, i.e. use default
3448          * setting that client always ship udesc to MDT if possible. to enable
3449          * it simply remove the following line */
3450         goto error_out;
3451
3452         ptr = strchr(param, '=');
3453         if (ptr == NULL)
3454                 goto error_out;
3455         *ptr++ = '\0';
3456
3457         if (strcmp(param, PARAM_SRPC_UDESC))
3458                 goto error_out;
3459
3460         if (strcmp(ptr, "yes") == 0) {
3461                 set_bit(FSDB_UDESC, &fsdb->fsdb_flags);
3462                 CWARN("Enable user descriptor shipping from client to MDT\n");
3463         } else if (strcmp(ptr, "no") == 0) {
3464                 clear_bit(FSDB_UDESC, &fsdb->fsdb_flags);
3465                 CWARN("Disable user descriptor shipping from client to MDT\n");
3466         } else {
3467                 *(ptr - 1) = '=';
3468                 goto error_out;
3469         }
3470         return 0;
3471
3472 error_out:
3473         CERROR("Invalid param: %s\n", param);
3474         return -EINVAL;
3475 }
3476
3477 static int mgs_srpc_set_param_mem(struct fs_db *fsdb,
3478                                   const char *svname,
3479                                   char *param)
3480 {
3481         struct sptlrpc_rule      rule;
3482         struct sptlrpc_rule_set *rset;
3483         int                      rc;
3484         ENTRY;
3485
3486         if (strncmp(param, PARAM_SRPC, sizeof(PARAM_SRPC) - 1) != 0) {
3487                 CERROR("Invalid sptlrpc parameter: %s\n", param);
3488                 RETURN(-EINVAL);
3489         }
3490
3491         if (strncmp(param, PARAM_SRPC_UDESC,
3492                     sizeof(PARAM_SRPC_UDESC) - 1) == 0) {
3493                 RETURN(mgs_srpc_set_param_udesc_mem(fsdb, param));
3494         }
3495
3496         if (strncmp(param, PARAM_SRPC_FLVR, sizeof(PARAM_SRPC_FLVR) - 1) != 0) {
3497                 CERROR("Invalid sptlrpc flavor parameter: %s\n", param);
3498                 RETURN(-EINVAL);
3499         }
3500
3501         param += sizeof(PARAM_SRPC_FLVR) - 1;
3502
3503         rc = sptlrpc_parse_rule(param, &rule);
3504         if (rc)
3505                 RETURN(rc);
3506
3507         /* mgs rules implies must be mgc->mgs */
3508         if (test_bit(FSDB_MGS_SELF, &fsdb->fsdb_flags)) {
3509                 if ((rule.sr_from != LUSTRE_SP_MGC &&
3510                      rule.sr_from != LUSTRE_SP_ANY) ||
3511                     (rule.sr_to != LUSTRE_SP_MGS &&
3512                      rule.sr_to != LUSTRE_SP_ANY))
3513                         RETURN(-EINVAL);
3514         }
3515
3516         /* preapre room for this coming rule. svcname format should be:
3517          * - fsname: general rule
3518          * - fsname-tgtname: target-specific rule
3519          */
3520         if (strchr(svname, '-')) {
3521                 struct mgs_tgt_srpc_conf *tgtconf;
3522                 int                       found = 0;
3523
3524                 for (tgtconf = fsdb->fsdb_srpc_tgt; tgtconf != NULL;
3525                      tgtconf = tgtconf->mtsc_next) {
3526                         if (!strcmp(tgtconf->mtsc_tgt, svname)) {
3527                                 found = 1;
3528                                 break;
3529                         }
3530                 }
3531
3532                 if (!found) {
3533                         int name_len;
3534
3535                         OBD_ALLOC_PTR(tgtconf);
3536                         if (tgtconf == NULL)
3537                                 RETURN(-ENOMEM);
3538
3539                         name_len = strlen(svname);
3540
3541                         OBD_ALLOC(tgtconf->mtsc_tgt, name_len + 1);
3542                         if (tgtconf->mtsc_tgt == NULL) {
3543                                 OBD_FREE_PTR(tgtconf);
3544                                 RETURN(-ENOMEM);
3545                         }
3546                         memcpy(tgtconf->mtsc_tgt, svname, name_len);
3547
3548                         tgtconf->mtsc_next = fsdb->fsdb_srpc_tgt;
3549                         fsdb->fsdb_srpc_tgt = tgtconf;
3550                 }
3551
3552                 rset = &tgtconf->mtsc_rset;
3553         } else if (strcmp(svname, MGSSELF_NAME) == 0) {
3554                 /* put _mgs related srpc rule directly in mgs ruleset */
3555                 rset = &fsdb->fsdb_mgs->mgs_lut.lut_sptlrpc_rset;
3556         } else {
3557                 rset = &fsdb->fsdb_srpc_gen;
3558         }
3559
3560         rc = sptlrpc_rule_set_merge(rset, &rule);
3561
3562         RETURN(rc);
3563 }
3564
3565 static int mgs_srpc_set_param(const struct lu_env *env,
3566                               struct mgs_device *mgs,
3567                               struct fs_db *fsdb,
3568                               struct mgs_target_info *mti,
3569                               char *param)
3570 {
3571         char                   *copy;
3572         int                     rc, copy_size;
3573         ENTRY;
3574
3575 #ifndef HAVE_GSS
3576         RETURN(-EINVAL);
3577 #endif
3578         /* keep a copy of original param, which could be destroied
3579          * during parsing */
3580         copy_size = strlen(param) + 1;
3581         OBD_ALLOC(copy, copy_size);
3582         if (copy == NULL)
3583                 return -ENOMEM;
3584         memcpy(copy, param, copy_size);
3585
3586         rc = mgs_srpc_set_param_mem(fsdb, mti->mti_svname, param);
3587         if (rc)
3588                 goto out_free;
3589
3590         /* previous steps guaranteed the syntax is correct */
3591         rc = mgs_srpc_set_param_disk(env, mgs, fsdb, mti, copy);
3592         if (rc)
3593                 goto out_free;
3594
3595         if (test_bit(FSDB_MGS_SELF, &fsdb->fsdb_flags)) {
3596                 /*
3597                  * for mgs rules, make them effective immediately.
3598                  */
3599                 LASSERT(fsdb->fsdb_srpc_tgt == NULL);
3600                 sptlrpc_target_update_exp_flavor(mgs->mgs_obd,
3601                                                  &fsdb->fsdb_srpc_gen);
3602         }
3603
3604 out_free:
3605         OBD_FREE(copy, copy_size);
3606         RETURN(rc);
3607 }
3608
3609 struct mgs_srpc_read_data {
3610         struct fs_db   *msrd_fsdb;
3611         int             msrd_skip;
3612 };
3613
3614 static int mgs_srpc_read_handler(const struct lu_env *env,
3615                                  struct llog_handle *llh,
3616                                  struct llog_rec_hdr *rec, void *data)
3617 {
3618         struct mgs_srpc_read_data *msrd = data;
3619         struct cfg_marker         *marker;
3620         struct lustre_cfg         *lcfg = REC_DATA(rec);
3621         char                      *svname, *param;
3622         int                        cfg_len, rc;
3623         ENTRY;
3624
3625         if (rec->lrh_type != OBD_CFG_REC) {
3626                 CERROR("unhandled lrh_type: %#x\n", rec->lrh_type);
3627                 RETURN(-EINVAL);
3628         }
3629
3630         cfg_len = REC_DATA_LEN(rec);
3631
3632         rc = lustre_cfg_sanity_check(lcfg, cfg_len);
3633         if (rc) {
3634                 CERROR("Insane cfg\n");
3635                 RETURN(rc);
3636         }
3637
3638         if (lcfg->lcfg_command == LCFG_MARKER) {
3639                 marker = lustre_cfg_buf(lcfg, 1);
3640
3641                 if (marker->cm_flags & CM_START &&
3642                     marker->cm_flags & CM_SKIP)
3643                         msrd->msrd_skip = 1;
3644                 if (marker->cm_flags & CM_END)
3645                         msrd->msrd_skip = 0;
3646
3647                 RETURN(0);
3648         }
3649
3650         if (msrd->msrd_skip)
3651                 RETURN(0);
3652
3653         if (lcfg->lcfg_command != LCFG_SPTLRPC_CONF) {
3654                 CERROR("invalid command (%x)\n", lcfg->lcfg_command);
3655                 RETURN(0);
3656         }
3657
3658         svname = lustre_cfg_string(lcfg, 0);
3659         if (svname == NULL) {
3660                 CERROR("svname is empty\n");
3661                 RETURN(0);
3662         }
3663
3664         param = lustre_cfg_string(lcfg, 1);
3665         if (param == NULL) {
3666                 CERROR("param is empty\n");
3667                 RETURN(0);
3668         }
3669
3670         rc = mgs_srpc_set_param_mem(msrd->msrd_fsdb, svname, param);
3671         if (rc)
3672                 CERROR("read sptlrpc record error (%d): %s\n", rc, param);
3673
3674         RETURN(0);
3675 }
3676
3677 int mgs_get_fsdb_srpc_from_llog(const struct lu_env *env,
3678                                 struct mgs_device *mgs,
3679                                 struct fs_db *fsdb)
3680 {
3681         struct llog_handle        *llh = NULL;
3682         struct llog_ctxt          *ctxt;
3683         char                      *logname;
3684         struct mgs_srpc_read_data  msrd;
3685         int                        rc;
3686         ENTRY;
3687
3688         /* construct log name */
3689         rc = name_create(&logname, fsdb->fsdb_name, "-sptlrpc");
3690         if (rc)
3691                 RETURN(rc);
3692
3693         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
3694         LASSERT(ctxt != NULL);
3695
3696         if (mgs_log_is_empty(env, mgs, logname))
3697                 GOTO(out, rc = 0);
3698
3699         rc = llog_open(env, ctxt, &llh, NULL, logname,
3700                        LLOG_OPEN_EXISTS);
3701         if (rc < 0) {
3702                 if (rc == -ENOENT)
3703                         rc = 0;
3704                 GOTO(out, rc);
3705         }
3706
3707         rc = llog_init_handle(env, llh, LLOG_F_IS_PLAIN, NULL);
3708         if (rc)
3709                 GOTO(out_close, rc);
3710
3711         if (llog_get_size(llh) <= 1)
3712                 GOTO(out_close, rc = 0);
3713
3714         msrd.msrd_fsdb = fsdb;
3715         msrd.msrd_skip = 0;
3716
3717         rc = llog_process(env, llh, mgs_srpc_read_handler, (void *)&msrd,
3718                           NULL);
3719
3720 out_close:
3721         llog_close(env, llh);
3722 out:
3723         llog_ctxt_put(ctxt);
3724         name_destroy(&logname);
3725
3726         if (rc)
3727                 CERROR("failed to read sptlrpc config database: %d\n", rc);
3728         RETURN(rc);
3729 }
3730
3731 static int mgs_write_log_param2(const struct lu_env *env,
3732                                 struct mgs_device *mgs,
3733                                 struct fs_db *fsdb,
3734                                 struct mgs_target_info *mti, char *ptr)
3735 {
3736         struct lustre_cfg_bufs bufs;
3737         int rc;
3738
3739         ENTRY;
3740         CDEBUG(D_MGS, "next param '%s'\n", ptr);
3741
3742         /* PARAM_MGSNODE and PARAM_NETWORK are set only when formating
3743          * or during the inital mount. It can never change after that.
3744          */
3745         if (!class_match_param(ptr, PARAM_MGSNODE, NULL) ||
3746             !class_match_param(ptr, PARAM_NETWORK, NULL)) {
3747                 rc = 0;
3748                 goto end;
3749         }
3750
3751         /* Processed in mgs_write_log_ost. Another value that can't
3752          * be changed by lctl set_param -P.
3753          */
3754         if (!class_match_param(ptr, PARAM_FAILMODE, NULL)) {
3755                 LCONSOLE_ERROR_MSG(0x169,
3756                                    "%s can only be changed with tunefs.lustre and --writeconf\n",
3757                                    ptr);
3758                 rc = -EPERM;
3759                 goto end;
3760         }
3761
3762         /* FIXME !!! Support for sptlrpc is incomplete. Currently the change
3763          * doesn't transmit to the client. See LU-7183.
3764          */
3765         if (!class_match_param(ptr, PARAM_SRPC, NULL)) {
3766                 rc = mgs_srpc_set_param(env, mgs, fsdb, mti, ptr);
3767                 goto end;
3768         }
3769
3770         /* Can't use class_match_param since ptr doesn't start with
3771          * PARAM_FAILNODE. So we look for PARAM_FAILNODE contained in ptr.
3772          */
3773         if (strstr(ptr, PARAM_FAILNODE)) {
3774                 /* Add a failover nidlist. We already processed failovers
3775                  * params for new targets in mgs_write_log_target.
3776                  */
3777                 const char *param;
3778
3779                 /* can't use wildcards with failover.node */
3780                 if (strchr(ptr, '*')) {
3781                         rc = -ENODEV;
3782                         goto end;
3783                 }
3784
3785                 param = strstr(ptr, PARAM_FAILNODE);
3786                 if (strlcpy(mti->mti_params, param, sizeof(mti->mti_params)) >=
3787                     sizeof(mti->mti_params)) {
3788                         rc = -E2BIG;
3789                         goto end;
3790                 }
3791
3792                 CDEBUG(D_MGS, "Adding failnode with param %s\n",
3793                        mti->mti_params);
3794                 rc = mgs_write_log_add_failnid(env, mgs, fsdb, mti);
3795                 goto end;
3796         }
3797
3798         rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, PARAMS_FILENAME, &bufs,
3799                           mti->mti_svname, ptr);
3800 end:
3801         RETURN(rc);
3802 }
3803
3804 /* Permanent settings of all parameters by writing into the appropriate
3805  * configuration logs.
3806  * A parameter with null value ("<param>='\0'") means to erase it out of
3807  * the logs.
3808  */
3809 static int mgs_write_log_param(const struct lu_env *env,
3810                                struct mgs_device *mgs, struct fs_db *fsdb,
3811                                struct mgs_target_info *mti, char *ptr)
3812 {
3813         struct mgs_thread_info *mgi = mgs_env_info(env);
3814         char *logname;
3815         char *tmp;
3816         int rc = 0;
3817         ENTRY;
3818
3819         /* For various parameter settings, we have to figure out which logs
3820            care about them (e.g. both mdt and client for lov settings) */
3821         CDEBUG(D_MGS, "next param '%s'\n", ptr);
3822
3823         /* The params are stored in MOUNT_DATA_FILE and modified via
3824            tunefs.lustre, or set using lctl conf_param */
3825
3826         /* Processed in lustre_start_mgc */
3827         if (class_match_param(ptr, PARAM_MGSNODE, NULL) == 0)
3828                 GOTO(end, rc);
3829
3830         /* Processed in ost/mdt */
3831         if (class_match_param(ptr, PARAM_NETWORK, NULL) == 0)
3832                 GOTO(end, rc);
3833
3834         /* Processed in mgs_write_log_ost */
3835         if (class_match_param(ptr, PARAM_FAILMODE, NULL) == 0) {
3836                 if (mti->mti_flags & LDD_F_PARAM) {
3837                         LCONSOLE_ERROR_MSG(0x169, "%s can only be "
3838                                            "changed with tunefs.lustre"
3839                                            "and --writeconf\n", ptr);
3840                         rc = -EPERM;
3841                 }
3842                 GOTO(end, rc);
3843         }
3844
3845         if (class_match_param(ptr, PARAM_SRPC, NULL) == 0) {
3846                 rc = mgs_srpc_set_param(env, mgs, fsdb, mti, ptr);
3847                 GOTO(end, rc);
3848         }
3849
3850         if (class_match_param(ptr, PARAM_FAILNODE, NULL) == 0) {
3851                 /* Add a failover nidlist */
3852                 rc = 0;
3853                 /* We already processed failovers params for new
3854                    targets in mgs_write_log_target */
3855                 if (mti->mti_flags & LDD_F_PARAM) {
3856                         CDEBUG(D_MGS, "Adding failnode\n");
3857                         rc = mgs_write_log_add_failnid(env, mgs, fsdb, mti);
3858                 }
3859                 GOTO(end, rc);
3860         }
3861
3862         if (class_match_param(ptr, PARAM_SYS, &tmp) == 0) {
3863                 rc = mgs_write_log_sys(env, mgs, fsdb, mti, ptr, tmp);
3864                 GOTO(end, rc);
3865         }
3866
3867         if (class_match_param(ptr, PARAM_QUOTA, &tmp) == 0) {
3868                 rc = mgs_write_log_quota(env, mgs, fsdb, mti, ptr, tmp);
3869                 GOTO(end, rc);
3870         }
3871
3872         if (class_match_param(ptr, PARAM_OSC PARAM_ACTIVE, &tmp) == 0 ||
3873             class_match_param(ptr, PARAM_MDC PARAM_ACTIVE, &tmp) == 0) {
3874                 /* active=0 means off, anything else means on */
3875                 int flag = (*tmp == '0') ? CM_EXCLUDE : 0;
3876                 bool deactive_osc = memcmp(ptr, PARAM_OSC PARAM_ACTIVE,
3877                                           strlen(PARAM_OSC PARAM_ACTIVE)) == 0;
3878                 int i;
3879
3880                 if (!deactive_osc) {
3881                         __u32   index;
3882
3883                         rc = server_name2index(mti->mti_svname, &index, NULL);
3884                         if (rc < 0)
3885                                 GOTO(end, rc);
3886
3887                         if (index == 0) {
3888                                 LCONSOLE_ERROR_MSG(0x144, "%s: MDC0 can not be"
3889                                                    " (de)activated.\n",
3890                                                    mti->mti_svname);
3891                                 GOTO(end, rc = -EPERM);
3892                         }
3893                 }
3894
3895                 LCONSOLE_WARN("Permanently %sactivating %s\n",
3896                               flag ? "de" : "re", mti->mti_svname);
3897                 /* Modify clilov */
3898                 rc = name_create(&logname, mti->mti_fsname, "-client");
3899                 if (rc < 0)
3900                         GOTO(end, rc);
3901                 rc = mgs_modify(env, mgs, fsdb, mti, logname,
3902                                 mti->mti_svname,
3903                                 deactive_osc ? "add osc" : "add mdc", flag);
3904                 name_destroy(&logname);
3905                 if (rc < 0)
3906                         goto active_err;
3907
3908                 /* Modify mdtlov */
3909                 /* Add to all MDT logs for DNE */
3910                 for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
3911                         if (!test_bit(i, fsdb->fsdb_mdt_index_map))
3912                                 continue;
3913                         rc = name_create_mdt(&logname, mti->mti_fsname, i);
3914                         if (rc < 0)
3915                                 GOTO(end, rc);
3916                         rc = mgs_modify(env, mgs, fsdb, mti, logname,
3917                                         mti->mti_svname,
3918                                         deactive_osc ? "add osc" : "add osp",
3919                                         flag);
3920                         name_destroy(&logname);
3921                         if (rc < 0)
3922                                 goto active_err;
3923                 }
3924 active_err:
3925                 if (rc < 0) {
3926                         LCONSOLE_ERROR_MSG(0x145, "Couldn't find %s in"
3927                                            "log (%d). No permanent "
3928                                            "changes were made to the "
3929                                            "config log.\n",
3930                                            mti->mti_svname, rc);
3931                         if (test_bit(FSDB_OLDLOG14, &fsdb->fsdb_flags))
3932                                 LCONSOLE_ERROR_MSG(0x146, "This may be"
3933                                                    " because the log"
3934                                                    "is in the old 1.4"
3935                                                    "style. Consider "
3936                                                    " --writeconf to "
3937                                                    "update the logs.\n");
3938                         GOTO(end, rc);
3939                 }
3940                 /* Fall through to osc/mdc proc for deactivating live
3941                    OSC/OSP on running MDT / clients. */
3942         }
3943         /* Below here, let obd's XXX_process_config methods handle it */
3944
3945         /* All lov. in proc */
3946         if (class_match_param(ptr, PARAM_LOV, NULL) == 0) {
3947                 char *mdtlovname;
3948
3949                 CDEBUG(D_MGS, "lov param %s\n", ptr);
3950                 if (!(mti->mti_flags & LDD_F_SV_TYPE_MDT)) {
3951                         LCONSOLE_ERROR_MSG(0x147, "LOV params must be "
3952                                            "set on the MDT, not %s. "
3953                                            "Ignoring.\n",
3954                                            mti->mti_svname);
3955                         GOTO(end, rc = 0);
3956                 }
3957
3958                 /* Modify mdtlov */
3959                 if (mgs_log_is_empty(env, mgs, mti->mti_svname))
3960                         GOTO(end, rc = -ENODEV);
3961
3962                 rc = name_create_mdt_and_lov(&logname, &mdtlovname, fsdb,
3963                                              mti->mti_stripe_index);
3964                 if (rc)
3965                         GOTO(end, rc);
3966                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, mti->mti_svname,
3967                                   &mgi->mgi_bufs, mdtlovname, ptr);
3968                 name_destroy(&logname);
3969                 name_destroy(&mdtlovname);
3970                 if (rc)
3971                         GOTO(end, rc);
3972
3973                 /* Modify clilov */
3974                 rc = name_create(&logname, mti->mti_fsname, "-client");
3975                 if (rc)
3976                         GOTO(end, rc);
3977                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, logname, &mgi->mgi_bufs,
3978                                   fsdb->fsdb_clilov, ptr);
3979                 name_destroy(&logname);
3980                 GOTO(end, rc);
3981         }
3982
3983         /* All osc., mdc., llite. params in proc */
3984         if ((class_match_param(ptr, PARAM_OSC, NULL) == 0) ||
3985             (class_match_param(ptr, PARAM_MDC, NULL) == 0) ||
3986             (class_match_param(ptr, PARAM_LLITE, NULL) == 0)) {
3987                 char *cname;
3988
3989                 if (test_bit(FSDB_OLDLOG14, &fsdb->fsdb_flags)) {
3990                         LCONSOLE_ERROR_MSG(0x148, "Upgraded client logs for %s"
3991                                            " cannot be modified. Consider"
3992                                            " updating the configuration with"
3993                                            " --writeconf\n",
3994                                            mti->mti_svname);
3995                         GOTO(end, rc = -EINVAL);
3996                 }
3997                 if (memcmp(ptr, PARAM_LLITE, strlen(PARAM_LLITE)) == 0) {
3998                         rc = name_create(&cname, mti->mti_fsname, "-client");
3999                         /* Add the client type to match the obdname in
4000                            class_config_llog_handler */
4001                 } else if (mti->mti_flags & LDD_F_SV_TYPE_MDT) {
4002                         rc = name_create(&cname, mti->mti_svname, "-mdc");
4003                 } else if (mti->mti_flags & LDD_F_SV_TYPE_OST) {
4004                         rc = name_create(&cname, mti->mti_svname, "-osc");
4005                 } else {
4006                         GOTO(end, rc = -EINVAL);
4007                 }
4008                 if (rc)
4009                         GOTO(end, rc);
4010
4011                 /* Forbid direct update of llite root squash parameters.
4012                  * These parameters are indirectly set via the MDT settings.
4013                  * See (LU-1778) */
4014                 if ((class_match_param(ptr, PARAM_LLITE, &tmp) == 0) &&
4015                     ((memcmp(tmp, "root_squash=", 12) == 0) ||
4016                      (memcmp(tmp, "nosquash_nids=", 14) == 0))) {
4017                         LCONSOLE_ERROR("%s: root squash parameters can only "
4018                                 "be updated through MDT component\n",
4019                                 mti->mti_fsname);
4020                         name_destroy(&cname);
4021                         GOTO(end, rc = -EINVAL);
4022                 }
4023
4024                 CDEBUG(D_MGS, "%.3s param %s\n", ptr, ptr + 4);
4025
4026                 /* Modify client */
4027                 rc = name_create(&logname, mti->mti_fsname, "-client");
4028                 if (rc) {
4029                         name_destroy(&cname);
4030                         GOTO(end, rc);
4031                 }
4032                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, logname, &mgi->mgi_bufs,
4033                                   cname, ptr);
4034
4035                 /* osc params affect the MDT as well */
4036                 if (!rc && (mti->mti_flags & LDD_F_SV_TYPE_OST)) {
4037                         int i;
4038
4039                         for (i = 0; i < INDEX_MAP_SIZE * 8; i++){
4040                                 if (!test_bit(i, fsdb->fsdb_mdt_index_map))
4041                                         continue;
4042                                 name_destroy(&cname);
4043                                 rc = name_create_mdt_osc(&cname, mti->mti_svname,
4044                                                          fsdb, i);
4045                                 name_destroy(&logname);
4046                                 if (rc)
4047                                         break;
4048                                 rc = name_create_mdt(&logname,
4049                                                      mti->mti_fsname, i);
4050                                 if (rc)
4051                                         break;
4052                                 if (!mgs_log_is_empty(env, mgs, logname)) {
4053                                         rc = mgs_wlp_lcfg(env, mgs, fsdb,
4054                                                           mti, logname,
4055                                                           &mgi->mgi_bufs,
4056                                                           cname, ptr);
4057                                         if (rc)
4058                                                 break;
4059                                 }
4060                         }
4061                 }
4062
4063                 /* For mdc activate/deactivate, it affects OSP on MDT as well */
4064                 if (class_match_param(ptr, PARAM_MDC PARAM_ACTIVE, &tmp) == 0 &&
4065                     rc == 0) {
4066                         char suffix[16];
4067                         char *lodname = NULL;
4068                         char *param_str = NULL;
4069                         int i;
4070                         int index;
4071
4072                         /* replace mdc with osp */
4073                         memcpy(ptr, PARAM_OSP, strlen(PARAM_OSP));
4074                         rc = server_name2index(mti->mti_svname, &index, NULL);
4075                         if (rc < 0) {
4076                                 memcpy(ptr, PARAM_MDC, strlen(PARAM_MDC));
4077                                 GOTO(end, rc);
4078                         }
4079
4080                         for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
4081                                 if (!test_bit(i, fsdb->fsdb_mdt_index_map))
4082                                         continue;
4083
4084                                 if (i == index)
4085                                         continue;
4086
4087                                 name_destroy(&logname);
4088                                 rc = name_create_mdt(&logname, mti->mti_fsname,
4089                                                      i);
4090                                 if (rc < 0)
4091                                         break;
4092
4093                                 if (mgs_log_is_empty(env, mgs, logname))
4094                                         continue;
4095
4096                                 snprintf(suffix, sizeof(suffix), "-osp-MDT%04x",
4097                                          i);
4098                                 name_destroy(&cname);
4099                                 rc = name_create(&cname, mti->mti_svname,
4100                                                  suffix);
4101                                 if (rc < 0)
4102                                         break;
4103
4104                                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, logname,
4105                                                   &mgi->mgi_bufs, cname, ptr);
4106                                 if (rc < 0)
4107                                         break;
4108
4109                                 /* Add configuration log for noitfying LOD
4110                                  * to active/deactive the OSP. */
4111                                 name_destroy(&param_str);
4112                                 rc = name_create(&param_str, cname,
4113                                                  (*tmp == '0') ?  ".active=0" :
4114                                                  ".active=1");
4115                                 if (rc < 0)
4116                                         break;
4117
4118                                 name_destroy(&lodname);
4119                                 rc = name_create(&lodname, logname, "-mdtlov");
4120                                 if (rc < 0)
4121                                         break;
4122
4123                                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, logname,
4124                                                   &mgi->mgi_bufs, lodname,
4125                                                   param_str);
4126                                 if (rc < 0)
4127                                         break;
4128                         }
4129                         memcpy(ptr, PARAM_MDC, strlen(PARAM_MDC));
4130                         name_destroy(&lodname);
4131                         name_destroy(&param_str);
4132                 }
4133
4134                 name_destroy(&logname);
4135                 name_destroy(&cname);
4136                 GOTO(end, rc);
4137         }
4138
4139         /* All mdt. params in proc */
4140         if (class_match_param(ptr, PARAM_MDT, &tmp) == 0) {
4141                 int i;
4142                 __u32 idx;
4143
4144                 CDEBUG(D_MGS, "%.3s param %s\n", ptr, ptr + 4);
4145                 if (strncmp(mti->mti_svname, mti->mti_fsname,
4146                             MTI_NAME_MAXLEN) == 0)
4147                         /* device is unspecified completely? */
4148                         rc = LDD_F_SV_TYPE_MDT | LDD_F_SV_ALL;
4149                 else
4150                         rc = server_name2index(mti->mti_svname, &idx, NULL);
4151                 if (rc < 0)
4152                         goto active_err;
4153                 if ((rc & LDD_F_SV_TYPE_MDT) == 0)
4154                         goto active_err;
4155                 if (rc & LDD_F_SV_ALL) {
4156                         for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
4157                                 if (!test_bit(i,
4158                                                   fsdb->fsdb_mdt_index_map))
4159                                         continue;
4160                                 rc = name_create_mdt(&logname,
4161                                                 mti->mti_fsname, i);
4162                                 if (rc)
4163                                         goto active_err;
4164                                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti,
4165                                                   logname, &mgi->mgi_bufs,
4166                                                   logname, ptr);
4167                                 name_destroy(&logname);
4168                                 if (rc)
4169                                         goto active_err;
4170                         }
4171                 } else {
4172                         if ((memcmp(tmp, "root_squash=", 12) == 0) ||
4173                             (memcmp(tmp, "nosquash_nids=", 14) == 0)) {
4174                                 LCONSOLE_ERROR("%s: root squash parameters "
4175                                         "cannot be applied to a single MDT\n",
4176                                         mti->mti_fsname);
4177                                 GOTO(end, rc = -EINVAL);
4178                         }
4179                         rc = mgs_wlp_lcfg(env, mgs, fsdb, mti,
4180                                           mti->mti_svname, &mgi->mgi_bufs,
4181                                           mti->mti_svname, ptr);
4182                         if (rc)
4183                                 goto active_err;
4184                 }
4185
4186                 /* root squash settings are also applied to llite
4187                  * config log (see LU-1778) */
4188                 if (rc == 0 &&
4189                     ((memcmp(tmp, "root_squash=", 12) == 0) ||
4190                      (memcmp(tmp, "nosquash_nids=", 14) == 0))) {
4191                         char *cname;
4192                         char *ptr2;
4193
4194                         rc = name_create(&cname, mti->mti_fsname, "-client");
4195                         if (rc)
4196                                 GOTO(end, rc);
4197                         rc = name_create(&logname, mti->mti_fsname, "-client");
4198                         if (rc) {
4199                                 name_destroy(&cname);
4200                                 GOTO(end, rc);
4201                         }
4202                         rc = name_create(&ptr2, PARAM_LLITE, tmp);
4203                         if (rc) {
4204                                 name_destroy(&cname);
4205                                 name_destroy(&logname);
4206                                 GOTO(end, rc);
4207                         }
4208                         rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, logname,
4209                                           &mgi->mgi_bufs, cname, ptr2);
4210                         name_destroy(&ptr2);
4211                         name_destroy(&logname);
4212                         name_destroy(&cname);
4213                 }
4214                 GOTO(end, rc);
4215         }
4216
4217         /* All mdd., ost. and osd. params in proc */
4218         if ((class_match_param(ptr, PARAM_MDD, 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 &
4270             (LDD_F_VIRGIN | LDD_F_UPGRADE14 | LDD_F_WRITECONF)) {
4271                 /* Generate a log from scratch */
4272                 if (mti->mti_flags & LDD_F_SV_TYPE_MDT) {
4273                         rc = mgs_write_log_mdt(env, mgs, fsdb, mti);
4274                 } else if (mti->mti_flags & LDD_F_SV_TYPE_OST) {
4275                         rc = mgs_write_log_ost(env, mgs, fsdb, mti);
4276                 } else {
4277                         CERROR("Unknown target type %#x, can't create log for "
4278                                "%s\n", mti->mti_flags, mti->mti_svname);
4279                 }
4280                 if (rc) {
4281                         CERROR("Can't write logs for %s (%d)\n",
4282                                mti->mti_svname, rc);
4283                         GOTO(out_up, rc);
4284                 }
4285         } else {
4286                 /* Just update the params from tunefs in mgs_write_log_params */
4287                 CDEBUG(D_MGS, "Update params for %s\n", mti->mti_svname);
4288                 mti->mti_flags |= LDD_F_PARAM;
4289         }
4290
4291         /* allocate temporary buffer, where class_get_next_param will
4292            make copy of a current  parameter */
4293         OBD_ALLOC(buf, strlen(mti->mti_params) + 1);
4294         if (buf == NULL)
4295                 GOTO(out_up, rc = -ENOMEM);
4296         params = mti->mti_params;
4297         while (params != NULL) {
4298                 rc = class_get_next_param(&params, buf);
4299                 if (rc) {
4300                         if (rc == 1)
4301                                 /* there is no next parameter, that is
4302                                    not an error */
4303                                 rc = 0;
4304                         break;
4305                 }
4306                 CDEBUG(D_MGS, "remaining string: '%s', param: '%s'\n",
4307                        params, buf);
4308                 rc = mgs_write_log_param(env, mgs, fsdb, mti, buf);
4309                 if (rc)
4310                         break;
4311         }
4312
4313         OBD_FREE(buf, strlen(mti->mti_params) + 1);
4314
4315 out_up:
4316         mutex_unlock(&fsdb->fsdb_mutex);
4317         RETURN(rc);
4318 }
4319
4320 int mgs_erase_log(const struct lu_env *env, struct mgs_device *mgs, char *name)
4321 {
4322         struct llog_ctxt        *ctxt;
4323         int                      rc = 0;
4324
4325         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
4326         if (ctxt == NULL) {
4327                 CERROR("%s: MGS config context doesn't exist\n",
4328                        mgs->mgs_obd->obd_name);
4329                 rc = -ENODEV;
4330         } else {
4331                 rc = llog_erase(env, ctxt, NULL, name);
4332                 /* llog may not exist */
4333                 if (rc == -ENOENT)
4334                         rc = 0;
4335                 llog_ctxt_put(ctxt);
4336         }
4337
4338         if (rc)
4339                 CERROR("%s: failed to clear log %s: %d\n",
4340                        mgs->mgs_obd->obd_name, name, rc);
4341
4342         return rc;
4343 }
4344
4345 /* erase all logs for the given fs */
4346 int mgs_erase_logs(const struct lu_env *env, struct mgs_device *mgs,
4347                    const char *fsname)
4348 {
4349         struct list_head log_list;
4350         struct mgs_direntry *dirent, *n;
4351         char barrier_name[20] = {};
4352         char *suffix;
4353         int count = 0;
4354         int rc, len = strlen(fsname);
4355         ENTRY;
4356
4357         mutex_lock(&mgs->mgs_mutex);
4358
4359         /* Find all the logs in the CONFIGS directory */
4360         rc = class_dentry_readdir(env, mgs, &log_list);
4361         if (rc) {
4362                 mutex_unlock(&mgs->mgs_mutex);
4363                 RETURN(rc);
4364         }
4365
4366         if (list_empty(&log_list)) {
4367                 mutex_unlock(&mgs->mgs_mutex);
4368                 RETURN(-ENOENT);
4369         }
4370
4371         snprintf(barrier_name, sizeof(barrier_name) - 1, "%s-%s",
4372                  fsname, BARRIER_FILENAME);
4373         /* Delete the barrier fsdb */
4374         mgs_remove_fsdb_by_name(mgs, barrier_name);
4375         /* Delete the fs db */
4376         mgs_remove_fsdb_by_name(mgs, fsname);
4377         mutex_unlock(&mgs->mgs_mutex);
4378
4379         list_for_each_entry_safe(dirent, n, &log_list, mde_list) {
4380                 list_del_init(&dirent->mde_list);
4381                 suffix = strrchr(dirent->mde_name, '-');
4382                 if (suffix != NULL) {
4383                         if ((len == suffix - dirent->mde_name) &&
4384                             (strncmp(fsname, dirent->mde_name, len) == 0)) {
4385                                 CDEBUG(D_MGS, "Removing log %s\n",
4386                                        dirent->mde_name);
4387                                 mgs_erase_log(env, mgs, dirent->mde_name);
4388                                 count++;
4389                         }
4390                 }
4391                 mgs_direntry_free(dirent);
4392         }
4393
4394         if (count == 0)
4395                 rc = -ENOENT;
4396
4397         RETURN(rc);
4398 }
4399
4400 /* list all logs for the given fs */
4401 int mgs_list_logs(const struct lu_env *env, struct mgs_device *mgs,
4402                   struct obd_ioctl_data *data)
4403 {
4404         struct list_head         log_list;
4405         struct mgs_direntry     *dirent, *n;
4406         char                    *out, *suffix;
4407         int                      l, remains, rc;
4408
4409         ENTRY;
4410
4411         /* Find all the logs in the CONFIGS directory */
4412         rc = class_dentry_readdir(env, mgs, &log_list);
4413         if (rc)
4414                 RETURN(rc);
4415
4416         out = data->ioc_bulk;
4417         remains = data->ioc_inllen1;
4418         list_for_each_entry_safe(dirent, n, &log_list, mde_list) {
4419                 list_del_init(&dirent->mde_list);
4420                 suffix = strrchr(dirent->mde_name, '-');
4421                 if (suffix != NULL) {
4422                         l = snprintf(out, remains, "config_log: %s\n",
4423                                      dirent->mde_name);
4424                         out += l;
4425                         remains -= l;
4426                 }
4427                 mgs_direntry_free(dirent);
4428                 if (remains <= 0)
4429                         break;
4430         }
4431         RETURN(rc);
4432 }
4433
4434 struct mgs_lcfg_fork_data {
4435         struct lustre_cfg_bufs   mlfd_bufs;
4436         struct mgs_device       *mlfd_mgs;
4437         struct llog_handle      *mlfd_llh;
4438         const char              *mlfd_oldname;
4439         const char              *mlfd_newname;
4440         char                     mlfd_data[0];
4441 };
4442
4443 static bool contain_valid_fsname(char *buf, const char *fsname,
4444                                  int buflen, int namelen)
4445 {
4446         if (buflen < namelen)
4447                 return false;
4448
4449         if (memcmp(buf, fsname, namelen) != 0)
4450                 return false;
4451
4452         if (buf[namelen] != '\0' && buf[namelen] != '-')
4453                 return false;
4454
4455         return true;
4456 }
4457
4458 static int mgs_lcfg_fork_handler(const struct lu_env *env,
4459                                  struct llog_handle *o_llh,
4460                                  struct llog_rec_hdr *o_rec, void *data)
4461 {
4462         struct mgs_lcfg_fork_data *mlfd = data;
4463         struct lustre_cfg_bufs *n_bufs = &mlfd->mlfd_bufs;
4464         struct lustre_cfg *o_lcfg = (struct lustre_cfg *)(o_rec + 1);
4465         struct llog_cfg_rec *lcr;
4466         char *o_buf;
4467         char *n_buf = mlfd->mlfd_data;
4468         int o_buflen;
4469         int o_namelen = strlen(mlfd->mlfd_oldname);
4470         int n_namelen = strlen(mlfd->mlfd_newname);
4471         int diff = n_namelen - o_namelen;
4472         __u32 cmd = o_lcfg->lcfg_command;
4473         __u32 cnt = o_lcfg->lcfg_bufcount;
4474         int rc;
4475         int i;
4476         ENTRY;
4477
4478         /* buf[0] */
4479         o_buf = lustre_cfg_buf(o_lcfg, 0);
4480         o_buflen = o_lcfg->lcfg_buflens[0];
4481         if (contain_valid_fsname(o_buf, mlfd->mlfd_oldname, o_buflen,
4482                                  o_namelen)) {
4483                 memcpy(n_buf, mlfd->mlfd_newname, n_namelen);
4484                 memcpy(n_buf + n_namelen, o_buf + o_namelen,
4485                        o_buflen - o_namelen);
4486                 lustre_cfg_bufs_reset(n_bufs, n_buf);
4487                 n_buf += cfs_size_round(o_buflen + diff);
4488         } else {
4489                 lustre_cfg_bufs_reset(n_bufs, o_buflen != 0 ? o_buf : NULL);
4490         }
4491
4492         switch (cmd) {
4493         case LCFG_MARKER: {
4494                 struct cfg_marker *o_marker;
4495                 struct cfg_marker *n_marker;
4496                 int tgt_namelen;
4497
4498                 if (cnt != 2) {
4499                         CDEBUG(D_MGS, "Unknown cfg marker entry with %d "
4500                                "buffers\n", cnt);
4501                         RETURN(-EINVAL);
4502                 }
4503
4504                 /* buf[1] is marker */
4505                 o_buf = lustre_cfg_buf(o_lcfg, 1);
4506                 o_buflen = o_lcfg->lcfg_buflens[1];
4507                 o_marker = (struct cfg_marker *)o_buf;
4508                 if (!contain_valid_fsname(o_marker->cm_tgtname,
4509                                           mlfd->mlfd_oldname,
4510                                           sizeof(o_marker->cm_tgtname),
4511                                           o_namelen)) {
4512                         lustre_cfg_bufs_set(n_bufs, 1, o_marker,
4513                                             sizeof(*o_marker));
4514                         break;
4515                 }
4516
4517                 n_marker = (struct cfg_marker *)n_buf;
4518                 *n_marker = *o_marker;
4519                 memcpy(n_marker->cm_tgtname, mlfd->mlfd_newname, n_namelen);
4520                 tgt_namelen = strlen(o_marker->cm_tgtname);
4521                 if (tgt_namelen > o_namelen)
4522                         memcpy(n_marker->cm_tgtname + n_namelen,
4523                                o_marker->cm_tgtname + o_namelen,
4524                                tgt_namelen - o_namelen);
4525                 n_marker->cm_tgtname[tgt_namelen + diff] = '\0';
4526                 lustre_cfg_bufs_set(n_bufs, 1, n_marker, sizeof(*n_marker));
4527                 break;
4528         }
4529         case LCFG_PARAM:
4530         case LCFG_SET_PARAM: {
4531                 for (i = 1; i < cnt; i++)
4532                         /* buf[i] is the param value, reuse it directly */
4533                         lustre_cfg_bufs_set(n_bufs, i,
4534                                             lustre_cfg_buf(o_lcfg, i),
4535                                             o_lcfg->lcfg_buflens[i]);
4536                 break;
4537         }
4538         case LCFG_POOL_NEW:
4539         case LCFG_POOL_ADD:
4540         case LCFG_POOL_REM:
4541         case LCFG_POOL_DEL: {
4542                 if (cnt < 3 || cnt > 4) {
4543                         CDEBUG(D_MGS, "Unknown cfg pool (%x) entry with %d "
4544                                "buffers\n", cmd, cnt);
4545                         RETURN(-EINVAL);
4546                 }
4547
4548                 /* buf[1] is fsname */
4549                 o_buf = lustre_cfg_buf(o_lcfg, 1);
4550                 o_buflen = o_lcfg->lcfg_buflens[1];
4551                 memcpy(n_buf, mlfd->mlfd_newname, n_namelen);
4552                 memcpy(n_buf + n_namelen, o_buf + o_namelen,
4553                        o_buflen - o_namelen);
4554                 lustre_cfg_bufs_set(n_bufs, 1, n_buf, o_buflen + diff);
4555                 n_buf += cfs_size_round(o_buflen + diff);
4556
4557                 /* buf[2] is the pool name, reuse it directly */
4558                 lustre_cfg_bufs_set(n_bufs, 2, lustre_cfg_buf(o_lcfg, 2),
4559                                     o_lcfg->lcfg_buflens[2]);
4560
4561                 if (cnt == 3)
4562                         break;
4563
4564                 /* buf[3] is ostname */
4565                 o_buf = lustre_cfg_buf(o_lcfg, 3);
4566                 o_buflen = o_lcfg->lcfg_buflens[3];
4567                 memcpy(n_buf, mlfd->mlfd_newname, n_namelen);
4568                 memcpy(n_buf + n_namelen, o_buf + o_namelen,
4569                        o_buflen - o_namelen);
4570                 lustre_cfg_bufs_set(n_bufs, 3, n_buf, o_buflen + diff);
4571                 break;
4572         }
4573         case LCFG_SETUP: {
4574                 if (cnt == 2) {
4575                         o_buflen = o_lcfg->lcfg_buflens[1];
4576                         if (o_buflen == sizeof(struct lov_desc) ||
4577                             o_buflen == sizeof(struct lmv_desc)) {
4578                                 char *o_uuid;
4579                                 char *n_uuid;
4580                                 int uuid_len;
4581
4582                                 /* buf[1] */
4583                                 o_buf = lustre_cfg_buf(o_lcfg, 1);
4584                                 if (o_buflen == sizeof(struct lov_desc)) {
4585                                         struct lov_desc *o_desc =
4586                                                 (struct lov_desc *)o_buf;
4587                                         struct lov_desc *n_desc =
4588                                                 (struct lov_desc *)n_buf;
4589
4590                                         *n_desc = *o_desc;
4591                                         o_uuid = o_desc->ld_uuid.uuid;
4592                                         n_uuid = n_desc->ld_uuid.uuid;
4593                                         uuid_len = sizeof(o_desc->ld_uuid.uuid);
4594                                 } else {
4595                                         struct lmv_desc *o_desc =
4596                                                 (struct lmv_desc *)o_buf;
4597                                         struct lmv_desc *n_desc =
4598                                                 (struct lmv_desc *)n_buf;
4599
4600                                         *n_desc = *o_desc;
4601                                         o_uuid = o_desc->ld_uuid.uuid;
4602                                         n_uuid = n_desc->ld_uuid.uuid;
4603                                         uuid_len = sizeof(o_desc->ld_uuid.uuid);
4604                                 }
4605
4606                                 if (unlikely(!contain_valid_fsname(o_uuid,
4607                                                 mlfd->mlfd_oldname, uuid_len,
4608                                                 o_namelen))) {
4609                                         lustre_cfg_bufs_set(n_bufs, 1, o_buf,
4610                                                             o_buflen);
4611                                         break;
4612                                 }
4613
4614                                 memcpy(n_uuid, mlfd->mlfd_newname, n_namelen);
4615                                 uuid_len = strlen(o_uuid);
4616                                 if (uuid_len > o_namelen)
4617                                         memcpy(n_uuid + n_namelen,
4618                                                o_uuid + o_namelen,
4619                                                uuid_len - o_namelen);
4620                                 n_uuid[uuid_len + diff] = '\0';
4621                                 lustre_cfg_bufs_set(n_bufs, 1, n_buf, o_buflen);
4622                                 break;
4623                         } /* else case fall through */
4624                 } /* else case fall through */
4625         }
4626         default: {
4627                 for (i = 1; i < cnt; i++) {
4628                         o_buflen = o_lcfg->lcfg_buflens[i];
4629                         if (o_buflen == 0)
4630                                 continue;
4631
4632                         o_buf = lustre_cfg_buf(o_lcfg, i);
4633                         if (!contain_valid_fsname(o_buf, mlfd->mlfd_oldname,
4634                                                   o_buflen, o_namelen)) {
4635                                 lustre_cfg_bufs_set(n_bufs, i, o_buf, o_buflen);
4636                                 continue;
4637                         }
4638
4639                         memcpy(n_buf, mlfd->mlfd_newname, n_namelen);
4640                         if (o_buflen == o_namelen) {
4641                                 lustre_cfg_bufs_set(n_bufs, i, n_buf,
4642                                                     n_namelen);
4643                                 n_buf += cfs_size_round(n_namelen);
4644                                 continue;
4645                         }
4646
4647                         memcpy(n_buf + n_namelen, o_buf + o_namelen,
4648                                o_buflen - o_namelen);
4649                         lustre_cfg_bufs_set(n_bufs, i, n_buf, o_buflen + diff);
4650                         n_buf += cfs_size_round(o_buflen + diff);
4651                 }
4652                 break;
4653         }
4654         }
4655
4656         lcr = lustre_cfg_rec_new(cmd, n_bufs);
4657         if (!lcr)
4658                 RETURN(-ENOMEM);
4659
4660         lcr->lcr_cfg = *o_lcfg;
4661         rc = llog_write(env, mlfd->mlfd_llh, &lcr->lcr_hdr, LLOG_NEXT_IDX);
4662         lustre_cfg_rec_free(lcr);
4663
4664         RETURN(rc);
4665 }
4666
4667 static int mgs_lcfg_fork_one(const struct lu_env *env, struct mgs_device *mgs,
4668                              struct mgs_direntry *mde, const char *oldname,
4669                              const char *newname)
4670 {
4671         struct llog_handle *old_llh = NULL;
4672         struct llog_handle *new_llh = NULL;
4673         struct llog_ctxt *ctxt = NULL;
4674         struct mgs_lcfg_fork_data *mlfd = NULL;
4675         char *name_buf = NULL;
4676         int name_buflen;
4677         int old_namelen = strlen(oldname);
4678         int new_namelen = strlen(newname);
4679         int rc;
4680         ENTRY;
4681
4682         name_buflen = mde->mde_len + new_namelen - old_namelen;
4683         OBD_ALLOC(name_buf, name_buflen);
4684         if (!name_buf)
4685                 RETURN(-ENOMEM);
4686
4687         memcpy(name_buf, newname, new_namelen);
4688         memcpy(name_buf + new_namelen, mde->mde_name + old_namelen,
4689                mde->mde_len - old_namelen);
4690
4691         CDEBUG(D_MGS, "Fork the config-log from %s to %s\n",
4692                mde->mde_name, name_buf);
4693
4694         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
4695         LASSERT(ctxt);
4696
4697         rc = llog_open_create(env, ctxt, &new_llh, NULL, name_buf);
4698         if (rc)
4699                 GOTO(out, rc);
4700
4701         rc = llog_init_handle(env, new_llh, LLOG_F_IS_PLAIN, NULL);
4702         if (rc)
4703                 GOTO(out, rc);
4704
4705         if (unlikely(mgs_log_is_empty(env, mgs, mde->mde_name)))
4706                 GOTO(out, rc = 0);
4707
4708         rc = llog_open(env, ctxt, &old_llh, NULL, mde->mde_name,
4709                        LLOG_OPEN_EXISTS);
4710         if (rc)
4711                 GOTO(out, rc);
4712
4713         rc = llog_init_handle(env, old_llh, LLOG_F_IS_PLAIN, NULL);
4714         if (rc)
4715                 GOTO(out, rc);
4716
4717         new_llh->lgh_hdr->llh_tgtuuid = old_llh->lgh_hdr->llh_tgtuuid;
4718
4719         OBD_ALLOC(mlfd, LLOG_MIN_CHUNK_SIZE);
4720         if (!mlfd)
4721                 GOTO(out, rc = -ENOMEM);
4722
4723         mlfd->mlfd_mgs = mgs;
4724         mlfd->mlfd_llh = new_llh;
4725         mlfd->mlfd_oldname = oldname;
4726         mlfd->mlfd_newname = newname;
4727
4728         rc = llog_process(env, old_llh, mgs_lcfg_fork_handler, mlfd, NULL);
4729         OBD_FREE(mlfd, LLOG_MIN_CHUNK_SIZE);
4730
4731         GOTO(out, rc);
4732
4733 out:
4734         if (old_llh)
4735                 llog_close(env, old_llh);
4736         if (new_llh)
4737                 llog_close(env, new_llh);
4738         if (name_buf)
4739                 OBD_FREE(name_buf, name_buflen);
4740         if (ctxt)
4741                 llog_ctxt_put(ctxt);
4742
4743         return rc;
4744 }
4745
4746 int mgs_lcfg_fork(const struct lu_env *env, struct mgs_device *mgs,
4747                   const char *oldname, const char *newname)
4748 {
4749         struct list_head log_list;
4750         struct mgs_direntry *dirent, *n;
4751         int olen = strlen(oldname);
4752         int nlen = strlen(newname);
4753         int count = 0;
4754         int rc = 0;
4755         ENTRY;
4756
4757         if (unlikely(!oldname || oldname[0] == '\0' ||
4758                      !newname || newname[0] == '\0'))
4759                 RETURN(-EINVAL);
4760
4761         if (strcmp(oldname, newname) == 0)
4762                 RETURN(-EINVAL);
4763
4764         /* lock it to prevent fork/erase/register in parallel. */
4765         mutex_lock(&mgs->mgs_mutex);
4766
4767         rc = class_dentry_readdir(env, mgs, &log_list);
4768         if (rc) {
4769                 mutex_unlock(&mgs->mgs_mutex);
4770                 RETURN(rc);
4771         }
4772
4773         if (list_empty(&log_list)) {
4774                 mutex_unlock(&mgs->mgs_mutex);
4775                 RETURN(-ENOENT);
4776         }
4777
4778         list_for_each_entry_safe(dirent, n, &log_list, mde_list) {
4779                 char *ptr;
4780
4781                 ptr = strrchr(dirent->mde_name, '-');
4782                 if (ptr) {
4783                         int tlen = ptr - dirent->mde_name;
4784
4785                         if (tlen == nlen &&
4786                             strncmp(newname, dirent->mde_name, tlen) == 0)
4787                                 GOTO(out, rc = -EEXIST);
4788
4789                         if (tlen == olen &&
4790                             strncmp(oldname, dirent->mde_name, tlen) == 0)
4791                                 continue;
4792                 }
4793
4794                 list_del_init(&dirent->mde_list);
4795                 mgs_direntry_free(dirent);
4796         }
4797
4798         if (list_empty(&log_list)) {
4799                 mutex_unlock(&mgs->mgs_mutex);
4800                 RETURN(-ENOENT);
4801         }
4802
4803         list_for_each_entry(dirent, &log_list, mde_list) {
4804                 rc = mgs_lcfg_fork_one(env, mgs, dirent, oldname, newname);
4805                 if (rc)
4806                         break;
4807
4808                 count++;
4809         }
4810
4811 out:
4812         mutex_unlock(&mgs->mgs_mutex);
4813
4814         list_for_each_entry_safe(dirent, n, &log_list, mde_list) {
4815                 list_del_init(&dirent->mde_list);
4816                 mgs_direntry_free(dirent);
4817         }
4818
4819         if (rc && count > 0)
4820                 mgs_erase_logs(env, mgs, newname);
4821
4822         RETURN(rc);
4823 }
4824
4825 int mgs_lcfg_erase(const struct lu_env *env, struct mgs_device *mgs,
4826                    const char *fsname)
4827 {
4828         int rc;
4829         ENTRY;
4830
4831         if (unlikely(!fsname || fsname[0] == '\0'))
4832                 RETURN(-EINVAL);
4833
4834         rc = mgs_erase_logs(env, mgs, fsname);
4835
4836         RETURN(rc);
4837 }
4838
4839 static int mgs_xattr_del(const struct lu_env *env, struct dt_object *obj)
4840 {
4841         struct dt_device *dev;
4842         struct thandle *th = NULL;
4843         int rc = 0;
4844
4845         ENTRY;
4846
4847         dev = container_of0(obj->do_lu.lo_dev, struct dt_device, dd_lu_dev);
4848         th = dt_trans_create(env, dev);
4849         if (IS_ERR(th))
4850                 RETURN(PTR_ERR(th));
4851
4852         rc = dt_declare_xattr_del(env, obj, XATTR_TARGET_RENAME, th);
4853         if (rc)
4854                 GOTO(stop, rc);
4855
4856         rc = dt_trans_start_local(env, dev, th);
4857         if (rc)
4858                 GOTO(stop, rc);
4859
4860         dt_write_lock(env, obj, 0);
4861         rc = dt_xattr_del(env, obj, XATTR_TARGET_RENAME, th);
4862
4863         GOTO(unlock, rc);
4864
4865 unlock:
4866         dt_write_unlock(env, obj);
4867
4868 stop:
4869         dt_trans_stop(env, dev, th);
4870
4871         return rc;
4872 }
4873
4874 int mgs_lcfg_rename(const struct lu_env *env, struct mgs_device *mgs)
4875 {
4876         struct list_head log_list;
4877         struct mgs_direntry *dirent, *n;
4878         char fsname[16];
4879         struct lu_buf buf = {
4880                 .lb_buf = fsname,
4881                 .lb_len = sizeof(fsname)
4882         };
4883         int rc = 0;
4884
4885         ENTRY;
4886
4887         rc = class_dentry_readdir(env, mgs, &log_list);
4888         if (rc)
4889                 RETURN(rc);
4890
4891         if (list_empty(&log_list))
4892                 RETURN(0);
4893
4894         list_for_each_entry_safe(dirent, n, &log_list, mde_list) {
4895                 struct dt_object *o = NULL;
4896                 char oldname[16];
4897                 char *ptr;
4898                 int len;
4899
4900                 list_del_init(&dirent->mde_list);
4901                 ptr = strrchr(dirent->mde_name, '-');
4902                 if (!ptr)
4903                         goto next;
4904
4905                 len = ptr - dirent->mde_name;
4906                 if (unlikely(len >= sizeof(oldname))) {
4907                         CDEBUG(D_MGS, "Skip invalid configuration file %s\n",
4908                                dirent->mde_name);
4909                         goto next;
4910                 }
4911
4912                 o = local_file_find(env, mgs->mgs_los, mgs->mgs_configs_dir,
4913                                     dirent->mde_name);
4914                 if (IS_ERR(o)) {
4915                         rc = PTR_ERR(o);
4916                         CDEBUG(D_MGS, "Fail to locate file %s: rc = %d\n",
4917                                dirent->mde_name, rc);
4918                         goto next;
4919                 }
4920
4921                 rc = dt_xattr_get(env, o, &buf, XATTR_TARGET_RENAME);
4922                 if (rc < 0) {
4923                         if (rc == -ENODATA)
4924                                 rc = 0;
4925                         else
4926                                 CDEBUG(D_MGS,
4927                                        "Fail to get EA for %s: rc = %d\n",
4928                                        dirent->mde_name, rc);
4929                         goto next;
4930                 }
4931
4932                 if (unlikely(rc == len &&
4933                              memcmp(fsname, dirent->mde_name, len) == 0)) {
4934                         /* The new fsname is the same as the old one. */
4935                         rc = mgs_xattr_del(env, o);
4936                         goto next;
4937                 }
4938
4939                 memcpy(oldname, dirent->mde_name, len);
4940                 oldname[len] = '\0';
4941                 fsname[rc] = '\0';
4942                 rc = mgs_lcfg_fork_one(env, mgs, dirent, oldname, fsname);
4943                 if (rc && rc != -EEXIST) {
4944                         CDEBUG(D_MGS, "Fail to fork %s: rc = %d\n",
4945                                dirent->mde_name, rc);
4946                         goto next;
4947                 }
4948
4949                 rc = mgs_erase_log(env, mgs, dirent->mde_name);
4950                 if (rc) {
4951                         CDEBUG(D_MGS, "Fail to erase old %s: rc = %d\n",
4952                                dirent->mde_name, rc);
4953                         /* keep it there if failed to remove it. */
4954                         rc = 0;
4955                 }
4956
4957 next:
4958                 if (o && !IS_ERR(o))
4959                         lu_object_put(env, &o->do_lu);
4960
4961                 mgs_direntry_free(dirent);
4962                 if (rc)
4963                         break;
4964         }
4965
4966         list_for_each_entry_safe(dirent, n, &log_list, mde_list) {
4967                 list_del_init(&dirent->mde_list);
4968                 mgs_direntry_free(dirent);
4969         }
4970
4971         RETURN(rc);
4972 }
4973
4974 /* Setup _mgs fsdb and log
4975  */
4976 int mgs__mgs_fsdb_setup(const struct lu_env *env, struct mgs_device *mgs)
4977 {
4978         struct fs_db *fsdb = NULL;
4979         int rc;
4980         ENTRY;
4981
4982         rc = mgs_find_or_make_fsdb(env, mgs, MGSSELF_NAME, &fsdb);
4983         if (!rc)
4984                 mgs_put_fsdb(mgs, fsdb);
4985
4986         RETURN(rc);
4987 }
4988
4989 /* Setup params fsdb and log
4990  */
4991 int mgs_params_fsdb_setup(const struct lu_env *env, struct mgs_device *mgs)
4992 {
4993         struct fs_db *fsdb = NULL;
4994         struct llog_handle *params_llh = NULL;
4995         int rc;
4996         ENTRY;
4997
4998         rc = mgs_find_or_make_fsdb(env, mgs, PARAMS_FILENAME, &fsdb);
4999         if (!rc) {
5000                 mutex_lock(&fsdb->fsdb_mutex);
5001                 rc = record_start_log(env, mgs, &params_llh, PARAMS_FILENAME);
5002                 if (!rc)
5003                         rc = record_end_log(env, &params_llh);
5004                 mutex_unlock(&fsdb->fsdb_mutex);
5005                 mgs_put_fsdb(mgs, fsdb);
5006         }
5007
5008         RETURN(rc);
5009 }
5010
5011 /* Cleanup params fsdb and log
5012  */
5013 int mgs_params_fsdb_cleanup(const struct lu_env *env, struct mgs_device *mgs)
5014 {
5015         int rc;
5016
5017         rc = mgs_erase_logs(env, mgs, PARAMS_FILENAME);
5018         return rc == -ENOENT ? 0 : rc;
5019 }
5020
5021 /**
5022  * Fill in the mgs_target_info based on data devname and param provide.
5023  *
5024  * @env         thread context
5025  * @mgs         mgs device
5026  * @mti         mgs target info. We want to set this based other paramters
5027  *              passed to this function. Once setup we write it to the config
5028  *              logs.
5029  * @devname     optional OBD device name
5030  * @param       string that contains both what tunable to set and the value to
5031  *              set it to.
5032  *
5033  * RETURN       0 for success
5034  *              negative error number on failure
5035  **/
5036 static int mgs_set_conf_param(const struct lu_env *env, struct mgs_device *mgs,
5037                               struct mgs_target_info *mti, const char *devname,
5038                               const char *param)
5039 {
5040         struct fs_db *fsdb = NULL;
5041         int dev_type;
5042         int rc = 0;
5043
5044         ENTRY;
5045         /* lustre, lustre-mdtlov, lustre-client, lustre-MDT0000 */
5046         if (!devname) {
5047                 size_t len;
5048
5049                 /* We have two possible cases here:
5050                  *
5051                  * 1) the device name embedded in the param:
5052                  *    lustre-OST0000.osc.max_dirty_mb=32
5053                  *
5054                  * 2) the file system name is embedded in
5055                  *    the param: lustre.sys.at.min=0
5056                  */
5057                 len = strcspn(param, ".=");
5058                 if (!len || param[len] == '=')
5059                         RETURN(-EINVAL);
5060
5061                 if (len >= sizeof(mti->mti_svname))
5062                         RETURN(-E2BIG);
5063
5064                 snprintf(mti->mti_svname, sizeof(mti->mti_svname),
5065                          "%.*s", (int)len, param);
5066                 param += len + 1;
5067         } else {
5068                 if (strlcpy(mti->mti_svname, devname, sizeof(mti->mti_svname)) >=
5069                     sizeof(mti->mti_svname))
5070                         RETURN(-E2BIG);
5071         }
5072
5073         if (!strlen(mti->mti_svname)) {
5074                 LCONSOLE_ERROR_MSG(0x14d, "No target specified: %s\n", param);
5075                 RETURN(-ENOSYS);
5076         }
5077
5078         dev_type = mgs_parse_devname(mti->mti_svname, mti->mti_fsname,
5079                                      &mti->mti_stripe_index);
5080         switch (dev_type) {
5081         /* For this case we have an invalid obd device name */
5082         case -ENXIO:
5083                 CDEBUG(D_MGS, "%s don't contain an index\n", mti->mti_svname);
5084                 strlcpy(mti->mti_fsname, mti->mti_svname, MTI_NAME_MAXLEN);
5085                 dev_type = 0;
5086                 break;
5087         /* Not an obd device, assume devname is the fsname.
5088          * User might of only provided fsname and not obd device
5089          */
5090         case -EINVAL:
5091                 CDEBUG(D_MGS, "%s is seen as a file system name\n", mti->mti_svname);
5092                 strlcpy(mti->mti_fsname, mti->mti_svname, MTI_NAME_MAXLEN);
5093                 dev_type = 0;
5094                 break;
5095         default:
5096                 if (dev_type < 0)
5097                         GOTO(out, rc = dev_type);
5098
5099                 /* param related to llite isn't allowed to set by OST or MDT */
5100                 if (dev_type & LDD_F_SV_TYPE_OST ||
5101                     dev_type & LDD_F_SV_TYPE_MDT) {
5102                         /* param related to llite isn't allowed to set by OST
5103                          * or MDT
5104                          */
5105                         if (!strncmp(param, PARAM_LLITE,
5106                                      sizeof(PARAM_LLITE) - 1))
5107                                 GOTO(out, rc = -EINVAL);
5108
5109                         /* Strip -osc or -mdc suffix from svname */
5110                         if (server_make_name(dev_type, mti->mti_stripe_index,
5111                                              mti->mti_fsname, mti->mti_svname,
5112                                              sizeof(mti->mti_svname)))
5113                                 GOTO(out, rc = -EINVAL);
5114                 }
5115                 break;
5116         }
5117
5118         if (strlcpy(mti->mti_params, param, sizeof(mti->mti_params)) >=
5119             sizeof(mti->mti_params))
5120                 GOTO(out, rc = -E2BIG);
5121
5122         CDEBUG(D_MGS, "set_conf_param fs='%s' device='%s' param='%s'\n",
5123                mti->mti_fsname, mti->mti_svname, mti->mti_params);
5124
5125         rc = mgs_find_or_make_fsdb(env, mgs, mti->mti_fsname, &fsdb);
5126         if (rc)
5127                 GOTO(out, rc);
5128
5129         if (!test_bit(FSDB_MGS_SELF, &fsdb->fsdb_flags) &&
5130             test_bit(FSDB_LOG_EMPTY, &fsdb->fsdb_flags)) {
5131                 CERROR("No filesystem targets for %s. cfg_device from lctl "
5132                        "is '%s'\n", mti->mti_fsname, mti->mti_svname);
5133                 mgs_unlink_fsdb(mgs, fsdb);
5134                 GOTO(out, rc = -EINVAL);
5135         }
5136
5137         /*
5138          * Revoke lock so everyone updates.  Should be alright if
5139          * someone was already reading while we were updating the logs,
5140          * so we don't really need to hold the lock while we're
5141          * writing (above).
5142          */
5143         mti->mti_flags = dev_type | LDD_F_PARAM;
5144         mutex_lock(&fsdb->fsdb_mutex);
5145         rc = mgs_write_log_param(env, mgs, fsdb, mti, mti->mti_params);
5146         mutex_unlock(&fsdb->fsdb_mutex);
5147         mgs_revoke_lock(mgs, fsdb, CONFIG_T_CONFIG);
5148
5149 out:
5150         if (fsdb)
5151                 mgs_put_fsdb(mgs, fsdb);
5152
5153         RETURN(rc);
5154 }
5155
5156 static int mgs_set_param2(const struct lu_env *env, struct mgs_device *mgs,
5157                           struct mgs_target_info *mti, const char *param)
5158 {
5159         struct fs_db *fsdb = NULL;
5160         int dev_type;
5161         size_t len;
5162         int rc;
5163
5164         if (strlcpy(mti->mti_params, param, sizeof(mti->mti_params)) >=
5165             sizeof(mti->mti_params))
5166                 GOTO(out, rc = -E2BIG);
5167
5168         /* obdname2fsname reports devname as an obd device */
5169         len = strcspn(param, ".=");
5170         if (len && param[len] != '=') {
5171                 char *ptr;
5172
5173                 param += len + 1;
5174                 ptr = strchr(param, '.');
5175
5176                 len = strlen(param);
5177                 if (ptr)
5178                         len -= strlen(ptr);
5179                 if (len >= sizeof(mti->mti_svname))
5180                         GOTO(out, rc = -E2BIG);
5181
5182                 snprintf(mti->mti_svname, sizeof(mti->mti_svname), "%.*s",
5183                         (int)len, param);
5184
5185                 obdname2fsname(mti->mti_svname, mti->mti_fsname,
5186                                sizeof(mti->mti_fsname));
5187         } else {
5188                 snprintf(mti->mti_svname, sizeof(mti->mti_svname), "general");
5189         }
5190
5191         CDEBUG(D_MGS, "set_param2 fs='%s' device='%s' param='%s'\n",
5192                mti->mti_fsname, mti->mti_svname, mti->mti_params);
5193
5194         /* The return value should be the device type i.e LDD_F_SV_TYPE_XXX.
5195          * A returned error tells us we don't have a target obd device.
5196          */
5197         dev_type = server_name2index(mti->mti_svname, &mti->mti_stripe_index,
5198                                      NULL);
5199         if (dev_type < 0)
5200                 dev_type = 0;
5201
5202         /* the return value should be the device type i.e LDD_F_SV_TYPE_XXX.
5203          * Strip -osc or -mdc suffix from svname
5204          */
5205         if ((dev_type & LDD_F_SV_TYPE_OST || dev_type & LDD_F_SV_TYPE_MDT) &&
5206             server_make_name(dev_type, mti->mti_stripe_index,
5207                              mti->mti_fsname, mti->mti_svname,
5208                              sizeof(mti->mti_svname)))
5209                 GOTO(out, rc = -EINVAL);
5210
5211         rc = mgs_find_or_make_fsdb(env, mgs, PARAMS_FILENAME, &fsdb);
5212         if (rc)
5213                 GOTO(out, rc);
5214         /*
5215          * Revoke lock so everyone updates.  Should be alright if
5216          * someone was already reading while we were updating the logs,
5217          * so we don't really need to hold the lock while we're
5218          * writing (above).
5219          */
5220         mti->mti_flags = dev_type | LDD_F_PARAM2;
5221         mutex_lock(&fsdb->fsdb_mutex);
5222         rc = mgs_write_log_param2(env, mgs, fsdb, mti, mti->mti_params);
5223         mutex_unlock(&fsdb->fsdb_mutex);
5224         mgs_revoke_lock(mgs, fsdb, CONFIG_T_PARAMS);
5225         mgs_put_fsdb(mgs, fsdb);
5226 out:
5227         RETURN(rc);
5228 }
5229
5230 /* Set a permanent (config log) param for a target or fs
5231  *
5232  * @lcfg buf0 may contain the device (testfs-MDT0000) name
5233  *       buf1 contains the single parameter
5234  */
5235 int mgs_set_param(const struct lu_env *env, struct mgs_device *mgs,
5236                   struct lustre_cfg *lcfg)
5237 {
5238         const char *param = lustre_cfg_string(lcfg, 1);
5239         struct mgs_target_info *mti;
5240         int rc;
5241
5242         /* Create a fake mti to hold everything */
5243         OBD_ALLOC_PTR(mti);
5244         if (!mti)
5245                 return -ENOMEM;
5246
5247         print_lustre_cfg(lcfg);
5248
5249         if (lcfg->lcfg_command == LCFG_PARAM) {
5250                 /* For the case of lctl conf_param devname can be
5251                  * lustre, lustre-mdtlov, lustre-client, lustre-MDT0000
5252                  */
5253                 const char *devname = lustre_cfg_string(lcfg, 0);
5254
5255                 rc = mgs_set_conf_param(env, mgs, mti, devname, param);
5256         } else {
5257                 /* In the case of lctl set_param -P lcfg[0] will always
5258                  * be 'general'. At least for now.
5259                  */
5260                 rc = mgs_set_param2(env, mgs, mti, param);
5261         }
5262
5263         OBD_FREE_PTR(mti);
5264
5265         return rc;
5266 }
5267
5268 static int mgs_write_log_pool(const struct lu_env *env,
5269                               struct mgs_device *mgs, char *logname,
5270                               struct fs_db *fsdb, char *tgtname,
5271                               enum lcfg_command_type cmd,
5272                               char *fsname, char *poolname,
5273                               char *ostname, char *comment)
5274 {
5275         struct llog_handle *llh = NULL;
5276         int rc;
5277
5278         rc = record_start_log(env, mgs, &llh, logname);
5279         if (rc)
5280                 return rc;
5281         rc = record_marker(env, llh, fsdb, CM_START, tgtname, comment);
5282         if (rc)
5283                 goto out;
5284         rc = record_base(env, llh, tgtname, 0, cmd,
5285                          fsname, poolname, ostname, NULL);
5286         if (rc)
5287                 goto out;
5288         rc = record_marker(env, llh, fsdb, CM_END, tgtname, comment);
5289 out:
5290         record_end_log(env, &llh);
5291         return rc;
5292 }
5293
5294 int mgs_nodemap_cmd(const struct lu_env *env, struct mgs_device *mgs,
5295                     enum lcfg_command_type cmd, const char *nodemap_name,
5296                     char *param)
5297 {
5298         lnet_nid_t      nid[2];
5299         __u32           idmap[2];
5300         bool            bool_switch;
5301         __u32           int_id;
5302         int             rc = 0;
5303         ENTRY;
5304
5305         switch (cmd) {
5306         case LCFG_NODEMAP_ADD:
5307                 rc = nodemap_add(nodemap_name);
5308                 break;
5309         case LCFG_NODEMAP_DEL:
5310                 rc = nodemap_del(nodemap_name);
5311                 break;
5312         case LCFG_NODEMAP_ADD_RANGE:
5313                 rc = nodemap_parse_range(param, nid);
5314                 if (rc != 0)
5315                         break;
5316                 rc = nodemap_add_range(nodemap_name, nid);
5317                 break;
5318         case LCFG_NODEMAP_DEL_RANGE:
5319                 rc = nodemap_parse_range(param, nid);
5320                 if (rc != 0)
5321                         break;
5322                 rc = nodemap_del_range(nodemap_name, nid);
5323                 break;
5324         case LCFG_NODEMAP_ADMIN:
5325                 bool_switch = simple_strtoul(param, NULL, 10);
5326                 rc = nodemap_set_allow_root(nodemap_name, bool_switch);
5327                 break;
5328         case LCFG_NODEMAP_DENY_UNKNOWN:
5329                 bool_switch = simple_strtoul(param, NULL, 10);
5330                 rc = nodemap_set_deny_unknown(nodemap_name, bool_switch);
5331                 break;
5332         case LCFG_NODEMAP_MAP_MODE:
5333                 if (strcmp("both", param) == 0)
5334                         rc = nodemap_set_mapping_mode(nodemap_name,
5335                                                       NODEMAP_MAP_BOTH);
5336                 else if (strcmp("uid_only", param) == 0)
5337                         rc = nodemap_set_mapping_mode(nodemap_name,
5338                                                       NODEMAP_MAP_UID_ONLY);
5339                 else if (strcmp("gid_only", param) == 0)
5340                         rc = nodemap_set_mapping_mode(nodemap_name,
5341                                                       NODEMAP_MAP_GID_ONLY);
5342                 else
5343                         rc = -EINVAL;
5344                 break;
5345         case LCFG_NODEMAP_TRUSTED:
5346                 bool_switch = simple_strtoul(param, NULL, 10);
5347                 rc = nodemap_set_trust_client_ids(nodemap_name, bool_switch);
5348                 break;
5349         case LCFG_NODEMAP_SQUASH_UID:
5350                 int_id = simple_strtoul(param, NULL, 10);
5351                 rc = nodemap_set_squash_uid(nodemap_name, int_id);
5352                 break;
5353         case LCFG_NODEMAP_SQUASH_GID:
5354                 int_id = simple_strtoul(param, NULL, 10);
5355                 rc = nodemap_set_squash_gid(nodemap_name, int_id);
5356                 break;
5357         case LCFG_NODEMAP_ADD_UIDMAP:
5358         case LCFG_NODEMAP_ADD_GIDMAP:
5359                 rc = nodemap_parse_idmap(param, idmap);
5360                 if (rc != 0)
5361                         break;
5362                 if (cmd == LCFG_NODEMAP_ADD_UIDMAP)
5363                         rc = nodemap_add_idmap(nodemap_name, NODEMAP_UID,
5364                                                idmap);
5365                 else
5366                         rc = nodemap_add_idmap(nodemap_name, NODEMAP_GID,
5367                                                idmap);
5368                 break;
5369         case LCFG_NODEMAP_DEL_UIDMAP:
5370         case LCFG_NODEMAP_DEL_GIDMAP:
5371                 rc = nodemap_parse_idmap(param, idmap);
5372                 if (rc != 0)
5373                         break;
5374                 if (cmd == LCFG_NODEMAP_DEL_UIDMAP)
5375                         rc = nodemap_del_idmap(nodemap_name, NODEMAP_UID,
5376                                                idmap);
5377                 else
5378                         rc = nodemap_del_idmap(nodemap_name, NODEMAP_GID,
5379                                                idmap);
5380                 break;
5381         case LCFG_NODEMAP_SET_FILESET:
5382                 rc = nodemap_set_fileset(nodemap_name, param);
5383                 break;
5384         default:
5385                 rc = -EINVAL;
5386         }
5387
5388         RETURN(rc);
5389 }
5390
5391 int mgs_pool_cmd(const struct lu_env *env, struct mgs_device *mgs,
5392                  enum lcfg_command_type cmd, char *fsname,
5393                  char *poolname, char *ostname)
5394 {
5395         struct fs_db *fsdb;
5396         char *lovname;
5397         char *logname;
5398         char *label = NULL, *canceled_label = NULL;
5399         int label_sz;
5400         struct mgs_target_info *mti = NULL;
5401         bool checked = false;
5402         bool locked = false;
5403         bool free = false;
5404         int rc, i;
5405         ENTRY;
5406
5407         rc = mgs_find_or_make_fsdb(env, mgs, fsname, &fsdb);
5408         if (rc) {
5409                 CERROR("Can't get db for %s\n", fsname);
5410                 RETURN(rc);
5411         }
5412         if (test_bit(FSDB_LOG_EMPTY, &fsdb->fsdb_flags)) {
5413                 CERROR("%s is not defined\n", fsname);
5414                 free = true;
5415                 GOTO(out_fsdb, rc = -EINVAL);
5416         }
5417
5418         label_sz = 10 + strlen(fsname) + strlen(poolname);
5419
5420         /* check if ostname match fsname */
5421         if (ostname != NULL) {
5422                 char *ptr;
5423
5424                 ptr = strrchr(ostname, '-');
5425                 if ((ptr == NULL) ||
5426                     (strncmp(fsname, ostname, ptr-ostname) != 0))
5427                         RETURN(-EINVAL);
5428                 label_sz += strlen(ostname);
5429         }
5430
5431         OBD_ALLOC(label, label_sz);
5432         if (!label)
5433                 GOTO(out_fsdb, rc = -ENOMEM);
5434
5435         switch(cmd) {
5436         case LCFG_POOL_NEW:
5437                 sprintf(label,
5438                         "new %s.%s", fsname, poolname);
5439                 break;
5440         case LCFG_POOL_ADD:
5441                 sprintf(label,
5442                         "add %s.%s.%s", fsname, poolname, ostname);
5443                 break;
5444         case LCFG_POOL_REM:
5445                 OBD_ALLOC(canceled_label, label_sz);
5446                 if (canceled_label == NULL)
5447                         GOTO(out_label, rc = -ENOMEM);
5448                 sprintf(label,
5449                         "rem %s.%s.%s", fsname, poolname, ostname);
5450                 sprintf(canceled_label,
5451                         "add %s.%s.%s", fsname, poolname, ostname);
5452                 break;
5453         case LCFG_POOL_DEL:
5454                 OBD_ALLOC(canceled_label, label_sz);
5455                 if (canceled_label == NULL)
5456                         GOTO(out_label, rc = -ENOMEM);
5457                 sprintf(label,
5458                         "del %s.%s", fsname, poolname);
5459                 sprintf(canceled_label,
5460                         "new %s.%s", fsname, poolname);
5461                 break;
5462         default:
5463                 break;
5464         }
5465
5466         OBD_ALLOC_PTR(mti);
5467         if (mti == NULL)
5468                 GOTO(out_cancel, rc = -ENOMEM);
5469         strncpy(mti->mti_svname, "lov pool", sizeof(mti->mti_svname));
5470
5471         mutex_lock(&fsdb->fsdb_mutex);
5472         locked = true;
5473         /* write pool def to all MDT logs */
5474         for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
5475                 if (test_bit(i,  fsdb->fsdb_mdt_index_map)) {
5476                         rc = name_create_mdt_and_lov(&logname, &lovname,
5477                                                      fsdb, i);
5478                         if (rc)
5479                                 GOTO(out_mti, rc);
5480
5481                         if (!checked && (canceled_label == NULL)) {
5482                                 rc = mgs_check_marker(env, mgs, fsdb, mti,
5483                                                 logname, lovname, label);
5484                                 if (rc) {
5485                                         name_destroy(&logname);
5486                                         name_destroy(&lovname);
5487                                         GOTO(out_mti,
5488                                                 rc = (rc == LLOG_PROC_BREAK ?
5489                                                         -EEXIST : rc));
5490                                 }
5491                                 checked = true;
5492                         }
5493                         if (canceled_label != NULL)
5494                                 rc = mgs_modify(env, mgs, fsdb, mti, logname,
5495                                                 lovname, canceled_label,
5496                                                 CM_SKIP);
5497
5498                         if (rc >= 0)
5499                                 rc = mgs_write_log_pool(env, mgs, logname,
5500                                                         fsdb, lovname, cmd,
5501                                                         fsname, poolname,
5502                                                         ostname, label);
5503                         name_destroy(&logname);
5504                         name_destroy(&lovname);
5505                         if (rc)
5506                                 GOTO(out_mti, rc);
5507                 }
5508         }
5509
5510         rc = name_create(&logname, fsname, "-client");
5511         if (rc)
5512                 GOTO(out_mti, rc);
5513
5514         if (!checked && (canceled_label == NULL)) {
5515                 rc = mgs_check_marker(env, mgs, fsdb, mti, logname,
5516                                 fsdb->fsdb_clilov, label);
5517                 if (rc) {
5518                         name_destroy(&logname);
5519                         GOTO(out_mti, rc = (rc == LLOG_PROC_BREAK ?
5520                                 -EEXIST : rc));
5521                 }
5522         }
5523         if (canceled_label != NULL) {
5524                 rc = mgs_modify(env, mgs, fsdb, mti, logname,
5525                                 fsdb->fsdb_clilov, canceled_label, CM_SKIP);
5526                 if (rc < 0) {
5527                         name_destroy(&logname);
5528                         GOTO(out_mti, rc);
5529                 }
5530         }
5531
5532         rc = mgs_write_log_pool(env, mgs, logname, fsdb, fsdb->fsdb_clilov,
5533                                 cmd, fsname, poolname, ostname, label);
5534         mutex_unlock(&fsdb->fsdb_mutex);
5535         locked = false;
5536         name_destroy(&logname);
5537         /* request for update */
5538         mgs_revoke_lock(mgs, fsdb, CONFIG_T_CONFIG);
5539
5540         GOTO(out_mti, rc);
5541
5542 out_mti:
5543         if (locked)
5544                 mutex_unlock(&fsdb->fsdb_mutex);
5545         if (mti != NULL)
5546                 OBD_FREE_PTR(mti);
5547 out_cancel:
5548         if (canceled_label != NULL)
5549                 OBD_FREE(canceled_label, label_sz);
5550 out_label:
5551         OBD_FREE(label, label_sz);
5552 out_fsdb:
5553         if (free)
5554                 mgs_unlink_fsdb(mgs, fsdb);
5555         mgs_put_fsdb(mgs, fsdb);
5556
5557         return rc;
5558 }