diff options
Diffstat (limited to 'src/rgw/driver/rados/rgw_tools.cc')
| -rw-r--r-- | src/rgw/driver/rados/rgw_tools.cc | 437 |
1 files changed, 437 insertions, 0 deletions
diff --git a/src/rgw/driver/rados/rgw_tools.cc b/src/rgw/driver/rados/rgw_tools.cc new file mode 100644 index 000000000..66651da5c --- /dev/null +++ b/src/rgw/driver/rados/rgw_tools.cc @@ -0,0 +1,437 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab ft=cpp + +#include "common/errno.h" +#include "librados/librados_asio.h" + +#include "include/stringify.h" + +#include "rgw_tools.h" +#include "rgw_acl_s3.h" +#include "rgw_aio_throttle.h" +#include "rgw_compression.h" +#include "common/BackTrace.h" + +#define dout_subsys ceph_subsys_rgw + +#define READ_CHUNK_LEN (512 * 1024) + +using namespace std; + +int rgw_init_ioctx(const DoutPrefixProvider *dpp, + librados::Rados *rados, const rgw_pool& pool, + librados::IoCtx& ioctx, bool create, + bool mostly_omap, + bool bulk) +{ + int r = rados->ioctx_create(pool.name.c_str(), ioctx); + if (r == -ENOENT && create) { + r = rados->pool_create(pool.name.c_str()); + if (r == -ERANGE) { + ldpp_dout(dpp, 0) + << __func__ + << " ERROR: librados::Rados::pool_create returned " << cpp_strerror(-r) + << " (this can be due to a pool or placement group misconfiguration, e.g." + << " pg_num < pgp_num or mon_max_pg_per_osd exceeded)" + << dendl; + } + if (r < 0 && r != -EEXIST) { + return r; + } + + r = rados->ioctx_create(pool.name.c_str(), ioctx); + if (r < 0) { + return r; + } + + r = ioctx.application_enable(pg_pool_t::APPLICATION_NAME_RGW, false); + if (r < 0 && r != -EOPNOTSUPP) { + return r; + } + + if (mostly_omap) { + // set pg_autoscale_bias + bufferlist inbl; + float bias = g_conf().get_val<double>("rgw_rados_pool_autoscale_bias"); + int r = rados->mon_command( + "{\"prefix\": \"osd pool set\", \"pool\": \"" + + pool.name + "\", \"var\": \"pg_autoscale_bias\", \"val\": \"" + + stringify(bias) + "\"}", + inbl, NULL, NULL); + if (r < 0) { + ldpp_dout(dpp, 10) << __func__ << " warning: failed to set pg_autoscale_bias on " + << pool.name << dendl; + } + // set recovery_priority + int p = g_conf().get_val<uint64_t>("rgw_rados_pool_recovery_priority"); + r = rados->mon_command( + "{\"prefix\": \"osd pool set\", \"pool\": \"" + + pool.name + "\", \"var\": \"recovery_priority\": \"" + + stringify(p) + "\"}", + inbl, NULL, NULL); + if (r < 0) { + ldpp_dout(dpp, 10) << __func__ << " warning: failed to set recovery_priority on " + << pool.name << dendl; + } + } + if (bulk) { + // set bulk + bufferlist inbl; + int r = rados->mon_command( + "{\"prefix\": \"osd pool set\", \"pool\": \"" + + pool.name + "\", \"var\": \"bulk\", \"val\": \"true\"}", + inbl, NULL, NULL); + if (r < 0) { + ldpp_dout(dpp, 10) << __func__ << " warning: failed to set 'bulk' on " + << pool.name << dendl; + } + } + } else if (r < 0) { + return r; + } + if (!pool.ns.empty()) { + ioctx.set_namespace(pool.ns); + } + return 0; +} + +map<string, bufferlist>* no_change_attrs() { + static map<string, bufferlist> no_change; + return &no_change; +} + +int rgw_put_system_obj(const DoutPrefixProvider *dpp, RGWSI_SysObj* svc_sysobj, + const rgw_pool& pool, const string& oid, bufferlist& data, bool exclusive, + RGWObjVersionTracker *objv_tracker, real_time set_mtime, optional_yield y, map<string, bufferlist> *pattrs) +{ + map<string,bufferlist> no_attrs; + if (!pattrs) { + pattrs = &no_attrs; + } + + rgw_raw_obj obj(pool, oid); + + auto sysobj = svc_sysobj->get_obj(obj); + int ret; + + if (pattrs != no_change_attrs()) { + ret = sysobj.wop() + .set_objv_tracker(objv_tracker) + .set_exclusive(exclusive) + .set_mtime(set_mtime) + .set_attrs(*pattrs) + .write(dpp, data, y); + } else { + ret = sysobj.wop() + .set_objv_tracker(objv_tracker) + .set_exclusive(exclusive) + .set_mtime(set_mtime) + .write_data(dpp, data, y); + } + + return ret; +} + +int rgw_stat_system_obj(const DoutPrefixProvider *dpp, RGWSI_SysObj* svc_sysobj, + const rgw_pool& pool, const std::string& key, + RGWObjVersionTracker *objv_tracker, + real_time *pmtime, optional_yield y, + std::map<std::string, bufferlist> *pattrs) +{ + rgw_raw_obj obj(pool, key); + auto sysobj = svc_sysobj->get_obj(obj); + return sysobj.rop() + .set_attrs(pattrs) + .set_last_mod(pmtime) + .stat(y, dpp); +} + + +int rgw_get_system_obj(RGWSI_SysObj* svc_sysobj, const rgw_pool& pool, const string& key, bufferlist& bl, + RGWObjVersionTracker *objv_tracker, real_time *pmtime, optional_yield y, + const DoutPrefixProvider *dpp, map<string, bufferlist> *pattrs, + rgw_cache_entry_info *cache_info, + boost::optional<obj_version> refresh_version, bool raw_attrs) +{ + const rgw_raw_obj obj(pool, key); + auto sysobj = svc_sysobj->get_obj(obj); + auto rop = sysobj.rop(); + return rop.set_attrs(pattrs) + .set_last_mod(pmtime) + .set_objv_tracker(objv_tracker) + .set_raw_attrs(raw_attrs) + .set_cache_info(cache_info) + .set_refresh_version(refresh_version) + .read(dpp, &bl, y); +} + +int rgw_delete_system_obj(const DoutPrefixProvider *dpp, + RGWSI_SysObj *sysobj_svc, const rgw_pool& pool, const string& oid, + RGWObjVersionTracker *objv_tracker, optional_yield y) +{ + auto sysobj = sysobj_svc->get_obj(rgw_raw_obj{pool, oid}); + rgw_raw_obj obj(pool, oid); + return sysobj.wop() + .set_objv_tracker(objv_tracker) + .remove(dpp, y); +} + +int rgw_rados_operate(const DoutPrefixProvider *dpp, librados::IoCtx& ioctx, const std::string& oid, + librados::ObjectReadOperation *op, bufferlist* pbl, + optional_yield y, int flags) +{ + // given a yield_context, call async_operate() to yield the coroutine instead + // of blocking + if (y) { + auto& context = y.get_io_context(); + auto& yield = y.get_yield_context(); + boost::system::error_code ec; + auto bl = librados::async_operate( + context, ioctx, oid, op, flags, yield[ec]); + if (pbl) { + *pbl = std::move(bl); + } + return -ec.value(); + } + // work on asio threads should be asynchronous, so warn when they block + if (is_asio_thread) { + ldpp_dout(dpp, 20) << "WARNING: blocking librados call" << dendl; +#ifdef _BACKTRACE_LOGGING + ldpp_dout(dpp, 20) << "BACKTRACE: " << __func__ << ": " << ClibBackTrace(0) << dendl; +#endif + } + return ioctx.operate(oid, op, nullptr, flags); +} + +int rgw_rados_operate(const DoutPrefixProvider *dpp, librados::IoCtx& ioctx, const std::string& oid, + librados::ObjectWriteOperation *op, optional_yield y, + int flags) +{ + if (y) { + auto& context = y.get_io_context(); + auto& yield = y.get_yield_context(); + boost::system::error_code ec; + librados::async_operate(context, ioctx, oid, op, flags, yield[ec]); + return -ec.value(); + } + if (is_asio_thread) { + ldpp_dout(dpp, 20) << "WARNING: blocking librados call" << dendl; +#ifdef _BACKTRACE_LOGGING + ldpp_dout(dpp, 20) << "BACKTRACE: " << __func__ << ": " << ClibBackTrace(0) << dendl; +#endif + } + return ioctx.operate(oid, op, flags); +} + +int rgw_rados_notify(const DoutPrefixProvider *dpp, librados::IoCtx& ioctx, const std::string& oid, + bufferlist& bl, uint64_t timeout_ms, bufferlist* pbl, + optional_yield y) +{ + if (y) { + auto& context = y.get_io_context(); + auto& yield = y.get_yield_context(); + boost::system::error_code ec; + auto reply = librados::async_notify(context, ioctx, oid, + bl, timeout_ms, yield[ec]); + if (pbl) { + *pbl = std::move(reply); + } + return -ec.value(); + } + if (is_asio_thread) { + ldpp_dout(dpp, 20) << "WARNING: blocking librados call" << dendl; +#ifdef _BACKTRACE_LOGGING + ldpp_dout(dpp, 20) << "BACKTRACE: " << __func__ << ": " << ClibBackTrace(0) << dendl; +#endif + } + return ioctx.notify2(oid, bl, timeout_ms, pbl); +} + +void rgw_filter_attrset(map<string, bufferlist>& unfiltered_attrset, const string& check_prefix, + map<string, bufferlist> *attrset) +{ + attrset->clear(); + map<string, bufferlist>::iterator iter; + for (iter = unfiltered_attrset.lower_bound(check_prefix); + iter != unfiltered_attrset.end(); ++iter) { + if (!boost::algorithm::starts_with(iter->first, check_prefix)) + break; + (*attrset)[iter->first] = iter->second; + } +} + +RGWDataAccess::RGWDataAccess(rgw::sal::Driver* _driver) : driver(_driver) +{ +} + + +int RGWDataAccess::Bucket::finish_init() +{ + auto iter = attrs.find(RGW_ATTR_ACL); + if (iter == attrs.end()) { + return 0; + } + + bufferlist::const_iterator bliter = iter->second.begin(); + try { + policy.decode(bliter); + } catch (buffer::error& err) { + return -EIO; + } + + return 0; +} + +int RGWDataAccess::Bucket::init(const DoutPrefixProvider *dpp, optional_yield y) +{ + std::unique_ptr<rgw::sal::Bucket> bucket; + int ret = sd->driver->get_bucket(dpp, nullptr, tenant, name, &bucket, y); + if (ret < 0) { + return ret; + } + + bucket_info = bucket->get_info(); + mtime = bucket->get_modification_time(); + attrs = bucket->get_attrs(); + + return finish_init(); +} + +int RGWDataAccess::Bucket::init(const RGWBucketInfo& _bucket_info, + const map<string, bufferlist>& _attrs) +{ + bucket_info = _bucket_info; + attrs = _attrs; + + return finish_init(); +} + +int RGWDataAccess::Bucket::get_object(const rgw_obj_key& key, + ObjectRef *obj) { + obj->reset(new Object(sd, shared_from_this(), key)); + return 0; +} + +int RGWDataAccess::Object::put(bufferlist& data, + map<string, bufferlist>& attrs, + const DoutPrefixProvider *dpp, + optional_yield y) +{ + rgw::sal::Driver* driver = sd->driver; + CephContext *cct = driver->ctx(); + + string tag; + append_rand_alpha(cct, tag, tag, 32); + + RGWBucketInfo& bucket_info = bucket->bucket_info; + + rgw::BlockingAioThrottle aio(driver->ctx()->_conf->rgw_put_obj_min_window_size); + + std::unique_ptr<rgw::sal::Bucket> b; + driver->get_bucket(NULL, bucket_info, &b); + std::unique_ptr<rgw::sal::Object> obj = b->get_object(key); + + auto& owner = bucket->policy.get_owner(); + + string req_id = driver->zone_unique_id(driver->get_new_req_id()); + + std::unique_ptr<rgw::sal::Writer> processor; + processor = driver->get_atomic_writer(dpp, y, obj.get(), + owner.get_id(), + nullptr, olh_epoch, req_id); + + int ret = processor->prepare(y); + if (ret < 0) + return ret; + + rgw::sal::DataProcessor *filter = processor.get(); + + CompressorRef plugin; + boost::optional<RGWPutObj_Compress> compressor; + + const auto& compression_type = driver->get_compression_type(bucket_info.placement_rule); + if (compression_type != "none") { + plugin = Compressor::create(driver->ctx(), compression_type); + if (!plugin) { + ldpp_dout(dpp, 1) << "Cannot load plugin for compression type " + << compression_type << dendl; + } else { + compressor.emplace(driver->ctx(), plugin, filter); + filter = &*compressor; + } + } + + off_t ofs = 0; + auto obj_size = data.length(); + + RGWMD5Etag etag_calc; + + do { + size_t read_len = std::min(data.length(), (unsigned int)cct->_conf->rgw_max_chunk_size); + + bufferlist bl; + + data.splice(0, read_len, &bl); + etag_calc.update(bl); + + ret = filter->process(std::move(bl), ofs); + if (ret < 0) + return ret; + + ofs += read_len; + } while (data.length() > 0); + + ret = filter->process({}, ofs); + if (ret < 0) { + return ret; + } + bool has_etag_attr = false; + auto iter = attrs.find(RGW_ATTR_ETAG); + if (iter != attrs.end()) { + bufferlist& bl = iter->second; + etag = bl.to_str(); + has_etag_attr = true; + } + + if (!aclbl) { + RGWAccessControlPolicy_S3 policy(cct); + + policy.create_canned(bucket->policy.get_owner(), bucket->policy.get_owner(), string()); /* default private policy */ + + policy.encode(aclbl.emplace()); + } + + if (etag.empty()) { + etag_calc.finish(&etag); + } + + if (!has_etag_attr) { + bufferlist etagbl; + etagbl.append(etag); + attrs[RGW_ATTR_ETAG] = etagbl; + } + attrs[RGW_ATTR_ACL] = *aclbl; + + string *puser_data = nullptr; + if (user_data) { + puser_data = &(*user_data); + } + + return processor->complete(obj_size, etag, + &mtime, mtime, + attrs, delete_at, + nullptr, nullptr, + puser_data, + nullptr, nullptr, y); +} + +void RGWDataAccess::Object::set_policy(const RGWAccessControlPolicy& policy) +{ + policy.encode(aclbl.emplace()); +} + +void rgw_complete_aio_completion(librados::AioCompletion* c, int r) { + auto pc = c->pc; + librados::CB_AioCompleteAndSafe cb(pc); + cb(r); +} |