rgw: use single entity in expanded sync determination

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

diff --git a/src/rgw/rgw_bucket_sync.cc b/src/rgw/rgw_bucket_sync.cc
index dce8839c13d00eb9c3ff7d60bb1975ed7a99e591..4302ca1785484170941e97a7a4c28e917d650100 100644 (file)
--- a/src/rgw/rgw_bucket_sync.cc
+++ b/src/rgw/rgw_bucket_sync.cc
@@ -10,6 +10,11 @@
 #define dout_subsys ceph_subsys_rgw
 
 
+string rgw_sync_bucket_entity::bucket_key() const
+{
+  return rgw_sync_bucket_entities::bucket_key(bucket);
+}
+
 void rgw_sync_bucket_entities::add_zones(const std::vector<string>& new_zones) {
   for (auto& z : new_zones) {
     if (z == "*") {
@@ -548,7 +553,8 @@ RGWBucketSyncFlowManager::flow_map_t::iterator RGWBucketSyncFlowManager::find_bu
   return m.find(rgw_bucket());
 }
 
-
+#warning cleanup
+#if 0
 void RGWBucketSyncFlowManager::update_flow_maps(const rgw_sync_bucket_pipes& pipe) {
   auto source_bucket = pipe.source.get_bucket();
   auto dest_bucket = pipe.dest.get_bucket();
@@ -579,11 +585,12 @@ void RGWBucketSyncFlowManager::update_flow_maps(const rgw_sync_bucket_pipes& pip
   }
 #endif
 }
+#endif
 
 void RGWBucketSyncFlowManager::init(const rgw_sync_policy_info& sync_policy) {
-    rgw_sync_bucket_entities entity;
-    entity.zones = std::set<string>( { zone_name } );
-    entity.bucket =  bucket;
+  rgw_sync_bucket_entity entity;
+  entity.zone = zone_name;
+  entity.bucket = bucket.value_or(rgw_bucket());
 
   for (auto& item : sync_policy.groups) {
     auto& group = item.second;
@@ -605,24 +612,24 @@ void RGWBucketSyncFlowManager::init(const rgw_sync_policy_info& sync_policy) {
                         });
 
     for (auto& entry : flow_group_map.sources) {
-      rgw_sync_bucket_pipes pipe;
-      rgw_sync_bucket_entities source;
-      pipe.source.zones = std::set<string>( { entry.first.zone } );
+      rgw_sync_bucket_pipe pipe;
+      rgw_sync_bucket_entity source;
+      pipe.source.zone = entry.first.zone;
       pipe.source.bucket = entry.first.bucket;
       pipe.dest = entity;
 
-      auto& by_source = flow_by_source[pipe.source.get_bucket()];
+      auto& by_source = flow_by_source[pipe.source.bucket];
       by_source.pipe.push_back(pipe);
     }
 
     for (auto& entry : flow_group_map.dests) {
-      rgw_sync_bucket_pipes pipe;
-      rgw_sync_bucket_entities dest;
-      pipe.dest.zones = std::set<string>( { entry.first.zone } );
+      rgw_sync_bucket_pipe pipe;
+      rgw_sync_bucket_entity dest;
+      pipe.dest.zone = entry.first.zone;
       pipe.dest.bucket = entry.first.bucket;
       pipe.source = entity;
 
-      auto& by_dest = flow_by_source[pipe.dest.get_bucket()];
+      auto& by_dest = flow_by_source[pipe.dest.bucket];
       by_dest.pipe.push_back(pipe);
     }
   }

diff --git a/src/rgw/rgw_bucket_sync.h b/src/rgw/rgw_bucket_sync.h
index e8b1af9d194c8084f0f01bdabfef07ffeaf3fc09..81547dea96f81b579be67042bcdb45d0f3a7399b 100644 (file)
--- a/src/rgw/rgw_bucket_sync.h
+++ b/src/rgw/rgw_bucket_sync.h
@@ -116,7 +116,7 @@ struct rgw_sync_group_pipe_map {
 class RGWBucketSyncFlowManager {
 public:
   struct pipe_flow {
-    std::vector<rgw_sync_bucket_pipes> pipe;
+    std::vector<rgw_sync_bucket_pipe> pipe;
 
     void dump(ceph::Formatter *f) const;
   };

diff --git a/src/rgw/rgw_json_enc.cc b/src/rgw/rgw_json_enc.cc
index ba3d574d4a4595453660730c4cbc73796a883335..980d23fd1257663e8e962eaa5828555af7b2bf43 100644 (file)
--- a/src/rgw/rgw_json_enc.cc
+++ b/src/rgw/rgw_json_enc.cc
@@ -865,6 +865,22 @@ void rgw_sync_symmetric_group::decode_json(JSONObj *obj)
   JSONDecoder::decode_json("zones", zones, obj);
 }
 
+void rgw_sync_bucket_entity::dump(Formatter *f) const
+{
+  encode_json("zone", zone, f);
+  encode_json("bucket", bucket_key(), f);
+}
+
+void rgw_sync_bucket_entity::decode_json(JSONObj *obj)
+{
+  JSONDecoder::decode_json("zone", zone, obj);
+  string s;
+  JSONDecoder::decode_json("bucket", s, obj);
+  int ret = rgw_bucket_parse_bucket_key(nullptr, s, &bucket, nullptr);
+  if (ret < 0) {
+    bucket = rgw_bucket();
+  }
+}
 void rgw_sync_bucket_entities::dump(Formatter *f) const
 {
   encode_json("bucket", rgw_sync_bucket_entities::bucket_key(bucket), f);
@@ -903,6 +919,18 @@ void rgw_sync_bucket_entities::decode_json(JSONObj *obj)
   JSONDecoder::decode_json("zones", zones, obj);
 }
 
+void rgw_sync_bucket_pipe::dump(Formatter *f) const
+{
+  encode_json("source", source, f);
+  encode_json("dest", dest, f);
+}
+
+void rgw_sync_bucket_pipe::decode_json(JSONObj *obj)
+{
+  JSONDecoder::decode_json("source", source, obj);
+  JSONDecoder::decode_json("dest", dest, obj);
+}
+
 void rgw_sync_bucket_pipes::dump(Formatter *f) const
 {
   encode_json("id", id, f);

diff --git a/src/rgw/rgw_sync_policy.h b/src/rgw/rgw_sync_policy.h
index 30513895a737c1f3de33f8f1ecc3ec1b4f380037..e96c833149a477476f7bbe373a9d759c3b6c43b6 100644 (file)
--- a/src/rgw/rgw_sync_policy.h
+++ b/src/rgw/rgw_sync_policy.h
@@ -219,6 +219,65 @@ struct rgw_sync_directional_rule {
 };
 WRITE_CLASS_ENCODER(rgw_sync_directional_rule)
 
+struct rgw_sync_bucket_entity {
+  string zone; /* define specific zones */
+  rgw_bucket bucket; /* define specific bucket */
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    encode(zone, bl);
+    encode(bucket, bl);
+    ENCODE_FINISH(bl);
+  }
+
+  void decode(bufferlist::const_iterator& bl) {
+    DECODE_START(1, bl);
+    decode(zone, bl);
+    decode(bucket, bl);
+    DECODE_FINISH(bl);
+  }
+
+  void dump(ceph::Formatter *f) const;
+  void decode_json(JSONObj *obj);
+
+  string bucket_key() const;
+
+  const bool operator<(const rgw_sync_bucket_entity& e) const {
+    if (zone < e.zone) {
+      return true;
+    }
+    if (zone > e.zone) {
+      return false;
+    }
+    return (bucket < e.bucket);
+  }
+};
+WRITE_CLASS_ENCODER(rgw_sync_bucket_entity)
+
+struct rgw_sync_bucket_pipe {
+public:
+  rgw_sync_bucket_entity source;
+  rgw_sync_bucket_entity dest;
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    encode(source, bl);
+    encode(dest, bl);
+    ENCODE_FINISH(bl);
+  }
+
+  void decode(bufferlist::const_iterator& bl) {
+    DECODE_START(1, bl);
+    decode(source, bl);
+    decode(dest, bl);
+    DECODE_FINISH(bl);
+  }
+
+  void dump(ceph::Formatter *f) const;
+  void decode_json(JSONObj *obj);
+};
+WRITE_CLASS_ENCODER(rgw_sync_bucket_pipe)
+
 struct rgw_sync_bucket_entities {
 private:
   bool match_str(const string& s1, const string& s2) const { /* empty string is wildcard */