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-admin: sync group pipe create and remove

Browse Source 
also rename sync group flow add to sync group flow create.

Signed-off-by: Yehuda Sadeh <yehuda@redhat.com>
This commit is contained in:
Yehuda Sadeh 2019-10-17 16:15:20 -07:00
parent e163cd0565
commit cb55d8ada1
5 changed files with 333 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

164
src/rgw/rgw_admin.cc
View File

@ -650,10 +650,13 @@ enum class OPT {
SYNC_ERROR_LIST,
SYNC_ERROR_TRIM,
SYNC_GROUP_CREATE,
SYNC_GROUP_MODIFY,
SYNC_GROUP_GET,
SYNC_GROUP_REMOVE,
SYNC_GROUP_FLOW_ADD,
SYNC_GROUP_FLOW_CREATE,
SYNC_GROUP_FLOW_REMOVE,
SYNC_GROUP_PIPE_CREATE,
SYNC_GROUP_PIPE_REMOVE,
SYNC_POLICY_GET,
BILOG_LIST,
BILOG_TRIM,
@ -850,10 +853,13 @@ static SimpleCmd::Commands all_cmds = {
{ "sync error trim", OPT::SYNC_ERROR_TRIM },
{ "sync policy get", OPT::SYNC_POLICY_GET },
{ "sync group create", OPT::SYNC_GROUP_CREATE },
{ "sync group modify", OPT::SYNC_GROUP_MODIFY },
{ "sync group get", OPT::SYNC_GROUP_GET },
{ "sync group remove", OPT::SYNC_GROUP_REMOVE },
{ "sync group flow add", OPT::SYNC_GROUP_FLOW_ADD },
{ "sync group flow create", OPT::SYNC_GROUP_FLOW_CREATE },
{ "sync group flow remove", OPT::SYNC_GROUP_FLOW_REMOVE },
{ "sync group pipe create", OPT::SYNC_GROUP_PIPE_CREATE },
{ "sync group pipe remove", OPT::SYNC_GROUP_PIPE_REMOVE },
{ "bilog list", OPT::BILOG_LIST },
{ "bilog trim", OPT::BILOG_TRIM },
{ "bilog status", OPT::BILOG_STATUS },
@ -2835,6 +2841,18 @@ int main(int argc, const char **argv)
std::optional<string> opt_flow_id;
std::optional<string> opt_source_zone;
std::optional<string> opt_dest_zone;
std::optional<vector<string> > opt_source_zones;
std::optional<vector<string> > opt_dest_zones;
std::optional<string> opt_pipe_id;
std::optional<string> opt_tenant;
std::optional<string> opt_bucket;
std::optional<string> opt_bucket_id;
std::optional<string> opt_source_tenant;
std::optional<string> opt_source_bucket;
std::optional<string> opt_source_bucket_id;
std::optional<string> opt_dest_tenant;
std::optional<string> opt_dest_bucket;
std::optional<string> opt_dest_bucket_id;
rgw::notify::EventTypeList event_types;
@ -2853,6 +2871,7 @@ int main(int argc, const char **argv)
new_user_id.from_str(val);
} else if (ceph_argparse_witharg(args, i, &val, "--tenant", (char*)NULL)) {
tenant = val;
opt_tenant = val;
} else if (ceph_argparse_witharg(args, i, &val, "--access-key", (char*)NULL)) {
access_key = val;
} else if (ceph_argparse_witharg(args, i, &val, "--subuser", (char*)NULL)) {
@ -2866,6 +2885,7 @@ int main(int argc, const char **argv)
display_name = val;
} else if (ceph_argparse_witharg(args, i, &val, "-b", "--bucket", (char*)NULL)) {
bucket_name = val;
opt_bucket = val;
} else if (ceph_argparse_witharg(args, i, &val, "-p", "--pool", (char*)NULL)) {
pool_name = val;
pool = rgw_pool(pool_name);
@ -2991,6 +3011,7 @@ int main(int argc, const char **argv)
set_temp_url_key = true;
} else if (ceph_argparse_witharg(args, i, &val, "--bucket-id", (char*)NULL)) {
bucket_id = val;
opt_bucket_id = val;
if (bucket_id.empty()) {
cerr << "bad bucket-id" << std::endl;
exit(1);
@ -3199,8 +3220,30 @@ int main(int argc, const char **argv)
vector<string> v;
get_str_vec(val, v);
opt_zones = std::move(v);
} else if (ceph_argparse_witharg(args, i, &val, "--source-zones", (char*)NULL)) {
vector<string> v;
get_str_vec(val, v);
opt_source_zones = std::move(v);
} else if (ceph_argparse_witharg(args, i, &val, "--dest-zones", (char*)NULL)) {
vector<string> v;
get_str_vec(val, v);
opt_dest_zones = std::move(v);
} else if (ceph_argparse_witharg(args, i, &val, "--flow-id", (char*)NULL)) {
opt_flow_id = val;
} else if (ceph_argparse_witharg(args, i, &val, "--pipe-id", (char*)NULL)) {
opt_pipe_id = val;
} else if (ceph_argparse_witharg(args, i, &val, "--source-tenant", (char*)NULL)) {
opt_source_tenant = val;
} else if (ceph_argparse_witharg(args, i, &val, "--source-bucket", (char*)NULL)) {
opt_source_bucket = val;
} else if (ceph_argparse_witharg(args, i, &val, "--source-bucket-id", (char*)NULL)) {
opt_source_bucket_id = val;
} else if (ceph_argparse_witharg(args, i, &val, "--dest-tenant", (char*)NULL)) {
opt_dest_tenant = val;
} else if (ceph_argparse_witharg(args, i, &val, "--dest-bucket", (char*)NULL)) {
opt_dest_bucket = val;
} else if (ceph_argparse_witharg(args, i, &val, "--dest-bucket-id", (char*)NULL)) {
opt_dest_bucket_id = val;
} else if (ceph_argparse_binary_flag(args, i, &detail, NULL, "--detail", (char*)NULL)) {
// do nothing
} else if (strncmp(*i, "-", 1) == 0) {
@ -7630,7 +7673,8 @@ next:
}
}
if (opt_cmd == OPT::SYNC_GROUP_CREATE) {
if (opt_cmd == OPT::SYNC_GROUP_CREATE ||
opt_cmd == OPT::SYNC_GROUP_MODIFY) {
CHECK_TRUE(require_opt(opt_group_id, "ERROR: --group-id not specified"), EINVAL);
CHECK_TRUE(require_opt(opt_status, "ERROR: --status is not specified (options: forbidden, enabled, activated)"), EINVAL);
@ -7641,6 +7685,14 @@ next:
return -ret;
}
if (opt_cmd == OPT::SYNC_GROUP_MODIFY) {
auto iter = zonegroup.sync_policy.groups.find(*opt_group_id);
if (iter != zonegroup.sync_policy.groups.end()) {
cerr << "ERROR: could not find group '" << *opt_group_id << "'" << std::endl;
return ENOENT;
}
}
auto& group = zonegroup.sync_policy.groups[*opt_group_id];
group.id = *opt_group_id;
@ -7714,7 +7766,7 @@ next:
formatter->flush(cout);
}
if (opt_cmd == OPT::SYNC_GROUP_FLOW_ADD) {
if (opt_cmd == OPT::SYNC_GROUP_FLOW_CREATE) {
CHECK_TRUE(require_opt(opt_group_id, "ERROR: --group-id not specified"), EINVAL);
CHECK_TRUE(require_opt(opt_flow_id, "ERROR: --flow-id not specified"), EINVAL);
CHECK_TRUE(require_opt(opt_flow_type,
@ -7806,6 +7858,110 @@ next:
show_result(zonegroup.sync_policy, formatter, cout);
}
if (opt_cmd == OPT::SYNC_GROUP_PIPE_CREATE) {
CHECK_TRUE(require_opt(opt_group_id, "ERROR: --group-id not specified"), EINVAL);
CHECK_TRUE(require_opt(opt_pipe_id, "ERROR: --pipe-id not specified"), EINVAL);
CHECK_TRUE(require_non_empty_opt(opt_source_zones, "ERROR: --source-zones not provided or is empty; should be list of zones or '*'"), EINVAL);
CHECK_TRUE(require_non_empty_opt(opt_dest_zones, "ERROR: --dest-zones not provided or is empty; should be list of zones or '*'"), EINVAL);
RGWZoneGroup zonegroup(zonegroup_id, zonegroup_name);
ret = zonegroup.init(g_ceph_context, store->svc()->sysobj);
if (ret < 0) {
cerr << "failed to init zonegroup: " << cpp_strerror(-ret) << std::endl;
return -ret;
}
auto iter = zonegroup.sync_policy.groups.find(*opt_group_id);
if (iter == zonegroup.sync_policy.groups.end()) {
cerr << "ERROR: could not find group '" << *opt_group_id << "'" << std::endl;
return ENOENT;
}
auto& group = iter->second;
rgw_sync_bucket_pipe *pipe;
group.find_pipe(*opt_pipe_id, true, &pipe);
pipe->source.add_zones(*opt_source_zones);
pipe->source.set_bucket(opt_source_tenant,
opt_source_bucket,
opt_source_bucket_id);
pipe->dest.add_zones(*opt_dest_zones);
pipe->dest.set_bucket(opt_dest_tenant,
opt_dest_bucket,
opt_dest_bucket_id);
ret = zonegroup.update();
if (ret < 0) {
cerr << "failed to update zonegroup: " << cpp_strerror(-ret) << std::endl;
return -ret;
}
show_result(zonegroup.sync_policy, formatter, cout);
}
if (opt_cmd == OPT::SYNC_GROUP_PIPE_REMOVE) {
CHECK_TRUE(require_opt(opt_group_id, "ERROR: --group-id not specified"), EINVAL);
CHECK_TRUE(require_opt(opt_pipe_id, "ERROR: --pipe-id not specified"), EINVAL);
RGWZoneGroup zonegroup(zonegroup_id, zonegroup_name);
ret = zonegroup.init(g_ceph_context, store->svc()->sysobj);
if (ret < 0) {
cerr << "failed to init zonegroup: " << cpp_strerror(-ret) << std::endl;
return -ret;
}
auto iter = zonegroup.sync_policy.groups.find(*opt_group_id);
if (iter == zonegroup.sync_policy.groups.end()) {
cerr << "ERROR: could not find group '" << *opt_group_id << "'" << std::endl;
return ENOENT;
}
auto& group = iter->second;
rgw_sync_bucket_pipe *pipe;
if (!group.find_pipe(*opt_pipe_id, false, &pipe)) {
cerr << "ERROR: could not find pipe '" << *opt_pipe_id << "'" << std::endl;
return ENOENT;
}
if (opt_source_zones) {
pipe->source.remove_zones(*opt_source_zones);
}
pipe->source.remove_bucket(opt_source_tenant,
opt_source_bucket,
opt_source_bucket_id);
if (opt_dest_zones) {
pipe->dest.remove_zones(*opt_dest_zones);
}
pipe->dest.remove_bucket(opt_dest_tenant,
opt_dest_bucket,
opt_dest_bucket_id);
if (!(opt_source_zones ||
opt_source_tenant ||
opt_source_bucket ||
opt_source_bucket_id ||
opt_dest_zones ||
opt_dest_tenant ||
opt_dest_bucket ||
opt_dest_bucket_id)) {
group.remove_pipe(*opt_pipe_id);
}
ret = zonegroup.update();
if (ret < 0) {
cerr << "failed to update zonegroup: " << cpp_strerror(-ret) << std::endl;
return -ret;
}
show_result(zonegroup.sync_policy, formatter, cout);
}
if (opt_cmd == OPT::SYNC_POLICY_GET) {
RGWZoneGroup zonegroup(zonegroup_id, zonegroup_name);
ret = zonegroup.init(g_ceph_context, store->svc()->sysobj);

6
src/rgw/rgw_bucket.cc
View File

@ -228,8 +228,10 @@ int rgw_bucket_parse_bucket_key(CephContext *cct, const string& key,
string err;
auto id = strict_strtol(shard.data(), 10, &err);
if (!err.empty()) {
ldout(cct, 0) << "ERROR: failed to parse bucket shard '"
<< instance.data() << "': " << err << dendl;
if (cct) {
ldout(cct, 0) << "ERROR: failed to parse bucket shard '"
<< instance.data() << "': " << err << dendl;
}
return -EINVAL;
}

113
src/rgw/rgw_bucket_sync.cc
View File

@ -10,6 +10,89 @@
#define dout_subsys ceph_subsys_rgw
void rgw_sync_bucket_entity::add_zones(const std::vector<string>& new_zones) {
for (auto& z : new_zones) {
if (z == "*") {
all_zones = true;
zones.reset();
return;
}
if (!zones) {
zones.emplace();
}
zones->insert(z);
}
}
void rgw_sync_bucket_entity::remove_zones(const std::vector<string>& rm_zones) {
all_zones = false;
if (!zones) {
return;
}
for (auto& z : rm_zones) {
zones->erase(z);
}
}
static void set_bucket_field(std::optional<string> source, string *field) {
if (!source) {
return;
}
if (source == "*") {
field->clear();
return;
}
*field = *source;
}
void rgw_sync_bucket_entity::set_bucket(std::optional<string> tenant,
std::optional<string> bucket_name,
std::optional<string> bucket_id)
{
if ((!bucket) && (tenant || bucket_name || bucket_id)) {
bucket.emplace();
}
set_bucket_field(tenant, &bucket->tenant);
set_bucket_field(bucket_name, &bucket->name);
set_bucket_field(bucket_id, &bucket->bucket_id);
if (bucket->tenant.empty() &&
bucket->name.empty() &&
bucket->bucket_id.empty()) {
bucket.reset();
}
}
void rgw_sync_bucket_entity::remove_bucket(std::optional<string> tenant,
std::optional<string> bucket_name,
std::optional<string> bucket_id)
{
if (!bucket) {
return;
}
if (tenant) {
bucket->tenant.clear();
}
if (bucket_name) {
bucket->name.clear();
}
if (bucket_id) {
bucket->bucket_id.clear();
}
if (bucket->tenant.empty() &&
bucket->name.empty() &&
bucket->bucket_id.empty()) {
bucket.reset();
}
}
bool rgw_sync_data_flow_group::find_symmetrical(const string& flow_id, bool create, rgw_sync_symmetric_group **flow_group)
{
if (!symmetrical) {
@ -117,6 +200,36 @@ void rgw_sync_data_flow_group::remove_directional(const string& source_zone, con
}
}
bool rgw_sync_policy_group::find_pipe(const string& pipe_id, bool create, rgw_sync_bucket_pipe **pipe)
{
for (auto& p : pipes) {
if (pipe_id == p.id) {
*pipe = &p;
return true;
}
}
if (!create) {
return false;
}
auto& p = pipes.emplace_back();
*pipe = &p;
p.id = pipe_id;
return true;
}
void rgw_sync_policy_group::remove_pipe(const string& pipe_id)
{
for (auto iter = pipes.begin(); iter != pipes.end(); ++iter) {
if (pipe_id == iter->id) {
pipes.erase(iter);
return;
}
}
}
static std::vector<rgw_sync_bucket_pipe> filter_relevant_pipes(const std::vector<rgw_sync_bucket_pipe>& pipes,
const string& source_zone,
const string& dest_zone)

36
src/rgw/rgw_json_enc.cc
View File

@ -867,24 +867,56 @@ void rgw_sync_symmetric_group::decode_json(JSONObj *obj)
void rgw_sync_bucket_entity::dump(Formatter *f) const
{
encode_json("bucket", bucket, f);
if (bucket) {
rgw_bucket b = *bucket;
if (b.name.empty()) {
b.name = "*";
}
encode_json("bucket", b.get_key(), f);
} else {
encode_json("bucket", "*", f);
}
encode_json("zones", zones, f);
}
void rgw_sync_bucket_entity::decode_json(JSONObj *obj)
{
JSONDecoder::decode_json("bucket", bucket, obj);
string s;
JSONDecoder::decode_json("bucket", s, obj);
if (s == "*") {
bucket.reset();
} else {
rgw_bucket b;
int ret = rgw_bucket_parse_bucket_key(nullptr, s, &b, nullptr);
if (ret < 0) {
bucket.reset();
} else {
if (b.tenant == "*") {
b.tenant.clear();
}
if (b.name == "*") {
b.name.clear();
}
if (b.bucket_id == "*") {
b.bucket_id.clear();
}
bucket = b;
}
}
JSONDecoder::decode_json("zones", zones, obj);
}
void rgw_sync_bucket_pipe::dump(Formatter *f) const
{
encode_json("id", id, f);
encode_json("source", source, f);
encode_json("dest", dest, f);
}
void rgw_sync_bucket_pipe::decode_json(JSONObj *obj)
{
JSONDecoder::decode_json("id", id, obj);
JSONDecoder::decode_json("source", source, obj);
JSONDecoder::decode_json("dest", dest, obj);
}

24
src/rgw/rgw_sync_policy.h
View File

@ -231,6 +231,8 @@ public:
std::optional<rgw_bucket> bucket; /* define specific bucket */
std::optional<std::set<string> > zones; /* define specific zones, if not set then all zones */
bool all_zones{false};
void encode(bufferlist& bl) const {
ENCODE_START(1, 1, bl);
encode(bucket, bl);
@ -262,9 +264,22 @@ public:
match_str(bucket->bucket_id, b->bucket_id));
}
void add_zones(const std::vector<string>& new_zones);
void remove_zones(const std::vector<string>& rm_zones);
void set_bucket(std::optional<string> tenant,
std::optional<string> bucket_name,
std::optional<string> bucket_id);
void remove_bucket(std::optional<string> tenant,
std::optional<string> bucket_name,
std::optional<string> bucket_id);
bool match_zone(const string& zone) const {
if (all_zones) {
return true;
}
if (!zones) { /* all zones */
return true;
return false;
}
return (zones->find(zone) != zones->end());
@ -273,7 +288,6 @@ public:
rgw_bucket get_bucket() const {
return bucket.value_or(rgw_bucket());
}
};
WRITE_CLASS_ENCODER(rgw_sync_bucket_entity)
@ -288,11 +302,13 @@ private:
}
public:
string id;
rgw_sync_bucket_entity source;
rgw_sync_bucket_entity dest;
void encode(bufferlist& bl) const {
ENCODE_START(1, 1, bl);
encode(id, bl);
encode(source, bl);
encode(dest, bl);
ENCODE_FINISH(bl);
@ -300,6 +316,7 @@ public:
void decode(bufferlist::const_iterator& bl) {
DECODE_START(1, bl);
decode(id, bl);
decode(source, bl);
decode(dest, bl);
DECODE_FINISH(bl);
@ -419,6 +436,9 @@ struct rgw_sync_policy_group {
return true;
}
bool find_pipe(const string& pipe_id, bool create, rgw_sync_bucket_pipe **pipe);
void remove_pipe(const string& pipe_id);
};
WRITE_CLASS_ENCODER(rgw_sync_policy_group)