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