lustre/mgs/mgs_llog.c

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