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