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