rgw_bucket_encryption.cc
rgw_tracer.cc
rgw_lua_background.cc
+ rgw_data_access.cc
driver/rados/cls_fifo_legacy.cc
driver/rados/rgw_bucket.cc
driver/rados/rgw_bucket_sync.cc
}
}
-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->load_bucket(dpp, rgw_bucket(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(bucket_info);
- 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.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 policy;
-
- const auto& owner = bucket->policy.get_owner();
- policy.create_default(owner.id, owner.display_name); // 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);
- }
-
- const req_context rctx{dpp, y, nullptr};
- return processor->complete(obj_size, etag,
- &mtime, mtime,
- attrs, delete_at,
- nullptr, nullptr,
- puser_data,
- nullptr, nullptr, rctx);
-}
-
-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);
int rgw_tools_init(const DoutPrefixProvider *dpp, CephContext *cct);
void rgw_tools_cleanup();
-template<class H, size_t S>
-class RGWEtag
-{
- H hash;
-
-public:
- RGWEtag() {
- if constexpr (std::is_same_v<H, MD5>) {
- // Allow use of MD5 digest in FIPS mode for non-cryptographic purposes
- hash.SetFlags(EVP_MD_CTX_FLAG_NON_FIPS_ALLOW);
- }
- }
-
- void update(const char *buf, size_t len) {
- hash.Update((const unsigned char *)buf, len);
- }
-
- void update(bufferlist& bl) {
- if (bl.length() > 0) {
- update(bl.c_str(), bl.length());
- }
- }
-
- void update(const std::string& s) {
- if (!s.empty()) {
- update(s.c_str(), s.size());
- }
- }
- void finish(std::string *etag) {
- char etag_buf[S];
- char etag_buf_str[S * 2 + 16];
-
- hash.Final((unsigned char *)etag_buf);
- buf_to_hex((const unsigned char *)etag_buf, S,
- etag_buf_str);
-
- *etag = etag_buf_str;
- }
-};
-
-using RGWMD5Etag = RGWEtag<MD5, CEPH_CRYPTO_MD5_DIGESTSIZE>;
-
-class RGWDataAccess
-{
- rgw::sal::Driver* driver;
-
-public:
- RGWDataAccess(rgw::sal::Driver* _driver);
-
- class Object;
- class Bucket;
-
- using BucketRef = std::shared_ptr<Bucket>;
- using ObjectRef = std::shared_ptr<Object>;
-
- class Bucket : public std::enable_shared_from_this<Bucket> {
- friend class RGWDataAccess;
- friend class Object;
-
- RGWDataAccess *sd{nullptr};
- RGWBucketInfo bucket_info;
- std::string tenant;
- std::string name;
- std::string bucket_id;
- ceph::real_time mtime;
- std::map<std::string, bufferlist> attrs;
-
- RGWAccessControlPolicy policy;
- int finish_init();
-
- Bucket(RGWDataAccess *_sd,
- const std::string& _tenant,
- const std::string& _name,
- const std::string& _bucket_id) : sd(_sd),
- tenant(_tenant),
- name(_name),
- bucket_id(_bucket_id) {}
- Bucket(RGWDataAccess *_sd) : sd(_sd) {}
- int init(const DoutPrefixProvider *dpp, optional_yield y);
- int init(const RGWBucketInfo& _bucket_info, const std::map<std::string, bufferlist>& _attrs);
- public:
- int get_object(const rgw_obj_key& key,
- ObjectRef *obj);
-
- };
-
-
- class Object {
- RGWDataAccess *sd{nullptr};
- BucketRef bucket;
- rgw_obj_key key;
-
- ceph::real_time mtime;
- std::string etag;
- uint64_t olh_epoch{0};
- ceph::real_time delete_at;
- std::optional<std::string> user_data;
-
- std::optional<bufferlist> aclbl;
-
- Object(RGWDataAccess *_sd,
- BucketRef&& _bucket,
- const rgw_obj_key& _key) : sd(_sd),
- bucket(_bucket),
- key(_key) {}
- public:
- int put(bufferlist& data, std::map<std::string, bufferlist>& attrs, const DoutPrefixProvider *dpp, optional_yield y); /* might modify attrs */
-
- void set_mtime(const ceph::real_time& _mtime) {
- mtime = _mtime;
- }
-
- void set_etag(const std::string& _etag) {
- etag = _etag;
- }
-
- void set_olh_epoch(uint64_t epoch) {
- olh_epoch = epoch;
- }
-
- void set_delete_at(ceph::real_time _delete_at) {
- delete_at = _delete_at;
- }
-
- void set_user_data(const std::string& _user_data) {
- user_data = _user_data;
- }
-
- void set_policy(const RGWAccessControlPolicy& policy);
-
- friend class Bucket;
- };
-
- int get_bucket(const DoutPrefixProvider *dpp,
- const std::string& tenant,
- const std::string name,
- const std::string bucket_id,
- BucketRef *bucket,
- optional_yield y) {
- bucket->reset(new Bucket(this, tenant, name, bucket_id));
- return (*bucket)->init(dpp, y);
- }
-
- int get_bucket(const RGWBucketInfo& bucket_info,
- const std::map<std::string, bufferlist>& attrs,
- BucketRef *bucket) {
- bucket->reset(new Bucket(this));
- return (*bucket)->init(bucket_info, attrs);
- }
- friend class Bucket;
- friend class Object;
-};
-
-using RGWDataAccessRef = std::shared_ptr<RGWDataAccess>;
-
/// Complete an AioCompletion. To return error values or otherwise
/// satisfy the caller. Useful for making complicated asynchronous
/// calls and error handling.
#include "rgw_lua.h"
#include "rgw_sal.h"
#include "rgw_sal_config.h"
+#include "rgw_data_access.h"
#include "services/svc_sync_modules.h"
#include "services/svc_cls.h"
--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab ft=cpp
+
+#include "rgw_data_access.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
+
+template<class H, size_t S>
+class RGWEtag
+{
+ H hash;
+
+public:
+ RGWEtag() {
+ if constexpr (std::is_same_v<H, MD5>) {
+ // Allow use of MD5 digest in FIPS mode for non-cryptographic purposes
+ hash.SetFlags(EVP_MD_CTX_FLAG_NON_FIPS_ALLOW);
+ }
+ }
+
+ void update(const char *buf, size_t len) {
+ hash.Update((const unsigned char *)buf, len);
+ }
+
+ void update(bufferlist& bl) {
+ if (bl.length() > 0) {
+ update(bl.c_str(), bl.length());
+ }
+ }
+
+ void update(const std::string& s) {
+ if (!s.empty()) {
+ update(s.c_str(), s.size());
+ }
+ }
+ void finish(std::string *etag) {
+ char etag_buf[S];
+ char etag_buf_str[S * 2 + 16];
+
+ hash.Final((unsigned char *)etag_buf);
+ buf_to_hex((const unsigned char *)etag_buf, S,
+ etag_buf_str);
+
+ *etag = etag_buf_str;
+ }
+};
+
+using RGWMD5Etag = RGWEtag<MD5, CEPH_CRYPTO_MD5_DIGESTSIZE>;
+
+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->load_bucket(dpp, rgw_bucket(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 std::map<std::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,
+ std::map<std::string, bufferlist>& attrs,
+ const DoutPrefixProvider *dpp,
+ optional_yield y)
+{
+ rgw::sal::Driver* driver = sd->driver;
+ CephContext *cct = driver->ctx();
+
+ std::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(bucket_info);
+ std::unique_ptr<rgw::sal::Object> obj = b->get_object(key);
+
+ auto& owner = bucket->policy.get_owner();
+
+ std::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.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 policy;
+
+ const auto& owner = bucket->policy.get_owner();
+ policy.create_default(owner.id, owner.display_name); // 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;
+
+ std::string *puser_data = nullptr;
+ if (user_data) {
+ puser_data = &(*user_data);
+ }
+
+ const req_context rctx{dpp, y, nullptr};
+ return processor->complete(obj_size, etag,
+ &mtime, mtime,
+ attrs, delete_at,
+ nullptr, nullptr,
+ puser_data,
+ nullptr, nullptr, rctx);
+}
+
+void RGWDataAccess::Object::set_policy(const RGWAccessControlPolicy& policy)
+{
+ policy.encode(aclbl.emplace());
+}
+
--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab ft=cpp
+
+#pragma once
+
+#include <string>
+#include "include/types.h"
+#include "common/ceph_time.h"
+#include "rgw_common.h"
+#include "rgw_sal_fwd.h"
+
+class RGWDataAccess
+{
+ rgw::sal::Driver* driver;
+
+public:
+ RGWDataAccess(rgw::sal::Driver* _driver);
+
+ class Object;
+ class Bucket;
+
+ using BucketRef = std::shared_ptr<Bucket>;
+ using ObjectRef = std::shared_ptr<Object>;
+
+ class Bucket : public std::enable_shared_from_this<Bucket> {
+ friend class RGWDataAccess;
+ friend class Object;
+
+ RGWDataAccess *sd{nullptr};
+ RGWBucketInfo bucket_info;
+ std::string tenant;
+ std::string name;
+ std::string bucket_id;
+ ceph::real_time mtime;
+ std::map<std::string, bufferlist> attrs;
+
+ RGWAccessControlPolicy policy;
+ int finish_init();
+
+ Bucket(RGWDataAccess *_sd,
+ const std::string& _tenant,
+ const std::string& _name,
+ const std::string& _bucket_id) : sd(_sd),
+ tenant(_tenant),
+ name(_name),
+ bucket_id(_bucket_id) {}
+ Bucket(RGWDataAccess *_sd) : sd(_sd) {}
+ int init(const DoutPrefixProvider *dpp, optional_yield y);
+ int init(const RGWBucketInfo& _bucket_info, const std::map<std::string, bufferlist>& _attrs);
+ public:
+ int get_object(const rgw_obj_key& key,
+ ObjectRef *obj);
+
+ };
+
+
+ class Object {
+ RGWDataAccess *sd{nullptr};
+ BucketRef bucket;
+ rgw_obj_key key;
+
+ ceph::real_time mtime;
+ std::string etag;
+ uint64_t olh_epoch{0};
+ ceph::real_time delete_at;
+ std::optional<std::string> user_data;
+
+ std::optional<bufferlist> aclbl;
+
+ Object(RGWDataAccess *_sd,
+ BucketRef&& _bucket,
+ const rgw_obj_key& _key) : sd(_sd),
+ bucket(_bucket),
+ key(_key) {}
+ public:
+ int put(bufferlist& data, std::map<std::string, bufferlist>& attrs, const DoutPrefixProvider *dpp, optional_yield y); /* might modify attrs */
+
+ void set_mtime(const ceph::real_time& _mtime) {
+ mtime = _mtime;
+ }
+
+ void set_etag(const std::string& _etag) {
+ etag = _etag;
+ }
+
+ void set_olh_epoch(uint64_t epoch) {
+ olh_epoch = epoch;
+ }
+
+ void set_delete_at(ceph::real_time _delete_at) {
+ delete_at = _delete_at;
+ }
+
+ void set_user_data(const std::string& _user_data) {
+ user_data = _user_data;
+ }
+
+ void set_policy(const RGWAccessControlPolicy& policy);
+
+ friend class Bucket;
+ };
+
+ int get_bucket(const DoutPrefixProvider *dpp,
+ const std::string& tenant,
+ const std::string name,
+ const std::string bucket_id,
+ BucketRef *bucket,
+ optional_yield y) {
+ bucket->reset(new Bucket(this, tenant, name, bucket_id));
+ return (*bucket)->init(dpp, y);
+ }
+
+ int get_bucket(const RGWBucketInfo& bucket_info,
+ const std::map<std::string, bufferlist>& attrs,
+ BucketRef *bucket) {
+ bucket->reset(new Bucket(this));
+ return (*bucket)->init(bucket_info, attrs);
+ }
+ friend class Bucket;
+ friend class Object;
+};
+
+using RGWDataAccessRef = std::shared_ptr<RGWDataAccess>;
+
projects / ceph.git / commitdiff