RepoMirrors/ceph
1
0
Fork 0
mirror of https://github.com/ceph/ceph synced 2025-02-21 18:17:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

rgw: sync: override dest object owner if needed

Browse Source 
Signed-off-by: Yehuda Sadeh <yehuda@redhat.com>
This commit is contained in:
Yehuda Sadeh 2019-11-25 16:17:24 -08:00
parent a4550801cb
commit a8ded23d8e
6 changed files with 103 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
src/rgw/rgw_admin.cc
View File

@ -3108,7 +3108,7 @@ int main(int argc, const char **argv)
string quota_scope;
string object_version;
string placement_id;
string storage_class;
std::optional<string> opt_storage_class;
list<string> tags;
list<string> tags_add;
list<string> tags_rm;
@ -3212,6 +3212,7 @@ int main(int argc, const char **argv)
std::optional<int> opt_priority;
std::optional<string> opt_mode;
std::optional<rgw_user> opt_dest_owner;
rgw::notify::EventTypeList event_types;
@ -3474,7 +3475,7 @@ int main(int argc, const char **argv)
} else if (ceph_argparse_witharg(args, i, &val, "--placement-id", (char*)NULL)) {
placement_id = val;
} else if (ceph_argparse_witharg(args, i, &val, "--storage-class", (char*)NULL)) {
storage_class = val;
opt_storage_class = val;
} else if (ceph_argparse_witharg(args, i, &val, "--tags", (char*)NULL)) {
get_str_list(val, ",", tags);
} else if (ceph_argparse_witharg(args, i, &val, "--tags-add", (char*)NULL)) {
@ -3621,6 +3622,9 @@ int main(int argc, const char **argv)
opt_priority = atoi(val.c_str());
} else if (ceph_argparse_witharg(args, i, &val, "--mode", (char*)NULL)) {
opt_mode = val;
} else if (ceph_argparse_witharg(args, i, &val, "--dest-owner", (char*)NULL)) {
opt_dest_owner.emplace(val);
opt_dest_owner = val;
} else if (ceph_argparse_binary_flag(args, i, &detail, NULL, "--detail", (char*)NULL)) {
// do nothing
} else if (strncmp(*i, "-", 1) == 0) {
@ -4759,12 +4763,12 @@ int main(int argc, const char **argv)
rgw_placement_rule rule;
rule.from_str(placement_id);
if (!rule.storage_class.empty() && !storage_class.empty() &&
rule.storage_class != storage_class) {
if (!rule.storage_class.empty() && opt_storage_class &&
rule.storage_class != *opt_storage_class) {
cerr << "ERROR: provided contradicting storage class configuration" << std::endl;
return EINVAL;
} else if (rule.storage_class.empty()) {
rule.storage_class = storage_class;
rule.storage_class = opt_storage_class.value_or(string());
}
RGWZoneGroup zonegroup(zonegroup_id, zonegroup_name);
@ -4792,13 +4796,14 @@ int main(int argc, const char **argv)
}
target.storage_classes.insert(rule.get_storage_class());
} else if (opt_cmd == OPT::ZONEGROUP_PLACEMENT_RM) {
if (storage_class.empty()) {
if (!opt_storage_class ||
opt_storage_class->empty()) {
zonegroup.placement_targets.erase(placement_id);
} else {
auto iter = zonegroup.placement_targets.find(placement_id);
if (iter != zonegroup.placement_targets.end()) {
RGWZoneGroupPlacementTarget& info = zonegroup.placement_targets[placement_id];
info.storage_classes.erase(storage_class);
info.storage_classes.erase(*opt_storage_class);
}
}
} else if (opt_cmd == OPT::ZONEGROUP_PLACEMENT_DEFAULT) {
@ -5255,7 +5260,7 @@ int main(int argc, const char **argv)
return EINVAL;
}
storage_class = rgw_placement_rule::get_canonical_storage_class(storage_class);
string storage_class = rgw_placement_rule::get_canonical_storage_class(storage_class);
if (ptiter->second.storage_classes.find(storage_class) == ptiter->second.storage_classes.end()) {
cerr << "ERROR: storage class '" << storage_class << "' is not defined in zonegroup '" << placement_id << "' placement target" << std::endl;
return EINVAL;
@ -5305,13 +5310,14 @@ int main(int argc, const char **argv)
return ret;
}
} else if (opt_cmd == OPT::ZONE_PLACEMENT_RM) {
if (storage_class.empty()) {
if (!opt_storage_class ||
opt_storage_class->empty()) {
zone.placement_pools.erase(placement_id);
} else {
auto iter = zone.placement_pools.find(placement_id);
if (iter != zone.placement_pools.end()) {
RGWZonePlacementInfo& info = zone.placement_pools[placement_id];
info.storage_classes.remove_storage_class(storage_class);
info.storage_classes.remove_storage_class(*opt_storage_class);
}
}
}
@ -8291,6 +8297,12 @@ next:
pipe->params.source.filter.set_prefix(opt_prefix, !!opt_prefix_rm);
pipe->params.source.filter.set_tags(tags_add, tags_rm);
if (opt_dest_owner) {
pipe->params.dest.set_owner(*opt_dest_owner);
}
if (opt_storage_class) {
pipe->params.dest.set_storage_class(*opt_storage_class);
}
if (opt_priority) {
pipe->params.priority = *opt_priority;
}

15
src/rgw/rgw_data_sync.cc
View File

@ -2166,6 +2166,7 @@ public:
const RGWBucketInfo& dest_bucket_info,
std::optional<rgw_placement_rule> dest_placement_rule,
const map<string, bufferlist>& obj_attrs,
std::optional<rgw_user> *poverride_owner,
const rgw_placement_rule **prule) override;
};
@ -2174,6 +2175,7 @@ int RGWFetchObjFilter_Sync::filter(CephContext *cct,
const RGWBucketInfo& dest_bucket_info,
std::optional<rgw_placement_rule> dest_placement_rule,
const map<string, bufferlist>& obj_attrs,
std::optional<rgw_user> *poverride_owner,
const rgw_placement_rule **prule)
{
int abort_err = -ERR_PRECONDITION_FAILED;
@ -2200,6 +2202,18 @@ int RGWFetchObjFilter_Sync::filter(CephContext *cct,
std::optional<std::map<string, bufferlist> > new_attrs;
if (params.dest.acl_translation) {
rgw_user& acl_translation_owner = params.dest.acl_translation->owner;
if (!acl_translation_owner.empty()) {
if (params.mode == rgw_sync_pipe_params::MODE_USER &&
acl_translation_owner != dest_bucket_info.owner) {
ldout(cct, 0) << "ERROR: " << __func__ << ": acl translation was requested, but user (" << acl_translation_owner
<< ") is not dest bucket owner (" << dest_bucket_info.owner << ")" << dendl;
return -EPERM;
}
*poverride_owner = acl_translation_owner;
}
}
if (params.mode == rgw_sync_pipe_params::MODE_USER) {
if (!bucket_perms->verify_object_permission(obj_attrs, RGW_PERM_READ)) {
ldout(cct, 0) << "ERROR: " << __func__ << ": permission check failed: user not allowed to fetch object" << dendl;
@ -2220,6 +2234,7 @@ int RGWFetchObjFilter_Sync::filter(CephContext *cct,
dest_bucket_info,
dest_placement_rule,
obj_attrs,
poverride_owner,
prule);
}

6
src/rgw/rgw_putobj_processor.h
View File

@ -122,7 +122,7 @@ class ManifestObjectProcessor : public HeadObjectProcessor,
rgw::sal::RGWRadosStore *const store;
const RGWBucketInfo& bucket_info;
rgw_placement_rule tail_placement_rule;
const rgw_user& owner;
rgw_user owner;
RGWObjectCtx& obj_ctx;
rgw_obj head_obj;
@ -154,6 +154,10 @@ class ManifestObjectProcessor : public HeadObjectProcessor,
}
}
void set_owner(const rgw_user& _owner) {
owner = _owner;
}
void set_tail_placement(const rgw_placement_rule& tpr) {
tail_placement_rule = tpr;
}

50
src/rgw/rgw_rados.cc
View File

@ -3224,7 +3224,7 @@ class RGWRadosPutObj : public RGWHTTPStreamRWRequest::ReceiveCB
map<string, bufferlist> src_attrs;
uint64_t ofs{0};
uint64_t lofs{0}; /* logical ofs */
std::function<int(const map<string, bufferlist>&)> attrs_handler;
std::function<int(map<string, bufferlist>&)> attrs_handler;
public:
RGWRadosPutObj(CephContext* cct,
CompressorRef& plugin,
@ -3232,7 +3232,7 @@ public:
rgw::putobj::ObjectProcessor *p,
void (*_progress_cb)(off_t, void *),
void *_progress_data,
std::function<int(const map<string, bufferlist>&)> _attrs_handler) :
std::function<int(map<string, bufferlist>&)> _attrs_handler) :
cct(cct),
filter(p),
compressor(compressor),
@ -3621,6 +3621,7 @@ int RGWFetchObjFilter_Default::filter(CephContext *cct,
const RGWBucketInfo& dest_bucket_info,
std::optional<rgw_placement_rule> dest_placement_rule,
const map<string, bufferlist>& obj_attrs,
std::optional<rgw_user> *poverride_owner,
const rgw_placement_rule **prule)
{
const rgw_placement_rule *ptail_rule = (dest_placement_rule ? &(*dest_placement_rule) : nullptr);
@ -3714,15 +3715,19 @@ int RGWRados::fetch_remote_obj(RGWObjectCtx& obj_ctx,
if (!filter) {
filter = &source_filter;
}
std::optional<rgw_user> override_owner;
RGWRadosPutObj cb(cct, plugin, compressor, &processor, progress_cb, progress_data,
[&](const map<string, bufferlist>& obj_attrs) {
[&](map<string, bufferlist>& obj_attrs) {
const rgw_placement_rule *ptail_rule;
int ret = filter->filter(cct,
src_obj.key,
dest_bucket_info,
dest_placement_rule,
obj_attrs,
&override_owner,
&ptail_rule);
if (ret < 0) {
ldout(cct, 5) << "Aborting fetch: source object filter returned ret=" << ret << dendl;
@ -3809,6 +3814,43 @@ int RGWRados::fetch_remote_obj(RGWObjectCtx& obj_ctx,
cb.get_attrs()[RGW_ATTR_COMPRESSION] = tmp;
}
if (override_owner) {
processor.set_owner(*override_owner);
auto& obj_attrs = cb.get_attrs();
RGWUserInfo owner_info;
if (ctl.user->get_info_by_uid(*override_owner, &owner_info, null_yield) < 0) {
ldout(cct, 10) << "owner info does not exist" << dendl;
return -EINVAL;
}
RGWAccessControlPolicy acl;
auto aiter = obj_attrs.find(RGW_ATTR_ACL);
if (aiter == obj_attrs.end()) {
ldout(cct, 0) << "WARNING: " << __func__ << "(): object doesn't have ACL attribute" << dendl;
} else {
auto iter = aiter->second.cbegin();
try {
acl.decode(iter);
} catch (buffer::error& err) {
ldout(cct, 0) << "ERROR: " << __func__ << "(): could not decode policy, caught buffer::error" << dendl;
return -EIO;
}
}
ACLOwner new_owner;
new_owner.set_id(*override_owner);
new_owner.set_name(owner_info.display_name);
acl.set_owner(new_owner);
bufferlist bl;
acl.encode(bl);
obj_attrs[RGW_ATTR_ACL] = std::move(bl);
}
if (source_zone.empty()) { /* need to preserve expiration if copy in the same zonegroup */
cb.get_attrs().erase(RGW_ATTR_DELETE_AT);
} else {

2
src/rgw/rgw_rados.h
View File

@ -211,6 +211,7 @@ public:
const RGWBucketInfo& dest_bucket_info,
std::optional<rgw_placement_rule> dest_placement_rule,
const map<string, bufferlist>& obj_attrs,
std::optional<rgw_user> *poverride_owner,
const rgw_placement_rule **prule) = 0;
};
@ -225,6 +226,7 @@ public:
const RGWBucketInfo& dest_bucket_info,
std::optional<rgw_placement_rule> dest_placement_rule,
const map<string, bufferlist>& obj_attrs,
std::optional<rgw_user> *poverride_owner,
const rgw_placement_rule **prule) override;
};

13
src/rgw/rgw_sync_policy.h
View File

@ -308,6 +308,19 @@ struct rgw_sync_pipe_dest_params {
DECODE_FINISH(bl);
}
void set_storage_class(const string& sc) {
storage_class = sc;
}
void set_owner(const rgw_user& owner) {
if (owner.empty()){
acl_translation.reset();
} else {
acl_translation.emplace();
acl_translation->owner = owner;
}
}
void dump(ceph::Formatter *f) const;
void decode_json(JSONObj *obj);
};