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