lustre/mgs/mgs_handler.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
  30  * Use is subject to license terms.
  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_handler.c
  37  *
  38  * Author: Nathan Rutman <nathan@clusterfs.com>
  39  */
  40
  41 #ifndef EXPORT_SYMTAB
  42 # define EXPORT_SYMTAB
  43 #endif
  44 #define DEBUG_SUBSYSTEM S_MGS
  45 #define D_MGS D_CONFIG/*|D_WARNING*/
  46
  47 #ifdef __KERNEL__
  48 # include <linux/module.h>
  49 # include <linux/pagemap.h>
  50 # include <linux/miscdevice.h>
  51 # include <linux/init.h>
  52 #else
  53 # include <liblustre.h>
  54 #endif
  55
  56 #include <obd_class.h>
  57 #include <lustre_dlm.h>
  58 #include <lprocfs_status.h>
  59 #include <lustre_fsfilt.h>
  60 #include <lustre_disk.h>
  61 #include "mgs_internal.h"
  62
  63
  64 /* Establish a connection to the MGS.*/
  65 static int mgs_connect(const struct lu_env *env,
  66                        struct obd_export **exp, struct obd_device *obd,
  67                        struct obd_uuid *cluuid, struct obd_connect_data *data,
  68                        void *localdata)
  69 {
  70         struct obd_export *lexp;
  71         struct lustre_handle conn = { 0 };
  72         int rc;
  73         ENTRY;
  74
  75         if (!exp || !obd || !cluuid)
  76                 RETURN(-EINVAL);
  77
  78         rc = class_connect(&conn, obd, cluuid);
  79         if (rc)
  80                 RETURN(rc);
  81
  82         lexp = class_conn2export(&conn);
  83         LASSERT(lexp);
  84
  85         mgs_counter_incr(lexp, LPROC_MGS_CONNECT);
  86
  87         if (data != NULL) {
  88                 data->ocd_connect_flags &= MGS_CONNECT_SUPPORTED;
  89                 lexp->exp_connect_flags = data->ocd_connect_flags;
  90                 data->ocd_version = LUSTRE_VERSION_CODE;
  91         }
  92
  93         rc = mgs_client_add(obd, lexp, localdata);
  94
  95         if (rc) {
  96                 class_disconnect(lexp);
  97         } else {
  98                 *exp = lexp;
  99         }
 100
 101         RETURN(rc);
 102 }
 103
 104 static int mgs_reconnect(const struct lu_env *env,
 105                          struct obd_export *exp, struct obd_device *obd,
 106                          struct obd_uuid *cluuid, struct obd_connect_data *data,
 107                          void *localdata)
 108 {
 109         ENTRY;
 110
 111         if (exp == NULL || obd == NULL || cluuid == NULL)
 112                 RETURN(-EINVAL);
 113
 114         mgs_counter_incr(exp, LPROC_MGS_CONNECT);
 115
 116         if (data != NULL) {
 117                 data->ocd_connect_flags &= MGS_CONNECT_SUPPORTED;
 118                 exp->exp_connect_flags = data->ocd_connect_flags;
 119                 data->ocd_version = LUSTRE_VERSION_CODE;
 120         }
 121
 122         RETURN(0);
 123 }
 124
 125 static int mgs_disconnect(struct obd_export *exp)
 126 {
 127         int rc;
 128         ENTRY;
 129
 130         LASSERT(exp);
 131
 132         class_export_get(exp);
 133         mgs_counter_incr(exp, LPROC_MGS_DISCONNECT);
 134
 135         /* Disconnect early so that clients can't keep using export */
 136         rc = class_disconnect(exp);
 137         ldlm_cancel_locks_for_export(exp);
 138
 139         lprocfs_exp_cleanup(exp);
 140
 141         /* complete all outstanding replies */
 142         spin_lock(&exp->exp_lock);
 143         while (!list_empty(&exp->exp_outstanding_replies)) {
 144                 struct ptlrpc_reply_state *rs =
 145                         list_entry(exp->exp_outstanding_replies.next,
 146                                    struct ptlrpc_reply_state, rs_exp_list);
 147                 struct ptlrpc_service *svc = rs->rs_service;
 148
 149                 spin_lock(&svc->srv_lock);
 150                 list_del_init(&rs->rs_exp_list);
 151                 spin_lock(&rs->rs_lock);
 152                 ptlrpc_schedule_difficult_reply(rs);
 153                 spin_unlock(&rs->rs_lock);
 154                 spin_unlock(&svc->srv_lock);
 155         }
 156         spin_unlock(&exp->exp_lock);
 157
 158         class_export_put(exp);
 159         RETURN(rc);
 160 }
 161
 162 static int mgs_cleanup(struct obd_device *obd);
 163 static int mgs_handle(struct ptlrpc_request *req);
 164
 165 static int mgs_llog_init(struct obd_device *obd, struct obd_llog_group *olg,
 166                          struct obd_device *tgt, int count,
 167                          struct llog_catid *logid, struct obd_uuid *uuid)
 168 {
 169         int rc;
 170         ENTRY;
 171
 172         LASSERT(olg == &obd->obd_olg);
 173         rc = llog_setup(obd, olg, LLOG_CONFIG_ORIG_CTXT, obd, 0, NULL,
 174                         &llog_lvfs_ops);
 175         RETURN(rc);
 176 }
 177
 178 static int mgs_llog_finish(struct obd_device *obd, int count)
 179 {
 180         struct llog_ctxt *ctxt;
 181         int rc = 0;
 182         ENTRY;
 183
 184         ctxt = llog_get_context(obd, LLOG_CONFIG_ORIG_CTXT);
 185         if (ctxt)
 186                 rc = llog_cleanup(ctxt);
 187
 188         RETURN(rc);
 189 }
 190
 191 /* Start the MGS obd */
 192 static int mgs_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
 193 {
 194         struct lprocfs_static_vars lvars;
 195         struct mgs_obd *mgs = &obd->u.mgs;
 196         struct lustre_mount_info *lmi;
 197         struct lustre_sb_info *lsi;
 198         struct vfsmount *mnt;
 199         int rc = 0;
 200         ENTRY;
 201
 202         CDEBUG(D_CONFIG, "Starting MGS\n");
 203
 204         /* Find our disk */
 205         lmi = server_get_mount(obd->obd_name);
 206         if (!lmi)
 207                 RETURN(rc = -EINVAL);
 208
 209         mnt = lmi->lmi_mnt;
 210         lsi = s2lsi(lmi->lmi_sb);
 211         obd->obd_fsops = fsfilt_get_ops(MT_STR(lsi->lsi_ldd));
 212         if (IS_ERR(obd->obd_fsops))
 213                 GOTO(err_put, rc = PTR_ERR(obd->obd_fsops));
 214
 215         /* namespace for mgs llog */
 216         obd->obd_namespace = ldlm_namespace_new(obd ,"MGS", LDLM_NAMESPACE_SERVER,
 217                                                 LDLM_NAMESPACE_MODEST);
 218         if (obd->obd_namespace == NULL)
 219                 GOTO(err_ops, rc = -ENOMEM);
 220
 221         /* ldlm setup */
 222         ptlrpc_init_client(LDLM_CB_REQUEST_PORTAL, LDLM_CB_REPLY_PORTAL,
 223                            "mgs_ldlm_client", &obd->obd_ldlm_client);
 224
 225         if (lvfs_check_rdonly(lvfs_sbdev(mnt->mnt_sb))) {
 226                 CERROR("%s: Underlying device is marked as read-only. "
 227                        "Setup failed\n", obd->obd_name);
 228                 GOTO(err_ops, rc = -EROFS);
 229         }
 230
 231         rc = mgs_fs_setup(obd, mnt);
 232         if (rc) {
 233                 CERROR("%s: MGS filesystem method init failed: rc = %d\n",
 234                        obd->obd_name, rc);
 235                 GOTO(err_ns, rc);
 236         }
 237
 238         rc = obd_llog_init(obd, &obd->obd_olg, obd, 0, NULL, NULL);
 239         if (rc)
 240                 GOTO(err_fs, rc);
 241
 242         /* No recovery for MGC's */
 243         obd->obd_replayable = 0;
 244
 245         /* Internal mgs setup */
 246         mgs_init_fsdb_list(obd);
 247         sema_init(&mgs->mgs_sem, 1);
 248
 249         /* Start the service threads */
 250         mgs->mgs_service =
 251                 ptlrpc_init_svc(MGS_NBUFS, MGS_BUFSIZE, MGS_MAXREQSIZE,
 252                                 MGS_MAXREPSIZE, MGS_REQUEST_PORTAL,
 253                                 MGC_REPLY_PORTAL, 2000,
 254                                 mgs_handle, LUSTRE_MGS_NAME,
 255                                 obd->obd_proc_entry, target_print_req,
 256                                 MGS_THREADS_AUTO_MIN, MGS_THREADS_AUTO_MAX,
 257                                 "ll_mgs", LCT_MD_THREAD, NULL);
 258
 259         if (!mgs->mgs_service) {
 260                 CERROR("failed to start service\n");
 261                 GOTO(err_llog, rc = -ENOMEM);
 262         }
 263
 264         rc = ptlrpc_start_threads(obd, mgs->mgs_service);
 265         if (rc)
 266                 GOTO(err_thread, rc);
 267
 268         /* Setup proc */
 269         lprocfs_mgs_init_vars(&lvars);
 270         if (lprocfs_obd_setup(obd, lvars.obd_vars) == 0) {
 271                 lproc_mgs_setup(obd);
 272         }
 273
 274         ping_evictor_start();
 275
 276         LCONSOLE_INFO("MGS %s started\n", obd->obd_name);
 277
 278         RETURN(0);
 279
 280 err_thread:
 281         ptlrpc_unregister_service(mgs->mgs_service);
 282 err_llog:
 283         obd_llog_finish(obd, 0);
 284 err_fs:
 285         /* No extra cleanup needed for llog_init_commit_thread() */
 286         mgs_fs_cleanup(obd);
 287 err_ns:
 288         ldlm_namespace_free(obd->obd_namespace, NULL, 0);
 289         obd->obd_namespace = NULL;
 290 err_ops:
 291         fsfilt_put_ops(obd->obd_fsops);
 292 err_put:
 293         server_put_mount(obd->obd_name, mnt);
 294         mgs->mgs_sb = 0;
 295         return rc;
 296 }
 297
 298 static int mgs_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
 299 {
 300         int rc = 0;
 301         ENTRY;
 302
 303         switch (stage) {
 304         case OBD_CLEANUP_EARLY:
 305         case OBD_CLEANUP_EXPORTS:
 306                 rc = obd_llog_finish(obd, 0);
 307                 break;
 308         }
 309         RETURN(rc);
 310 }
 311
 312 /**
 313  * Performs cleanup procedures for passed \a obd given it is mgs obd.
 314  */
 315 static int mgs_cleanup(struct obd_device *obd)
 316 {
 317         struct mgs_obd *mgs = &obd->u.mgs;
 318         ENTRY;
 319
 320         if (mgs->mgs_sb == NULL)
 321                 RETURN(0);
 322
 323         ping_evictor_stop();
 324
 325         ptlrpc_unregister_service(mgs->mgs_service);
 326
 327         mgs_cleanup_fsdb_list(obd);
 328         lproc_mgs_cleanup(obd);
 329         mgs_fs_cleanup(obd);
 330
 331         server_put_mount(obd->obd_name, mgs->mgs_vfsmnt);
 332         mgs->mgs_sb = NULL;
 333
 334         ldlm_namespace_free(obd->obd_namespace, NULL, 1);
 335         obd->obd_namespace = NULL;
 336
 337         fsfilt_put_ops(obd->obd_fsops);
 338
 339         LCONSOLE_INFO("%s has stopped.\n", obd->obd_name);
 340         RETURN(0);
 341 }
 342
 343 /* similar to filter_prepare_destroy */
 344 static int mgs_get_cfg_lock(struct obd_device *obd, char *fsname,
 345                             struct lustre_handle *lockh)
 346 {
 347         struct ldlm_res_id res_id;
 348         int rc, flags = 0;
 349         ENTRY;
 350
 351         rc = mgc_fsname2resid(fsname, &res_id);
 352         if (!rc)
 353                 rc = ldlm_cli_enqueue_local(obd->obd_namespace, &res_id,
 354                                             LDLM_PLAIN, NULL, LCK_EX,
 355                                             &flags, ldlm_blocking_ast,
 356                                             ldlm_completion_ast, NULL,
 357                                             fsname, 0, NULL, NULL, lockh);
 358         if (rc)
 359                 CERROR("can't take cfg lock for %s (%d)\n", fsname, rc);
 360
 361         RETURN(rc);
 362 }
 363
 364 static int mgs_put_cfg_lock(struct lustre_handle *lockh)
 365 {
 366         ENTRY;
 367         ldlm_lock_decref(lockh, LCK_EX);
 368         RETURN(0);
 369 }
 370
 371 static void mgs_revoke_lock(struct obd_device *obd, char *fsname,
 372                             struct lustre_handle *lockh)
 373 {
 374         int lockrc;
 375
 376         if (fsname[0]) {
 377                 lockrc = mgs_get_cfg_lock(obd, fsname, lockh);
 378                 if (lockrc != ELDLM_OK)
 379                         CERROR("lock error %d for fs %s\n", lockrc,
 380                                fsname);
 381                 else
 382                         mgs_put_cfg_lock(lockh);
 383         }
 384 }
 385
 386 /* rc=0 means ok
 387       1 means update
 388      <0 means error */
 389 static int mgs_check_target(struct obd_device *obd, struct mgs_target_info *mti)
 390 {
 391         int rc;
 392         ENTRY;
 393
 394         rc = mgs_check_index(obd, mti);
 395         if (rc == 0) {
 396                 LCONSOLE_ERROR_MSG(0x13b, "%s claims to have registered, but "
 397                                    "this MGS does not know about it, preventing "
 398                                    "registration.\n", mti->mti_svname);
 399                 rc = -ENOENT;
 400         } else if (rc == -1) {
 401                 LCONSOLE_ERROR_MSG(0x13c, "Client log %s-client has "
 402                                    "disappeared! Regenerating all logs.\n",
 403                                    mti->mti_fsname);
 404                 mti->mti_flags |= LDD_F_WRITECONF;
 405                 rc = 1;
 406         } else {
 407                 /* Index is correctly marked as used */
 408
 409                 /* If the logs don't contain the mti_nids then add
 410                    them as failover nids */
 411                 rc = mgs_check_failnid(obd, mti);
 412         }
 413
 414         RETURN(rc);
 415 }
 416
 417 /* Called whenever a target starts up.  Flags indicate first connect, etc. */
 418 static int mgs_handle_target_reg(struct ptlrpc_request *req)
 419 {
 420         struct obd_device *obd = req->rq_export->exp_obd;
 421         struct lustre_handle lockh;
 422         struct mgs_target_info *mti, *rep_mti;
 423         int rc = 0, lockrc;
 424         ENTRY;
 425
 426         mgs_counter_incr(req->rq_export, LPROC_MGS_TARGET_REG);
 427
 428         mti = req_capsule_client_get(&req->rq_pill, &RMF_MGS_TARGET_INFO);
 429         if (!(mti->mti_flags & (LDD_F_WRITECONF | LDD_F_UPGRADE14 |
 430                                 LDD_F_UPDATE))) {
 431                 /* We're just here as a startup ping. */
 432                 CDEBUG(D_MGS, "Server %s is running on %s\n",
 433                        mti->mti_svname, obd_export_nid2str(req->rq_export));
 434                 rc = mgs_check_target(obd, mti);
 435                 /* above will set appropriate mti flags */
 436                 if (rc <= 0)
 437                         /* Nothing wrong, or fatal error */
 438                         GOTO(out_nolock, rc);
 439         }
 440
 441         /* Revoke the config lock to make sure nobody is reading. */
 442         /* Although actually I think it should be alright if
 443            someone was reading while we were updating the logs - if we
 444            revoke at the end they will just update from where they left off. */
 445         lockrc = mgs_get_cfg_lock(obd, mti->mti_fsname, &lockh);
 446         if (lockrc != ELDLM_OK) {
 447                 LCONSOLE_ERROR_MSG(0x13d, "%s: Can't signal other nodes to "
 448                                    "update their configuration (%d). Updating "
 449                                    "local logs anyhow; you might have to "
 450                                    "manually restart other nodes to get the "
 451                                    "latest configuration.\n",
 452                                    obd->obd_name, lockrc);
 453         }
 454
 455         OBD_FAIL_TIMEOUT(OBD_FAIL_MGS_PAUSE_TARGET_REG, 10);
 456
 457         /* Log writing contention is handled by the fsdb_sem */
 458
 459         if (mti->mti_flags & LDD_F_WRITECONF) {
 460                 if (mti->mti_flags & LDD_F_SV_TYPE_MDT &&
 461                     mti->mti_stripe_index == 0) {
 462                         rc = mgs_erase_logs(obd, mti->mti_fsname);
 463                         LCONSOLE_WARN("%s: Logs for fs %s were removed by user "
 464                                       "request.  All servers must be restarted "
 465                                       "in order to regenerate the logs."
 466                                       "\n", obd->obd_name, mti->mti_fsname);
 467                 } else if (mti->mti_flags &
 468                            (LDD_F_SV_TYPE_OST | LDD_F_SV_TYPE_MDT)) {
 469                         rc = mgs_erase_log(obd, mti->mti_svname);
 470                         LCONSOLE_WARN("%s: Regenerating %s log by user "
 471                                       "request.\n",
 472                                       obd->obd_name, mti->mti_svname);
 473                 }
 474                 mti->mti_flags |= LDD_F_UPDATE;
 475                 /* Erased logs means start from scratch. */
 476                 mti->mti_flags &= ~LDD_F_UPGRADE14;
 477         }
 478
 479         /* COMPAT_146 */
 480         if (mti->mti_flags & LDD_F_UPGRADE14) {
 481                 rc = mgs_upgrade_sv_14(obd, mti);
 482                 if (rc) {
 483                         CERROR("Can't upgrade from 1.4 (%d)\n", rc);
 484                         GOTO(out, rc);
 485                 }
 486
 487                 /* We're good to go */
 488                 mti->mti_flags |= LDD_F_UPDATE;
 489         }
 490         /* end COMPAT_146 */
 491
 492         if (mti->mti_flags & LDD_F_UPDATE) {
 493                 CDEBUG(D_MGS, "updating %s, index=%d\n", mti->mti_svname,
 494                        mti->mti_stripe_index);
 495
 496                 /* create or update the target log
 497                    and update the client/mdt logs */
 498                 rc = mgs_write_log_target(obd, mti);
 499                 if (rc) {
 500                         CERROR("Failed to write %s log (%d)\n",
 501                                mti->mti_svname, rc);
 502                         GOTO(out, rc);
 503                 }
 504
 505                 mti->mti_flags &= ~(LDD_F_VIRGIN | LDD_F_UPDATE |
 506                                     LDD_F_NEED_INDEX | LDD_F_WRITECONF |
 507                                     LDD_F_UPGRADE14);
 508                 mti->mti_flags |= LDD_F_REWRITE_LDD;
 509         }
 510
 511 out:
 512         /* done with log update */
 513         if (lockrc == ELDLM_OK)
 514                 mgs_put_cfg_lock(&lockh);
 515 out_nolock:
 516         CDEBUG(D_MGS, "replying with %s, index=%d, rc=%d\n", mti->mti_svname,
 517                mti->mti_stripe_index, rc);
 518         rc = req_capsule_server_pack(&req->rq_pill);
 519         if (rc)
 520                 RETURN(rc);
 521
 522         /* send back the whole mti in the reply */
 523         rep_mti = req_capsule_server_get(&req->rq_pill, &RMF_MGS_TARGET_INFO);
 524         *rep_mti = *mti;
 525
 526         /* Flush logs to disk */
 527         fsfilt_sync(obd, obd->u.mgs.mgs_sb);
 528         RETURN(rc);
 529 }
 530
 531 static int mgs_set_info_rpc(struct ptlrpc_request *req)
 532 {
 533         struct obd_device *obd = req->rq_export->exp_obd;
 534         struct mgs_send_param *msp, *rep_msp;
 535         struct lustre_handle lockh;
 536         int rc;
 537         struct lustre_cfg_bufs bufs;
 538         struct lustre_cfg *lcfg;
 539         char fsname[MTI_NAME_MAXLEN];
 540         ENTRY;
 541
 542         msp = req_capsule_client_get(&req->rq_pill, &RMF_MGS_SEND_PARAM);
 543         LASSERT(msp);
 544
 545         /* Construct lustre_cfg structure to pass to function mgs_setparam */
 546         lustre_cfg_bufs_reset(&bufs, NULL);
 547         lustre_cfg_bufs_set_string(&bufs, 1, msp->mgs_param);
 548         lcfg = lustre_cfg_new(LCFG_PARAM, &bufs);
 549         rc = mgs_setparam(obd, lcfg, fsname);
 550         if (rc) {
 551                 CERROR("Error %d in setting the parameter %s for fs %s\n",
 552                        rc, msp->mgs_param, fsname);
 553                 RETURN(rc);
 554         }
 555
 556         /* request for update */
 557         mgs_revoke_lock(obd, fsname, &lockh);
 558
 559         lustre_cfg_free(lcfg);
 560
 561         rc = req_capsule_server_pack(&req->rq_pill);
 562         if (rc == 0) {
 563                 rep_msp = req_capsule_server_get(&req->rq_pill, &RMF_MGS_SEND_PARAM);
 564                 rep_msp = msp;
 565         }
 566         RETURN(rc);
 567 }
 568
 569 /*
 570  * similar as in ost_connect_check_sptlrpc()
 571  */
 572 static int mgs_connect_check_sptlrpc(struct ptlrpc_request *req)
 573 {
 574         struct obd_export     *exp = req->rq_export;
 575         struct obd_device     *obd = exp->exp_obd;
 576         struct fs_db          *fsdb;
 577         struct sptlrpc_flavor  flvr;
 578         int                    rc = 0;
 579
 580         if (exp->exp_flvr.sf_rpc == SPTLRPC_FLVR_INVALID) {
 581                 rc = mgs_find_or_make_fsdb(obd, MGSSELF_NAME, &fsdb);
 582                 if (rc)
 583                         return rc;
 584
 585                 down(&fsdb->fsdb_sem);
 586                 if (sptlrpc_rule_set_choose(&fsdb->fsdb_srpc_gen,
 587                                             LUSTRE_SP_MGC, LUSTRE_SP_MGS,
 588                                             req->rq_peer.nid,
 589                                             &flvr) == 0) {
 590                         /* by defualt allow any flavors */
 591                         flvr.sf_rpc = SPTLRPC_FLVR_ANY;
 592                 }
 593                 up(&fsdb->fsdb_sem);
 594
 595                 spin_lock(&exp->exp_lock);
 596
 597                 exp->exp_sp_peer = req->rq_sp_from;
 598                 exp->exp_flvr = flvr;
 599
 600                 if (exp->exp_flvr.sf_rpc != SPTLRPC_FLVR_ANY &&
 601                     exp->exp_flvr.sf_rpc != req->rq_flvr.sf_rpc) {
 602                         CERROR("invalid rpc flavor %x, expect %x, from %s\n",
 603                                req->rq_flvr.sf_rpc, exp->exp_flvr.sf_rpc,
 604                                libcfs_nid2str(req->rq_peer.nid));
 605                         rc = -EACCES;
 606                 }
 607
 608                 spin_unlock(&exp->exp_lock);
 609         } else {
 610                 if (exp->exp_sp_peer != req->rq_sp_from) {
 611                         CERROR("RPC source %s doesn't match %s\n",
 612                                sptlrpc_part2name(req->rq_sp_from),
 613                                sptlrpc_part2name(exp->exp_sp_peer));
 614                         rc = -EACCES;
 615                 } else {
 616                         rc = sptlrpc_target_export_check(exp, req);
 617                 }
 618         }
 619
 620         return rc;
 621 }
 622
 623 /* Called whenever a target cleans up. */
 624 /* XXX - Currently unused */
 625 static int mgs_handle_target_del(struct ptlrpc_request *req)
 626 {
 627         ENTRY;
 628         mgs_counter_incr(req->rq_export, LPROC_MGS_TARGET_DEL);
 629         RETURN(0);
 630 }
 631
 632 /* XXX - Currently unused */
 633 static int mgs_handle_exception(struct ptlrpc_request *req)
 634 {
 635         ENTRY;
 636         mgs_counter_incr(req->rq_export, LPROC_MGS_EXCEPTION);
 637         RETURN(0);
 638 }
 639
 640 /* TODO: handle requests in a similar way as MDT: see mdt_handle_common() */
 641 int mgs_handle(struct ptlrpc_request *req)
 642 {
 643         int fail = OBD_FAIL_MGS_ALL_REPLY_NET;
 644         int opc, rc = 0;
 645         ENTRY;
 646
 647         req_capsule_init(&req->rq_pill, req, RCL_SERVER);
 648         OBD_FAIL_TIMEOUT_MS(OBD_FAIL_MGS_PAUSE_REQ, obd_fail_val);
 649         if (OBD_FAIL_CHECK(OBD_FAIL_MGS_ALL_REQUEST_NET))
 650                 RETURN(0);
 651
 652         LASSERT(current->journal_info == NULL);
 653         opc = lustre_msg_get_opc(req->rq_reqmsg);
 654
 655         if (opc == SEC_CTX_INIT ||
 656             opc == SEC_CTX_INIT_CONT ||
 657             opc == SEC_CTX_FINI)
 658                 GOTO(out, rc = 0);
 659
 660         if (opc != MGS_CONNECT) {
 661                 if (req->rq_export == NULL) {
 662                         CERROR("lustre_mgs: operation %d on unconnected MGS\n",
 663                                opc);
 664                         req->rq_status = -ENOTCONN;
 665                         GOTO(out, rc = -ENOTCONN);
 666                 }
 667         }
 668
 669         switch (opc) {
 670         case MGS_CONNECT:
 671                 DEBUG_REQ(D_MGS, req, "connect");
 672                 /* MGS and MDS have same request format for connect */
 673                 req_capsule_set(&req->rq_pill, &RQF_MDS_CONNECT);
 674                 rc = target_handle_connect(req);
 675                 if (rc == 0)
 676                         rc = mgs_connect_check_sptlrpc(req);
 677
 678                 if (!rc && (lustre_msg_get_conn_cnt(req->rq_reqmsg) > 1))
 679                         /* Make clients trying to reconnect after a MGS restart
 680                            happy; also requires obd_replayable */
 681                         lustre_msg_add_op_flags(req->rq_repmsg,
 682                                                 MSG_CONNECT_RECONNECT);
 683                 break;
 684         case MGS_DISCONNECT:
 685                 DEBUG_REQ(D_MGS, req, "disconnect");
 686                 /* MGS and MDS have same request format for disconnect */
 687                 req_capsule_set(&req->rq_pill, &RQF_MDS_DISCONNECT);
 688                 rc = target_handle_disconnect(req);
 689                 req->rq_status = rc;            /* superfluous? */
 690                 break;
 691         case MGS_EXCEPTION:
 692                 DEBUG_REQ(D_MGS, req, "exception");
 693                 rc = mgs_handle_exception(req);
 694                 break;
 695         case MGS_TARGET_REG:
 696                 DEBUG_REQ(D_MGS, req, "target add");
 697                 req_capsule_set(&req->rq_pill, &RQF_MGS_TARGET_REG);
 698                 rc = mgs_handle_target_reg(req);
 699                 break;
 700         case MGS_TARGET_DEL:
 701                 DEBUG_REQ(D_MGS, req, "target del");
 702                 rc = mgs_handle_target_del(req);
 703                 break;
 704         case MGS_SET_INFO:
 705                 DEBUG_REQ(D_MGS, req, "set_info");
 706                 req_capsule_set(&req->rq_pill, &RQF_MGS_SET_INFO);
 707                 rc = mgs_set_info_rpc(req);
 708                 break;
 709
 710         case LDLM_ENQUEUE:
 711                 DEBUG_REQ(D_MGS, req, "enqueue");
 712                 req_capsule_set(&req->rq_pill, &RQF_LDLM_ENQUEUE);
 713                 rc = ldlm_handle_enqueue(req, ldlm_server_completion_ast,
 714                                          ldlm_server_blocking_ast, NULL);
 715                 break;
 716         case LDLM_BL_CALLBACK:
 717         case LDLM_CP_CALLBACK:
 718                 DEBUG_REQ(D_MGS, req, "callback");
 719                 CERROR("callbacks should not happen on MGS\n");
 720                 LBUG();
 721                 break;
 722
 723         case OBD_PING:
 724                 DEBUG_REQ(D_INFO, req, "ping");
 725                 req_capsule_set(&req->rq_pill, &RQF_OBD_PING);
 726                 rc = target_handle_ping(req);
 727                 break;
 728         case OBD_LOG_CANCEL:
 729                 DEBUG_REQ(D_MGS, req, "log cancel");
 730                 rc = -ENOTSUPP; /* la la la */
 731                 break;
 732
 733         case LLOG_ORIGIN_HANDLE_CREATE:
 734                 DEBUG_REQ(D_MGS, req, "llog_init");
 735                 req_capsule_set(&req->rq_pill, &RQF_LLOG_ORIGIN_HANDLE_CREATE);
 736                 rc = llog_origin_handle_create(req);
 737                 break;
 738         case LLOG_ORIGIN_HANDLE_NEXT_BLOCK:
 739                 DEBUG_REQ(D_MGS, req, "llog next block");
 740                 req_capsule_set(&req->rq_pill,
 741                                 &RQF_LLOG_ORIGIN_HANDLE_NEXT_BLOCK);
 742                 rc = llog_origin_handle_next_block(req);
 743                 break;
 744         case LLOG_ORIGIN_HANDLE_READ_HEADER:
 745                 DEBUG_REQ(D_MGS, req, "llog read header");
 746                 req_capsule_set(&req->rq_pill,
 747                                 &RQF_LLOG_ORIGIN_HANDLE_READ_HEADER);
 748                 rc = llog_origin_handle_read_header(req);
 749                 break;
 750         case LLOG_ORIGIN_HANDLE_CLOSE:
 751                 DEBUG_REQ(D_MGS, req, "llog close");
 752                 rc = llog_origin_handle_close(req);
 753                 break;
 754         case LLOG_CATINFO:
 755                 DEBUG_REQ(D_MGS, req, "llog catinfo");
 756                 req_capsule_set(&req->rq_pill, &RQF_LLOG_CATINFO);
 757                 rc = llog_catinfo(req);
 758                 break;
 759         default:
 760                 req->rq_status = -ENOTSUPP;
 761                 rc = ptlrpc_error(req);
 762                 RETURN(rc);
 763         }
 764
 765         LASSERT(current->journal_info == NULL);
 766
 767         if (rc)
 768                 CERROR("MGS handle cmd=%d rc=%d\n", opc, rc);
 769
 770 out:
 771         target_send_reply(req, rc, fail);
 772         RETURN(0);
 773 }
 774
 775 static inline int mgs_init_export(struct obd_export *exp)
 776 {
 777         return ldlm_init_export(exp);
 778 }
 779
 780 static inline int mgs_destroy_export(struct obd_export *exp)
 781 {
 782         ENTRY;
 783
 784         target_destroy_export(exp);
 785         mgs_client_free(exp);
 786         ldlm_destroy_export(exp);
 787
 788         RETURN(0);
 789 }
 790
 791 static int mgs_extract_fs_pool(char * arg, char *fsname, char *poolname)
 792 {
 793         char *ptr;
 794
 795         ENTRY;
 796         for (ptr = arg;  (*ptr != '\0') && (*ptr != '.'); ptr++ ) {
 797                 *fsname = *ptr;
 798                 fsname++;
 799         }
 800         if (*ptr == '\0')
 801                 return -EINVAL;
 802         *fsname = '\0';
 803         ptr++;
 804         strcpy(poolname, ptr);
 805
 806         RETURN(0);
 807 }
 808
 809 static int mgs_iocontrol_pool(struct obd_device *obd,
 810                               struct obd_ioctl_data *data)
 811 {
 812         int rc;
 813         struct lustre_handle lockh;
 814         struct lustre_cfg *lcfg = NULL;
 815         struct llog_rec_hdr rec;
 816         char *fsname = NULL;
 817         char *poolname = NULL;
 818         ENTRY;
 819
 820         OBD_ALLOC(fsname, MTI_NAME_MAXLEN);
 821         if (fsname == NULL)
 822                 RETURN(-ENOMEM);
 823
 824         OBD_ALLOC(poolname, LOV_MAXPOOLNAME + 1);
 825         if (poolname == NULL) {
 826                 rc = -ENOMEM;
 827                 GOTO(out_pool, rc);
 828         }
 829         rec.lrh_len = llog_data_len(data->ioc_plen1);
 830
 831         if (data->ioc_type == LUSTRE_CFG_TYPE) {
 832                 rec.lrh_type = OBD_CFG_REC;
 833         } else {
 834                 CERROR("unknown cfg record type:%d \n", data->ioc_type);
 835                 rc = -EINVAL;
 836                 GOTO(out_pool, rc);
 837         }
 838
 839         if (data->ioc_plen1 > CFS_PAGE_SIZE) {
 840                 rc = -E2BIG;
 841                 GOTO(out_pool, rc);
 842         }
 843
 844         OBD_ALLOC(lcfg, data->ioc_plen1);
 845         if (lcfg == NULL)
 846                 GOTO(out_pool, rc = -ENOMEM);
 847
 848         if (copy_from_user(lcfg, data->ioc_pbuf1, data->ioc_plen1))
 849                 GOTO(out_pool, rc = -EFAULT);
 850
 851         if (lcfg->lcfg_bufcount < 2) {
 852                 GOTO(out_pool, rc = -EFAULT);
 853         }
 854
 855         /* first arg is always <fsname>.<poolname> */
 856         mgs_extract_fs_pool(lustre_cfg_string(lcfg, 1), fsname,
 857                             poolname);
 858
 859         switch (lcfg->lcfg_command) {
 860         case LCFG_POOL_NEW: {
 861                 if (lcfg->lcfg_bufcount != 2)
 862                         RETURN(-EINVAL);
 863                 rc = mgs_pool_cmd(obd, LCFG_POOL_NEW, fsname,
 864                                   poolname, NULL);
 865                 break;
 866         }
 867         case LCFG_POOL_ADD: {
 868                 if (lcfg->lcfg_bufcount != 3)
 869                         RETURN(-EINVAL);
 870                 rc = mgs_pool_cmd(obd, LCFG_POOL_ADD, fsname, poolname,
 871                                   lustre_cfg_string(lcfg, 2));
 872                 break;
 873         }
 874         case LCFG_POOL_REM: {
 875                 if (lcfg->lcfg_bufcount != 3)
 876                         RETURN(-EINVAL);
 877                 rc = mgs_pool_cmd(obd, LCFG_POOL_REM, fsname, poolname,
 878                                   lustre_cfg_string(lcfg, 2));
 879                 break;
 880         }
 881         case LCFG_POOL_DEL: {
 882                 if (lcfg->lcfg_bufcount != 2)
 883                         RETURN(-EINVAL);
 884                 rc = mgs_pool_cmd(obd, LCFG_POOL_DEL, fsname,
 885                                   poolname, NULL);
 886                 break;
 887         }
 888         default: {
 889                  rc = -EINVAL;
 890                  GOTO(out_pool, rc);
 891         }
 892         }
 893
 894         if (rc) {
 895                 CERROR("OBD_IOC_POOL err %d, cmd %X for pool %s.%s\n",
 896                        rc, lcfg->lcfg_command, fsname, poolname);
 897                 GOTO(out_pool, rc);
 898         }
 899
 900         /* request for update */
 901         mgs_revoke_lock(obd, fsname, &lockh);
 902
 903 out_pool:
 904         if (lcfg != NULL)
 905                 OBD_FREE(lcfg, data->ioc_plen1);
 906
 907         if (fsname != NULL)
 908                 OBD_FREE(fsname, MTI_NAME_MAXLEN);
 909
 910         if (poolname != NULL)
 911                 OBD_FREE(poolname, LOV_MAXPOOLNAME + 1);
 912
 913         RETURN(rc);
 914 }
 915
 916 /* from mdt_iocontrol */
 917 int mgs_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
 918                   void *karg, void *uarg)
 919 {
 920         struct obd_device *obd = exp->exp_obd;
 921         struct obd_ioctl_data *data = karg;
 922         struct lvfs_run_ctxt saved;
 923         int rc = 0;
 924
 925         ENTRY;
 926         CDEBUG(D_IOCTL, "handling ioctl cmd %#x\n", cmd);
 927
 928         switch (cmd) {
 929
 930         case OBD_IOC_PARAM: {
 931                 struct lustre_handle lockh;
 932                 struct lustre_cfg *lcfg;
 933                 struct llog_rec_hdr rec;
 934                 char fsname[MTI_NAME_MAXLEN];
 935
 936                 rec.lrh_len = llog_data_len(data->ioc_plen1);
 937
 938                 if (data->ioc_type == LUSTRE_CFG_TYPE) {
 939                         rec.lrh_type = OBD_CFG_REC;
 940                 } else {
 941                         CERROR("unknown cfg record type:%d \n", data->ioc_type);
 942                         RETURN(-EINVAL);
 943                 }
 944
 945                 OBD_ALLOC(lcfg, data->ioc_plen1);
 946                 if (lcfg == NULL)
 947                         RETURN(-ENOMEM);
 948                 if (copy_from_user(lcfg, data->ioc_pbuf1, data->ioc_plen1))
 949                         GOTO(out_free, rc = -EFAULT);
 950
 951                 if (lcfg->lcfg_bufcount < 1)
 952                         GOTO(out_free, rc = -EINVAL);
 953
 954                 rc = mgs_setparam(obd, lcfg, fsname);
 955                 if (rc) {
 956                         CERROR("setparam err %d\n", rc);
 957                         GOTO(out_free, rc);
 958                 }
 959
 960                 /* Revoke lock so everyone updates.  Should be alright if
 961                    someone was already reading while we were updating the logs,
 962                    so we don't really need to hold the lock while we're
 963                    writing (above). */
 964                 mgs_revoke_lock(obd, fsname, &lockh);
 965
 966 out_free:
 967                 OBD_FREE(lcfg, data->ioc_plen1);
 968                 RETURN(rc);
 969         }
 970
 971         case OBD_IOC_POOL: {
 972                 RETURN(mgs_iocontrol_pool(obd, data));
 973         }
 974
 975         case OBD_IOC_DUMP_LOG: {
 976                 struct llog_ctxt *ctxt;
 977                 ctxt = llog_get_context(obd, LLOG_CONFIG_ORIG_CTXT);
 978                 push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 979                 rc = class_config_dump_llog(ctxt, data->ioc_inlbuf1, NULL);
 980                 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 981                 llog_ctxt_put(ctxt);
 982
 983                 RETURN(rc);
 984         }
 985
 986         case OBD_IOC_LLOG_CHECK:
 987         case OBD_IOC_LLOG_INFO:
 988         case OBD_IOC_LLOG_PRINT: {
 989                 struct llog_ctxt *ctxt;
 990                 ctxt = llog_get_context(obd, LLOG_CONFIG_ORIG_CTXT);
 991
 992                 push_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
 993                 rc = llog_ioctl(ctxt, cmd, data);
 994                 pop_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
 995                 llog_ctxt_put(ctxt);
 996
 997                 RETURN(rc);
 998         }
 999
1000         default:
1001                 CDEBUG(D_INFO, "unknown command %x\n", cmd);
1002                 RETURN(-EINVAL);
1003         }
1004         RETURN(0);
1005 }
1006
1007 /* use obd ops to offer management infrastructure */
1008 static struct obd_ops mgs_obd_ops = {
1009         .o_owner           = THIS_MODULE,
1010         .o_connect         = mgs_connect,
1011         .o_reconnect       = mgs_reconnect,
1012         .o_disconnect      = mgs_disconnect,
1013         .o_setup           = mgs_setup,
1014         .o_precleanup      = mgs_precleanup,
1015         .o_cleanup         = mgs_cleanup,
1016         .o_init_export     = mgs_init_export,
1017         .o_destroy_export  = mgs_destroy_export,
1018         .o_iocontrol       = mgs_iocontrol,
1019         .o_llog_init       = mgs_llog_init,
1020         .o_llog_finish     = mgs_llog_finish
1021 };
1022
1023 static int __init mgs_init(void)
1024 {
1025         struct lprocfs_static_vars lvars;
1026
1027         lprocfs_mgs_init_vars(&lvars);
1028         class_register_type(&mgs_obd_ops, NULL,
1029                             lvars.module_vars, LUSTRE_MGS_NAME, NULL);
1030
1031         return 0;
1032 }
1033
1034 static void /*__exit*/ mgs_exit(void)
1035 {
1036         class_unregister_type(LUSTRE_MGS_NAME);
1037 }
1038
1039 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
1040 MODULE_DESCRIPTION("Lustre  Management Server (MGS)");
1041 MODULE_LICENSE("GPL");
1042
1043 module_init(mgs_init);
1044 module_exit(mgs_exit);