lustre/mgs/mgs_llog.c

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