ceph/src/rgw/librgw.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab ft=cpp
   3
   4 /*
   5  * Ceph - scalable distributed file system
   6  *
   7  * Copyright (C) 2011 New Dream Network
   8  *
   9  * This is free software; you can redistribute it and/or
  10  * modify it under the terms of the GNU Lesser General Public
  11  * License version 2.1, as published by the Free Software
  12  * Foundation.  See file COPYING.
  13  *
  14  */
  15
  16 #include "include/compat.h"
  17 #include <sys/types.h>
  18 #include <string.h>
  19 #include <chrono>
  20
  21 #include "include/types.h"
  22 #include "include/rados/librgw.h"
  23 #include "rgw/rgw_acl_s3.h"
  24 #include "rgw_acl.h"
  25
  26 #include "include/str_list.h"
  27 #include "include/stringify.h"
  28 #include "global/global_init.h"
  29 #include "global/signal_handler.h"
  30 #include "common/config.h"
  31 #include "common/errno.h"
  32 #include "common/Timer.h"
  33 #include "common/Throttle.h"
  34 #include "common/WorkQueue.h"
  35 #include "common/ceph_argparse.h"
  36 #include "common/ceph_context.h"
  37 #include "common/common_init.h"
  38 #include "common/dout.h"
  39
  40 #include "rgw_rados.h"
  41 #include "rgw_resolve.h"
  42 #include "rgw_op.h"
  43 #include "rgw_rest.h"
  44 #include "rgw_frontend.h"
  45 #include "rgw_request.h"
  46 #include "rgw_process.h"
  47 #include "rgw_rest_user.h"
  48 #include "rgw_rest_s3.h"
  49 #include "rgw_os_lib.h"
  50 #include "rgw_auth.h"
  51 #include "rgw_auth_s3.h"
  52 #include "rgw_lib.h"
  53 #include "rgw_lib_frontend.h"
  54 #include "rgw_http_client.h"
  55 #include "rgw_http_client_curl.h"
  56 #include "rgw_perf_counters.h"
  57
  58 #include "services/svc_zone.h"
  59
  60 #include <errno.h>
  61 #include <thread>
  62 #include <string>
  63 #include <mutex>
  64
  65
  66 #define dout_subsys ceph_subsys_rgw
  67
  68 bool global_stop = false;
  69
  70 static void handle_sigterm(int signum)
  71 {
  72   dout(20) << __func__ << " SIGUSR1 ignored" << dendl;
  73 }
  74
  75 namespace rgw {
  76
  77   using std::string;
  78
  79   static std::mutex librgw_mtx;
  80
  81   RGWLib rgwlib;
  82
  83   class C_InitTimeout : public Context {
  84   public:
  85     C_InitTimeout() {}
  86     void finish(int r) override {
  87       derr << "Initialization timeout, failed to initialize" << dendl;
  88       exit(1);
  89     }
  90   };
  91
  92   void RGWLibProcess::checkpoint()
  93   {
  94     m_tp.drain(&req_wq);
  95   }
  96
  97 #define MIN_EXPIRE_S 120
  98
  99   void RGWLibProcess::run()
 100   {
 101     /* write completion interval */
 102     RGWLibFS::write_completion_interval_s =
 103       cct->_conf->rgw_nfs_write_completion_interval_s;
 104
 105     /* start write timer */
 106     RGWLibFS::write_timer.resume();
 107
 108     /* gc loop */
 109     while (! shutdown) {
 110       lsubdout(cct, rgw, 5) << "RGWLibProcess GC" << dendl;
 111
 112       /* dirent invalidate timeout--basically, the upper-bound on
 113        * inconsistency with the S3 namespace */
 114       auto expire_s = cct->_conf->rgw_nfs_namespace_expire_secs;
 115
 116       /* delay between gc cycles */
 117       auto delay_s = std::max(int64_t(1), std::min(int64_t(MIN_EXPIRE_S), expire_s/2));
 118
 119       unique_lock uniq(mtx);
 120     restart:
 121       int cur_gen = gen;
 122       for (auto iter = mounted_fs.begin(); iter != mounted_fs.end();
 123            ++iter) {
 124         RGWLibFS* fs = iter->first->ref();
 125         uniq.unlock();
 126         fs->gc();
 127         fs->update_user();
 128         fs->rele();
 129         uniq.lock();
 130         if (cur_gen != gen)
 131           goto restart; /* invalidated */
 132       }
 133       uniq.unlock();
 134       std::this_thread::sleep_for(std::chrono::seconds(delay_s));
 135     }
 136   }
 137
 138   void RGWLibProcess::handle_request(RGWRequest* r)
 139   {
 140     /*
 141      * invariant: valid requests are derived from RGWLibRequst
 142      */
 143     RGWLibRequest* req = static_cast<RGWLibRequest*>(r);
 144
 145     // XXX move RGWLibIO and timing setup into process_request
 146
 147 #if 0 /* XXX */
 148     utime_t tm = ceph_clock_now();
 149 #endif
 150
 151     RGWLibIO io_ctx;
 152
 153     int ret = process_request(req, &io_ctx);
 154     if (ret < 0) {
 155       /* we don't really care about return code */
 156       dout(20) << "process_request() returned " << ret << dendl;
 157
 158     }
 159     delete req;
 160   } /* handle_request */
 161
 162   int RGWLibProcess::process_request(RGWLibRequest* req)
 163   {
 164     // XXX move RGWLibIO and timing setup into process_request
 165
 166 #if 0 /* XXX */
 167     utime_t tm = ceph_clock_now();
 168 #endif
 169
 170     RGWLibIO io_ctx;
 171
 172     int ret = process_request(req, &io_ctx);
 173     if (ret < 0) {
 174       /* we don't really care about return code */
 175       dout(20) << "process_request() returned " << ret << dendl;
 176     }
 177     return ret;
 178   } /* process_request */
 179
 180   static inline void abort_req(struct req_state *s, RGWOp *op, int err_no)
 181   {
 182     if (!s)
 183       return;
 184
 185     /* XXX the dump_errno and dump_bucket_from_state behaviors in
 186      * the abort_early (rgw_rest.cc) might be valuable, but aren't
 187      * safe to call presently as they return HTTP data */
 188
 189     perfcounter->inc(l_rgw_failed_req);
 190   } /* abort_req */
 191
 192   int RGWLibProcess::process_request(RGWLibRequest* req, RGWLibIO* io)
 193   {
 194     int ret = 0;
 195     bool should_log = true; // XXX
 196
 197     dout(1) << "====== " << __func__
 198             << " starting new request req=" << hex << req << dec
 199             << " ======" << dendl;
 200
 201     /*
 202      * invariant: valid requests are derived from RGWOp--well-formed
 203      * requests should have assigned RGWRequest::op in their descendant
 204      * constructor--if not, the compiler can find it, at the cost of
 205      * a runtime check
 206      */
 207     RGWOp *op = (req->op) ? req->op : dynamic_cast<RGWOp*>(req);
 208     if (! op) {
 209       dout(1) << "failed to derive cognate RGWOp (invalid op?)" << dendl;
 210       return -EINVAL;
 211     }
 212
 213     io->init(req->cct);
 214
 215     perfcounter->inc(l_rgw_req);
 216
 217     RGWEnv& rgw_env = io->get_env();
 218
 219     /* XXX
 220      * until major refactoring of req_state and req_info, we need
 221      * to build their RGWEnv boilerplate from the RGWLibRequest,
 222      * pre-staging any strings (HTTP_HOST) that provoke a crash when
 223      * not found
 224      */
 225
 226     /* XXX for now, use "";  could be a legit hostname, or, in future,
 227      * perhaps a tenant (Yehuda) */
 228     rgw_env.set("HTTP_HOST", "");
 229
 230     /* XXX and -then- bloat up req_state with string copies from it */
 231     struct req_state rstate(req->cct, &rgw_env, req->get_user(), req->id);
 232     struct req_state *s = &rstate;
 233
 234     // XXX fix this
 235     s->cio = io;
 236
 237     RGWObjectCtx rados_ctx(store, s); // XXX holds std::map
 238
 239     auto sysobj_ctx = store->svc()->sysobj->init_obj_ctx();
 240     s->sysobj_ctx = &sysobj_ctx;
 241
 242     /* XXX and -then- stash req_state pointers everywhere they are needed */
 243     ret = req->init(rgw_env, &rados_ctx, io, s);
 244     if (ret < 0) {
 245       dout(10) << "failed to initialize request" << dendl;
 246       abort_req(s, op, ret);
 247       goto done;
 248     }
 249
 250     /* req is-a RGWOp, currently initialized separately */
 251     ret = req->op_init();
 252     if (ret < 0) {
 253       dout(10) << "failed to initialize RGWOp" << dendl;
 254       abort_req(s, op, ret);
 255       goto done;
 256     }
 257
 258     /* now expected by rgw_log_op() */
 259     rgw_env.set("REQUEST_METHOD", s->info.method);
 260     rgw_env.set("REQUEST_URI", s->info.request_uri);
 261     rgw_env.set("QUERY_STRING", "");
 262
 263     try {
 264       /* XXX authorize does less here then in the REST path, e.g.,
 265        * the user's info is cached, but still incomplete */
 266       ldpp_dout(s, 2) << "authorizing" << dendl;
 267       ret = req->authorize(op);
 268       if (ret < 0) {
 269         dout(10) << "failed to authorize request" << dendl;
 270         abort_req(s, op, ret);
 271         goto done;
 272       }
 273
 274       /* FIXME: remove this after switching all handlers to the new
 275        * authentication infrastructure. */
 276       if (! s->auth.identity) {
 277         s->auth.identity = rgw::auth::transform_old_authinfo(s);
 278       }
 279
 280       ldpp_dout(s, 2) << "reading op permissions" << dendl;
 281       ret = req->read_permissions(op);
 282       if (ret < 0) {
 283         abort_req(s, op, ret);
 284         goto done;
 285       }
 286
 287       ldpp_dout(s, 2) << "init op" << dendl;
 288       ret = op->init_processing();
 289       if (ret < 0) {
 290         abort_req(s, op, ret);
 291         goto done;
 292       }
 293
 294       ldpp_dout(s, 2) << "verifying op mask" << dendl;
 295       ret = op->verify_op_mask();
 296       if (ret < 0) {
 297         abort_req(s, op, ret);
 298         goto done;
 299       }
 300
 301       ldpp_dout(s, 2) << "verifying op permissions" << dendl;
 302       ret = op->verify_permission();
 303       if (ret < 0) {
 304         if (s->system_request) {
 305           dout(2) << "overriding permissions due to system operation" << dendl;
 306         } else if (s->auth.identity->is_admin_of(s->user->get_id())) {
 307           dout(2) << "overriding permissions due to admin operation" << dendl;
 308         } else {
 309           abort_req(s, op, ret);
 310           goto done;
 311         }
 312       }
 313
 314       ldpp_dout(s, 2) << "verifying op params" << dendl;
 315       ret = op->verify_params();
 316       if (ret < 0) {
 317         abort_req(s, op, ret);
 318         goto done;
 319       }
 320
 321       ldpp_dout(s, 2) << "executing" << dendl;
 322       op->pre_exec();
 323       op->execute();
 324       op->complete();
 325
 326     } catch (const ceph::crypto::DigestException& e) {
 327       dout(0) << "authentication failed" << e.what() << dendl;
 328       abort_req(s, op, -ERR_INVALID_SECRET_KEY);
 329     }
 330
 331   done:
 332     try {
 333       io->complete_request();
 334     } catch (rgw::io::Exception& e) {
 335       dout(0) << "ERROR: io->complete_request() returned "
 336               << e.what() << dendl;
 337     }
 338     if (should_log) {
 339       rgw_log_op(store->getRados(), nullptr /* !rest */, s,
 340                  (op ? op->name() : "unknown"), olog);
 341     }
 342
 343     int http_ret = s->err.http_ret;
 344
 345     ldpp_dout(s, 2) << "http status=" << http_ret << dendl;
 346
 347     dout(1) << "====== " << __func__
 348             << " req done req=" << hex << req << dec << " http_status="
 349             << http_ret
 350             << " ======" << dendl;
 351
 352     return (ret < 0 ? ret : s->err.ret);
 353   } /* process_request */
 354
 355   int RGWLibProcess::start_request(RGWLibContinuedReq* req)
 356   {
 357
 358     dout(1) << "====== " << __func__
 359             << " starting new continued request req=" << hex << req << dec
 360             << " ======" << dendl;
 361
 362     /*
 363      * invariant: valid requests are derived from RGWOp--well-formed
 364      * requests should have assigned RGWRequest::op in their descendant
 365      * constructor--if not, the compiler can find it, at the cost of
 366      * a runtime check
 367      */
 368     RGWOp *op = (req->op) ? req->op : dynamic_cast<RGWOp*>(req);
 369     if (! op) {
 370       dout(1) << "failed to derive cognate RGWOp (invalid op?)" << dendl;
 371       return -EINVAL;
 372     }
 373
 374     struct req_state* s = req->get_state();
 375
 376     /* req is-a RGWOp, currently initialized separately */
 377     int ret = req->op_init();
 378     if (ret < 0) {
 379       dout(10) << "failed to initialize RGWOp" << dendl;
 380       abort_req(s, op, ret);
 381       goto done;
 382     }
 383
 384     /* XXX authorize does less here then in the REST path, e.g.,
 385      * the user's info is cached, but still incomplete */
 386     ldpp_dout(s, 2) << "authorizing" << dendl;
 387     ret = req->authorize(op);
 388     if (ret < 0) {
 389       dout(10) << "failed to authorize request" << dendl;
 390       abort_req(s, op, ret);
 391       goto done;
 392     }
 393
 394     /* FIXME: remove this after switching all handlers to the new authentication
 395      * infrastructure. */
 396     if (! s->auth.identity) {
 397       s->auth.identity = rgw::auth::transform_old_authinfo(s);
 398     }
 399
 400     ldpp_dout(s, 2) << "reading op permissions" << dendl;
 401     ret = req->read_permissions(op);
 402     if (ret < 0) {
 403       abort_req(s, op, ret);
 404       goto done;
 405     }
 406
 407     ldpp_dout(s, 2) << "init op" << dendl;
 408     ret = op->init_processing();
 409     if (ret < 0) {
 410       abort_req(s, op, ret);
 411       goto done;
 412     }
 413
 414     ldpp_dout(s, 2) << "verifying op mask" << dendl;
 415     ret = op->verify_op_mask();
 416     if (ret < 0) {
 417       abort_req(s, op, ret);
 418       goto done;
 419     }
 420
 421     ldpp_dout(s, 2) << "verifying op permissions" << dendl;
 422     ret = op->verify_permission();
 423     if (ret < 0) {
 424       if (s->system_request) {
 425         dout(2) << "overriding permissions due to system operation" << dendl;
 426       } else if (s->auth.identity->is_admin_of(s->user->get_id())) {
 427         dout(2) << "overriding permissions due to admin operation" << dendl;
 428       } else {
 429         abort_req(s, op, ret);
 430         goto done;
 431       }
 432     }
 433
 434     ldpp_dout(s, 2) << "verifying op params" << dendl;
 435     ret = op->verify_params();
 436     if (ret < 0) {
 437       abort_req(s, op, ret);
 438       goto done;
 439     }
 440
 441     op->pre_exec();
 442     req->exec_start();
 443
 444   done:
 445     return (ret < 0 ? ret : s->err.ret);
 446   }
 447
 448   int RGWLibProcess::finish_request(RGWLibContinuedReq* req)
 449   {
 450     RGWOp *op = (req->op) ? req->op : dynamic_cast<RGWOp*>(req);
 451     if (! op) {
 452       dout(1) << "failed to derive cognate RGWOp (invalid op?)" << dendl;
 453       return -EINVAL;
 454     }
 455
 456     int ret = req->exec_finish();
 457     int op_ret = op->get_ret();
 458
 459     dout(1) << "====== " << __func__
 460             << " finishing continued request req=" << hex << req << dec
 461             << " op status=" << op_ret
 462             << " ======" << dendl;
 463
 464     return ret;
 465   }
 466
 467   int RGWLibFrontend::init()
 468   {
 469     pprocess = new RGWLibProcess(g_ceph_context, &env,
 470                                  g_conf()->rgw_thread_pool_size, conf);
 471     return 0;
 472   }
 473
 474   int RGWLib::init()
 475   {
 476     vector<const char*> args;
 477     return init(args);
 478   }
 479
 480   int RGWLib::init(vector<const char*>& args)
 481   {
 482     int r = 0;
 483
 484     /* alternative default for module */
 485     map<string,string> defaults = {
 486       { "debug_rgw", "1/5" },
 487       { "keyring", "$rgw_data/keyring" },
 488       { "log_file", "/var/log/radosgw/$cluster-$name.log" }
 489     };
 490
 491     cct = global_init(&defaults, args,
 492                       CEPH_ENTITY_TYPE_CLIENT,
 493                       CODE_ENVIRONMENT_DAEMON,
 494                       CINIT_FLAG_UNPRIVILEGED_DAEMON_DEFAULTS);
 495
 496     ceph::mutex mutex = ceph::make_mutex("main");
 497     SafeTimer init_timer(g_ceph_context, mutex);
 498     init_timer.init();
 499     mutex.lock();
 500     init_timer.add_event_after(g_conf()->rgw_init_timeout, new C_InitTimeout);
 501     mutex.unlock();
 502
 503     common_init_finish(g_ceph_context);
 504
 505     rgw_tools_init(g_ceph_context);
 506
 507     rgw_init_resolver();
 508     rgw::curl::setup_curl(boost::none);
 509     rgw_http_client_init(g_ceph_context);
 510
 511     auto run_gc =
 512       g_conf()->rgw_enable_gc_threads &&
 513       g_conf()->rgw_nfs_run_gc_threads;
 514
 515     auto run_lc =
 516       g_conf()->rgw_enable_lc_threads &&
 517       g_conf()->rgw_nfs_run_lc_threads;
 518
 519     auto run_quota =
 520       g_conf()->rgw_enable_quota_threads &&
 521       g_conf()->rgw_nfs_run_quota_threads;
 522
 523     auto run_sync =
 524       g_conf()->rgw_run_sync_thread &&
 525       g_conf()->rgw_nfs_run_sync_thread;
 526
 527     store = RGWStoreManager::get_storage(g_ceph_context,
 528                                          run_gc,
 529                                          run_lc,
 530                                          run_quota,
 531                                          run_sync,
 532                                          g_conf().get_val<bool>("rgw_dynamic_resharding"));
 533
 534     if (!store) {
 535       mutex.lock();
 536       init_timer.cancel_all_events();
 537       init_timer.shutdown();
 538       mutex.unlock();
 539
 540       derr << "Couldn't init storage provider (RADOS)" << dendl;
 541       return -EIO;
 542     }
 543
 544     r = rgw_perf_start(g_ceph_context);
 545
 546     rgw_rest_init(g_ceph_context, store->svc()->zone->get_zonegroup());
 547
 548     mutex.lock();
 549     init_timer.cancel_all_events();
 550     init_timer.shutdown();
 551     mutex.unlock();
 552
 553     if (r)
 554       return -EIO;
 555
 556     const string& ldap_uri = store->ctx()->_conf->rgw_ldap_uri;
 557     const string& ldap_binddn = store->ctx()->_conf->rgw_ldap_binddn;
 558     const string& ldap_searchdn = store->ctx()->_conf->rgw_ldap_searchdn;
 559     const string& ldap_searchfilter = store->ctx()->_conf->rgw_ldap_searchfilter;
 560     const string& ldap_dnattr =
 561       store->ctx()->_conf->rgw_ldap_dnattr;
 562     std::string ldap_bindpw = parse_rgw_ldap_bindpw(store->ctx());
 563
 564     ldh = new rgw::LDAPHelper(ldap_uri, ldap_binddn, ldap_bindpw.c_str(),
 565                               ldap_searchdn, ldap_searchfilter, ldap_dnattr);
 566     ldh->init();
 567     ldh->bind();
 568
 569     rgw_log_usage_init(g_ceph_context, store->getRados());
 570
 571     // XXX ex-RGWRESTMgr_lib, mgr->set_logging(true)
 572
 573     if (!g_conf()->rgw_ops_log_socket_path.empty()) {
 574       olog = new OpsLogSocket(g_ceph_context, g_conf()->rgw_ops_log_data_backlog);
 575       olog->init(g_conf()->rgw_ops_log_socket_path);
 576     }
 577
 578     int port = 80;
 579     RGWProcessEnv env = { store, &rest, olog, port };
 580
 581     string fe_count{"0"};
 582     fec = new RGWFrontendConfig("rgwlib");
 583     fe = new RGWLibFrontend(env, fec);
 584
 585     init_async_signal_handler();
 586     register_async_signal_handler(SIGUSR1, handle_sigterm);
 587
 588     map<string, string> service_map_meta;
 589     service_map_meta["pid"] = stringify(getpid());
 590     service_map_meta["frontend_type#" + fe_count] = "rgw-nfs";
 591     service_map_meta["frontend_config#" + fe_count] = fec->get_config();
 592
 593     fe->init();
 594     if (r < 0) {
 595       derr << "ERROR: failed initializing frontend" << dendl;
 596       return r;
 597     }
 598
 599     fe->run();
 600
 601     r = store->getRados()->register_to_service_map("rgw-nfs", service_map_meta);
 602     if (r < 0) {
 603       derr << "ERROR: failed to register to service map: " << cpp_strerror(-r) << dendl;
 604       /* ignore error */
 605     }
 606
 607     return 0;
 608   } /* RGWLib::init() */
 609
 610   int RGWLib::stop()
 611   {
 612     derr << "shutting down" << dendl;
 613
 614     fe->stop();
 615
 616     fe->join();
 617
 618     delete fe;
 619     delete fec;
 620     delete ldh;
 621
 622     unregister_async_signal_handler(SIGUSR1, handle_sigterm);
 623     shutdown_async_signal_handler();
 624
 625     rgw_log_usage_finalize();
 626
 627     delete olog;
 628
 629     RGWStoreManager::close_storage(store);
 630
 631     rgw_tools_cleanup();
 632     rgw_shutdown_resolver();
 633     rgw_http_client_cleanup();
 634     rgw::curl::cleanup_curl();
 635
 636     rgw_perf_stop(g_ceph_context);
 637
 638     dout(1) << "final shutdown" << dendl;
 639     cct.reset();
 640
 641     return 0;
 642   } /* RGWLib::stop() */
 643
 644   int RGWLibIO::set_uid(rgw::sal::RGWRadosStore *store, const rgw_user& uid)
 645   {
 646     int ret = store->ctl()->user->get_info_by_uid(uid, &user_info, null_yield);
 647     if (ret < 0) {
 648       derr << "ERROR: failed reading user info: uid=" << uid << " ret="
 649            << ret << dendl;
 650     }
 651     return ret;
 652   }
 653
 654   int RGWLibRequest::read_permissions(RGWOp* op) {
 655     /* bucket and object ops */
 656     int ret =
 657       rgw_build_bucket_policies(rgwlib.get_store(), get_state());
 658     if (ret < 0) {
 659       ldout(get_state()->cct, 10) << "read_permissions (bucket policy) on "
 660                                   << get_state()->bucket << ":"
 661                                   << get_state()->object
 662                                   << " only_bucket=" << only_bucket()
 663                                   << " ret=" << ret << dendl;
 664       if (ret == -ENODATA)
 665         ret = -EACCES;
 666     } else if (! only_bucket()) {
 667       /* object ops */
 668       ret = rgw_build_object_policies(rgwlib.get_store(), get_state(),
 669                                       op->prefetch_data());
 670       if (ret < 0) {
 671         ldout(get_state()->cct, 10) << "read_permissions (object policy) on"
 672                                     << get_state()->bucket << ":"
 673                                     << get_state()->object
 674                                     << " ret=" << ret << dendl;
 675         if (ret == -ENODATA)
 676           ret = -EACCES;
 677       }
 678     }
 679     return ret;
 680   } /* RGWLibRequest::read_permissions */
 681
 682   int RGWHandler_Lib::authorize(const DoutPrefixProvider *dpp)
 683   {
 684     /* TODO: handle
 685      *  1. subusers
 686      *  2. anonymous access
 687      *  3. system access
 688      *  4. ?
 689      *
 690      *  Much or all of this depends on handling the cached authorization
 691      *  correctly (e.g., dealing with keystone) at mount time.
 692      */
 693     s->perm_mask = RGW_PERM_FULL_CONTROL;
 694
 695     // populate the owner info
 696     s->owner.set_id(s->user->get_id());
 697     s->owner.set_name(s->user->get_display_name());
 698
 699     return 0;
 700   } /* RGWHandler_Lib::authorize */
 701
 702 } /* namespace rgw */
 703
 704 extern "C" {
 705
 706 int librgw_create(librgw_t* rgw, int argc, char **argv)
 707 {
 708   using namespace rgw;
 709
 710   int rc = -EINVAL;
 711
 712   if (! g_ceph_context) {
 713     std::lock_guard<std::mutex> lg(librgw_mtx);
 714     if (! g_ceph_context) {
 715       vector<const char*> args;
 716       std::vector<std::string> spl_args;
 717       // last non-0 argument will be split and consumed
 718       if (argc > 1) {
 719         const std::string spl_arg{argv[(--argc)]};
 720         get_str_vec(spl_arg, " \t", spl_args);
 721       }
 722       argv_to_vec(argc, const_cast<const char**>(argv), args);
 723       // append split args, if any
 724       for (const auto& elt : spl_args) {
 725         args.push_back(elt.c_str());
 726       }
 727       rc = rgwlib.init(args);
 728     }
 729   }
 730
 731   *rgw = g_ceph_context->get();
 732
 733   return rc;
 734 }
 735
 736 void librgw_shutdown(librgw_t rgw)
 737 {
 738   using namespace rgw;
 739
 740   CephContext* cct = static_cast<CephContext*>(rgw);
 741   rgwlib.stop();
 742   cct->put();
 743 }
 744
 745 } /* extern "C" */