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
74 bool global_stop
= false;
76 static void handle_sigterm(int signum
)
78 dout(20) << __func__
<< " SIGUSR1 ignored" << dendl
;
85 static std::mutex librgw_mtx
;
89 class C_InitTimeout
: public Context
{
92 void finish(int r
) override
{
93 derr
<< "Initialization timeout, failed to initialize" << dendl
;
98 void RGWLibProcess::checkpoint()
103 #define MIN_EXPIRE_S 120
105 void RGWLibProcess::run()
107 /* write completion interval */
108 RGWLibFS::write_completion_interval_s
=
109 cct
->_conf
->rgw_nfs_write_completion_interval_s
;
111 /* start write timer */
112 RGWLibFS::write_timer
.resume();
116 lsubdout(cct
, rgw
, 5) << "RGWLibProcess GC" << dendl
;
118 /* dirent invalidate timeout--basically, the upper-bound on
119 * inconsistency with the S3 namespace */
120 auto expire_s
= cct
->_conf
->rgw_nfs_namespace_expire_secs
;
122 /* delay between gc cycles */
123 auto delay_s
= std::max(int64_t(1), std::min(int64_t(MIN_EXPIRE_S
), expire_s
/2));
125 unique_lock
uniq(mtx
);
128 for (auto iter
= mounted_fs
.begin(); iter
!= mounted_fs
.end();
130 RGWLibFS
* fs
= iter
->first
->ref();
133 const DoutPrefix
dp(cct
, dout_subsys
, "librgw: ");
134 fs
->update_user(&dp
);
138 goto restart
; /* invalidated */
140 cv
.wait_for(uniq
, std::chrono::seconds(delay_s
));
145 void RGWLibProcess::handle_request(const DoutPrefixProvider
*dpp
, RGWRequest
* r
)
148 * invariant: valid requests are derived from RGWLibRequst
150 RGWLibRequest
* req
= static_cast<RGWLibRequest
*>(r
);
152 // XXX move RGWLibIO and timing setup into process_request
155 utime_t tm
= ceph_clock_now();
160 int ret
= process_request(req
, &io_ctx
);
162 /* we don't really care about return code */
163 dout(20) << "process_request() returned " << ret
<< dendl
;
167 } /* handle_request */
169 int RGWLibProcess::process_request(RGWLibRequest
* req
)
171 // XXX move RGWLibIO and timing setup into process_request
174 utime_t tm
= ceph_clock_now();
179 int ret
= process_request(req
, &io_ctx
);
181 /* we don't really care about return code */
182 dout(20) << "process_request() returned " << ret
<< dendl
;
185 } /* process_request */
187 static inline void abort_req(struct req_state
*s
, RGWOp
*op
, int err_no
)
192 /* XXX the dump_errno and dump_bucket_from_state behaviors in
193 * the abort_early (rgw_rest.cc) might be valuable, but aren't
194 * safe to call presently as they return HTTP data */
196 perfcounter
->inc(l_rgw_failed_req
);
199 int RGWLibProcess::process_request(RGWLibRequest
* req
, RGWLibIO
* io
)
202 bool should_log
= true; // XXX
204 dout(1) << "====== " << __func__
205 << " starting new request req=" << hex
<< req
<< dec
206 << " ======" << dendl
;
209 * invariant: valid requests are derived from RGWOp--well-formed
210 * requests should have assigned RGWRequest::op in their descendant
211 * constructor--if not, the compiler can find it, at the cost of
214 RGWOp
*op
= (req
->op
) ? req
->op
: dynamic_cast<RGWOp
*>(req
);
216 ldpp_dout(op
, 1) << "failed to derive cognate RGWOp (invalid op?)" << dendl
;
222 perfcounter
->inc(l_rgw_req
);
224 RGWEnv
& rgw_env
= io
->get_env();
227 * until major refactoring of req_state and req_info, we need
228 * to build their RGWEnv boilerplate from the RGWLibRequest,
229 * pre-staging any strings (HTTP_HOST) that provoke a crash when
233 /* XXX for now, use ""; could be a legit hostname, or, in future,
234 * perhaps a tenant (Yehuda) */
235 rgw_env
.set("HTTP_HOST", "");
237 /* XXX and -then- bloat up req_state with string copies from it */
238 struct req_state
rstate(req
->cct
, &rgw_env
, req
->id
);
239 struct req_state
*s
= &rstate
;
244 RGWObjectCtx
rados_ctx(store
, s
); // XXX holds std::map
246 /* XXX and -then- stash req_state pointers everywhere they are needed */
247 ret
= req
->init(rgw_env
, &rados_ctx
, io
, s
);
249 ldpp_dout(op
, 10) << "failed to initialize request" << dendl
;
250 abort_req(s
, op
, ret
);
254 /* req is-a RGWOp, currently initialized separately */
255 ret
= req
->op_init();
257 dout(10) << "failed to initialize RGWOp" << dendl
;
258 abort_req(s
, op
, ret
);
262 /* now expected by rgw_log_op() */
263 rgw_env
.set("REQUEST_METHOD", s
->info
.method
);
264 rgw_env
.set("REQUEST_URI", s
->info
.request_uri
);
265 rgw_env
.set("QUERY_STRING", "");
268 /* XXX authorize does less here then in the REST path, e.g.,
269 * the user's info is cached, but still incomplete */
270 ldpp_dout(s
, 2) << "authorizing" << dendl
;
271 ret
= req
->authorize(op
, null_yield
);
273 dout(10) << "failed to authorize request" << dendl
;
274 abort_req(s
, op
, ret
);
278 /* FIXME: remove this after switching all handlers to the new
279 * authentication infrastructure. */
280 if (! s
->auth
.identity
) {
281 s
->auth
.identity
= rgw::auth::transform_old_authinfo(s
);
284 ldpp_dout(s
, 2) << "reading op permissions" << dendl
;
285 ret
= req
->read_permissions(op
, null_yield
);
287 abort_req(s
, op
, ret
);
291 ldpp_dout(s
, 2) << "init op" << dendl
;
292 ret
= op
->init_processing(null_yield
);
294 abort_req(s
, op
, ret
);
298 ldpp_dout(s
, 2) << "verifying op mask" << dendl
;
299 ret
= op
->verify_op_mask();
301 abort_req(s
, op
, ret
);
305 ldpp_dout(s
, 2) << "verifying op permissions" << dendl
;
306 ret
= op
->verify_permission(null_yield
);
308 if (s
->system_request
) {
309 ldpp_dout(op
, 2) << "overriding permissions due to system operation" << dendl
;
310 } else if (s
->auth
.identity
->is_admin_of(s
->user
->get_id())) {
311 ldpp_dout(op
, 2) << "overriding permissions due to admin operation" << dendl
;
313 abort_req(s
, op
, ret
);
318 ldpp_dout(s
, 2) << "verifying op params" << dendl
;
319 ret
= op
->verify_params();
321 abort_req(s
, op
, ret
);
325 ldpp_dout(s
, 2) << "executing" << dendl
;
327 op
->execute(null_yield
);
330 } catch (const ceph::crypto::DigestException
& e
) {
331 dout(0) << "authentication failed" << e
.what() << dendl
;
332 abort_req(s
, op
, -ERR_INVALID_SECRET_KEY
);
337 io
->complete_request();
338 } catch (rgw::io::Exception
& e
) {
339 dout(0) << "ERROR: io->complete_request() returned "
340 << e
.what() << dendl
;
343 rgw_log_op(nullptr /* !rest */, s
, op
, olog
);
346 int http_ret
= s
->err
.http_ret
;
348 ldpp_dout(s
, 2) << "http status=" << http_ret
<< dendl
;
350 ldpp_dout(op
, 1) << "====== " << __func__
351 << " req done req=" << hex
<< req
<< dec
<< " http_status="
353 << " ======" << dendl
;
355 return (ret
< 0 ? ret
: s
->err
.ret
);
356 } /* process_request */
358 int RGWLibProcess::start_request(RGWLibContinuedReq
* req
)
361 dout(1) << "====== " << __func__
362 << " starting new continued request req=" << hex
<< req
<< dec
363 << " ======" << dendl
;
366 * invariant: valid requests are derived from RGWOp--well-formed
367 * requests should have assigned RGWRequest::op in their descendant
368 * constructor--if not, the compiler can find it, at the cost of
371 RGWOp
*op
= (req
->op
) ? req
->op
: dynamic_cast<RGWOp
*>(req
);
373 ldpp_dout(op
, 1) << "failed to derive cognate RGWOp (invalid op?)" << dendl
;
377 struct req_state
* s
= req
->get_state();
378 RGWLibIO
& io_ctx
= req
->get_io();
379 RGWEnv
& rgw_env
= io_ctx
.get_env();
380 RGWObjectCtx
& rados_ctx
= req
->get_octx();
382 rgw_env
.set("HTTP_HOST", "");
384 int ret
= req
->init(rgw_env
, &rados_ctx
, &io_ctx
, s
);
386 ldpp_dout(op
, 10) << "failed to initialize request" << dendl
;
387 abort_req(s
, op
, ret
);
391 /* req is-a RGWOp, currently initialized separately */
392 ret
= req
->op_init();
394 dout(10) << "failed to initialize RGWOp" << dendl
;
395 abort_req(s
, op
, ret
);
399 /* XXX authorize does less here then in the REST path, e.g.,
400 * the user's info is cached, but still incomplete */
401 ldpp_dout(s
, 2) << "authorizing" << dendl
;
402 ret
= req
->authorize(op
, null_yield
);
404 dout(10) << "failed to authorize request" << dendl
;
405 abort_req(s
, op
, ret
);
409 /* FIXME: remove this after switching all handlers to the new authentication
411 if (! s
->auth
.identity
) {
412 s
->auth
.identity
= rgw::auth::transform_old_authinfo(s
);
415 ldpp_dout(s
, 2) << "reading op permissions" << dendl
;
416 ret
= req
->read_permissions(op
, null_yield
);
418 abort_req(s
, op
, ret
);
422 ldpp_dout(s
, 2) << "init op" << dendl
;
423 ret
= op
->init_processing(null_yield
);
425 abort_req(s
, op
, ret
);
429 ldpp_dout(s
, 2) << "verifying op mask" << dendl
;
430 ret
= op
->verify_op_mask();
432 abort_req(s
, op
, ret
);
436 ldpp_dout(s
, 2) << "verifying op permissions" << dendl
;
437 ret
= op
->verify_permission(null_yield
);
439 if (s
->system_request
) {
440 ldpp_dout(op
, 2) << "overriding permissions due to system operation" << dendl
;
441 } else if (s
->auth
.identity
->is_admin_of(s
->user
->get_id())) {
442 ldpp_dout(op
, 2) << "overriding permissions due to admin operation" << dendl
;
444 abort_req(s
, op
, ret
);
449 ldpp_dout(s
, 2) << "verifying op params" << dendl
;
450 ret
= op
->verify_params();
452 abort_req(s
, op
, ret
);
460 return (ret
< 0 ? ret
: s
->err
.ret
);
463 int RGWLibProcess::finish_request(RGWLibContinuedReq
* req
)
465 RGWOp
*op
= (req
->op
) ? req
->op
: dynamic_cast<RGWOp
*>(req
);
467 ldpp_dout(op
, 1) << "failed to derive cognate RGWOp (invalid op?)" << dendl
;
471 int ret
= req
->exec_finish();
472 int op_ret
= op
->get_ret();
474 ldpp_dout(op
, 1) << "====== " << __func__
475 << " finishing continued request req=" << hex
<< req
<< dec
476 << " op status=" << op_ret
477 << " ======" << dendl
;
479 perfcounter
->inc(l_rgw_req
);
484 int RGWLibFrontend::init()
486 pprocess
= new RGWLibProcess(g_ceph_context
, &env
,
487 g_conf()->rgw_thread_pool_size
, conf
);
493 vector
<const char*> args
;
497 int RGWLib::init(vector
<const char*>& args
)
501 /* alternative default for module */
502 map
<string
,string
> defaults
= {
503 { "debug_rgw", "1/5" },
504 { "keyring", "$rgw_data/keyring" },
505 { "log_file", "/var/log/radosgw/$cluster-$name.log" }
508 cct
= global_init(&defaults
, args
,
509 CEPH_ENTITY_TYPE_CLIENT
,
510 CODE_ENVIRONMENT_DAEMON
,
511 CINIT_FLAG_UNPRIVILEGED_DAEMON_DEFAULTS
);
513 ceph::mutex mutex
= ceph::make_mutex("main");
514 SafeTimer
init_timer(g_ceph_context
, mutex
);
517 init_timer
.add_event_after(g_conf()->rgw_init_timeout
, new C_InitTimeout
);
520 common_init_finish(g_ceph_context
);
522 rgw_tools_init(this, g_ceph_context
);
525 rgw::curl::setup_curl(boost::none
);
526 rgw_http_client_init(g_ceph_context
);
529 g_conf()->rgw_enable_gc_threads
&&
530 g_conf()->rgw_nfs_run_gc_threads
;
533 g_conf()->rgw_enable_lc_threads
&&
534 g_conf()->rgw_nfs_run_lc_threads
;
537 g_conf()->rgw_enable_quota_threads
&&
538 g_conf()->rgw_nfs_run_quota_threads
;
541 g_conf()->rgw_run_sync_thread
&&
542 g_conf()->rgw_nfs_run_sync_thread
;
544 store
= StoreManager::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
->get_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
);
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
));
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
->register_to_service_map(this, "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 StoreManager::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::Store
* store
, const rgw_user
& uid
)
691 const DoutPrefix
dp(store
->ctx(), dout_subsys
, "librgw: ");
692 std::unique_ptr
<rgw::sal::User
> user
= store
->get_user(uid
);
693 /* object exists, but policy is broken */
694 int ret
= user
->load_user(&dp
, null_yield
);
696 derr
<< "ERROR: failed reading user info: uid=" << uid
<< " ret="
699 user_info
= user
->get_info();
703 int RGWLibRequest::read_permissions(RGWOp
* op
, optional_yield y
) {
704 /* bucket and object ops */
706 rgw_build_bucket_policies(op
, rgwlib
.get_store(), get_state(), y
);
708 ldpp_dout(op
, 10) << "read_permissions (bucket policy) on "
709 << get_state()->bucket
<< ":"
710 << get_state()->object
711 << " only_bucket=" << only_bucket()
712 << " ret=" << ret
<< dendl
;
715 } else if (! only_bucket()) {
717 ret
= rgw_build_object_policies(op
, rgwlib
.get_store(), get_state(),
718 op
->prefetch_data(), y
);
720 ldpp_dout(op
, 10) << "read_permissions (object policy) on"
721 << get_state()->bucket
<< ":"
722 << get_state()->object
723 << " ret=" << ret
<< dendl
;
729 } /* RGWLibRequest::read_permissions */
731 int RGWHandler_Lib::authorize(const DoutPrefixProvider
*dpp
, optional_yield y
)
735 * 2. anonymous access
739 * Much or all of this depends on handling the cached authorization
740 * correctly (e.g., dealing with keystone) at mount time.
742 s
->perm_mask
= RGW_PERM_FULL_CONTROL
;
744 // populate the owner info
745 s
->owner
.set_id(s
->user
->get_id());
746 s
->owner
.set_name(s
->user
->get_display_name());
749 } /* RGWHandler_Lib::authorize */
751 } /* namespace rgw */
755 int librgw_create(librgw_t
* rgw
, int argc
, char **argv
)
761 if (! g_ceph_context
) {
762 std::lock_guard
<std::mutex
> lg(librgw_mtx
);
763 if (! g_ceph_context
) {
764 std::vector
<std::string
> spl_args
;
765 // last non-0 argument will be split and consumed
767 const std::string spl_arg
{argv
[(--argc
)]};
768 get_str_vec(spl_arg
, " \t", spl_args
);
770 auto args
= argv_to_vec(argc
, argv
);
771 // append split args, if any
772 for (const auto& elt
: spl_args
) {
773 args
.push_back(elt
.c_str());
775 rc
= rgwlib
.init(args
);
779 *rgw
= g_ceph_context
->get();
784 void librgw_shutdown(librgw_t rgw
)
788 CephContext
* cct
= static_cast<CephContext
*>(rgw
);