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                     strcmp(tmp, "none") != 0) {
3132                         CERROR("enable option(%s) isn't supported\n", tmp);
3133                         return -EINVAL;
3134                 }
3135         }
3136
3137         lustre_cfg_bufs_reset(&mgi->mgi_bufs, mti->mti_fsname);
3138         lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 1, quota);
3139         lcr = lustre_cfg_rec_new(cmd, &mgi->mgi_bufs);
3140         if (lcr == NULL)
3141                 return -ENOMEM;
3142
3143         /* truncate the comment to the parameter name */
3144         ptr = tmp - 1;
3145         sep = *ptr;
3146         *ptr = '\0';
3147
3148         /* XXX we duplicated quota enable information in all server
3149          *     config logs, it should be moved to a separate config
3150          *     log once we cleanup the config log for global param. */
3151         /* modify all servers */
3152         rc = mgs_write_log_direct_all(env, mgs, fsdb, mti,
3153                                       *tmp == '\0' ? NULL : lcr,
3154                                       mti->mti_fsname, quota, 1);
3155         *ptr = sep;
3156         lustre_cfg_rec_free(lcr);
3157         return rc < 0 ? rc : 0;
3158 }
3159
3160 static int mgs_srpc_set_param_disk(const struct lu_env *env,
3161                                    struct mgs_device *mgs,
3162                                    struct fs_db *fsdb,
3163                                    struct mgs_target_info *mti,
3164                                    char *param)
3165 {
3166         struct mgs_thread_info  *mgi = mgs_env_info(env);
3167         struct llog_cfg_rec     *lcr;
3168         struct llog_handle      *llh = NULL;
3169         char                    *logname;
3170         char                    *comment, *ptr;
3171         int                      rc, len;
3172
3173         ENTRY;
3174
3175         /* get comment */
3176         ptr = strchr(param, '=');
3177         LASSERT(ptr != NULL);
3178         len = ptr - param;
3179
3180         OBD_ALLOC(comment, len + 1);
3181         if (comment == NULL)
3182                 RETURN(-ENOMEM);
3183         strncpy(comment, param, len);
3184         comment[len] = '\0';
3185
3186         /* prepare lcfg */
3187         lustre_cfg_bufs_reset(&mgi->mgi_bufs, mti->mti_svname);
3188         lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 1, param);
3189         lcr = lustre_cfg_rec_new(LCFG_SPTLRPC_CONF, &mgi->mgi_bufs);
3190         if (lcr == NULL)
3191                 GOTO(out_comment, rc = -ENOMEM);
3192
3193         /* construct log name */
3194         rc = name_create(&logname, mti->mti_fsname, "-sptlrpc");
3195         if (rc < 0)
3196                 GOTO(out_lcfg, rc);
3197
3198         if (mgs_log_is_empty(env, mgs, logname)) {
3199                 rc = record_start_log(env, mgs, &llh, logname);
3200                 if (rc < 0)
3201                         GOTO(out, rc);
3202                 record_end_log(env, &llh);
3203         }
3204
3205         /* obsolete old one */
3206         rc = mgs_modify(env, mgs, fsdb, mti, logname, mti->mti_svname,
3207                         comment, CM_SKIP);
3208         if (rc < 0)
3209                 GOTO(out, rc);
3210         /* write the new one */
3211         rc = mgs_write_log_direct(env, mgs, fsdb, logname, lcr,
3212                                   mti->mti_svname, comment);
3213         if (rc)
3214                 CERROR("%s: error writing log %s: rc = %d\n",
3215                        mgs->mgs_obd->obd_name, logname, rc);
3216 out:
3217         name_destroy(&logname);
3218 out_lcfg:
3219         lustre_cfg_rec_free(lcr);
3220 out_comment:
3221         OBD_FREE(comment, len + 1);
3222         RETURN(rc);
3223 }
3224
3225 static int mgs_srpc_set_param_udesc_mem(struct fs_db *fsdb,
3226                                         char *param)
3227 {
3228         char    *ptr;
3229
3230         /* disable the adjustable udesc parameter for now, i.e. use default
3231          * setting that client always ship udesc to MDT if possible. to enable
3232          * it simply remove the following line */
3233         goto error_out;
3234
3235         ptr = strchr(param, '=');
3236         if (ptr == NULL)
3237                 goto error_out;
3238         *ptr++ = '\0';
3239
3240         if (strcmp(param, PARAM_SRPC_UDESC))
3241                 goto error_out;
3242
3243         if (strcmp(ptr, "yes") == 0) {
3244                 set_bit(FSDB_UDESC, &fsdb->fsdb_flags);
3245                 CWARN("Enable user descriptor shipping from client to MDT\n");
3246         } else if (strcmp(ptr, "no") == 0) {
3247                 clear_bit(FSDB_UDESC, &fsdb->fsdb_flags);
3248                 CWARN("Disable user descriptor shipping from client to MDT\n");
3249         } else {
3250                 *(ptr - 1) = '=';
3251                 goto error_out;
3252         }
3253         return 0;
3254
3255 error_out:
3256         CERROR("Invalid param: %s\n", param);
3257         return -EINVAL;
3258 }
3259
3260 static int mgs_srpc_set_param_mem(struct fs_db *fsdb,
3261                                   const char *svname,
3262                                   char *param)
3263 {
3264         struct sptlrpc_rule      rule;
3265         struct sptlrpc_rule_set *rset;
3266         int                      rc;
3267         ENTRY;
3268
3269         if (strncmp(param, PARAM_SRPC, sizeof(PARAM_SRPC) - 1) != 0) {
3270                 CERROR("Invalid sptlrpc parameter: %s\n", param);
3271                 RETURN(-EINVAL);
3272         }
3273
3274         if (strncmp(param, PARAM_SRPC_UDESC,
3275                     sizeof(PARAM_SRPC_UDESC) - 1) == 0) {
3276                 RETURN(mgs_srpc_set_param_udesc_mem(fsdb, param));
3277         }
3278
3279         if (strncmp(param, PARAM_SRPC_FLVR, sizeof(PARAM_SRPC_FLVR) - 1) != 0) {
3280                 CERROR("Invalid sptlrpc flavor parameter: %s\n", param);
3281                 RETURN(-EINVAL);
3282         }
3283
3284         param += sizeof(PARAM_SRPC_FLVR) - 1;
3285
3286         rc = sptlrpc_parse_rule(param, &rule);
3287         if (rc)
3288                 RETURN(rc);
3289
3290         /* mgs rules implies must be mgc->mgs */
3291         if (test_bit(FSDB_MGS_SELF, &fsdb->fsdb_flags)) {
3292                 if ((rule.sr_from != LUSTRE_SP_MGC &&
3293                      rule.sr_from != LUSTRE_SP_ANY) ||
3294                     (rule.sr_to != LUSTRE_SP_MGS &&
3295                      rule.sr_to != LUSTRE_SP_ANY))
3296                         RETURN(-EINVAL);
3297         }
3298
3299         /* preapre room for this coming rule. svcname format should be:
3300          * - fsname: general rule
3301          * - fsname-tgtname: target-specific rule
3302          */
3303         if (strchr(svname, '-')) {
3304                 struct mgs_tgt_srpc_conf *tgtconf;
3305                 int                       found = 0;
3306
3307                 for (tgtconf = fsdb->fsdb_srpc_tgt; tgtconf != NULL;
3308                      tgtconf = tgtconf->mtsc_next) {
3309                         if (!strcmp(tgtconf->mtsc_tgt, svname)) {
3310                                 found = 1;
3311                                 break;
3312                         }
3313                 }
3314
3315                 if (!found) {
3316                         int name_len;
3317
3318                         OBD_ALLOC_PTR(tgtconf);
3319                         if (tgtconf == NULL)
3320                                 RETURN(-ENOMEM);
3321
3322                         name_len = strlen(svname);
3323
3324                         OBD_ALLOC(tgtconf->mtsc_tgt, name_len + 1);
3325                         if (tgtconf->mtsc_tgt == NULL) {
3326                                 OBD_FREE_PTR(tgtconf);
3327                                 RETURN(-ENOMEM);
3328                         }
3329                         memcpy(tgtconf->mtsc_tgt, svname, name_len);
3330
3331                         tgtconf->mtsc_next = fsdb->fsdb_srpc_tgt;
3332                         fsdb->fsdb_srpc_tgt = tgtconf;
3333                 }
3334
3335                 rset = &tgtconf->mtsc_rset;
3336         } else if (strcmp(svname, MGSSELF_NAME) == 0) {
3337                 /* put _mgs related srpc rule directly in mgs ruleset */
3338                 rset = &fsdb->fsdb_mgs->mgs_lut.lut_sptlrpc_rset;
3339         } else {
3340                 rset = &fsdb->fsdb_srpc_gen;
3341         }
3342
3343         rc = sptlrpc_rule_set_merge(rset, &rule);
3344
3345         RETURN(rc);
3346 }
3347
3348 static int mgs_srpc_set_param(const struct lu_env *env,
3349                               struct mgs_device *mgs,
3350                               struct fs_db *fsdb,
3351                               struct mgs_target_info *mti,
3352                               char *param)
3353 {
3354         char                   *copy;
3355         int                     rc, copy_size;
3356         ENTRY;
3357
3358 #ifndef HAVE_GSS
3359         RETURN(-EINVAL);
3360 #endif
3361         /* keep a copy of original param, which could be destroied
3362          * during parsing */
3363         copy_size = strlen(param) + 1;
3364         OBD_ALLOC(copy, copy_size);
3365         if (copy == NULL)
3366                 return -ENOMEM;
3367         memcpy(copy, param, copy_size);
3368
3369         rc = mgs_srpc_set_param_mem(fsdb, mti->mti_svname, param);
3370         if (rc)
3371                 goto out_free;
3372
3373         /* previous steps guaranteed the syntax is correct */
3374         rc = mgs_srpc_set_param_disk(env, mgs, fsdb, mti, copy);
3375         if (rc)
3376                 goto out_free;
3377
3378         if (test_bit(FSDB_MGS_SELF, &fsdb->fsdb_flags)) {
3379                 /*
3380                  * for mgs rules, make them effective immediately.
3381                  */
3382                 LASSERT(fsdb->fsdb_srpc_tgt == NULL);
3383                 sptlrpc_target_update_exp_flavor(mgs->mgs_obd,
3384                                                  &fsdb->fsdb_srpc_gen);
3385         }
3386
3387 out_free:
3388         OBD_FREE(copy, copy_size);
3389         RETURN(rc);
3390 }
3391
3392 struct mgs_srpc_read_data {
3393         struct fs_db   *msrd_fsdb;
3394         int             msrd_skip;
3395 };
3396
3397 static int mgs_srpc_read_handler(const struct lu_env *env,
3398                                  struct llog_handle *llh,
3399                                  struct llog_rec_hdr *rec, void *data)
3400 {
3401         struct mgs_srpc_read_data *msrd = data;
3402         struct cfg_marker         *marker;
3403         struct lustre_cfg         *lcfg = REC_DATA(rec);
3404         char                      *svname, *param;
3405         int                        cfg_len, rc;
3406         ENTRY;
3407
3408         if (rec->lrh_type != OBD_CFG_REC) {
3409                 CERROR("unhandled lrh_type: %#x\n", rec->lrh_type);
3410                 RETURN(-EINVAL);
3411         }
3412
3413         cfg_len = REC_DATA_LEN(rec);
3414
3415         rc = lustre_cfg_sanity_check(lcfg, cfg_len);
3416         if (rc) {
3417                 CERROR("Insane cfg\n");
3418                 RETURN(rc);
3419         }
3420
3421         if (lcfg->lcfg_command == LCFG_MARKER) {
3422                 marker = lustre_cfg_buf(lcfg, 1);
3423
3424                 if (marker->cm_flags & CM_START &&
3425                     marker->cm_flags & CM_SKIP)
3426                         msrd->msrd_skip = 1;
3427                 if (marker->cm_flags & CM_END)
3428                         msrd->msrd_skip = 0;
3429
3430                 RETURN(0);
3431         }
3432
3433         if (msrd->msrd_skip)
3434                 RETURN(0);
3435
3436         if (lcfg->lcfg_command != LCFG_SPTLRPC_CONF) {
3437                 CERROR("invalid command (%x)\n", lcfg->lcfg_command);
3438                 RETURN(0);
3439         }
3440
3441         svname = lustre_cfg_string(lcfg, 0);
3442         if (svname == NULL) {
3443                 CERROR("svname is empty\n");
3444                 RETURN(0);
3445         }
3446
3447         param = lustre_cfg_string(lcfg, 1);
3448         if (param == NULL) {
3449                 CERROR("param is empty\n");
3450                 RETURN(0);
3451         }
3452
3453         rc = mgs_srpc_set_param_mem(msrd->msrd_fsdb, svname, param);
3454         if (rc)
3455                 CERROR("read sptlrpc record error (%d): %s\n", rc, param);
3456
3457         RETURN(0);
3458 }
3459
3460 int mgs_get_fsdb_srpc_from_llog(const struct lu_env *env,
3461                                 struct mgs_device *mgs,
3462                                 struct fs_db *fsdb)
3463 {
3464         struct llog_handle        *llh = NULL;
3465         struct llog_ctxt          *ctxt;
3466         char                      *logname;
3467         struct mgs_srpc_read_data  msrd;
3468         int                        rc;
3469         ENTRY;
3470
3471         /* construct log name */
3472         rc = name_create(&logname, fsdb->fsdb_name, "-sptlrpc");
3473         if (rc)
3474                 RETURN(rc);
3475
3476         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
3477         LASSERT(ctxt != NULL);
3478
3479         if (mgs_log_is_empty(env, mgs, logname))
3480                 GOTO(out, rc = 0);
3481
3482         rc = llog_open(env, ctxt, &llh, NULL, logname,
3483                        LLOG_OPEN_EXISTS);
3484         if (rc < 0) {
3485                 if (rc == -ENOENT)
3486                         rc = 0;
3487                 GOTO(out, rc);
3488         }
3489
3490         rc = llog_init_handle(env, llh, LLOG_F_IS_PLAIN, NULL);
3491         if (rc)
3492                 GOTO(out_close, rc);
3493
3494         if (llog_get_size(llh) <= 1)
3495                 GOTO(out_close, rc = 0);
3496
3497         msrd.msrd_fsdb = fsdb;
3498         msrd.msrd_skip = 0;
3499
3500         rc = llog_process(env, llh, mgs_srpc_read_handler, (void *)&msrd,
3501                           NULL);
3502
3503 out_close:
3504         llog_close(env, llh);
3505 out:
3506         llog_ctxt_put(ctxt);
3507         name_destroy(&logname);
3508
3509         if (rc)
3510                 CERROR("failed to read sptlrpc config database: %d\n", rc);
3511         RETURN(rc);
3512 }
3513
3514 /* Permanent settings of all parameters by writing into the appropriate
3515  * configuration logs.
3516  * A parameter with null value ("<param>='\0'") means to erase it out of
3517  * the logs.
3518  */
3519 static int mgs_write_log_param(const struct lu_env *env,
3520                                struct mgs_device *mgs, struct fs_db *fsdb,
3521                                struct mgs_target_info *mti, char *ptr)
3522 {
3523         struct mgs_thread_info *mgi = mgs_env_info(env);
3524         char *logname;
3525         char *tmp;
3526         int rc = 0;
3527         ENTRY;
3528
3529         /* For various parameter settings, we have to figure out which logs
3530            care about them (e.g. both mdt and client for lov settings) */
3531         CDEBUG(D_MGS, "next param '%s'\n", ptr);
3532
3533         /* The params are stored in MOUNT_DATA_FILE and modified via
3534            tunefs.lustre, or set using lctl conf_param */
3535
3536         /* Processed in lustre_start_mgc */
3537         if (class_match_param(ptr, PARAM_MGSNODE, NULL) == 0)
3538                 GOTO(end, rc);
3539
3540         /* Processed in ost/mdt */
3541         if (class_match_param(ptr, PARAM_NETWORK, NULL) == 0)
3542                 GOTO(end, rc);
3543
3544         /* Processed in mgs_write_log_ost */
3545         if (class_match_param(ptr, PARAM_FAILMODE, NULL) == 0) {
3546                 if (mti->mti_flags & LDD_F_PARAM) {
3547                         LCONSOLE_ERROR_MSG(0x169, "%s can only be "
3548                                            "changed with tunefs.lustre"
3549                                            "and --writeconf\n", ptr);
3550                         rc = -EPERM;
3551                 }
3552                 GOTO(end, rc);
3553         }
3554
3555         if (class_match_param(ptr, PARAM_SRPC, NULL) == 0) {
3556                 rc = mgs_srpc_set_param(env, mgs, fsdb, mti, ptr);
3557                 GOTO(end, rc);
3558         }
3559
3560         if (class_match_param(ptr, PARAM_FAILNODE, NULL) == 0) {
3561                 /* Add a failover nidlist */
3562                 rc = 0;
3563                 /* We already processed failovers params for new
3564                    targets in mgs_write_log_target */
3565                 if (mti->mti_flags & LDD_F_PARAM) {
3566                         CDEBUG(D_MGS, "Adding failnode\n");
3567                         rc = mgs_write_log_add_failnid(env, mgs, fsdb, mti);
3568                 }
3569                 GOTO(end, rc);
3570         }
3571
3572         if (class_match_param(ptr, PARAM_SYS, &tmp) == 0) {
3573                 rc = mgs_write_log_sys(env, mgs, fsdb, mti, ptr, tmp);
3574                 GOTO(end, rc);
3575         }
3576
3577         if (class_match_param(ptr, PARAM_QUOTA, &tmp) == 0) {
3578                 rc = mgs_write_log_quota(env, mgs, fsdb, mti, ptr, tmp);
3579                 GOTO(end, rc);
3580         }
3581
3582         if (class_match_param(ptr, PARAM_OSC PARAM_ACTIVE, &tmp) == 0 ||
3583             class_match_param(ptr, PARAM_MDC PARAM_ACTIVE, &tmp) == 0) {
3584                 /* active=0 means off, anything else means on */
3585                 int flag = (*tmp == '0') ? CM_EXCLUDE : 0;
3586                 bool deactive_osc = memcmp(ptr, PARAM_OSC PARAM_ACTIVE,
3587                                           strlen(PARAM_OSC PARAM_ACTIVE)) == 0;
3588                 int i;
3589
3590                 if (!deactive_osc) {
3591                         __u32   index;
3592
3593                         rc = server_name2index(mti->mti_svname, &index, NULL);
3594                         if (rc < 0)
3595                                 GOTO(end, rc);
3596
3597                         if (index == 0) {
3598                                 LCONSOLE_ERROR_MSG(0x144, "%s: MDC0 can not be"
3599                                                    " (de)activated.\n",
3600                                                    mti->mti_svname);
3601                                 GOTO(end, rc = -EINVAL);
3602                         }
3603                 }
3604
3605                 LCONSOLE_WARN("Permanently %sactivating %s\n",
3606                               flag ? "de" : "re", mti->mti_svname);
3607                 /* Modify clilov */
3608                 rc = name_create(&logname, mti->mti_fsname, "-client");
3609                 if (rc < 0)
3610                         GOTO(end, rc);
3611                 rc = mgs_modify(env, mgs, fsdb, mti, logname,
3612                                 mti->mti_svname,
3613                                 deactive_osc ? "add osc" : "add mdc", flag);
3614                 name_destroy(&logname);
3615                 if (rc < 0)
3616                         goto active_err;
3617
3618                 /* Modify mdtlov */
3619                 /* Add to all MDT logs for DNE */
3620                 for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
3621                         if (!test_bit(i, fsdb->fsdb_mdt_index_map))
3622                                 continue;
3623                         rc = name_create_mdt(&logname, mti->mti_fsname, i);
3624                         if (rc < 0)
3625                                 GOTO(end, rc);
3626                         rc = mgs_modify(env, mgs, fsdb, mti, logname,
3627                                         mti->mti_svname,
3628                                         deactive_osc ? "add osc" : "add osp",
3629                                         flag);
3630                         name_destroy(&logname);
3631                         if (rc < 0)
3632                                 goto active_err;
3633                 }
3634 active_err:
3635                 if (rc < 0) {
3636                         LCONSOLE_ERROR_MSG(0x145, "Couldn't find %s in"
3637                                            "log (%d). No permanent "
3638                                            "changes were made to the "
3639                                            "config log.\n",
3640                                            mti->mti_svname, rc);
3641                         if (test_bit(FSDB_OLDLOG14, &fsdb->fsdb_flags))
3642                                 LCONSOLE_ERROR_MSG(0x146, "This may be"
3643                                                    " because the log"
3644                                                    "is in the old 1.4"
3645                                                    "style. Consider "
3646                                                    " --writeconf to "
3647                                                    "update the logs.\n");
3648                         GOTO(end, rc);
3649                 }
3650                 /* Fall through to osc/mdc proc for deactivating live
3651                    OSC/OSP on running MDT / clients. */
3652         }
3653         /* Below here, let obd's XXX_process_config methods handle it */
3654
3655         /* All lov. in proc */
3656         if (class_match_param(ptr, PARAM_LOV, NULL) == 0) {
3657                 char *mdtlovname;
3658
3659                 CDEBUG(D_MGS, "lov param %s\n", ptr);
3660                 if (!(mti->mti_flags & LDD_F_SV_TYPE_MDT)) {
3661                         LCONSOLE_ERROR_MSG(0x147, "LOV params must be "
3662                                            "set on the MDT, not %s. "
3663                                            "Ignoring.\n",
3664                                            mti->mti_svname);
3665                         GOTO(end, rc = 0);
3666                 }
3667
3668                 /* Modify mdtlov */
3669                 if (mgs_log_is_empty(env, mgs, mti->mti_svname))
3670                         GOTO(end, rc = -ENODEV);
3671
3672                 rc = name_create_mdt_and_lov(&logname, &mdtlovname, fsdb,
3673                                              mti->mti_stripe_index);
3674                 if (rc)
3675                         GOTO(end, rc);
3676                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, mti->mti_svname,
3677                                   &mgi->mgi_bufs, mdtlovname, ptr);
3678                 name_destroy(&logname);
3679                 name_destroy(&mdtlovname);
3680                 if (rc)
3681                         GOTO(end, rc);
3682
3683                 /* Modify clilov */
3684                 rc = name_create(&logname, mti->mti_fsname, "-client");
3685                 if (rc)
3686                         GOTO(end, rc);
3687                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, logname, &mgi->mgi_bufs,
3688                                   fsdb->fsdb_clilov, ptr);
3689                 name_destroy(&logname);
3690                 GOTO(end, rc);
3691         }
3692
3693         /* All osc., mdc., llite. params in proc */
3694         if ((class_match_param(ptr, PARAM_OSC, NULL) == 0) ||
3695             (class_match_param(ptr, PARAM_MDC, NULL) == 0) ||
3696             (class_match_param(ptr, PARAM_LLITE, NULL) == 0)) {
3697                 char *cname;
3698
3699                 if (test_bit(FSDB_OLDLOG14, &fsdb->fsdb_flags)) {
3700                         LCONSOLE_ERROR_MSG(0x148, "Upgraded client logs for %s"
3701                                            " cannot be modified. Consider"
3702                                            " updating the configuration with"
3703                                            " --writeconf\n",
3704                                            mti->mti_svname);
3705                         GOTO(end, rc = -EINVAL);
3706                 }
3707                 if (memcmp(ptr, PARAM_LLITE, strlen(PARAM_LLITE)) == 0) {
3708                         rc = name_create(&cname, mti->mti_fsname, "-client");
3709                         /* Add the client type to match the obdname in
3710                            class_config_llog_handler */
3711                 } else if (mti->mti_flags & LDD_F_SV_TYPE_MDT) {
3712                         rc = name_create(&cname, mti->mti_svname, "-mdc");
3713                 } else if (mti->mti_flags & LDD_F_SV_TYPE_OST) {
3714                         rc = name_create(&cname, mti->mti_svname, "-osc");
3715                 } else {
3716                         GOTO(end, rc = -EINVAL);
3717                 }
3718                 if (rc)
3719                         GOTO(end, rc);
3720
3721                 /* Forbid direct update of llite root squash parameters.
3722                  * These parameters are indirectly set via the MDT settings.
3723                  * See (LU-1778) */
3724                 if ((class_match_param(ptr, PARAM_LLITE, &tmp) == 0) &&
3725                     ((memcmp(tmp, "root_squash=", 12) == 0) ||
3726                      (memcmp(tmp, "nosquash_nids=", 14) == 0))) {
3727                         LCONSOLE_ERROR("%s: root squash parameters can only "
3728                                 "be updated through MDT component\n",
3729                                 mti->mti_fsname);
3730                         name_destroy(&cname);
3731                         GOTO(end, rc = -EINVAL);
3732                 }
3733
3734                 CDEBUG(D_MGS, "%.3s param %s\n", ptr, ptr + 4);
3735
3736                 /* Modify client */
3737                 rc = name_create(&logname, mti->mti_fsname, "-client");
3738                 if (rc) {
3739                         name_destroy(&cname);
3740                         GOTO(end, rc);
3741                 }
3742                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, logname, &mgi->mgi_bufs,
3743                                   cname, ptr);
3744
3745                 /* osc params affect the MDT as well */
3746                 if (!rc && (mti->mti_flags & LDD_F_SV_TYPE_OST)) {
3747                         int i;
3748
3749                         for (i = 0; i < INDEX_MAP_SIZE * 8; i++){
3750                                 if (!test_bit(i, fsdb->fsdb_mdt_index_map))
3751                                         continue;
3752                                 name_destroy(&cname);
3753                                 rc = name_create_mdt_osc(&cname, mti->mti_svname,
3754                                                          fsdb, i);
3755                                 name_destroy(&logname);
3756                                 if (rc)
3757                                         break;
3758                                 rc = name_create_mdt(&logname,
3759                                                      mti->mti_fsname, i);
3760                                 if (rc)
3761                                         break;
3762                                 if (!mgs_log_is_empty(env, mgs, logname)) {
3763                                         rc = mgs_wlp_lcfg(env, mgs, fsdb,
3764                                                           mti, logname,
3765                                                           &mgi->mgi_bufs,
3766                                                           cname, ptr);
3767                                         if (rc)
3768                                                 break;
3769                                 }
3770                         }
3771                 }
3772
3773                 /* For mdc activate/deactivate, it affects OSP on MDT as well */
3774                 if (class_match_param(ptr, PARAM_MDC PARAM_ACTIVE, &tmp) == 0 &&
3775                     rc == 0) {
3776                         char suffix[16];
3777                         char *lodname = NULL;
3778                         char *param_str = NULL;
3779                         int i;
3780                         int index;
3781
3782                         /* replace mdc with osp */
3783                         memcpy(ptr, PARAM_OSP, strlen(PARAM_OSP));
3784                         rc = server_name2index(mti->mti_svname, &index, NULL);
3785                         if (rc < 0) {
3786                                 memcpy(ptr, PARAM_MDC, strlen(PARAM_MDC));
3787                                 GOTO(end, rc);
3788                         }
3789
3790                         for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
3791                                 if (!test_bit(i, fsdb->fsdb_mdt_index_map))
3792                                         continue;
3793
3794                                 if (i == index)
3795                                         continue;
3796
3797                                 name_destroy(&logname);
3798                                 rc = name_create_mdt(&logname, mti->mti_fsname,
3799                                                      i);
3800                                 if (rc < 0)
3801                                         break;
3802
3803                                 if (mgs_log_is_empty(env, mgs, logname))
3804                                         continue;
3805
3806                                 snprintf(suffix, sizeof(suffix), "-osp-MDT%04x",
3807                                          i);
3808                                 name_destroy(&cname);
3809                                 rc = name_create(&cname, mti->mti_svname,
3810                                                  suffix);
3811                                 if (rc < 0)
3812                                         break;
3813
3814                                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, logname,
3815                                                   &mgi->mgi_bufs, cname, ptr);
3816                                 if (rc < 0)
3817                                         break;
3818
3819                                 /* Add configuration log for noitfying LOD
3820                                  * to active/deactive the OSP. */
3821                                 name_destroy(&param_str);
3822                                 rc = name_create(&param_str, cname,
3823                                                  (*tmp == '0') ?  ".active=0" :
3824                                                  ".active=1");
3825                                 if (rc < 0)
3826                                         break;
3827
3828                                 name_destroy(&lodname);
3829                                 rc = name_create(&lodname, logname, "-mdtlov");
3830                                 if (rc < 0)
3831                                         break;
3832
3833                                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, logname,
3834                                                   &mgi->mgi_bufs, lodname,
3835                                                   param_str);
3836                                 if (rc < 0)
3837                                         break;
3838                         }
3839                         memcpy(ptr, PARAM_MDC, strlen(PARAM_MDC));
3840                         name_destroy(&lodname);
3841                         name_destroy(&param_str);
3842                 }
3843
3844                 name_destroy(&logname);
3845                 name_destroy(&cname);
3846                 GOTO(end, rc);
3847         }
3848
3849         /* All mdt. params in proc */
3850         if (class_match_param(ptr, PARAM_MDT, &tmp) == 0) {
3851                 int i;
3852                 __u32 idx;
3853
3854                 CDEBUG(D_MGS, "%.3s param %s\n", ptr, ptr + 4);
3855                 if (strncmp(mti->mti_svname, mti->mti_fsname,
3856                             MTI_NAME_MAXLEN) == 0)
3857                         /* device is unspecified completely? */
3858                         rc = LDD_F_SV_TYPE_MDT | LDD_F_SV_ALL;
3859                 else
3860                         rc = server_name2index(mti->mti_svname, &idx, NULL);
3861                 if (rc < 0)
3862                         goto active_err;
3863                 if ((rc & LDD_F_SV_TYPE_MDT) == 0)
3864                         goto active_err;
3865                 if (rc & LDD_F_SV_ALL) {
3866                         for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
3867                                 if (!test_bit(i,
3868                                                   fsdb->fsdb_mdt_index_map))
3869                                         continue;
3870                                 rc = name_create_mdt(&logname,
3871                                                 mti->mti_fsname, i);
3872                                 if (rc)
3873                                         goto active_err;
3874                                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti,
3875                                                   logname, &mgi->mgi_bufs,
3876                                                   logname, ptr);
3877                                 name_destroy(&logname);
3878                                 if (rc)
3879                                         goto active_err;
3880                         }
3881                 } else {
3882                         if ((memcmp(tmp, "root_squash=", 12) == 0) ||
3883                             (memcmp(tmp, "nosquash_nids=", 14) == 0)) {
3884                                 LCONSOLE_ERROR("%s: root squash parameters "
3885                                         "cannot be applied to a single MDT\n",
3886                                         mti->mti_fsname);
3887                                 GOTO(end, rc = -EINVAL);
3888                         }
3889                         rc = mgs_wlp_lcfg(env, mgs, fsdb, mti,
3890                                           mti->mti_svname, &mgi->mgi_bufs,
3891                                           mti->mti_svname, ptr);
3892                         if (rc)
3893                                 goto active_err;
3894                 }
3895
3896                 /* root squash settings are also applied to llite
3897                  * config log (see LU-1778) */
3898                 if (rc == 0 &&
3899                     ((memcmp(tmp, "root_squash=", 12) == 0) ||
3900                      (memcmp(tmp, "nosquash_nids=", 14) == 0))) {
3901                         char *cname;
3902                         char *ptr2;
3903
3904                         rc = name_create(&cname, mti->mti_fsname, "-client");
3905                         if (rc)
3906                                 GOTO(end, rc);
3907                         rc = name_create(&logname, mti->mti_fsname, "-client");
3908                         if (rc) {
3909                                 name_destroy(&cname);
3910                                 GOTO(end, rc);
3911                         }
3912                         rc = name_create(&ptr2, PARAM_LLITE, tmp);
3913                         if (rc) {
3914                                 name_destroy(&cname);
3915                                 name_destroy(&logname);
3916                                 GOTO(end, rc);
3917                         }
3918                         rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, logname,
3919                                           &mgi->mgi_bufs, cname, ptr2);
3920                         name_destroy(&ptr2);
3921                         name_destroy(&logname);
3922                         name_destroy(&cname);
3923                 }
3924                 GOTO(end, rc);
3925         }
3926
3927         /* All mdd., ost. and osd. params in proc */
3928         if ((class_match_param(ptr, PARAM_MDD, NULL) == 0) ||
3929             (class_match_param(ptr, PARAM_OST, NULL) == 0) ||
3930             (class_match_param(ptr, PARAM_OSD, NULL) == 0)) {
3931                 CDEBUG(D_MGS, "%.3s param %s\n", ptr, ptr + 4);
3932                 if (mgs_log_is_empty(env, mgs, mti->mti_svname))
3933                         GOTO(end, rc = -ENODEV);
3934
3935                 rc = mgs_wlp_lcfg(env, mgs, fsdb, mti, mti->mti_svname,
3936                                   &mgi->mgi_bufs, mti->mti_svname, ptr);
3937                 GOTO(end, rc);
3938         }
3939
3940         LCONSOLE_WARN("Ignoring unrecognized param '%s'\n", ptr);
3941
3942 end:
3943         if (rc)
3944                 CERROR("err %d on param '%s'\n", rc, ptr);
3945
3946         RETURN(rc);
3947 }
3948
3949 int mgs_write_log_target(const struct lu_env *env, struct mgs_device *mgs,
3950                          struct mgs_target_info *mti, struct fs_db *fsdb)
3951 {
3952         char    *buf, *params;
3953         int      rc = -EINVAL;
3954
3955         ENTRY;
3956
3957         /* set/check the new target index */
3958         rc = mgs_set_index(env, mgs, mti);
3959         if (rc < 0)
3960                 RETURN(rc);
3961
3962         if (rc == EALREADY) {
3963                 LCONSOLE_WARN("Found index %d for %s, updating log\n",
3964                               mti->mti_stripe_index, mti->mti_svname);
3965                 /* We would like to mark old log sections as invalid
3966                    and add new log sections in the client and mdt logs.
3967                    But if we add new sections, then live clients will
3968                    get repeat setup instructions for already running
3969                    osc's. So don't update the client/mdt logs. */
3970                 mti->mti_flags &= ~LDD_F_UPDATE;
3971                 rc = 0;
3972         }
3973
3974         OBD_FAIL_TIMEOUT(OBD_FAIL_MGS_WRITE_TARGET_DELAY, cfs_fail_val > 0 ?
3975                          cfs_fail_val : 10);
3976
3977         mutex_lock(&fsdb->fsdb_mutex);
3978
3979         if (mti->mti_flags &
3980             (LDD_F_VIRGIN | LDD_F_UPGRADE14 | LDD_F_WRITECONF)) {
3981                 /* Generate a log from scratch */
3982                 if (mti->mti_flags & LDD_F_SV_TYPE_MDT) {
3983                         rc = mgs_write_log_mdt(env, mgs, fsdb, mti);
3984                 } else if (mti->mti_flags & LDD_F_SV_TYPE_OST) {
3985                         rc = mgs_write_log_ost(env, mgs, fsdb, mti);
3986                 } else {
3987                         CERROR("Unknown target type %#x, can't create log for "
3988                                "%s\n", mti->mti_flags, mti->mti_svname);
3989                 }
3990                 if (rc) {
3991                         CERROR("Can't write logs for %s (%d)\n",
3992                                mti->mti_svname, rc);
3993                         GOTO(out_up, rc);
3994                 }
3995         } else {
3996                 /* Just update the params from tunefs in mgs_write_log_params */
3997                 CDEBUG(D_MGS, "Update params for %s\n", mti->mti_svname);
3998                 mti->mti_flags |= LDD_F_PARAM;
3999         }
4000
4001         /* allocate temporary buffer, where class_get_next_param will
4002            make copy of a current  parameter */
4003         OBD_ALLOC(buf, strlen(mti->mti_params) + 1);
4004         if (buf == NULL)
4005                 GOTO(out_up, rc = -ENOMEM);
4006         params = mti->mti_params;
4007         while (params != NULL) {
4008                 rc = class_get_next_param(&params, buf);
4009                 if (rc) {
4010                         if (rc == 1)
4011                                 /* there is no next parameter, that is
4012                                    not an error */
4013                                 rc = 0;
4014                         break;
4015                 }
4016                 CDEBUG(D_MGS, "remaining string: '%s', param: '%s'\n",
4017                        params, buf);
4018                 rc = mgs_write_log_param(env, mgs, fsdb, mti, buf);
4019                 if (rc)
4020                         break;
4021         }
4022
4023         OBD_FREE(buf, strlen(mti->mti_params) + 1);
4024
4025 out_up:
4026         mutex_unlock(&fsdb->fsdb_mutex);
4027         RETURN(rc);
4028 }
4029
4030 int mgs_erase_log(const struct lu_env *env, struct mgs_device *mgs, char *name)
4031 {
4032         struct llog_ctxt        *ctxt;
4033         int                      rc = 0;
4034
4035         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
4036         if (ctxt == NULL) {
4037                 CERROR("%s: MGS config context doesn't exist\n",
4038                        mgs->mgs_obd->obd_name);
4039                 rc = -ENODEV;
4040         } else {
4041                 rc = llog_erase(env, ctxt, NULL, name);
4042                 /* llog may not exist */
4043                 if (rc == -ENOENT)
4044                         rc = 0;
4045                 llog_ctxt_put(ctxt);
4046         }
4047
4048         if (rc)
4049                 CERROR("%s: failed to clear log %s: %d\n",
4050                        mgs->mgs_obd->obd_name, name, rc);
4051
4052         return rc;
4053 }
4054
4055 /* erase all logs for the given fs */
4056 int mgs_erase_logs(const struct lu_env *env, struct mgs_device *mgs,
4057                    const char *fsname)
4058 {
4059         struct list_head log_list;
4060         struct mgs_direntry *dirent, *n;
4061         char barrier_name[20] = {};
4062         char *suffix;
4063         int count = 0;
4064         int rc, len = strlen(fsname);
4065         ENTRY;
4066
4067         mutex_lock(&mgs->mgs_mutex);
4068
4069         /* Find all the logs in the CONFIGS directory */
4070         rc = class_dentry_readdir(env, mgs, &log_list);
4071         if (rc) {
4072                 mutex_unlock(&mgs->mgs_mutex);
4073                 RETURN(rc);
4074         }
4075
4076         if (list_empty(&log_list)) {
4077                 mutex_unlock(&mgs->mgs_mutex);
4078                 RETURN(-ENOENT);
4079         }
4080
4081         snprintf(barrier_name, sizeof(barrier_name) - 1, "%s-%s",
4082                  fsname, BARRIER_FILENAME);
4083         /* Delete the barrier fsdb */
4084         mgs_remove_fsdb_by_name(mgs, barrier_name);
4085         /* Delete the fs db */
4086         mgs_remove_fsdb_by_name(mgs, fsname);
4087         mutex_unlock(&mgs->mgs_mutex);
4088
4089         list_for_each_entry_safe(dirent, n, &log_list, mde_list) {
4090                 list_del_init(&dirent->mde_list);
4091                 suffix = strrchr(dirent->mde_name, '-');
4092                 if (suffix != NULL) {
4093                         if ((len == suffix - dirent->mde_name) &&
4094                             (strncmp(fsname, dirent->mde_name, len) == 0)) {
4095                                 CDEBUG(D_MGS, "Removing log %s\n",
4096                                        dirent->mde_name);
4097                                 mgs_erase_log(env, mgs, dirent->mde_name);
4098                                 count++;
4099                         }
4100                 }
4101                 mgs_direntry_free(dirent);
4102         }
4103
4104         if (count == 0)
4105                 rc = -ENOENT;
4106
4107         RETURN(rc);
4108 }
4109
4110 /* list all logs for the given fs */
4111 int mgs_list_logs(const struct lu_env *env, struct mgs_device *mgs,
4112                   struct obd_ioctl_data *data)
4113 {
4114         struct list_head         log_list;
4115         struct mgs_direntry     *dirent, *n;
4116         char                    *out, *suffix;
4117         int                      l, remains, rc;
4118
4119         ENTRY;
4120
4121         /* Find all the logs in the CONFIGS directory */
4122         rc = class_dentry_readdir(env, mgs, &log_list);
4123         if (rc)
4124                 RETURN(rc);
4125
4126         out = data->ioc_bulk;
4127         remains = data->ioc_inllen1;
4128         list_for_each_entry_safe(dirent, n, &log_list, mde_list) {
4129                 list_del_init(&dirent->mde_list);
4130                 suffix = strrchr(dirent->mde_name, '-');
4131                 if (suffix != NULL) {
4132                         l = snprintf(out, remains, "config_log: %s\n",
4133                                      dirent->mde_name);
4134                         out += l;
4135                         remains -= l;
4136                 }
4137                 mgs_direntry_free(dirent);
4138                 if (remains <= 0)
4139                         break;
4140         }
4141         RETURN(rc);
4142 }
4143
4144 struct mgs_lcfg_fork_data {
4145         struct lustre_cfg_bufs   mlfd_bufs;
4146         struct mgs_device       *mlfd_mgs;
4147         struct llog_handle      *mlfd_llh;
4148         const char              *mlfd_oldname;
4149         const char              *mlfd_newname;
4150         char                     mlfd_data[0];
4151 };
4152
4153 static bool contain_valid_fsname(char *buf, const char *fsname,
4154                                  int buflen, int namelen)
4155 {
4156         if (buflen < namelen)
4157                 return false;
4158
4159         if (memcmp(buf, fsname, namelen) != 0)
4160                 return false;
4161
4162         if (buf[namelen] != '\0' && buf[namelen] != '-')
4163                 return false;
4164
4165         return true;
4166 }
4167
4168 static int mgs_lcfg_fork_handler(const struct lu_env *env,
4169                                  struct llog_handle *o_llh,
4170                                  struct llog_rec_hdr *o_rec, void *data)
4171 {
4172         struct mgs_lcfg_fork_data *mlfd = data;
4173         struct lustre_cfg_bufs *n_bufs = &mlfd->mlfd_bufs;
4174         struct lustre_cfg *o_lcfg = (struct lustre_cfg *)(o_rec + 1);
4175         struct llog_cfg_rec *lcr;
4176         char *o_buf;
4177         char *n_buf = mlfd->mlfd_data;
4178         int o_buflen;
4179         int o_namelen = strlen(mlfd->mlfd_oldname);
4180         int n_namelen = strlen(mlfd->mlfd_newname);
4181         int diff = n_namelen - o_namelen;
4182         __u32 cmd = o_lcfg->lcfg_command;
4183         __u32 cnt = o_lcfg->lcfg_bufcount;
4184         int rc;
4185         int i;
4186         ENTRY;
4187
4188         /* buf[0] */
4189         o_buf = lustre_cfg_buf(o_lcfg, 0);
4190         o_buflen = o_lcfg->lcfg_buflens[0];
4191         if (contain_valid_fsname(o_buf, mlfd->mlfd_oldname, o_buflen,
4192                                  o_namelen)) {
4193                 memcpy(n_buf, mlfd->mlfd_newname, n_namelen);
4194                 memcpy(n_buf + n_namelen, o_buf + o_namelen,
4195                        o_buflen - o_namelen);
4196                 lustre_cfg_bufs_reset(n_bufs, n_buf);
4197                 n_buf += cfs_size_round(o_buflen + diff);
4198         } else {
4199                 lustre_cfg_bufs_reset(n_bufs, o_buflen != 0 ? o_buf : NULL);
4200         }
4201
4202         switch (cmd) {
4203         case LCFG_MARKER: {
4204                 struct cfg_marker *o_marker;
4205                 struct cfg_marker *n_marker;
4206                 int tgt_namelen;
4207
4208                 if (cnt != 2) {
4209                         CDEBUG(D_MGS, "Unknown cfg marker entry with %d "
4210                                "buffers\n", cnt);
4211                         RETURN(-EINVAL);
4212                 }
4213
4214                 /* buf[1] is marker */
4215                 o_buf = lustre_cfg_buf(o_lcfg, 1);
4216                 o_buflen = o_lcfg->lcfg_buflens[1];
4217                 o_marker = (struct cfg_marker *)o_buf;
4218                 if (!contain_valid_fsname(o_marker->cm_tgtname,
4219                                           mlfd->mlfd_oldname,
4220                                           sizeof(o_marker->cm_tgtname),
4221                                           o_namelen)) {
4222                         lustre_cfg_bufs_set(n_bufs, 1, o_marker,
4223                                             sizeof(*o_marker));
4224                         break;
4225                 }
4226
4227                 n_marker = (struct cfg_marker *)n_buf;
4228                 *n_marker = *o_marker;
4229                 memcpy(n_marker->cm_tgtname, mlfd->mlfd_newname, n_namelen);
4230                 tgt_namelen = strlen(o_marker->cm_tgtname);
4231                 if (tgt_namelen > o_namelen)
4232                         memcpy(n_marker->cm_tgtname + n_namelen,
4233                                o_marker->cm_tgtname + o_namelen,
4234                                tgt_namelen - o_namelen);
4235                 n_marker->cm_tgtname[tgt_namelen + diff] = '\0';
4236                 lustre_cfg_bufs_set(n_bufs, 1, n_marker, sizeof(*n_marker));
4237                 break;
4238         }
4239         case LCFG_PARAM:
4240         case LCFG_SET_PARAM: {
4241                 for (i = 1; i < cnt; i++)
4242                         /* buf[i] is the param value, reuse it directly */
4243                         lustre_cfg_bufs_set(n_bufs, i,
4244                                             lustre_cfg_buf(o_lcfg, i),
4245                                             o_lcfg->lcfg_buflens[i]);
4246                 break;
4247         }
4248         case LCFG_POOL_NEW:
4249         case LCFG_POOL_ADD:
4250         case LCFG_POOL_REM:
4251         case LCFG_POOL_DEL: {
4252                 if (cnt < 3 || cnt > 4) {
4253                         CDEBUG(D_MGS, "Unknown cfg pool (%x) entry with %d "
4254                                "buffers\n", cmd, cnt);
4255                         RETURN(-EINVAL);
4256                 }
4257
4258                 /* buf[1] is fsname */
4259                 o_buf = lustre_cfg_buf(o_lcfg, 1);
4260                 o_buflen = o_lcfg->lcfg_buflens[1];
4261                 memcpy(n_buf, mlfd->mlfd_newname, n_namelen);
4262                 memcpy(n_buf + n_namelen, o_buf + o_namelen,
4263                        o_buflen - o_namelen);
4264                 lustre_cfg_bufs_set(n_bufs, 1, n_buf, o_buflen + diff);
4265                 n_buf += cfs_size_round(o_buflen + diff);
4266
4267                 /* buf[2] is the pool name, reuse it directly */
4268                 lustre_cfg_bufs_set(n_bufs, 2, lustre_cfg_buf(o_lcfg, 2),
4269                                     o_lcfg->lcfg_buflens[2]);
4270
4271                 if (cnt == 3)
4272                         break;
4273
4274                 /* buf[3] is ostname */
4275                 o_buf = lustre_cfg_buf(o_lcfg, 3);
4276                 o_buflen = o_lcfg->lcfg_buflens[3];
4277                 memcpy(n_buf, mlfd->mlfd_newname, n_namelen);
4278                 memcpy(n_buf + n_namelen, o_buf + o_namelen,
4279                        o_buflen - o_namelen);
4280                 lustre_cfg_bufs_set(n_bufs, 3, n_buf, o_buflen + diff);
4281                 break;
4282         }
4283         case LCFG_SETUP: {
4284                 if (cnt == 2) {
4285                         o_buflen = o_lcfg->lcfg_buflens[1];
4286                         if (o_buflen == sizeof(struct lov_desc) ||
4287                             o_buflen == sizeof(struct lmv_desc)) {
4288                                 char *o_uuid;
4289                                 char *n_uuid;
4290                                 int uuid_len;
4291
4292                                 /* buf[1] */
4293                                 o_buf = lustre_cfg_buf(o_lcfg, 1);
4294                                 if (o_buflen == sizeof(struct lov_desc)) {
4295                                         struct lov_desc *o_desc =
4296                                                 (struct lov_desc *)o_buf;
4297                                         struct lov_desc *n_desc =
4298                                                 (struct lov_desc *)n_buf;
4299
4300                                         *n_desc = *o_desc;
4301                                         o_uuid = o_desc->ld_uuid.uuid;
4302                                         n_uuid = n_desc->ld_uuid.uuid;
4303                                         uuid_len = sizeof(o_desc->ld_uuid.uuid);
4304                                 } else {
4305                                         struct lmv_desc *o_desc =
4306                                                 (struct lmv_desc *)o_buf;
4307                                         struct lmv_desc *n_desc =
4308                                                 (struct lmv_desc *)n_buf;
4309
4310                                         *n_desc = *o_desc;
4311                                         o_uuid = o_desc->ld_uuid.uuid;
4312                                         n_uuid = n_desc->ld_uuid.uuid;
4313                                         uuid_len = sizeof(o_desc->ld_uuid.uuid);
4314                                 }
4315
4316                                 if (unlikely(!contain_valid_fsname(o_uuid,
4317                                                 mlfd->mlfd_oldname, uuid_len,
4318                                                 o_namelen))) {
4319                                         lustre_cfg_bufs_set(n_bufs, 1, o_buf,
4320                                                             o_buflen);
4321                                         break;
4322                                 }
4323
4324                                 memcpy(n_uuid, mlfd->mlfd_newname, n_namelen);
4325                                 uuid_len = strlen(o_uuid);
4326                                 if (uuid_len > o_namelen)
4327                                         memcpy(n_uuid + n_namelen,
4328                                                o_uuid + o_namelen,
4329                                                uuid_len - o_namelen);
4330                                 n_uuid[uuid_len + diff] = '\0';
4331                                 lustre_cfg_bufs_set(n_bufs, 1, n_buf, o_buflen);
4332                                 break;
4333                         } /* else case fall through */
4334                 } /* else case fall through */
4335         }
4336         default: {
4337                 for (i = 1; i < cnt; i++) {
4338                         o_buflen = o_lcfg->lcfg_buflens[i];
4339                         if (o_buflen == 0)
4340                                 continue;
4341
4342                         o_buf = lustre_cfg_buf(o_lcfg, i);
4343                         if (!contain_valid_fsname(o_buf, mlfd->mlfd_oldname,
4344                                                   o_buflen, o_namelen)) {
4345                                 lustre_cfg_bufs_set(n_bufs, i, o_buf, o_buflen);
4346                                 continue;
4347                         }
4348
4349                         memcpy(n_buf, mlfd->mlfd_newname, n_namelen);
4350                         if (o_buflen == o_namelen) {
4351                                 lustre_cfg_bufs_set(n_bufs, i, n_buf,
4352                                                     n_namelen);
4353                                 n_buf += cfs_size_round(n_namelen);
4354                                 continue;
4355                         }
4356
4357                         memcpy(n_buf + n_namelen, o_buf + o_namelen,
4358                                o_buflen - o_namelen);
4359                         lustre_cfg_bufs_set(n_bufs, i, n_buf, o_buflen + diff);
4360                         n_buf += cfs_size_round(o_buflen + diff);
4361                 }
4362                 break;
4363         }
4364         }
4365
4366         lcr = lustre_cfg_rec_new(cmd, n_bufs);
4367         if (!lcr)
4368                 RETURN(-ENOMEM);
4369
4370         lcr->lcr_cfg = *o_lcfg;
4371         rc = llog_write(env, mlfd->mlfd_llh, &lcr->lcr_hdr, LLOG_NEXT_IDX);
4372         lustre_cfg_rec_free(lcr);
4373
4374         RETURN(rc);
4375 }
4376
4377 static int mgs_lcfg_fork_one(const struct lu_env *env, struct mgs_device *mgs,
4378                              struct mgs_direntry *mde, const char *oldname,
4379                              const char *newname)
4380 {
4381         struct llog_handle *old_llh = NULL;
4382         struct llog_handle *new_llh = NULL;
4383         struct llog_ctxt *ctxt = NULL;
4384         struct mgs_lcfg_fork_data *mlfd = NULL;
4385         char *name_buf = NULL;
4386         int name_buflen;
4387         int old_namelen = strlen(oldname);
4388         int new_namelen = strlen(newname);
4389         int rc;
4390         ENTRY;
4391
4392         name_buflen = mde->mde_len + new_namelen - old_namelen;
4393         OBD_ALLOC(name_buf, name_buflen);
4394         if (!name_buf)
4395                 RETURN(-ENOMEM);
4396
4397         memcpy(name_buf, newname, new_namelen);
4398         memcpy(name_buf + new_namelen, mde->mde_name + old_namelen,
4399                mde->mde_len - old_namelen);
4400
4401         CDEBUG(D_MGS, "Fork the config-log from %s to %s\n",
4402                mde->mde_name, name_buf);
4403
4404         ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
4405         LASSERT(ctxt);
4406
4407         rc = llog_open_create(env, ctxt, &new_llh, NULL, name_buf);
4408         if (rc)
4409                 GOTO(out, rc);
4410
4411         rc = llog_init_handle(env, new_llh, LLOG_F_IS_PLAIN, NULL);
4412         if (rc)
4413                 GOTO(out, rc);
4414
4415         if (unlikely(mgs_log_is_empty(env, mgs, mde->mde_name)))
4416                 GOTO(out, rc = 0);
4417
4418         rc = llog_open(env, ctxt, &old_llh, NULL, mde->mde_name,
4419                        LLOG_OPEN_EXISTS);
4420         if (rc)
4421                 GOTO(out, rc);
4422
4423         rc = llog_init_handle(env, old_llh, LLOG_F_IS_PLAIN, NULL);
4424         if (rc)
4425                 GOTO(out, rc);
4426
4427         new_llh->lgh_hdr->llh_tgtuuid = old_llh->lgh_hdr->llh_tgtuuid;
4428
4429         OBD_ALLOC(mlfd, LLOG_MIN_CHUNK_SIZE);
4430         if (!mlfd)
4431                 GOTO(out, rc = -ENOMEM);
4432
4433         mlfd->mlfd_mgs = mgs;
4434         mlfd->mlfd_llh = new_llh;
4435         mlfd->mlfd_oldname = oldname;
4436         mlfd->mlfd_newname = newname;
4437
4438         rc = llog_process(env, old_llh, mgs_lcfg_fork_handler, mlfd, NULL);
4439         OBD_FREE(mlfd, LLOG_MIN_CHUNK_SIZE);
4440
4441         GOTO(out, rc);
4442
4443 out:
4444         if (old_llh)
4445                 llog_close(env, old_llh);
4446         if (new_llh)
4447                 llog_close(env, new_llh);
4448         if (name_buf)
4449                 OBD_FREE(name_buf, name_buflen);
4450         if (ctxt)
4451                 llog_ctxt_put(ctxt);
4452
4453         return rc;
4454 }
4455
4456 int mgs_lcfg_fork(const struct lu_env *env, struct mgs_device *mgs,
4457                   const char *oldname, const char *newname)
4458 {
4459         struct list_head log_list;
4460         struct mgs_direntry *dirent, *n;
4461         int olen = strlen(oldname);
4462         int nlen = strlen(newname);
4463         int count = 0;
4464         int rc = 0;
4465         ENTRY;
4466
4467         if (unlikely(!oldname || oldname[0] == '\0' ||
4468                      !newname || newname[0] == '\0'))
4469                 RETURN(-EINVAL);
4470
4471         if (strcmp(oldname, newname) == 0)
4472                 RETURN(-EINVAL);
4473
4474         /* lock it to prevent fork/erase/register in parallel. */
4475         mutex_lock(&mgs->mgs_mutex);
4476
4477         rc = class_dentry_readdir(env, mgs, &log_list);
4478         if (rc) {
4479                 mutex_unlock(&mgs->mgs_mutex);
4480                 RETURN(rc);
4481         }
4482
4483         if (list_empty(&log_list)) {
4484                 mutex_unlock(&mgs->mgs_mutex);
4485                 RETURN(-ENOENT);
4486         }
4487
4488         list_for_each_entry_safe(dirent, n, &log_list, mde_list) {
4489                 char *ptr;
4490
4491                 ptr = strrchr(dirent->mde_name, '-');
4492                 if (ptr) {
4493                         int tlen = ptr - dirent->mde_name;
4494
4495                         if (tlen == nlen &&
4496                             strncmp(newname, dirent->mde_name, tlen) == 0)
4497                                 GOTO(out, rc = -EEXIST);
4498
4499                         if (tlen == olen &&
4500                             strncmp(oldname, dirent->mde_name, tlen) == 0)
4501                                 continue;
4502                 }
4503
4504                 list_del_init(&dirent->mde_list);
4505                 mgs_direntry_free(dirent);
4506         }
4507
4508         if (list_empty(&log_list)) {
4509                 mutex_unlock(&mgs->mgs_mutex);
4510                 RETURN(-ENOENT);
4511         }
4512
4513         list_for_each_entry(dirent, &log_list, mde_list) {
4514                 rc = mgs_lcfg_fork_one(env, mgs, dirent, oldname, newname);
4515                 if (rc)
4516                         break;
4517
4518                 count++;
4519         }
4520
4521 out:
4522         mutex_unlock(&mgs->mgs_mutex);
4523
4524         list_for_each_entry_safe(dirent, n, &log_list, mde_list) {
4525                 list_del_init(&dirent->mde_list);
4526                 mgs_direntry_free(dirent);
4527         }
4528
4529         if (rc && count > 0)
4530                 mgs_erase_logs(env, mgs, newname);
4531
4532         RETURN(rc);
4533 }
4534
4535 int mgs_lcfg_erase(const struct lu_env *env, struct mgs_device *mgs,
4536                    const char *fsname)
4537 {
4538         int rc;
4539         ENTRY;
4540
4541         if (unlikely(!fsname || fsname[0] == '\0'))
4542                 RETURN(-EINVAL);
4543
4544         rc = mgs_erase_logs(env, mgs, fsname);
4545
4546         RETURN(rc);
4547 }
4548
4549 static int mgs_xattr_del(const struct lu_env *env, struct dt_object *obj)
4550 {
4551         struct dt_device *dev;
4552         struct thandle *th = NULL;
4553         int rc = 0;
4554
4555         ENTRY;
4556
4557         dev = container_of0(obj->do_lu.lo_dev, struct dt_device, dd_lu_dev);
4558         th = dt_trans_create(env, dev);
4559         if (IS_ERR(th))
4560                 RETURN(PTR_ERR(th));
4561
4562         rc = dt_declare_xattr_del(env, obj, XATTR_TARGET_RENAME, th);
4563         if (rc)
4564                 GOTO(stop, rc);
4565
4566         rc = dt_trans_start_local(env, dev, th);
4567         if (rc)
4568                 GOTO(stop, rc);
4569
4570         dt_write_lock(env, obj, 0);
4571         rc = dt_xattr_del(env, obj, XATTR_TARGET_RENAME, th);
4572
4573         GOTO(unlock, rc);
4574
4575 unlock:
4576         dt_write_unlock(env, obj);
4577
4578 stop:
4579         dt_trans_stop(env, dev, th);
4580
4581         return rc;
4582 }
4583
4584 int mgs_lcfg_rename(const struct lu_env *env, struct mgs_device *mgs)
4585 {
4586         struct list_head log_list;
4587         struct mgs_direntry *dirent, *n;
4588         char fsname[16];
4589         struct lu_buf buf = {
4590                 .lb_buf = fsname,
4591                 .lb_len = sizeof(fsname)
4592         };
4593         int rc = 0;
4594
4595         ENTRY;
4596
4597         rc = class_dentry_readdir(env, mgs, &log_list);
4598         if (rc)
4599                 RETURN(rc);
4600
4601         if (list_empty(&log_list))
4602                 RETURN(0);
4603
4604         list_for_each_entry_safe(dirent, n, &log_list, mde_list) {
4605                 struct dt_object *o = NULL;
4606                 char oldname[16];
4607                 char *ptr;
4608                 int len;
4609
4610                 list_del_init(&dirent->mde_list);
4611                 ptr = strrchr(dirent->mde_name, '-');
4612                 if (!ptr)
4613                         goto next;
4614
4615                 len = ptr - dirent->mde_name;
4616                 if (unlikely(len >= sizeof(oldname))) {
4617                         CDEBUG(D_MGS, "Skip invalid configuration file %s\n",
4618                                dirent->mde_name);
4619                         goto next;
4620                 }
4621
4622                 o = local_file_find(env, mgs->mgs_los, mgs->mgs_configs_dir,
4623                                     dirent->mde_name);
4624                 if (IS_ERR(o)) {
4625                         rc = PTR_ERR(o);
4626                         CDEBUG(D_MGS, "Fail to locate file %s: rc = %d\n",
4627                                dirent->mde_name, rc);
4628                         goto next;
4629                 }
4630
4631                 rc = dt_xattr_get(env, o, &buf, XATTR_TARGET_RENAME);
4632                 if (rc < 0) {
4633                         if (rc == -ENODATA)
4634                                 rc = 0;
4635                         else
4636                                 CDEBUG(D_MGS,
4637                                        "Fail to get EA for %s: rc = %d\n",
4638                                        dirent->mde_name, rc);
4639                         goto next;
4640                 }
4641
4642                 if (unlikely(rc == len &&
4643                              memcmp(fsname, dirent->mde_name, len) == 0)) {
4644                         /* The new fsname is the same as the old one. */
4645                         rc = mgs_xattr_del(env, o);
4646                         goto next;
4647                 }
4648
4649                 memcpy(oldname, dirent->mde_name, len);
4650                 oldname[len] = '\0';
4651                 fsname[rc] = '\0';
4652                 rc = mgs_lcfg_fork_one(env, mgs, dirent, oldname, fsname);
4653                 if (rc && rc != -EEXIST) {
4654                         CDEBUG(D_MGS, "Fail to fork %s: rc = %d\n",
4655                                dirent->mde_name, rc);
4656                         goto next;
4657                 }
4658
4659                 rc = mgs_erase_log(env, mgs, dirent->mde_name);
4660                 if (rc) {
4661                         CDEBUG(D_MGS, "Fail to erase old %s: rc = %d\n",
4662                                dirent->mde_name, rc);
4663                         /* keep it there if failed to remove it. */
4664                         rc = 0;
4665                 }
4666
4667 next:
4668                 if (o && !IS_ERR(o))
4669                         lu_object_put(env, &o->do_lu);
4670
4671                 mgs_direntry_free(dirent);
4672                 if (rc)
4673                         break;
4674         }
4675
4676         list_for_each_entry_safe(dirent, n, &log_list, mde_list) {
4677                 list_del_init(&dirent->mde_list);
4678                 mgs_direntry_free(dirent);
4679         }
4680
4681         RETURN(rc);
4682 }
4683
4684 /* from llog_swab */
4685 static void print_lustre_cfg(struct lustre_cfg *lcfg)
4686 {
4687         int i;
4688         ENTRY;
4689
4690         CDEBUG(D_MGS, "lustre_cfg: %p\n", lcfg);
4691         CDEBUG(D_MGS, "\tlcfg->lcfg_version: %#x\n", lcfg->lcfg_version);
4692
4693         CDEBUG(D_MGS, "\tlcfg->lcfg_command: %#x\n", lcfg->lcfg_command);
4694         CDEBUG(D_MGS, "\tlcfg->lcfg_num: %#x\n", lcfg->lcfg_num);
4695         CDEBUG(D_MGS, "\tlcfg->lcfg_flags: %#x\n", lcfg->lcfg_flags);
4696         CDEBUG(D_MGS, "\tlcfg->lcfg_nid: %s\n", libcfs_nid2str(lcfg->lcfg_nid));
4697
4698         CDEBUG(D_MGS, "\tlcfg->lcfg_bufcount: %d\n", lcfg->lcfg_bufcount);
4699         if (lcfg->lcfg_bufcount < LUSTRE_CFG_MAX_BUFCOUNT)
4700                 for (i = 0; i < lcfg->lcfg_bufcount; i++) {
4701                         CDEBUG(D_MGS, "\tlcfg->lcfg_buflens[%d]: %d %s\n",
4702                                i, lcfg->lcfg_buflens[i],
4703                                lustre_cfg_string(lcfg, i));
4704                 }
4705         EXIT;
4706 }
4707
4708 /* Setup _mgs fsdb and log
4709  */
4710 int mgs__mgs_fsdb_setup(const struct lu_env *env, struct mgs_device *mgs)
4711 {
4712         struct fs_db *fsdb = NULL;
4713         int rc;
4714         ENTRY;
4715
4716         rc = mgs_find_or_make_fsdb(env, mgs, MGSSELF_NAME, &fsdb);
4717         if (!rc)
4718                 mgs_put_fsdb(mgs, fsdb);
4719
4720         RETURN(rc);
4721 }
4722
4723 /* Setup params fsdb and log
4724  */
4725 int mgs_params_fsdb_setup(const struct lu_env *env, struct mgs_device *mgs)
4726 {
4727         struct fs_db *fsdb = NULL;
4728         struct llog_handle *params_llh = NULL;
4729         int rc;
4730         ENTRY;
4731
4732         rc = mgs_find_or_make_fsdb(env, mgs, PARAMS_FILENAME, &fsdb);
4733         if (!rc) {
4734                 mutex_lock(&fsdb->fsdb_mutex);
4735                 rc = record_start_log(env, mgs, &params_llh, PARAMS_FILENAME);
4736                 if (!rc)
4737                         rc = record_end_log(env, &params_llh);
4738                 mutex_unlock(&fsdb->fsdb_mutex);
4739                 mgs_put_fsdb(mgs, fsdb);
4740         }
4741
4742         RETURN(rc);
4743 }
4744
4745 /* Cleanup params fsdb and log
4746  */
4747 int mgs_params_fsdb_cleanup(const struct lu_env *env, struct mgs_device *mgs)
4748 {
4749         int rc;
4750
4751         rc = mgs_erase_logs(env, mgs, PARAMS_FILENAME);
4752         return rc == -ENOENT ? 0 : rc;
4753 }
4754
4755 /* Set a permanent (config log) param for a target or fs
4756  * \param lcfg buf0 may contain the device (testfs-MDT0000) name
4757  *             buf1 contains the single parameter
4758  */
4759 int mgs_setparam(const struct lu_env *env, struct mgs_device *mgs,
4760                  struct lustre_cfg *lcfg, char *fsname)
4761 {
4762         struct fs_db *fsdb = NULL;
4763         struct mgs_target_info *mti = NULL;
4764         char *devname, *param;
4765         char *ptr;
4766         const char *tmp;
4767         __u32 index;
4768         int rc = 0;
4769         bool free = false;
4770         ENTRY;
4771
4772         print_lustre_cfg(lcfg);
4773
4774         /* lustre, lustre-mdtlov, lustre-client, lustre-MDT0000 */
4775         devname = lustre_cfg_string(lcfg, 0);
4776         param = lustre_cfg_string(lcfg, 1);
4777         if (!devname) {
4778                 /* Assume device name embedded in param:
4779                    lustre-OST0000.osc.max_dirty_mb=32 */
4780                 ptr = strchr(param, '.');
4781                 if (ptr) {
4782                         devname = param;
4783                         *ptr = 0;
4784                         param = ptr + 1;
4785                 }
4786         }
4787         if (!devname) {
4788                 LCONSOLE_ERROR_MSG(0x14d, "No target specified: %s\n", param);
4789                 RETURN(-ENOSYS);
4790         }
4791
4792         rc = mgs_parse_devname(devname, fsname, NULL);
4793         if (rc == 0 && !mgs_parse_devname(devname, NULL, &index)) {
4794                 /* param related to llite isn't allowed to set by OST or MDT */
4795                 if (rc == 0 && strncmp(param, PARAM_LLITE,
4796                                        sizeof(PARAM_LLITE) - 1) == 0)
4797                         RETURN(-EINVAL);
4798         } else {
4799                 /* assume devname is the fsname */
4800                 strlcpy(fsname, devname, MTI_NAME_MAXLEN);
4801         }
4802         CDEBUG(D_MGS, "setparam fs='%s' device='%s'\n", fsname, devname);
4803
4804         rc = mgs_find_or_make_fsdb(env, mgs,
4805                                    lcfg->lcfg_command == LCFG_SET_PARAM ?
4806                                    PARAMS_FILENAME : fsname, &fsdb);
4807         if (rc)
4808                 RETURN(rc);
4809
4810         if (lcfg->lcfg_command != LCFG_SET_PARAM &&
4811             !test_bit(FSDB_MGS_SELF, &fsdb->fsdb_flags) &&
4812             test_bit(FSDB_LOG_EMPTY, &fsdb->fsdb_flags)) {
4813                 CERROR("No filesystem targets for %s. cfg_device from lctl "
4814                        "is '%s'\n", fsname, devname);
4815                 free = true;
4816                 GOTO(out, rc = -EINVAL);
4817         }
4818
4819         /* Create a fake mti to hold everything */
4820         OBD_ALLOC_PTR(mti);
4821         if (!mti)
4822                 GOTO(out, rc = -ENOMEM);
4823         if (strlcpy(mti->mti_fsname, fsname, sizeof(mti->mti_fsname))
4824             >= sizeof(mti->mti_fsname))
4825                 GOTO(out, rc = -E2BIG);
4826         if (strlcpy(mti->mti_svname, devname, sizeof(mti->mti_svname))
4827             >= sizeof(mti->mti_svname))
4828                 GOTO(out, rc = -E2BIG);
4829         if (strlcpy(mti->mti_params, param, sizeof(mti->mti_params))
4830             >= sizeof(mti->mti_params))
4831                 GOTO(out, rc = -E2BIG);
4832         rc = server_name2index(mti->mti_svname, &mti->mti_stripe_index, &tmp);
4833         if (rc < 0)
4834                 /* Not a valid server; may be only fsname */
4835                 rc = 0;
4836         else
4837                 /* Strip -osc or -mdc suffix from svname */
4838                 if (server_make_name(rc, mti->mti_stripe_index, mti->mti_fsname,
4839                                      mti->mti_svname, sizeof(mti->mti_svname)))
4840                         GOTO(out, rc = -EINVAL);
4841         /*
4842          * Revoke lock so everyone updates.  Should be alright if
4843          * someone was already reading while we were updating the logs,
4844          * so we don't really need to hold the lock while we're
4845          * writing (above).
4846          */
4847         if (lcfg->lcfg_command == LCFG_SET_PARAM) {
4848                 mti->mti_flags = rc | LDD_F_PARAM2;
4849                 mutex_lock(&fsdb->fsdb_mutex);
4850                 rc = mgs_write_log_param2(env, mgs, fsdb, mti, mti->mti_params);
4851                 mutex_unlock(&fsdb->fsdb_mutex);
4852                 mgs_revoke_lock(mgs, fsdb, CONFIG_T_PARAMS);
4853         } else {
4854                 mti->mti_flags = rc | LDD_F_PARAM;
4855                 mutex_lock(&fsdb->fsdb_mutex);
4856                 rc = mgs_write_log_param(env, mgs, fsdb, mti, mti->mti_params);
4857                 mutex_unlock(&fsdb->fsdb_mutex);
4858                 mgs_revoke_lock(mgs, fsdb, CONFIG_T_CONFIG);
4859         }
4860
4861 out:
4862         if (mti)
4863                 OBD_FREE_PTR(mti);
4864
4865         if (fsdb) {
4866                 if (free)
4867                         mgs_unlink_fsdb(mgs, fsdb);
4868                 mgs_put_fsdb(mgs, fsdb);
4869         }
4870
4871         RETURN(rc);
4872 }
4873
4874 static int mgs_write_log_pool(const struct lu_env *env,
4875                               struct mgs_device *mgs, char *logname,
4876                               struct fs_db *fsdb, char *tgtname,
4877                               enum lcfg_command_type cmd,
4878                               char *fsname, char *poolname,
4879                               char *ostname, char *comment)
4880 {
4881         struct llog_handle *llh = NULL;
4882         int rc;
4883
4884         rc = record_start_log(env, mgs, &llh, logname);
4885         if (rc)
4886                 return rc;
4887         rc = record_marker(env, llh, fsdb, CM_START, tgtname, comment);
4888         if (rc)
4889                 goto out;
4890         rc = record_base(env, llh, tgtname, 0, cmd,
4891                          fsname, poolname, ostname, NULL);
4892         if (rc)
4893                 goto out;
4894         rc = record_marker(env, llh, fsdb, CM_END, tgtname, comment);
4895 out:
4896         record_end_log(env, &llh);
4897         return rc;
4898 }
4899
4900 int mgs_nodemap_cmd(const struct lu_env *env, struct mgs_device *mgs,
4901                     enum lcfg_command_type cmd, const char *nodemap_name,
4902                     char *param)
4903 {
4904         lnet_nid_t      nid[2];
4905         __u32           idmap[2];
4906         bool            bool_switch;
4907         __u32           int_id;
4908         int             rc = 0;
4909         ENTRY;
4910
4911         switch (cmd) {
4912         case LCFG_NODEMAP_ADD:
4913                 rc = nodemap_add(nodemap_name);
4914                 break;
4915         case LCFG_NODEMAP_DEL:
4916                 rc = nodemap_del(nodemap_name);
4917                 break;
4918         case LCFG_NODEMAP_ADD_RANGE:
4919                 rc = nodemap_parse_range(param, nid);
4920                 if (rc != 0)
4921                         break;
4922                 rc = nodemap_add_range(nodemap_name, nid);
4923                 break;
4924         case LCFG_NODEMAP_DEL_RANGE:
4925                 rc = nodemap_parse_range(param, nid);
4926                 if (rc != 0)
4927                         break;
4928                 rc = nodemap_del_range(nodemap_name, nid);
4929                 break;
4930         case LCFG_NODEMAP_ADMIN:
4931                 bool_switch = simple_strtoul(param, NULL, 10);
4932                 rc = nodemap_set_allow_root(nodemap_name, bool_switch);
4933                 break;
4934         case LCFG_NODEMAP_DENY_UNKNOWN:
4935                 bool_switch = simple_strtoul(param, NULL, 10);
4936                 rc = nodemap_set_deny_unknown(nodemap_name, bool_switch);
4937                 break;
4938         case LCFG_NODEMAP_TRUSTED:
4939                 bool_switch = simple_strtoul(param, NULL, 10);
4940                 rc = nodemap_set_trust_client_ids(nodemap_name, bool_switch);
4941                 break;
4942         case LCFG_NODEMAP_SQUASH_UID:
4943                 int_id = simple_strtoul(param, NULL, 10);
4944                 rc = nodemap_set_squash_uid(nodemap_name, int_id);
4945                 break;
4946         case LCFG_NODEMAP_SQUASH_GID:
4947                 int_id = simple_strtoul(param, NULL, 10);
4948                 rc = nodemap_set_squash_gid(nodemap_name, int_id);
4949                 break;
4950         case LCFG_NODEMAP_ADD_UIDMAP:
4951         case LCFG_NODEMAP_ADD_GIDMAP:
4952                 rc = nodemap_parse_idmap(param, idmap);
4953                 if (rc != 0)
4954                         break;
4955                 if (cmd == LCFG_NODEMAP_ADD_UIDMAP)
4956                         rc = nodemap_add_idmap(nodemap_name, NODEMAP_UID,
4957                                                idmap);
4958                 else
4959                         rc = nodemap_add_idmap(nodemap_name, NODEMAP_GID,
4960                                                idmap);
4961                 break;
4962         case LCFG_NODEMAP_DEL_UIDMAP:
4963         case LCFG_NODEMAP_DEL_GIDMAP:
4964                 rc = nodemap_parse_idmap(param, idmap);
4965                 if (rc != 0)
4966                         break;
4967                 if (cmd == LCFG_NODEMAP_DEL_UIDMAP)
4968                         rc = nodemap_del_idmap(nodemap_name, NODEMAP_UID,
4969                                                idmap);
4970                 else
4971                         rc = nodemap_del_idmap(nodemap_name, NODEMAP_GID,
4972                                                idmap);
4973                 break;
4974         case LCFG_NODEMAP_SET_FILESET:
4975                 rc = nodemap_set_fileset(nodemap_name, param);
4976                 break;
4977         default:
4978                 rc = -EINVAL;
4979         }
4980
4981         RETURN(rc);
4982 }
4983
4984 int mgs_pool_cmd(const struct lu_env *env, struct mgs_device *mgs,
4985                  enum lcfg_command_type cmd, char *fsname,
4986                  char *poolname, char *ostname)
4987 {
4988         struct fs_db *fsdb;
4989         char *lovname;
4990         char *logname;
4991         char *label = NULL, *canceled_label = NULL;
4992         int label_sz;
4993         struct mgs_target_info *mti = NULL;
4994         bool checked = false;
4995         bool locked = false;
4996         bool free = false;
4997         int rc, i;
4998         ENTRY;
4999
5000         rc = mgs_find_or_make_fsdb(env, mgs, fsname, &fsdb);
5001         if (rc) {
5002                 CERROR("Can't get db for %s\n", fsname);
5003                 RETURN(rc);
5004         }
5005         if (test_bit(FSDB_LOG_EMPTY, &fsdb->fsdb_flags)) {
5006                 CERROR("%s is not defined\n", fsname);
5007                 free = true;
5008                 GOTO(out_fsdb, rc = -EINVAL);
5009         }
5010
5011         label_sz = 10 + strlen(fsname) + strlen(poolname);
5012
5013         /* check if ostname match fsname */
5014         if (ostname != NULL) {
5015                 char *ptr;
5016
5017                 ptr = strrchr(ostname, '-');
5018                 if ((ptr == NULL) ||
5019                     (strncmp(fsname, ostname, ptr-ostname) != 0))
5020                         RETURN(-EINVAL);
5021                 label_sz += strlen(ostname);
5022         }
5023
5024         OBD_ALLOC(label, label_sz);
5025         if (!label)
5026                 GOTO(out_fsdb, rc = -ENOMEM);
5027
5028         switch(cmd) {
5029         case LCFG_POOL_NEW:
5030                 sprintf(label,
5031                         "new %s.%s", fsname, poolname);
5032                 break;
5033         case LCFG_POOL_ADD:
5034                 sprintf(label,
5035                         "add %s.%s.%s", fsname, poolname, ostname);
5036                 break;
5037         case LCFG_POOL_REM:
5038                 OBD_ALLOC(canceled_label, label_sz);
5039                 if (canceled_label == NULL)
5040                         GOTO(out_label, rc = -ENOMEM);
5041                 sprintf(label,
5042                         "rem %s.%s.%s", fsname, poolname, ostname);
5043                 sprintf(canceled_label,
5044                         "add %s.%s.%s", fsname, poolname, ostname);
5045                 break;
5046         case LCFG_POOL_DEL:
5047                 OBD_ALLOC(canceled_label, label_sz);
5048                 if (canceled_label == NULL)
5049                         GOTO(out_label, rc = -ENOMEM);
5050                 sprintf(label,
5051                         "del %s.%s", fsname, poolname);
5052                 sprintf(canceled_label,
5053                         "new %s.%s", fsname, poolname);
5054                 break;
5055         default:
5056                 break;
5057         }
5058
5059         OBD_ALLOC_PTR(mti);
5060         if (mti == NULL)
5061                 GOTO(out_cancel, rc = -ENOMEM);
5062         strncpy(mti->mti_svname, "lov pool", sizeof(mti->mti_svname));
5063
5064         mutex_lock(&fsdb->fsdb_mutex);
5065         locked = true;
5066         /* write pool def to all MDT logs */
5067         for (i = 0; i < INDEX_MAP_SIZE * 8; i++) {
5068                 if (test_bit(i,  fsdb->fsdb_mdt_index_map)) {
5069                         rc = name_create_mdt_and_lov(&logname, &lovname,
5070                                                      fsdb, i);
5071                         if (rc)
5072                                 GOTO(out_mti, rc);
5073
5074                         if (!checked && (canceled_label == NULL)) {
5075                                 rc = mgs_check_marker(env, mgs, fsdb, mti,
5076                                                 logname, lovname, label);
5077                                 if (rc) {
5078                                         name_destroy(&logname);
5079                                         name_destroy(&lovname);
5080                                         GOTO(out_mti,
5081                                                 rc = (rc == LLOG_PROC_BREAK ?
5082                                                         -EEXIST : rc));
5083                                 }
5084                                 checked = true;
5085                         }
5086                         if (canceled_label != NULL)
5087                                 rc = mgs_modify(env, mgs, fsdb, mti, logname,
5088                                                 lovname, canceled_label,
5089                                                 CM_SKIP);
5090
5091                         if (rc >= 0)
5092                                 rc = mgs_write_log_pool(env, mgs, logname,
5093                                                         fsdb, lovname, cmd,
5094                                                         fsname, poolname,
5095                                                         ostname, label);
5096                         name_destroy(&logname);
5097                         name_destroy(&lovname);
5098                         if (rc)
5099                                 GOTO(out_mti, rc);
5100                 }
5101         }
5102
5103         rc = name_create(&logname, fsname, "-client");
5104         if (rc)
5105                 GOTO(out_mti, rc);
5106
5107         if (!checked && (canceled_label == NULL)) {
5108                 rc = mgs_check_marker(env, mgs, fsdb, mti, logname,
5109                                 fsdb->fsdb_clilov, label);
5110                 if (rc) {
5111                         name_destroy(&logname);
5112                         GOTO(out_mti, rc = (rc == LLOG_PROC_BREAK ?
5113                                 -EEXIST : rc));
5114                 }
5115         }
5116         if (canceled_label != NULL) {
5117                 rc = mgs_modify(env, mgs, fsdb, mti, logname,
5118                                 fsdb->fsdb_clilov, canceled_label, CM_SKIP);
5119                 if (rc < 0) {
5120                         name_destroy(&logname);
5121                         GOTO(out_mti, rc);
5122                 }
5123         }
5124
5125         rc = mgs_write_log_pool(env, mgs, logname, fsdb, fsdb->fsdb_clilov,
5126                                 cmd, fsname, poolname, ostname, label);
5127         mutex_unlock(&fsdb->fsdb_mutex);
5128         locked = false;
5129         name_destroy(&logname);
5130         /* request for update */
5131         mgs_revoke_lock(mgs, fsdb, CONFIG_T_CONFIG);
5132
5133         GOTO(out_mti, rc);
5134
5135 out_mti:
5136         if (locked)
5137                 mutex_unlock(&fsdb->fsdb_mutex);
5138         if (mti != NULL)
5139                 OBD_FREE_PTR(mti);
5140 out_cancel:
5141         if (canceled_label != NULL)
5142                 OBD_FREE(canceled_label, label_sz);
5143 out_label:
5144         OBD_FREE(label, label_sz);
5145 out_fsdb:
5146         if (free)
5147                 mgs_unlink_fsdb(mgs, fsdb);
5148         mgs_put_fsdb(mgs, fsdb);
5149
5150         return rc;
5151 }