rgw-admin: sync group pipe create and remove

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

Signed-off-by: Yehuda Sadeh <yehuda@redhat.com>

diff --git a/src/rgw/rgw_admin.cc b/src/rgw/rgw_admin.cc
index c7b71d94267a53f0eb1aa33ab7aece51ac1c8734..741e9203c2cc7fe16f0ba7d9a25829c9e1659d15 100644 (file)
--- a/src/rgw/rgw_admin.cc
+++ b/src/rgw/rgw_admin.cc
@@ -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);

diff --git a/src/rgw/rgw_bucket.cc b/src/rgw/rgw_bucket.cc
index 3ba6326e35b9168152d1a15ed26bc0aeede0bd5e..cbbf015ee1dad20bdcaaf4324703a3a6331df509 100644 (file)
--- a/src/rgw/rgw_bucket.cc
+++ b/src/rgw/rgw_bucket.cc
@@ -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;
   }
 

diff --git a/src/rgw/rgw_bucket_sync.cc b/src/rgw/rgw_bucket_sync.cc
index 89035df5c26d43d0b6c1f9d54d442293677538ae..42c01a2c02ea214a9620478df59157b02f9fe046 100644 (file)
--- a/src/rgw/rgw_bucket_sync.cc
+++ b/src/rgw/rgw_bucket_sync.cc
@@ -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)

diff --git a/src/rgw/rgw_json_enc.cc b/src/rgw/rgw_json_enc.cc
index 35bbbebad0f7c503f4885e0054ab899384d3c43a..c331bdf8f88050b99ea64a0b9c26ddbcca1822c0 100644 (file)
--- a/src/rgw/rgw_json_enc.cc
+++ b/src/rgw/rgw_json_enc.cc
@@ -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);
 }

diff --git a/src/rgw/rgw_sync_policy.h b/src/rgw/rgw_sync_policy.h
index c8f319add06912009a9619ac67b68cfde0b1a17f..37ded0181823d4dcf31d580b08b0f09084d10384 100644 (file)
--- a/src/rgw/rgw_sync_policy.h
+++ b/src/rgw/rgw_sync_policy.h
@@ -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)