1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
2 // vim: ts=8 sw=2 smarttab ft=cpp
5 * Ceph - scalable distributed file system
7 * Copyright (C) 2011 New Dream Network
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.
16 #include "include/compat.h"
17 #include <sys/types.h>
21 #include "include/types.h"
22 #include "include/rados/librgw.h"
23 #include "rgw/rgw_acl_s3.h"
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"
40 #include "rgw_resolve.h"
43 #include "rgw_frontend.h"
44 #include "rgw_request.h"
45 #include "rgw_process.h"
46 #include "rgw_rest_user.h"
47 #include "rgw_rest_s3.h"
48 #include "rgw_os_lib.h"
50 #include "rgw_auth_s3.h"
52 #include "rgw_lib_frontend.h"
53 #include "rgw_http_client.h"
54 #include "rgw_http_client_curl.h"
55 #include "rgw_perf_counters.h"
56 #ifdef WITH_RADOSGW_AMQP_ENDPOINT
59 #ifdef WITH_RADOSGW_KAFKA_ENDPOINT
60 #include "rgw_kafka.h"
63 #include "services/svc_zone.h"
70 #define dout_subsys ceph_subsys_rgw
72 bool global_stop
= false;
74 static void handle_sigterm(int signum
)
76 dout(20) << __func__
<< " SIGUSR1 ignored" << dendl
;
83 static std::mutex librgw_mtx
;
87 class C_InitTimeout
: public Context
{
90 void finish(int r
) override
{
91 derr
<< "Initialization timeout, failed to initialize" << dendl
;
96 void RGWLibProcess::checkpoint()
101 #define MIN_EXPIRE_S 120
103 void RGWLibProcess::run()
105 /* write completion interval */
106 RGWLibFS::write_completion_interval_s
=
107 cct
->_conf
->rgw_nfs_write_completion_interval_s
;
109 /* start write timer */
110 RGWLibFS::write_timer
.resume();
114 lsubdout(cct
, rgw
, 5) << "RGWLibProcess GC" << dendl
;
116 /* dirent invalidate timeout--basically, the upper-bound on
117 * inconsistency with the S3 namespace */
118 auto expire_s
= cct
->_conf
->rgw_nfs_namespace_expire_secs
;
120 /* delay between gc cycles */
121 auto delay_s
= std::max(int64_t(1), std::min(int64_t(MIN_EXPIRE_S
), expire_s
/2));
123 unique_lock
uniq(mtx
);
126 for (auto iter
= mounted_fs
.begin(); iter
!= mounted_fs
.end();
128 RGWLibFS
* fs
= iter
->first
->ref();
131 const DoutPrefix
dp(cct
, dout_subsys
, "librgw: ");
132 fs
->update_user(&dp
);
136 goto restart
; /* invalidated */
138 cv
.wait_for(uniq
, std::chrono::seconds(delay_s
));
143 void RGWLibProcess::handle_request(const DoutPrefixProvider
*dpp
, RGWRequest
* r
)
146 * invariant: valid requests are derived from RGWLibRequst
148 RGWLibRequest
* req
= static_cast<RGWLibRequest
*>(r
);
150 // XXX move RGWLibIO and timing setup into process_request
153 utime_t tm
= ceph_clock_now();
158 int ret
= process_request(req
, &io_ctx
);
160 /* we don't really care about return code */
161 dout(20) << "process_request() returned " << ret
<< dendl
;
165 } /* handle_request */
167 int RGWLibProcess::process_request(RGWLibRequest
* req
)
169 // XXX move RGWLibIO and timing setup into process_request
172 utime_t tm
= ceph_clock_now();
177 int ret
= process_request(req
, &io_ctx
);
179 /* we don't really care about return code */
180 dout(20) << "process_request() returned " << ret
<< dendl
;
183 } /* process_request */
185 static inline void abort_req(struct req_state
*s
, RGWOp
*op
, int err_no
)
190 /* XXX the dump_errno and dump_bucket_from_state behaviors in
191 * the abort_early (rgw_rest.cc) might be valuable, but aren't
192 * safe to call presently as they return HTTP data */
194 perfcounter
->inc(l_rgw_failed_req
);
197 int RGWLibProcess::process_request(RGWLibRequest
* req
, RGWLibIO
* io
)
200 bool should_log
= true; // XXX
202 dout(1) << "====== " << __func__
203 << " starting new request req=" << hex
<< req
<< dec
204 << " ======" << dendl
;
207 * invariant: valid requests are derived from RGWOp--well-formed
208 * requests should have assigned RGWRequest::op in their descendant
209 * constructor--if not, the compiler can find it, at the cost of
212 RGWOp
*op
= (req
->op
) ? req
->op
: dynamic_cast<RGWOp
*>(req
);
214 ldpp_dout(op
, 1) << "failed to derive cognate RGWOp (invalid op?)" << dendl
;
220 perfcounter
->inc(l_rgw_req
);
222 RGWEnv
& rgw_env
= io
->get_env();
225 * until major refactoring of req_state and req_info, we need
226 * to build their RGWEnv boilerplate from the RGWLibRequest,
227 * pre-staging any strings (HTTP_HOST) that provoke a crash when
231 /* XXX for now, use ""; could be a legit hostname, or, in future,
232 * perhaps a tenant (Yehuda) */
233 rgw_env
.set("HTTP_HOST", "");
235 /* XXX and -then- bloat up req_state with string copies from it */
236 struct req_state
rstate(req
->cct
, &rgw_env
, req
->id
);
237 struct req_state
*s
= &rstate
;
242 RGWObjectCtx
rados_ctx(store
, s
); // XXX holds std::map
244 auto sysobj_ctx
= store
->svc()->sysobj
->init_obj_ctx();
245 s
->sysobj_ctx
= &sysobj_ctx
;
247 /* XXX and -then- stash req_state pointers everywhere they are needed */
248 ret
= req
->init(rgw_env
, &rados_ctx
, io
, s
);
250 ldpp_dout(op
, 10) << "failed to initialize request" << dendl
;
251 abort_req(s
, op
, ret
);
255 /* req is-a RGWOp, currently initialized separately */
256 ret
= req
->op_init();
258 dout(10) << "failed to initialize RGWOp" << dendl
;
259 abort_req(s
, op
, ret
);
263 /* now expected by rgw_log_op() */
264 rgw_env
.set("REQUEST_METHOD", s
->info
.method
);
265 rgw_env
.set("REQUEST_URI", s
->info
.request_uri
);
266 rgw_env
.set("QUERY_STRING", "");
269 /* XXX authorize does less here then in the REST path, e.g.,
270 * the user's info is cached, but still incomplete */
271 ldpp_dout(s
, 2) << "authorizing" << dendl
;
272 ret
= req
->authorize(op
, null_yield
);
274 dout(10) << "failed to authorize request" << dendl
;
275 abort_req(s
, op
, ret
);
279 /* FIXME: remove this after switching all handlers to the new
280 * authentication infrastructure. */
281 if (! s
->auth
.identity
) {
282 s
->auth
.identity
= rgw::auth::transform_old_authinfo(s
);
285 ldpp_dout(s
, 2) << "reading op permissions" << dendl
;
286 ret
= req
->read_permissions(op
, null_yield
);
288 abort_req(s
, op
, ret
);
292 ldpp_dout(s
, 2) << "init op" << dendl
;
293 ret
= op
->init_processing(null_yield
);
295 abort_req(s
, op
, ret
);
299 ldpp_dout(s
, 2) << "verifying op mask" << dendl
;
300 ret
= op
->verify_op_mask();
302 abort_req(s
, op
, ret
);
306 ldpp_dout(s
, 2) << "verifying op permissions" << dendl
;
307 ret
= op
->verify_permission(null_yield
);
309 if (s
->system_request
) {
310 ldpp_dout(op
, 2) << "overriding permissions due to system operation" << dendl
;
311 } else if (s
->auth
.identity
->is_admin_of(s
->user
->get_id())) {
312 ldpp_dout(op
, 2) << "overriding permissions due to admin operation" << dendl
;
314 abort_req(s
, op
, ret
);
319 ldpp_dout(s
, 2) << "verifying op params" << dendl
;
320 ret
= op
->verify_params();
322 abort_req(s
, op
, ret
);
326 ldpp_dout(s
, 2) << "executing" << dendl
;
328 op
->execute(null_yield
);
331 } catch (const ceph::crypto::DigestException
& e
) {
332 dout(0) << "authentication failed" << e
.what() << dendl
;
333 abort_req(s
, op
, -ERR_INVALID_SECRET_KEY
);
338 io
->complete_request();
339 } catch (rgw::io::Exception
& e
) {
340 dout(0) << "ERROR: io->complete_request() returned "
341 << e
.what() << dendl
;
344 rgw_log_op(nullptr /* !rest */, s
, (op
? op
->name() : "unknown"), olog
);
347 int http_ret
= s
->err
.http_ret
;
349 ldpp_dout(s
, 2) << "http status=" << http_ret
<< dendl
;
351 ldpp_dout(op
, 1) << "====== " << __func__
352 << " req done req=" << hex
<< req
<< dec
<< " http_status="
354 << " ======" << dendl
;
356 return (ret
< 0 ? ret
: s
->err
.ret
);
357 } /* process_request */
359 int RGWLibProcess::start_request(RGWLibContinuedReq
* req
)
362 dout(1) << "====== " << __func__
363 << " starting new continued request req=" << hex
<< req
<< dec
364 << " ======" << dendl
;
367 * invariant: valid requests are derived from RGWOp--well-formed
368 * requests should have assigned RGWRequest::op in their descendant
369 * constructor--if not, the compiler can find it, at the cost of
372 RGWOp
*op
= (req
->op
) ? req
->op
: dynamic_cast<RGWOp
*>(req
);
374 ldpp_dout(op
, 1) << "failed to derive cognate RGWOp (invalid op?)" << dendl
;
378 struct req_state
* s
= req
->get_state();
379 RGWLibIO
& io_ctx
= req
->get_io();
380 RGWEnv
& rgw_env
= io_ctx
.get_env();
381 RGWObjectCtx
& rados_ctx
= req
->get_octx();
383 rgw_env
.set("HTTP_HOST", "");
385 int ret
= req
->init(rgw_env
, &rados_ctx
, &io_ctx
, s
);
387 ldpp_dout(op
, 10) << "failed to initialize request" << dendl
;
388 abort_req(s
, op
, ret
);
392 /* req is-a RGWOp, currently initialized separately */
393 ret
= req
->op_init();
395 dout(10) << "failed to initialize RGWOp" << dendl
;
396 abort_req(s
, op
, ret
);
400 /* XXX authorize does less here then in the REST path, e.g.,
401 * the user's info is cached, but still incomplete */
402 ldpp_dout(s
, 2) << "authorizing" << dendl
;
403 ret
= req
->authorize(op
, null_yield
);
405 dout(10) << "failed to authorize request" << dendl
;
406 abort_req(s
, op
, ret
);
410 /* FIXME: remove this after switching all handlers to the new authentication
412 if (! s
->auth
.identity
) {
413 s
->auth
.identity
= rgw::auth::transform_old_authinfo(s
);
416 ldpp_dout(s
, 2) << "reading op permissions" << dendl
;
417 ret
= req
->read_permissions(op
, null_yield
);
419 abort_req(s
, op
, ret
);
423 ldpp_dout(s
, 2) << "init op" << dendl
;
424 ret
= op
->init_processing(null_yield
);
426 abort_req(s
, op
, ret
);
430 ldpp_dout(s
, 2) << "verifying op mask" << dendl
;
431 ret
= op
->verify_op_mask();
433 abort_req(s
, op
, ret
);
437 ldpp_dout(s
, 2) << "verifying op permissions" << dendl
;
438 ret
= op
->verify_permission(null_yield
);
440 if (s
->system_request
) {
441 ldpp_dout(op
, 2) << "overriding permissions due to system operation" << dendl
;
442 } else if (s
->auth
.identity
->is_admin_of(s
->user
->get_id())) {
443 ldpp_dout(op
, 2) << "overriding permissions due to admin operation" << dendl
;
445 abort_req(s
, op
, ret
);
450 ldpp_dout(s
, 2) << "verifying op params" << dendl
;
451 ret
= op
->verify_params();
453 abort_req(s
, op
, ret
);
461 return (ret
< 0 ? ret
: s
->err
.ret
);
464 int RGWLibProcess::finish_request(RGWLibContinuedReq
* req
)
466 RGWOp
*op
= (req
->op
) ? req
->op
: dynamic_cast<RGWOp
*>(req
);
468 ldpp_dout(op
, 1) << "failed to derive cognate RGWOp (invalid op?)" << dendl
;
472 int ret
= req
->exec_finish();
473 int op_ret
= op
->get_ret();
475 ldpp_dout(op
, 1) << "====== " << __func__
476 << " finishing continued request req=" << hex
<< req
<< dec
477 << " op status=" << op_ret
478 << " ======" << dendl
;
480 perfcounter
->inc(l_rgw_req
);
485 int RGWLibFrontend::init()
487 pprocess
= new RGWLibProcess(g_ceph_context
, &env
,
488 g_conf()->rgw_thread_pool_size
, conf
);
494 vector
<const char*> args
;
498 int RGWLib::init(vector
<const char*>& args
)
502 /* alternative default for module */
503 map
<string
,string
> defaults
= {
504 { "debug_rgw", "1/5" },
505 { "keyring", "$rgw_data/keyring" },
506 { "log_file", "/var/log/radosgw/$cluster-$name.log" }
509 cct
= global_init(&defaults
, args
,
510 CEPH_ENTITY_TYPE_CLIENT
,
511 CODE_ENVIRONMENT_DAEMON
,
512 CINIT_FLAG_UNPRIVILEGED_DAEMON_DEFAULTS
);
514 ceph::mutex mutex
= ceph::make_mutex("main");
515 SafeTimer
init_timer(g_ceph_context
, mutex
);
518 init_timer
.add_event_after(g_conf()->rgw_init_timeout
, new C_InitTimeout
);
521 common_init_finish(g_ceph_context
);
523 rgw_tools_init(g_ceph_context
);
526 rgw::curl::setup_curl(boost::none
);
527 rgw_http_client_init(g_ceph_context
);
530 g_conf()->rgw_enable_gc_threads
&&
531 g_conf()->rgw_nfs_run_gc_threads
;
534 g_conf()->rgw_enable_lc_threads
&&
535 g_conf()->rgw_nfs_run_lc_threads
;
538 g_conf()->rgw_enable_quota_threads
&&
539 g_conf()->rgw_nfs_run_quota_threads
;
542 g_conf()->rgw_run_sync_thread
&&
543 g_conf()->rgw_nfs_run_sync_thread
;
545 store
= RGWStoreManager::get_storage(this, g_ceph_context
,
550 g_conf().get_val
<bool>("rgw_dynamic_resharding"));
554 init_timer
.cancel_all_events();
555 init_timer
.shutdown();
558 derr
<< "Couldn't init storage provider (RADOS)" << dendl
;
562 r
= rgw_perf_start(g_ceph_context
);
564 rgw_rest_init(g_ceph_context
, store
->svc()->zone
->get_zonegroup());
567 init_timer
.cancel_all_events();
568 init_timer
.shutdown();
574 const string
& ldap_uri
= store
->ctx()->_conf
->rgw_ldap_uri
;
575 const string
& ldap_binddn
= store
->ctx()->_conf
->rgw_ldap_binddn
;
576 const string
& ldap_searchdn
= store
->ctx()->_conf
->rgw_ldap_searchdn
;
577 const string
& ldap_searchfilter
= store
->ctx()->_conf
->rgw_ldap_searchfilter
;
578 const string
& ldap_dnattr
=
579 store
->ctx()->_conf
->rgw_ldap_dnattr
;
580 std::string ldap_bindpw
= parse_rgw_ldap_bindpw(store
->ctx());
582 ldh
= new rgw::LDAPHelper(ldap_uri
, ldap_binddn
, ldap_bindpw
.c_str(),
583 ldap_searchdn
, ldap_searchfilter
, ldap_dnattr
);
587 rgw_log_usage_init(g_ceph_context
, store
->getRados());
589 // XXX ex-RGWRESTMgr_lib, mgr->set_logging(true)
591 OpsLogManifold
* olog_manifold
= new OpsLogManifold();
592 if (!g_conf()->rgw_ops_log_socket_path
.empty()) {
593 OpsLogSocket
* olog_socket
= new OpsLogSocket(g_ceph_context
, g_conf()->rgw_ops_log_data_backlog
);
594 olog_socket
->init(g_conf()->rgw_ops_log_socket_path
);
595 olog_manifold
->add_sink(olog_socket
);
597 OpsLogFile
* ops_log_file
;
598 if (!g_conf()->rgw_ops_log_file_path
.empty()) {
599 ops_log_file
= new OpsLogFile(g_ceph_context
, g_conf()->rgw_ops_log_file_path
, g_conf()->rgw_ops_log_data_backlog
);
600 ops_log_file
->start();
601 olog_manifold
->add_sink(ops_log_file
);
603 olog_manifold
->add_sink(new OpsLogRados(store
->getRados()));
604 olog
= olog_manifold
;
607 RGWProcessEnv env
= { store
, &rest
, olog
, port
};
609 string fe_count
{"0"};
610 fec
= new RGWFrontendConfig("rgwlib");
611 fe
= new RGWLibFrontend(env
, fec
);
613 init_async_signal_handler();
614 register_async_signal_handler(SIGUSR1
, handle_sigterm
);
616 map
<string
, string
> service_map_meta
;
617 service_map_meta
["pid"] = stringify(getpid());
618 service_map_meta
["frontend_type#" + fe_count
] = "rgw-nfs";
619 service_map_meta
["frontend_config#" + fe_count
] = fec
->get_config();
623 derr
<< "ERROR: failed initializing frontend" << dendl
;
629 r
= store
->getRados()->register_to_service_map("rgw-nfs", service_map_meta
);
631 derr
<< "ERROR: failed to register to service map: " << cpp_strerror(-r
) << dendl
;
635 #ifdef WITH_RADOSGW_AMQP_ENDPOINT
636 if (!rgw::amqp::init(cct
.get())) {
637 derr
<< "ERROR: failed to initialize AMQP manager" << dendl
;
640 #ifdef WITH_RADOSGW_KAFKA_ENDPOINT
641 if (!rgw::kafka::init(cct
.get())) {
642 derr
<< "ERROR: failed to initialize Kafka manager" << dendl
;
647 } /* RGWLib::init() */
651 derr
<< "shutting down" << dendl
;
661 unregister_async_signal_handler(SIGUSR1
, handle_sigterm
);
662 shutdown_async_signal_handler();
664 rgw_log_usage_finalize();
668 RGWStoreManager::close_storage(store
);
671 rgw_shutdown_resolver();
672 rgw_http_client_cleanup();
673 rgw::curl::cleanup_curl();
674 #ifdef WITH_RADOSGW_AMQP_ENDPOINT
675 rgw::amqp::shutdown();
677 #ifdef WITH_RADOSGW_KAFKA_ENDPOINT
678 rgw::kafka::shutdown();
681 rgw_perf_stop(g_ceph_context
);
683 dout(1) << "final shutdown" << dendl
;
687 } /* RGWLib::stop() */
689 int RGWLibIO::set_uid(rgw::sal::RGWRadosStore
*store
, const rgw_user
& uid
)
691 const DoutPrefix
dp(store
->ctx(), dout_subsys
, "librgw: ");
692 int ret
= store
->ctl()->user
->get_info_by_uid(&dp
, uid
, &user_info
, null_yield
);
694 derr
<< "ERROR: failed reading user info: uid=" << uid
<< " ret="
700 int RGWLibRequest::read_permissions(RGWOp
* op
, optional_yield y
) {
701 /* bucket and object ops */
703 rgw_build_bucket_policies(op
, rgwlib
.get_store(), get_state(), y
);
705 ldpp_dout(op
, 10) << "read_permissions (bucket policy) on "
706 << get_state()->bucket
<< ":"
707 << get_state()->object
708 << " only_bucket=" << only_bucket()
709 << " ret=" << ret
<< dendl
;
712 } else if (! only_bucket()) {
714 ret
= rgw_build_object_policies(op
, rgwlib
.get_store(), get_state(),
715 op
->prefetch_data(), y
);
717 ldpp_dout(op
, 10) << "read_permissions (object policy) on"
718 << get_state()->bucket
<< ":"
719 << get_state()->object
720 << " ret=" << ret
<< dendl
;
726 } /* RGWLibRequest::read_permissions */
728 int RGWHandler_Lib::authorize(const DoutPrefixProvider
*dpp
, optional_yield y
)
732 * 2. anonymous access
736 * Much or all of this depends on handling the cached authorization
737 * correctly (e.g., dealing with keystone) at mount time.
739 s
->perm_mask
= RGW_PERM_FULL_CONTROL
;
741 // populate the owner info
742 s
->owner
.set_id(s
->user
->get_id());
743 s
->owner
.set_name(s
->user
->get_display_name());
746 } /* RGWHandler_Lib::authorize */
748 } /* namespace rgw */
752 int librgw_create(librgw_t
* rgw
, int argc
, char **argv
)
758 if (! g_ceph_context
) {
759 std::lock_guard
<std::mutex
> lg(librgw_mtx
);
760 if (! g_ceph_context
) {
761 vector
<const char*> args
;
762 std::vector
<std::string
> spl_args
;
763 // last non-0 argument will be split and consumed
765 const std::string spl_arg
{argv
[(--argc
)]};
766 get_str_vec(spl_arg
, " \t", spl_args
);
768 argv_to_vec(argc
, const_cast<const char**>(argv
), args
);
769 // append split args, if any
770 for (const auto& elt
: spl_args
) {
771 args
.push_back(elt
.c_str());
773 rc
= rgwlib
.init(args
);
777 *rgw
= g_ceph_context
->get();
782 void librgw_shutdown(librgw_t rgw
)
786 CephContext
* cct
= static_cast<CephContext
*>(rgw
);