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