summaryrefslogtreecommitdiffstats
path: root/src/rgw/driver/rados/rgw_lc_tier.cc
diff options
context:
space:
mode:
Diffstat (limited to 'src/rgw/driver/rados/rgw_lc_tier.cc')
-rw-r--r--src/rgw/driver/rados/rgw_lc_tier.cc1310
1 files changed, 1310 insertions, 0 deletions
diff --git a/src/rgw/driver/rados/rgw_lc_tier.cc b/src/rgw/driver/rados/rgw_lc_tier.cc
new file mode 100644
index 000000000..c52acef65
--- /dev/null
+++ b/src/rgw/driver/rados/rgw_lc_tier.cc
@@ -0,0 +1,1310 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab ft=cpp
+
+#include <string.h>
+#include <iostream>
+#include <map>
+
+#include "common/Formatter.h"
+#include <common/errno.h>
+#include "rgw_lc.h"
+#include "rgw_lc_tier.h"
+#include "rgw_string.h"
+#include "rgw_zone.h"
+#include "rgw_common.h"
+#include "rgw_rest.h"
+#include "svc_zone.h"
+
+#include <boost/algorithm/string/split.hpp>
+#include <boost/algorithm/string.hpp>
+#include <boost/algorithm/string/predicate.hpp>
+
+#define dout_context g_ceph_context
+#define dout_subsys ceph_subsys_rgw
+
+using namespace std;
+
+struct rgw_lc_multipart_part_info {
+ int part_num{0};
+ uint64_t ofs{0};
+ uint64_t size{0};
+ std::string etag;
+};
+
+struct rgw_lc_obj_properties {
+ ceph::real_time mtime;
+ std::string etag;
+ uint64_t versioned_epoch{0};
+ std::map<std::string, RGWTierACLMapping>& target_acl_mappings;
+ std::string target_storage_class;
+
+ rgw_lc_obj_properties(ceph::real_time _mtime, std::string _etag,
+ uint64_t _versioned_epoch, std::map<std::string,
+ RGWTierACLMapping>& _t_acl_mappings,
+ std::string _t_storage_class) :
+ mtime(_mtime), etag(_etag),
+ versioned_epoch(_versioned_epoch),
+ target_acl_mappings(_t_acl_mappings),
+ target_storage_class(_t_storage_class) {}
+};
+
+struct rgw_lc_multipart_upload_info {
+ std::string upload_id;
+ uint64_t obj_size;
+ ceph::real_time mtime;
+ std::string etag;
+
+ void encode(bufferlist& bl) const {
+ ENCODE_START(1, 1, bl);
+ encode(upload_id, bl);
+ encode(obj_size, bl);
+ encode(mtime, bl);
+ encode(etag, bl);
+ ENCODE_FINISH(bl);
+ }
+
+ void decode(bufferlist::const_iterator& bl) {
+ DECODE_START(1, bl);
+ decode(upload_id, bl);
+ decode(obj_size, bl);
+ decode(mtime, bl);
+ decode(etag, bl);
+ DECODE_FINISH(bl);
+ }
+};
+WRITE_CLASS_ENCODER(rgw_lc_multipart_upload_info)
+
+static inline string get_key_instance(const rgw_obj_key& key)
+{
+ if (!key.instance.empty() &&
+ !key.have_null_instance()) {
+ return "-" + key.instance;
+ }
+ return "";
+}
+
+static inline string get_key_oid(const rgw_obj_key& key)
+{
+ string oid = key.name;
+ if (!key.instance.empty() &&
+ !key.have_null_instance()) {
+ oid += string("-") + key.instance;
+ }
+ return oid;
+}
+
+static inline string obj_to_aws_path(const rgw_obj& obj)
+{
+ string path = obj.bucket.name + "/" + get_key_oid(obj.key);
+ return path;
+}
+
+static int read_upload_status(const DoutPrefixProvider *dpp, rgw::sal::Driver *driver,
+ const rgw_raw_obj *status_obj, rgw_lc_multipart_upload_info *status)
+{
+ int ret = 0;
+ rgw::sal::RadosStore *rados = dynamic_cast<rgw::sal::RadosStore*>(driver);
+
+ if (!rados) {
+ ldpp_dout(dpp, 0) << "ERROR: Not a RadosStore. Cannot be transitioned to cloud." << dendl;
+ return -1;
+ }
+
+ auto& pool = status_obj->pool;
+ const auto oid = status_obj->oid;
+ auto sysobj = rados->svc()->sysobj;
+ bufferlist bl;
+
+ ret = rgw_get_system_obj(sysobj, pool, oid, bl, nullptr, nullptr,
+ null_yield, dpp);
+
+ if (ret < 0) {
+ return ret;
+ }
+
+ if (bl.length() > 0) {
+ try {
+ auto p = bl.cbegin();
+ status->decode(p);
+ } catch (buffer::error& e) {
+ ldpp_dout(dpp, 10) << "failed to decode status obj: "
+ << e.what() << dendl;
+ return -EIO;
+ }
+ } else {
+ return -EIO;
+ }
+
+ return 0;
+}
+
+static int put_upload_status(const DoutPrefixProvider *dpp, rgw::sal::Driver *driver,
+ const rgw_raw_obj *status_obj, rgw_lc_multipart_upload_info *status)
+{
+ int ret = 0;
+ rgw::sal::RadosStore *rados = dynamic_cast<rgw::sal::RadosStore*>(driver);
+
+ if (!rados) {
+ ldpp_dout(dpp, 0) << "ERROR: Not a RadosStore. Cannot be transitioned to cloud." << dendl;
+ return -1;
+ }
+
+ auto& pool = status_obj->pool;
+ const auto oid = status_obj->oid;
+ auto sysobj = rados->svc()->sysobj;
+ bufferlist bl;
+ status->encode(bl);
+
+ ret = rgw_put_system_obj(dpp, sysobj, pool, oid, bl, true, nullptr,
+ real_time{}, null_yield);
+
+ return ret;
+}
+
+static int delete_upload_status(const DoutPrefixProvider *dpp, rgw::sal::Driver *driver,
+ const rgw_raw_obj *status_obj)
+{
+ int ret = 0;
+ rgw::sal::RadosStore *rados = dynamic_cast<rgw::sal::RadosStore*>(driver);
+
+ if (!rados) {
+ ldpp_dout(dpp, 0) << "ERROR: Not a RadosStore. Cannot be transitioned to cloud." << dendl;
+ return -1;
+ }
+
+ auto& pool = status_obj->pool;
+ const auto oid = status_obj->oid;
+ auto sysobj = rados->svc()->sysobj;
+
+ ret = rgw_delete_system_obj(dpp, sysobj, pool, oid, nullptr, null_yield);
+
+ return ret;
+}
+
+static std::set<string> keep_headers = { "CONTENT_TYPE",
+ "CONTENT_ENCODING",
+ "CONTENT_DISPOSITION",
+ "CONTENT_LANGUAGE" };
+
+/*
+ * mapping between rgw object attrs and output http fields
+ *
+ static const struct rgw_http_attr base_rgw_to_http_attrs[] = {
+ { RGW_ATTR_CONTENT_LANG, "Content-Language" },
+ { RGW_ATTR_EXPIRES, "Expires" },
+ { RGW_ATTR_CACHE_CONTROL, "Cache-Control" },
+ { RGW_ATTR_CONTENT_DISP, "Content-Disposition" },
+ { RGW_ATTR_CONTENT_ENC, "Content-Encoding" },
+ { RGW_ATTR_USER_MANIFEST, "X-Object-Manifest" },
+ { RGW_ATTR_X_ROBOTS_TAG , "X-Robots-Tag" },
+ { RGW_ATTR_STORAGE_CLASS , "X-Amz-Storage-Class" },
+// RGW_ATTR_AMZ_WEBSITE_REDIRECT_LOCATION header depends on access mode:
+// S3 endpoint: x-amz-website-redirect-location
+// S3Website endpoint: Location
+{ RGW_ATTR_AMZ_WEBSITE_REDIRECT_LOCATION, "x-amz-website-redirect-location" },
+}; */
+
+static void init_headers(map<string, bufferlist>& attrs,
+ map<string, string>& headers)
+{
+ for (auto& kv : attrs) {
+ const char * name = kv.first.c_str();
+ const auto aiter = rgw_to_http_attrs.find(name);
+
+ if (aiter != std::end(rgw_to_http_attrs)) {
+ headers[aiter->second] = rgw_bl_str(kv.second);
+ } else if (strncmp(name, RGW_ATTR_META_PREFIX,
+ sizeof(RGW_ATTR_META_PREFIX)-1) == 0) {
+ name += sizeof(RGW_ATTR_META_PREFIX) - 1;
+ string sname(name);
+ string name_prefix = RGW_ATTR_META_PREFIX;
+ char full_name_buf[name_prefix.size() + sname.size() + 1];
+ snprintf(full_name_buf, sizeof(full_name_buf), "%.*s%.*s",
+ static_cast<int>(name_prefix.length()),
+ name_prefix.data(),
+ static_cast<int>(sname.length()),
+ sname.data());
+ headers[full_name_buf] = rgw_bl_str(kv.second);
+ } else if (strcmp(name,RGW_ATTR_CONTENT_TYPE) == 0) {
+ headers["CONTENT_TYPE"] = rgw_bl_str(kv.second);
+ }
+ }
+}
+
+/* Read object or just head from remote endpoint. For now initializes only headers,
+ * but can be extended to fetch etag, mtime etc if needed.
+ */
+static int cloud_tier_get_object(RGWLCCloudTierCtx& tier_ctx, bool head,
+ std::map<std::string, std::string>& headers) {
+ RGWRESTConn::get_obj_params req_params;
+ std::string target_obj_name;
+ int ret = 0;
+ rgw_lc_obj_properties obj_properties(tier_ctx.o.meta.mtime, tier_ctx.o.meta.etag,
+ tier_ctx.o.versioned_epoch, tier_ctx.acl_mappings,
+ tier_ctx.target_storage_class);
+ std::string etag;
+ RGWRESTStreamRWRequest *in_req;
+
+ rgw_bucket dest_bucket;
+ dest_bucket.name = tier_ctx.target_bucket_name;
+ target_obj_name = tier_ctx.bucket_info.bucket.name + "/" +
+ tier_ctx.obj->get_name();
+ if (!tier_ctx.o.is_current()) {
+ target_obj_name += get_key_instance(tier_ctx.obj->get_key());
+ }
+
+ rgw_obj dest_obj(dest_bucket, rgw_obj_key(target_obj_name));
+
+ /* init input connection */
+ req_params.get_op = !head;
+ req_params.prepend_metadata = true;
+ req_params.rgwx_stat = true;
+ req_params.sync_manifest = true;
+ req_params.skip_decrypt = true;
+
+ ret = tier_ctx.conn.get_obj(tier_ctx.dpp, dest_obj, req_params, true /* send */, &in_req);
+ if (ret < 0) {
+ ldpp_dout(tier_ctx.dpp, 0) << "ERROR: " << __func__ << "(): conn.get_obj() returned ret=" << ret << dendl;
+ return ret;
+ }
+
+ /* fetch headers */
+ ret = tier_ctx.conn.complete_request(in_req, nullptr, nullptr, nullptr, nullptr, &headers, null_yield);
+ if (ret < 0 && ret != -ENOENT) {
+ ldpp_dout(tier_ctx.dpp, 20) << "ERROR: " << __func__ << "(): conn.complete_request() returned ret=" << ret << dendl;
+ return ret;
+ }
+ return 0;
+}
+
+static bool is_already_tiered(const DoutPrefixProvider *dpp,
+ std::map<std::string, std::string>& headers,
+ ceph::real_time& mtime) {
+ char buf[32];
+ map<string, string> attrs = headers;
+
+ for (const auto& a : attrs) {
+ ldpp_dout(dpp, 20) << "GetCrf attr[" << a.first << "] = " << a.second <<dendl;
+ }
+ utime_t ut(mtime);
+ snprintf(buf, sizeof(buf), "%lld.%09lld",
+ (long long)ut.sec(),
+ (long long)ut.nsec());
+
+ string s = attrs["X_AMZ_META_RGWX_SOURCE_MTIME"];
+
+ if (s.empty())
+ s = attrs["x_amz_meta_rgwx_source_mtime"];
+
+ ldpp_dout(dpp, 20) << "is_already_tiered attrs[X_AMZ_META_RGWX_SOURCE_MTIME] = " << s <<dendl;
+ ldpp_dout(dpp, 20) << "is_already_tiered mtime buf = " << buf <<dendl;
+
+ if (!s.empty() && !strcmp(s.c_str(), buf)){
+ return 1;
+ }
+ return 0;
+}
+
+/* Read object locally & also initialize dest rest obj based on read attrs */
+class RGWLCStreamRead
+{
+ CephContext *cct;
+ const DoutPrefixProvider *dpp;
+ std::map<std::string, bufferlist> attrs;
+ uint64_t obj_size;
+ rgw::sal::Object *obj;
+ const real_time &mtime;
+
+ bool multipart{false};
+ uint64_t m_part_size{0};
+ off_t m_part_off{0};
+ off_t m_part_end{0};
+
+ std::unique_ptr<rgw::sal::Object::ReadOp> read_op;
+ off_t ofs{0};
+ off_t end{0};
+ rgw_rest_obj rest_obj;
+
+ int retcode{0};
+
+ public:
+ RGWLCStreamRead(CephContext *_cct, const DoutPrefixProvider *_dpp,
+ rgw::sal::Object *_obj, const real_time &_mtime) :
+ cct(_cct), dpp(_dpp), obj(_obj), mtime(_mtime),
+ read_op(obj->get_read_op()) {}
+
+ ~RGWLCStreamRead() {};
+ int set_range(off_t _ofs, off_t _end);
+ int get_range(off_t &_ofs, off_t &_end);
+ rgw_rest_obj& get_rest_obj();
+ void set_multipart(uint64_t part_size, off_t part_off, off_t part_end);
+ int init();
+ int init_rest_obj();
+ int read(off_t ofs, off_t end, RGWGetDataCB *out_cb);
+};
+
+/* Send PUT op to remote endpoint */
+class RGWLCCloudStreamPut
+{
+ const DoutPrefixProvider *dpp;
+ rgw_lc_obj_properties obj_properties;
+ RGWRESTConn& conn;
+ const rgw_obj& dest_obj;
+ std::string etag;
+ RGWRESTStreamS3PutObj *out_req{nullptr};
+
+ struct multipart_info {
+ bool is_multipart{false};
+ std::string upload_id;
+ int part_num{0};
+ uint64_t part_size;
+ } multipart;
+
+ int retcode;
+
+ public:
+ RGWLCCloudStreamPut(const DoutPrefixProvider *_dpp,
+ const rgw_lc_obj_properties& _obj_properties,
+ RGWRESTConn& _conn,
+ const rgw_obj& _dest_obj) :
+ dpp(_dpp), obj_properties(_obj_properties), conn(_conn), dest_obj(_dest_obj) {
+ }
+ int init();
+ static bool keep_attr(const std::string& h);
+ static void init_send_attrs(const DoutPrefixProvider *dpp, const rgw_rest_obj& rest_obj,
+ const rgw_lc_obj_properties& obj_properties,
+ std::map<std::string, std::string>& attrs);
+ void send_ready(const DoutPrefixProvider *dpp, const rgw_rest_obj& rest_obj);
+ void handle_headers(const std::map<std::string, std::string>& headers);
+ bool get_etag(std::string *petag);
+ void set_multipart(const std::string& upload_id, int part_num, uint64_t part_size);
+ int send();
+ RGWGetDataCB *get_cb();
+ int complete_request();
+};
+
+int RGWLCStreamRead::set_range(off_t _ofs, off_t _end) {
+ ofs = _ofs;
+ end = _end;
+
+ return 0;
+}
+
+int RGWLCStreamRead::get_range(off_t &_ofs, off_t &_end) {
+ _ofs = ofs;
+ _end = end;
+
+ return 0;
+}
+
+rgw_rest_obj& RGWLCStreamRead::get_rest_obj() {
+ return rest_obj;
+}
+
+void RGWLCStreamRead::set_multipart(uint64_t part_size, off_t part_off, off_t part_end) {
+ multipart = true;
+ m_part_size = part_size;
+ m_part_off = part_off;
+ m_part_end = part_end;
+}
+
+int RGWLCStreamRead::init() {
+ optional_yield y = null_yield;
+ real_time read_mtime;
+
+ read_op->params.lastmod = &read_mtime;
+
+ int ret = read_op->prepare(y, dpp);
+ if (ret < 0) {
+ ldpp_dout(dpp, 0) << "ERROR: fail to prepare read_op, ret = " << ret << dendl;
+ return ret;
+ }
+
+ if (read_mtime != mtime) {
+ /* raced */
+ return -ECANCELED;
+ }
+
+ attrs = obj->get_attrs();
+ obj_size = obj->get_obj_size();
+
+ ret = init_rest_obj();
+ if (ret < 0) {
+ ldpp_dout(dpp, 0) << "ERROR: fail to initialize rest_obj, ret = " << ret << dendl;
+ return ret;
+ }
+
+ if (!multipart) {
+ set_range(0, obj_size - 1);
+ } else {
+ set_range(m_part_off, m_part_end);
+ }
+ return 0;
+}
+
+int RGWLCStreamRead::init_rest_obj() {
+ /* Initialize rgw_rest_obj.
+ * Reference: do_decode_rest_obj
+ * Check how to copy headers content */
+ rest_obj.init(obj->get_key());
+
+ if (!multipart) {
+ rest_obj.content_len = obj_size;
+ } else {
+ rest_obj.content_len = m_part_size;
+ }
+
+ /* For mulitpart attrs are sent as part of InitMultipartCR itself */
+ if (multipart) {
+ return 0;
+ }
+
+ /*
+ * XXX: verify if its right way to copy attrs into rest obj
+ */
+ init_headers(attrs, rest_obj.attrs);
+
+ rest_obj.acls.set_ctx(cct);
+ const auto aiter = attrs.find(RGW_ATTR_ACL);
+ if (aiter != attrs.end()) {
+ bufferlist& bl = aiter->second;
+ auto bliter = bl.cbegin();
+ try {
+ rest_obj.acls.decode(bliter);
+ } catch (buffer::error& err) {
+ ldpp_dout(dpp, 0) << "ERROR: failed to decode policy off attrs" << dendl;
+ return -EIO;
+ }
+ } else {
+ ldpp_dout(dpp, 0) << "WARNING: acl attrs not provided" << dendl;
+ }
+ return 0;
+}
+
+int RGWLCStreamRead::read(off_t ofs, off_t end, RGWGetDataCB *out_cb) {
+ int ret = read_op->iterate(dpp, ofs, end, out_cb, null_yield);
+ return ret;
+}
+
+int RGWLCCloudStreamPut::init() {
+ /* init output connection */
+ if (multipart.is_multipart) {
+ char buf[32];
+ snprintf(buf, sizeof(buf), "%d", multipart.part_num);
+ rgw_http_param_pair params[] = { { "uploadId", multipart.upload_id.c_str() },
+ { "partNumber", buf },
+ { nullptr, nullptr } };
+ conn.put_obj_send_init(dest_obj, params, &out_req);
+ } else {
+ conn.put_obj_send_init(dest_obj, nullptr, &out_req);
+ }
+
+ return 0;
+}
+
+bool RGWLCCloudStreamPut::keep_attr(const string& h) {
+ return (keep_headers.find(h) != keep_headers.end());
+}
+
+void RGWLCCloudStreamPut::init_send_attrs(const DoutPrefixProvider *dpp,
+ const rgw_rest_obj& rest_obj,
+ const rgw_lc_obj_properties& obj_properties,
+ std::map<string, string>& attrs) {
+
+ map<string, RGWTierACLMapping>& acl_mappings(obj_properties.target_acl_mappings);
+ const std::string& target_storage_class = obj_properties.target_storage_class;
+
+ attrs.clear();
+
+ for (auto& hi : rest_obj.attrs) {
+ if (keep_attr(hi.first)) {
+ attrs.insert(hi);
+ } else {
+ std::string s1 = boost::algorithm::to_lower_copy(hi.first);
+ const char* k = std::strstr(s1.c_str(), "x-amz");
+ if (k) {
+ attrs[k] = hi.second;
+ }
+ }
+ }
+
+ const auto acl = rest_obj.acls.get_acl();
+
+ map<int, vector<string> > access_map;
+
+ if (!acl_mappings.empty()) {
+ for (auto& grant : acl.get_grant_map()) {
+ auto& orig_grantee = grant.first;
+ auto& perm = grant.second;
+
+ string grantee;
+
+ const auto& am = acl_mappings;
+
+ const auto iter = am.find(orig_grantee);
+ if (iter == am.end()) {
+ ldpp_dout(dpp, 20) << "acl_mappings: Could not find " << orig_grantee << " .. ignoring" << dendl;
+ continue;
+ }
+
+ grantee = iter->second.dest_id;
+
+ string type;
+
+ switch (iter->second.type) {
+ case ACL_TYPE_CANON_USER:
+ type = "id";
+ break;
+ case ACL_TYPE_EMAIL_USER:
+ type = "emailAddress";
+ break;
+ case ACL_TYPE_GROUP:
+ type = "uri";
+ break;
+ default:
+ continue;
+ }
+
+ string tv = type + "=" + grantee;
+
+ int flags = perm.get_permission().get_permissions();
+ if ((flags & RGW_PERM_FULL_CONTROL) == RGW_PERM_FULL_CONTROL) {
+ access_map[flags].push_back(tv);
+ continue;
+ }
+
+ for (int i = 1; i <= RGW_PERM_WRITE_ACP; i <<= 1) {
+ if (flags & i) {
+ access_map[i].push_back(tv);
+ }
+ }
+ }
+ }
+
+ for (const auto& aiter : access_map) {
+ int grant_type = aiter.first;
+
+ string header_str("x-amz-grant-");
+
+ switch (grant_type) {
+ case RGW_PERM_READ:
+ header_str.append("read");
+ break;
+ case RGW_PERM_WRITE:
+ header_str.append("write");
+ break;
+ case RGW_PERM_READ_ACP:
+ header_str.append("read-acp");
+ break;
+ case RGW_PERM_WRITE_ACP:
+ header_str.append("write-acp");
+ break;
+ case RGW_PERM_FULL_CONTROL:
+ header_str.append("full-control");
+ break;
+ }
+
+ string s;
+
+ for (const auto& viter : aiter.second) {
+ if (!s.empty()) {
+ s.append(", ");
+ }
+ s.append(viter);
+ }
+
+ ldpp_dout(dpp, 20) << "acl_mappings: set acl: " << header_str << "=" << s << dendl;
+
+ attrs[header_str] = s;
+ }
+
+ /* Copy target storage class */
+ if (!target_storage_class.empty()) {
+ attrs["x-amz-storage-class"] = target_storage_class;
+ } else {
+ attrs["x-amz-storage-class"] = "STANDARD";
+ }
+
+ /* New attribute to specify its transitioned from RGW */
+ attrs["x-amz-meta-rgwx-source"] = "rgw";
+ attrs["x-rgw-cloud"] = "true";
+ attrs["x-rgw-cloud-keep-attrs"] = "true";
+
+ char buf[32];
+ snprintf(buf, sizeof(buf), "%llu", (long long)obj_properties.versioned_epoch);
+ attrs["x-amz-meta-rgwx-versioned-epoch"] = buf;
+
+ utime_t ut(obj_properties.mtime);
+ snprintf(buf, sizeof(buf), "%lld.%09lld",
+ (long long)ut.sec(),
+ (long long)ut.nsec());
+
+ attrs["x-amz-meta-rgwx-source-mtime"] = buf;
+ attrs["x-amz-meta-rgwx-source-etag"] = obj_properties.etag;
+ attrs["x-amz-meta-rgwx-source-key"] = rest_obj.key.name;
+ if (!rest_obj.key.instance.empty()) {
+ attrs["x-amz-meta-rgwx-source-version-id"] = rest_obj.key.instance;
+ }
+ for (const auto& a : attrs) {
+ ldpp_dout(dpp, 30) << "init_send_attrs attr[" << a.first << "] = " << a.second <<dendl;
+ }
+}
+
+void RGWLCCloudStreamPut::send_ready(const DoutPrefixProvider *dpp, const rgw_rest_obj& rest_obj) {
+ auto r = static_cast<RGWRESTStreamS3PutObj *>(out_req);
+
+ std::map<std::string, std::string> new_attrs;
+ if (!multipart.is_multipart) {
+ init_send_attrs(dpp, rest_obj, obj_properties, new_attrs);
+ }
+
+ r->set_send_length(rest_obj.content_len);
+
+ RGWAccessControlPolicy policy;
+
+ r->send_ready(dpp, conn.get_key(), new_attrs, policy);
+}
+
+void RGWLCCloudStreamPut::handle_headers(const map<string, string>& headers) {
+ for (const auto& h : headers) {
+ if (h.first == "ETAG") {
+ etag = h.second;
+ }
+ }
+}
+
+bool RGWLCCloudStreamPut::get_etag(string *petag) {
+ if (etag.empty()) {
+ return false;
+ }
+ *petag = etag;
+ return true;
+}
+
+void RGWLCCloudStreamPut::set_multipart(const string& upload_id, int part_num, uint64_t part_size) {
+ multipart.is_multipart = true;
+ multipart.upload_id = upload_id;
+ multipart.part_num = part_num;
+ multipart.part_size = part_size;
+}
+
+int RGWLCCloudStreamPut::send() {
+ int ret = RGWHTTP::send(out_req);
+ return ret;
+}
+
+RGWGetDataCB *RGWLCCloudStreamPut::get_cb() {
+ return out_req->get_out_cb();
+}
+
+int RGWLCCloudStreamPut::complete_request() {
+ int ret = conn.complete_request(out_req, etag, &obj_properties.mtime, null_yield);
+ return ret;
+}
+
+/* Read local copy and write to Cloud endpoint */
+static int cloud_tier_transfer_object(const DoutPrefixProvider* dpp,
+ RGWLCStreamRead* readf, RGWLCCloudStreamPut* writef) {
+ std::string url;
+ bufferlist bl;
+ bool sent_attrs{false};
+ int ret{0};
+ off_t ofs;
+ off_t end;
+
+ ret = readf->init();
+ if (ret < 0) {
+ ldpp_dout(dpp, 0) << "ERROR: fail to initialize in_crf, ret = " << ret << dendl;
+ return ret;
+ }
+ readf->get_range(ofs, end);
+ rgw_rest_obj& rest_obj = readf->get_rest_obj();
+ if (!sent_attrs) {
+ ret = writef->init();
+ if (ret < 0) {
+ ldpp_dout(dpp, 0) << "ERROR: fail to initialize out_crf, ret = " << ret << dendl;
+ return ret;
+ }
+
+ writef->send_ready(dpp, rest_obj);
+ ret = writef->send();
+ if (ret < 0) {
+ return ret;
+ }
+ sent_attrs = true;
+ }
+
+ ret = readf->read(ofs, end, writef->get_cb());
+
+ if (ret < 0) {
+ ldpp_dout(dpp, 0) << "ERROR: fail to read from in_crf, ret = " << ret << dendl;
+ return ret;
+ }
+
+ ret = writef->complete_request();
+ if (ret < 0) {
+ ldpp_dout(dpp, 0) << "ERROR: fail to complete request, ret = " << ret << dendl;
+ return ret;
+ }
+
+ return 0;
+}
+
+static int cloud_tier_plain_transfer(RGWLCCloudTierCtx& tier_ctx) {
+ int ret;
+
+ rgw_lc_obj_properties obj_properties(tier_ctx.o.meta.mtime, tier_ctx.o.meta.etag,
+ tier_ctx.o.versioned_epoch, tier_ctx.acl_mappings,
+ tier_ctx.target_storage_class);
+ std::string target_obj_name;
+
+ rgw_bucket dest_bucket;
+ dest_bucket.name = tier_ctx.target_bucket_name;
+
+ target_obj_name = tier_ctx.bucket_info.bucket.name + "/" +
+ tier_ctx.obj->get_name();
+ if (!tier_ctx.o.is_current()) {
+ target_obj_name += get_key_instance(tier_ctx.obj->get_key());
+ }
+
+ rgw_obj dest_obj(dest_bucket, rgw_obj_key(target_obj_name));
+
+ tier_ctx.obj->set_atomic();
+
+ /* Prepare Read from source */
+ /* TODO: Define readf, writef as stack variables. For some reason,
+ * when used as stack variables (esp., readf), the transition seems to
+ * be taking lot of time eventually erroring out at times.
+ */
+ std::shared_ptr<RGWLCStreamRead> readf;
+ readf.reset(new RGWLCStreamRead(tier_ctx.cct, tier_ctx.dpp,
+ tier_ctx.obj, tier_ctx.o.meta.mtime));
+
+ std::shared_ptr<RGWLCCloudStreamPut> writef;
+ writef.reset(new RGWLCCloudStreamPut(tier_ctx.dpp, obj_properties, tier_ctx.conn,
+ dest_obj));
+
+ /* actual Read & Write */
+ ret = cloud_tier_transfer_object(tier_ctx.dpp, readf.get(), writef.get());
+
+ return ret;
+}
+
+static int cloud_tier_send_multipart_part(RGWLCCloudTierCtx& tier_ctx,
+ const std::string& upload_id,
+ const rgw_lc_multipart_part_info& part_info,
+ std::string *petag) {
+ int ret;
+
+ rgw_lc_obj_properties obj_properties(tier_ctx.o.meta.mtime, tier_ctx.o.meta.etag,
+ tier_ctx.o.versioned_epoch, tier_ctx.acl_mappings,
+ tier_ctx.target_storage_class);
+ std::string target_obj_name;
+ off_t end;
+
+ rgw_bucket dest_bucket;
+ dest_bucket.name = tier_ctx.target_bucket_name;
+
+ target_obj_name = tier_ctx.bucket_info.bucket.name + "/" +
+ tier_ctx.obj->get_name();
+ if (!tier_ctx.o.is_current()) {
+ target_obj_name += get_key_instance(tier_ctx.obj->get_key());
+ }
+
+ rgw_obj dest_obj(dest_bucket, rgw_obj_key(target_obj_name));
+
+ tier_ctx.obj->set_atomic();
+
+ /* TODO: Define readf, writef as stack variables. For some reason,
+ * when used as stack variables (esp., readf), the transition seems to
+ * be taking lot of time eventually erroring out at times. */
+ std::shared_ptr<RGWLCStreamRead> readf;
+ readf.reset(new RGWLCStreamRead(tier_ctx.cct, tier_ctx.dpp,
+ tier_ctx.obj, tier_ctx.o.meta.mtime));
+
+ std::shared_ptr<RGWLCCloudStreamPut> writef;
+ writef.reset(new RGWLCCloudStreamPut(tier_ctx.dpp, obj_properties, tier_ctx.conn,
+ dest_obj));
+
+ /* Prepare Read from source */
+ end = part_info.ofs + part_info.size - 1;
+ readf->set_multipart(part_info.size, part_info.ofs, end);
+
+ /* Prepare write */
+ writef->set_multipart(upload_id, part_info.part_num, part_info.size);
+
+ /* actual Read & Write */
+ ret = cloud_tier_transfer_object(tier_ctx.dpp, readf.get(), writef.get());
+ if (ret < 0) {
+ return ret;
+ }
+
+ if (!(writef->get_etag(petag))) {
+ ldpp_dout(tier_ctx.dpp, 0) << "ERROR: failed to get etag from PUT request" << dendl;
+ return -EIO;
+ }
+
+ return 0;
+}
+
+static int cloud_tier_abort_multipart(const DoutPrefixProvider *dpp,
+ RGWRESTConn& dest_conn, const rgw_obj& dest_obj,
+ const std::string& upload_id) {
+ int ret;
+ bufferlist out_bl;
+ bufferlist bl;
+ rgw_http_param_pair params[] = { { "uploadId", upload_id.c_str() }, {nullptr, nullptr} };
+
+ string resource = obj_to_aws_path(dest_obj);
+ ret = dest_conn.send_resource(dpp, "DELETE", resource, params, nullptr,
+ out_bl, &bl, nullptr, null_yield);
+
+
+ if (ret < 0) {
+ ldpp_dout(dpp, 0) << "ERROR: failed to abort multipart upload for dest object=" << dest_obj << " (ret=" << ret << ")" << dendl;
+ return ret;
+ }
+
+ return 0;
+}
+
+static int cloud_tier_init_multipart(const DoutPrefixProvider *dpp,
+ RGWRESTConn& dest_conn, const rgw_obj& dest_obj,
+ uint64_t obj_size, std::map<std::string, std::string>& attrs,
+ std::string& upload_id) {
+ bufferlist out_bl;
+ bufferlist bl;
+
+ struct InitMultipartResult {
+ std::string bucket;
+ std::string key;
+ std::string upload_id;
+
+ void decode_xml(XMLObj *obj) {
+ RGWXMLDecoder::decode_xml("Bucket", bucket, obj);
+ RGWXMLDecoder::decode_xml("Key", key, obj);
+ RGWXMLDecoder::decode_xml("UploadId", upload_id, obj);
+ }
+ } result;
+
+ int ret;
+ rgw_http_param_pair params[] = { { "uploads", nullptr }, {nullptr, nullptr} };
+
+ string resource = obj_to_aws_path(dest_obj);
+
+ ret = dest_conn.send_resource(dpp, "POST", resource, params, &attrs,
+ out_bl, &bl, nullptr, null_yield);
+
+ if (ret < 0) {
+ ldpp_dout(dpp, 0) << "ERROR: failed to initialize multipart upload for dest object=" << dest_obj << dendl;
+ return ret;
+ }
+ /*
+ * If one of the following fails we cannot abort upload, as we cannot
+ * extract the upload id. If one of these fail it's very likely that that's
+ * the least of our problem.
+ */
+ RGWXMLDecoder::XMLParser parser;
+ if (!parser.init()) {
+ ldpp_dout(dpp, 0) << "ERROR: failed to initialize xml parser for parsing multipart init response from server" << dendl;
+ return -EIO;
+ }
+
+ if (!parser.parse(out_bl.c_str(), out_bl.length(), 1)) {
+ string str(out_bl.c_str(), out_bl.length());
+ ldpp_dout(dpp, 5) << "ERROR: failed to parse xml initmultipart: " << str << dendl;
+ return -EIO;
+ }
+
+ try {
+ RGWXMLDecoder::decode_xml("InitiateMultipartUploadResult", result, &parser, true);
+ } catch (RGWXMLDecoder::err& err) {
+ string str(out_bl.c_str(), out_bl.length());
+ ldpp_dout(dpp, 5) << "ERROR: unexpected xml: " << str << dendl;
+ return -EIO;
+ }
+
+ ldpp_dout(dpp, 20) << "init multipart result: bucket=" << result.bucket << " key=" << result.key << " upload_id=" << result.upload_id << dendl;
+
+ upload_id = result.upload_id;
+
+ return 0;
+}
+
+static int cloud_tier_complete_multipart(const DoutPrefixProvider *dpp,
+ RGWRESTConn& dest_conn, const rgw_obj& dest_obj,
+ std::string& upload_id,
+ const std::map<int, rgw_lc_multipart_part_info>& parts) {
+ rgw_http_param_pair params[] = { { "uploadId", upload_id.c_str() }, {nullptr, nullptr} };
+
+ stringstream ss;
+ XMLFormatter formatter;
+ int ret;
+
+ bufferlist bl, out_bl;
+ string resource = obj_to_aws_path(dest_obj);
+
+ struct CompleteMultipartReq {
+ std::map<int, rgw_lc_multipart_part_info> parts;
+
+ explicit CompleteMultipartReq(const std::map<int, rgw_lc_multipart_part_info>& _parts) : parts(_parts) {}
+
+ void dump_xml(Formatter *f) const {
+ for (const auto& p : parts) {
+ f->open_object_section("Part");
+ encode_xml("PartNumber", p.first, f);
+ encode_xml("ETag", p.second.etag, f);
+ f->close_section();
+ };
+ }
+ } req_enc(parts);
+
+ struct CompleteMultipartResult {
+ std::string location;
+ std::string bucket;
+ std::string key;
+ std::string etag;
+
+ void decode_xml(XMLObj *obj) {
+ RGWXMLDecoder::decode_xml("Location", bucket, obj);
+ RGWXMLDecoder::decode_xml("Bucket", bucket, obj);
+ RGWXMLDecoder::decode_xml("Key", key, obj);
+ RGWXMLDecoder::decode_xml("ETag", etag, obj);
+ }
+ } result;
+
+ encode_xml("CompleteMultipartUpload", req_enc, &formatter);
+
+ formatter.flush(ss);
+ bl.append(ss.str());
+
+ ret = dest_conn.send_resource(dpp, "POST", resource, params, nullptr,
+ out_bl, &bl, nullptr, null_yield);
+
+
+ if (ret < 0) {
+ ldpp_dout(dpp, 0) << "ERROR: failed to complete multipart upload for dest object=" << dest_obj << dendl;
+ return ret;
+ }
+ /*
+ * If one of the following fails we cannot abort upload, as we cannot
+ * extract the upload id. If one of these fail it's very likely that that's
+ * the least of our problem.
+ */
+ RGWXMLDecoder::XMLParser parser;
+ if (!parser.init()) {
+ ldpp_dout(dpp, 0) << "ERROR: failed to initialize xml parser for parsing multipart init response from server" << dendl;
+ return -EIO;
+ }
+
+ if (!parser.parse(out_bl.c_str(), out_bl.length(), 1)) {
+ string str(out_bl.c_str(), out_bl.length());
+ ldpp_dout(dpp, 5) << "ERROR: failed to parse xml Completemultipart: " << str << dendl;
+ return -EIO;
+ }
+
+ try {
+ RGWXMLDecoder::decode_xml("CompleteMultipartUploadResult", result, &parser, true);
+ } catch (RGWXMLDecoder::err& err) {
+ string str(out_bl.c_str(), out_bl.length());
+ ldpp_dout(dpp, 5) << "ERROR: unexpected xml: " << str << dendl;
+ return -EIO;
+ }
+
+ ldpp_dout(dpp, 20) << "complete multipart result: location=" << result.location << " bucket=" << result.bucket << " key=" << result.key << " etag=" << result.etag << dendl;
+
+ return ret;
+}
+
+static int cloud_tier_abort_multipart_upload(RGWLCCloudTierCtx& tier_ctx,
+ const rgw_obj& dest_obj, const rgw_raw_obj& status_obj,
+ const std::string& upload_id) {
+ int ret;
+
+ ret = cloud_tier_abort_multipart(tier_ctx.dpp, tier_ctx.conn, dest_obj, upload_id);
+
+ if (ret < 0) {
+ ldpp_dout(tier_ctx.dpp, 0) << "ERROR: failed to abort multipart upload dest obj=" << dest_obj << " upload_id=" << upload_id << " ret=" << ret << dendl;
+ /* ignore error, best effort */
+ }
+ /* remove status obj */
+ ret = delete_upload_status(tier_ctx.dpp, tier_ctx.driver, &status_obj);
+ if (ret < 0) {
+ ldpp_dout(tier_ctx.dpp, 0) << "ERROR: failed to remove sync status obj obj=" << status_obj << " ret=" << ret << dendl;
+ // ignore error, best effort
+ }
+ return 0;
+}
+
+static int cloud_tier_multipart_transfer(RGWLCCloudTierCtx& tier_ctx) {
+ rgw_obj src_obj;
+ rgw_obj dest_obj;
+
+ uint64_t obj_size;
+ std::string src_etag;
+ rgw_rest_obj rest_obj;
+
+ rgw_lc_multipart_upload_info status;
+
+ std::map<std::string, std::string> new_attrs;
+
+ rgw_raw_obj status_obj;
+
+ RGWBucketInfo b;
+ std::string target_obj_name;
+ rgw_bucket target_bucket;
+
+ int ret;
+
+ rgw_lc_obj_properties obj_properties(tier_ctx.o.meta.mtime, tier_ctx.o.meta.etag,
+ tier_ctx.o.versioned_epoch, tier_ctx.acl_mappings,
+ tier_ctx.target_storage_class);
+
+ uint32_t part_size{0};
+ uint32_t num_parts{0};
+
+ int cur_part{0};
+ uint64_t cur_ofs{0};
+ std::map<int, rgw_lc_multipart_part_info> parts;
+
+ obj_size = tier_ctx.o.meta.size;
+
+ target_bucket.name = tier_ctx.target_bucket_name;
+
+ target_obj_name = tier_ctx.bucket_info.bucket.name + "/" +
+ tier_ctx.obj->get_name();
+ if (!tier_ctx.o.is_current()) {
+ target_obj_name += get_key_instance(tier_ctx.obj->get_key());
+ }
+ dest_obj.init(target_bucket, target_obj_name);
+
+ rgw_pool pool = static_cast<rgw::sal::RadosStore*>(tier_ctx.driver)->svc()->zone->get_zone_params().log_pool;
+ status_obj = rgw_raw_obj(pool, "lc_multipart_" + tier_ctx.obj->get_oid());
+
+ ret = read_upload_status(tier_ctx.dpp, tier_ctx.driver, &status_obj, &status);
+
+ if (ret < 0 && ret != -ENOENT) {
+ ldpp_dout(tier_ctx.dpp, 0) << "ERROR: failed to read sync status of object " << src_obj << " ret=" << ret << dendl;
+ return ret;
+ }
+
+ if (ret >= 0) {
+ // check here that mtime and size did not change
+ if (status.mtime != obj_properties.mtime || status.obj_size != obj_size ||
+ status.etag != obj_properties.etag) {
+ cloud_tier_abort_multipart_upload(tier_ctx, dest_obj, status_obj, status.upload_id);
+ ret = -ENOENT;
+ }
+ }
+
+ if (ret == -ENOENT) {
+ RGWLCStreamRead readf(tier_ctx.cct, tier_ctx.dpp, tier_ctx.obj, tier_ctx.o.meta.mtime);
+
+ readf.init();
+
+ rest_obj = readf.get_rest_obj();
+
+ RGWLCCloudStreamPut::init_send_attrs(tier_ctx.dpp, rest_obj, obj_properties, new_attrs);
+
+ ret = cloud_tier_init_multipart(tier_ctx.dpp, tier_ctx.conn, dest_obj, obj_size, new_attrs, status.upload_id);
+ if (ret < 0) {
+ return ret;
+ }
+
+ status.obj_size = obj_size;
+ status.mtime = obj_properties.mtime;
+ status.etag = obj_properties.etag;
+
+ ret = put_upload_status(tier_ctx.dpp, tier_ctx.driver, &status_obj, &status);
+
+ if (ret < 0) {
+ ldpp_dout(tier_ctx.dpp, 0) << "ERROR: failed to driver multipart upload state, ret=" << ret << dendl;
+ // continue with upload anyway
+ }
+
+#define MULTIPART_MAX_PARTS 10000
+#define MULTIPART_MAX_PARTS 10000
+ uint64_t min_part_size = obj_size / MULTIPART_MAX_PARTS;
+ uint64_t min_conf_size = tier_ctx.multipart_min_part_size;
+
+ if (min_conf_size < MULTIPART_MIN_POSSIBLE_PART_SIZE) {
+ min_conf_size = MULTIPART_MIN_POSSIBLE_PART_SIZE;
+ }
+
+ part_size = std::max(min_conf_size, min_part_size);
+ num_parts = (obj_size + part_size - 1) / part_size;
+ cur_part = 1;
+ cur_ofs = 0;
+ }
+
+ for (; (uint32_t)cur_part <= num_parts; ++cur_part) {
+ ldpp_dout(tier_ctx.dpp, 20) << "cur_part = "<< cur_part << ", info.ofs = " << cur_ofs << ", info.size = " << part_size << ", obj size = " << obj_size<< ", num_parts:" << num_parts << dendl;
+ rgw_lc_multipart_part_info& cur_part_info = parts[cur_part];
+ cur_part_info.part_num = cur_part;
+ cur_part_info.ofs = cur_ofs;
+ cur_part_info.size = std::min((uint64_t)part_size, obj_size - cur_ofs);
+
+ cur_ofs += cur_part_info.size;
+
+ ret = cloud_tier_send_multipart_part(tier_ctx,
+ status.upload_id,
+ cur_part_info,
+ &cur_part_info.etag);
+
+ if (ret < 0) {
+ ldpp_dout(tier_ctx.dpp, 0) << "ERROR: failed to send multipart part of obj=" << tier_ctx.obj << ", sync via multipart upload, upload_id=" << status.upload_id << " part number " << cur_part << " (error: " << cpp_strerror(-ret) << ")" << dendl;
+ cloud_tier_abort_multipart_upload(tier_ctx, dest_obj, status_obj, status.upload_id);
+ return ret;
+ }
+
+ }
+
+ ret = cloud_tier_complete_multipart(tier_ctx.dpp, tier_ctx.conn, dest_obj, status.upload_id, parts);
+ if (ret < 0) {
+ ldpp_dout(tier_ctx.dpp, 0) << "ERROR: failed to complete multipart upload of obj=" << tier_ctx.obj << " (error: " << cpp_strerror(-ret) << ")" << dendl;
+ cloud_tier_abort_multipart_upload(tier_ctx, dest_obj, status_obj, status.upload_id);
+ return ret;
+ }
+
+ /* remove status obj */
+ ret = delete_upload_status(tier_ctx.dpp, tier_ctx.driver, &status_obj);
+ if (ret < 0) {
+ ldpp_dout(tier_ctx.dpp, 0) << "ERROR: failed to abort multipart upload obj=" << tier_ctx.obj << " upload_id=" << status.upload_id << " part number " << cur_part << " (" << cpp_strerror(-ret) << ")" << dendl;
+ // ignore error, best effort
+ }
+ return 0;
+}
+
+/* Check if object has already been transitioned */
+static int cloud_tier_check_object(RGWLCCloudTierCtx& tier_ctx, bool& already_tiered) {
+ int ret;
+ std::map<std::string, std::string> headers;
+
+ /* Fetch Head object */
+ ret = cloud_tier_get_object(tier_ctx, true, headers);
+
+ if (ret < 0) {
+ ldpp_dout(tier_ctx.dpp, 0) << "ERROR: failed to fetch HEAD from cloud for obj=" << tier_ctx.obj << " , ret = " << ret << dendl;
+ return ret;
+ }
+
+ already_tiered = is_already_tiered(tier_ctx.dpp, headers, tier_ctx.o.meta.mtime);
+
+ if (already_tiered) {
+ ldpp_dout(tier_ctx.dpp, 20) << "is_already_tiered true" << dendl;
+ } else {
+ ldpp_dout(tier_ctx.dpp, 20) << "is_already_tiered false..going with out_crf writing" << dendl;
+ }
+
+ return ret;
+}
+
+static int cloud_tier_create_bucket(RGWLCCloudTierCtx& tier_ctx) {
+ bufferlist out_bl;
+ int ret = 0;
+ pair<string, string> key(tier_ctx.storage_class, tier_ctx.target_bucket_name);
+ struct CreateBucketResult {
+ std::string code;
+
+ void decode_xml(XMLObj *obj) {
+ RGWXMLDecoder::decode_xml("Code", code, obj);
+ }
+ } result;
+
+ ldpp_dout(tier_ctx.dpp, 30) << "Cloud_tier_ctx: creating bucket:" << tier_ctx.target_bucket_name << dendl;
+ bufferlist bl;
+ string resource = tier_ctx.target_bucket_name;
+
+ ret = tier_ctx.conn.send_resource(tier_ctx.dpp, "PUT", resource, nullptr, nullptr,
+ out_bl, &bl, nullptr, null_yield);
+
+ if (ret < 0 ) {
+ ldpp_dout(tier_ctx.dpp, 0) << "create target bucket : " << tier_ctx.target_bucket_name << " returned ret:" << ret << dendl;
+ }
+ if (out_bl.length() > 0) {
+ RGWXMLDecoder::XMLParser parser;
+ if (!parser.init()) {
+ ldpp_dout(tier_ctx.dpp, 0) << "ERROR: failed to initialize xml parser for parsing create_bucket response from server" << dendl;
+ return -EIO;
+ }
+
+ if (!parser.parse(out_bl.c_str(), out_bl.length(), 1)) {
+ string str(out_bl.c_str(), out_bl.length());
+ ldpp_dout(tier_ctx.dpp, 5) << "ERROR: failed to parse xml createbucket: " << str << dendl;
+ return -EIO;
+ }
+
+ try {
+ RGWXMLDecoder::decode_xml("Error", result, &parser, true);
+ } catch (RGWXMLDecoder::err& err) {
+ string str(out_bl.c_str(), out_bl.length());
+ ldpp_dout(tier_ctx.dpp, 5) << "ERROR: unexpected xml: " << str << dendl;
+ return -EIO;
+ }
+
+ if (result.code != "BucketAlreadyOwnedByYou" && result.code != "BucketAlreadyExists") {
+ ldpp_dout(tier_ctx.dpp, 0) << "ERROR: Creating target bucket failed with error: " << result.code << dendl;
+ return -EIO;
+ }
+ }
+
+ return 0;
+}
+
+int rgw_cloud_tier_transfer_object(RGWLCCloudTierCtx& tier_ctx, std::set<std::string>& cloud_targets) {
+ int ret = 0;
+
+ // check if target_path is already created
+ std::set<std::string>::iterator it;
+
+ it = cloud_targets.find(tier_ctx.target_bucket_name);
+ tier_ctx.target_bucket_created = (it != cloud_targets.end());
+
+ /* If run first time attempt to create the target bucket */
+ if (!tier_ctx.target_bucket_created) {
+ ret = cloud_tier_create_bucket(tier_ctx);
+
+ if (ret < 0) {
+ ldpp_dout(tier_ctx.dpp, 0) << "ERROR: failed to create target bucket on the cloud endpoint ret=" << ret << dendl;
+ return ret;
+ }
+ tier_ctx.target_bucket_created = true;
+ cloud_targets.insert(tier_ctx.target_bucket_name);
+ }
+
+ /* Since multiple zones may try to transition the same object to the cloud,
+ * verify if the object is already transitioned. And since its just a best
+ * effort, do not bail out in case of any errors.
+ */
+ bool already_tiered = false;
+ ret = cloud_tier_check_object(tier_ctx, already_tiered);
+
+ if (ret < 0) {
+ ldpp_dout(tier_ctx.dpp, 0) << "ERROR: failed to check object on the cloud endpoint ret=" << ret << dendl;
+ }
+
+ if (already_tiered) {
+ ldpp_dout(tier_ctx.dpp, 20) << "Object (" << tier_ctx.o.key << ") is already tiered" << dendl;
+ return 0;
+ }
+
+ uint64_t size = tier_ctx.o.meta.size;
+ uint64_t multipart_sync_threshold = tier_ctx.multipart_sync_threshold;
+
+ if (multipart_sync_threshold < MULTIPART_MIN_POSSIBLE_PART_SIZE) {
+ multipart_sync_threshold = MULTIPART_MIN_POSSIBLE_PART_SIZE;
+ }
+
+ if (size < multipart_sync_threshold) {
+ ret = cloud_tier_plain_transfer(tier_ctx);
+ } else {
+ tier_ctx.is_multipart_upload = true;
+ ret = cloud_tier_multipart_transfer(tier_ctx);
+ }
+
+ if (ret < 0) {
+ ldpp_dout(tier_ctx.dpp, 0) << "ERROR: failed to transition object ret=" << ret << dendl;
+ }
+
+ return ret;
+}
generated by cgit v1.2.3 (git 2.39.1) at 2024-07-17 18:09:01 +0000