rgw: sync: extend data structures to extend bucket sync functionality

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

diff --git a/src/rgw/rgw_json_enc.cc b/src/rgw/rgw_json_enc.cc
index b2d571e53aa616381419e3e8e4abe9487d9d5360..06c93880571435ee88f7959d598e86c578267bb7 100644 (file)
--- a/src/rgw/rgw_json_enc.cc
+++ b/src/rgw/rgw_json_enc.cc
@@ -885,6 +885,68 @@ void rgw_sync_bucket_entity::decode_json(JSONObj *obj)
     }
   }
 }
+
+void rgw_sync_pipe_filter::_tag::dump(Formatter *f) const
+{
+  encode_json("key", key, f);
+  encode_json("value", value, f);
+}
+
+void rgw_sync_pipe_filter::_tag::decode_json(JSONObj *obj)
+{
+  JSONDecoder::decode_json("key", key, obj);
+  JSONDecoder::decode_json("value", value, obj);
+}
+
+void rgw_sync_pipe_filter::dump(Formatter *f) const
+{
+  encode_json("prefix", prefix, f);
+  encode_json("tags", tags, f);
+}
+
+void rgw_sync_pipe_filter::decode_json(JSONObj *obj)
+{
+  JSONDecoder::decode_json("prefix", prefix, obj);
+  JSONDecoder::decode_json("tags", tags, obj);
+}
+
+void rgw_sync_pipe_acl_translation::dump(Formatter *f) const
+{
+  encode_json("owner", owner, f);
+}
+
+void rgw_sync_pipe_acl_translation::decode_json(JSONObj *obj)
+{
+  JSONDecoder::decode_json("owner", owner, obj);
+}
+
+void rgw_sync_pipe_dest_params::dump(Formatter *f) const
+{
+  encode_json("acl_translation", acl_translation, f);
+  encode_json("storage_class", storage_class, f);
+}
+
+void rgw_sync_pipe_dest_params::decode_json(JSONObj *obj)
+{
+  JSONDecoder::decode_json("acl_translation", acl_translation, obj);
+  JSONDecoder::decode_json("storage_class", storage_class, obj);
+}
+
+void rgw_sync_pipe_params::dump(Formatter *f) const
+{
+  encode_json("filter", filter, f);
+  encode_json("dest_params", dest_params, f);
+  encode_json("priority", priority, f);
+}
+
+void rgw_sync_pipe_params::decode_json(JSONObj *obj)
+{
+  JSONDecoder::decode_json("filter", filter, obj);
+  JSONDecoder::decode_json("dest_params", dest_params, obj);
+  JSONDecoder::decode_json("priority", priority, obj);
+}
+
+
 void rgw_sync_bucket_entities::dump(Formatter *f) const
 {
   encode_json("bucket", rgw_sync_bucket_entities::bucket_key(bucket), f);
@@ -932,14 +994,18 @@ void rgw_sync_bucket_entities::decode_json(JSONObj *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);
+  encode_json("params", params, 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);
+  JSONDecoder::decode_json("params", params, obj);
 }
 
 void rgw_sync_bucket_pipes::dump(Formatter *f) const
@@ -947,6 +1013,7 @@ void rgw_sync_bucket_pipes::dump(Formatter *f) const
   encode_json("id", id, f);
   encode_json("source", source, f);
   encode_json("dest", dest, f);
+  encode_json("params", params, f);
 }
 
 void rgw_sync_bucket_pipes::decode_json(JSONObj *obj)
@@ -954,6 +1021,7 @@ void rgw_sync_bucket_pipes::decode_json(JSONObj *obj)
   JSONDecoder::decode_json("id", id, obj);
   JSONDecoder::decode_json("source", source, obj);
   JSONDecoder::decode_json("dest", dest, obj);
+  JSONDecoder::decode_json("params", params, obj);
 }
 
 void rgw_sync_data_flow_group::dump(Formatter *f) const

diff --git a/src/rgw/rgw_sync_policy.cc b/src/rgw/rgw_sync_policy.cc
index 5e8d24f6f8fbfccca89976bc1ba127a68aeb2fd0..cbc979df789c0d4d5559262442b1afcc50390441 100644 (file)
--- a/src/rgw/rgw_sync_policy.cc
+++ b/src/rgw/rgw_sync_policy.cc
@@ -10,6 +10,71 @@ string rgw_sync_bucket_entity::bucket_key() const
 {
   return rgw_sync_bucket_entities::bucket_key(bucket);
 }
+
+bool rgw_sync_pipe_filter::_tag::from_str(const string& s)
+{
+  if (s.empty()) {
+    return false;
+  }
+
+  auto pos = s.find('=');
+  if (pos == string::npos) {
+    key = s;
+    return true;
+  }
+
+  key = s.substr(0, pos);
+  if (pos < s.size() - 1) {
+    value = s.substr(pos + 1);
+  }
+
+  return true;
+}
+
+void rgw_sync_pipe_filter::encode(bufferlist& bl) const
+{
+  ENCODE_START(1, 1, bl);
+  encode(prefix, bl);
+  encode(tags, bl);
+  ENCODE_FINISH(bl);
+}
+
+void rgw_sync_pipe_filter::decode(bufferlist::const_iterator& bl)
+{
+  DECODE_START(1, bl);
+  decode(prefix, bl);
+  decode(tags, bl);
+  DECODE_FINISH(bl);
+}
+
+void rgw_sync_pipe_filter::set_prefix(std::optional<std::string> opt_prefix,
+                                      bool prefix_rm)
+{
+  if (opt_prefix) {
+    prefix = *opt_prefix;    
+  } else if (prefix_rm) {
+    prefix.reset();
+  }
+}
+
+void rgw_sync_pipe_filter::set_tags(std::list<std::string>& tags_add,
+                                    std::list<std::string>& tags_rm)
+{
+  for (auto& t : tags_rm) {
+    _tag tag;
+    if (tag.from_str(t)) {
+      tags.erase(tag);
+    }
+  }
+
+  for (auto& t : tags_add) {
+    _tag tag;
+    if (tag.from_str(t)) {
+      tags.insert(tag);
+    }
+  }
+}
+
 void rgw_sync_bucket_entity::apply_bucket(std::optional<rgw_bucket> b)
 {
   if (!b) {
@@ -158,8 +223,10 @@ std::vector<rgw_sync_bucket_pipe> rgw_sync_bucket_pipes::expand() const
   for (auto& s : sources) {
     for (auto& d : dests) {
       rgw_sync_bucket_pipe pipe;
+      pipe.id = id;
       pipe.source = s;
       pipe.dest = d;
+      pipe.params = params;
       result.push_back(pipe);
     }
   }

diff --git a/src/rgw/rgw_sync_policy.h b/src/rgw/rgw_sync_policy.h
index a869a88acdb191d937f2fe9024de24415123082b..0fca7408fa59fd4cede3aeb0cfdc7ef03f6065fd 100644 (file)
--- a/src/rgw/rgw_sync_policy.h
+++ b/src/rgw/rgw_sync_policy.h
@@ -176,30 +176,163 @@ struct rgw_sync_bucket_entity {
 };
 WRITE_CLASS_ENCODER(rgw_sync_bucket_entity)
 
+struct rgw_sync_pipe_filter {
+  struct _tag {
+    string key;
+    string value;
+
+    void encode(bufferlist& bl) const {
+      ENCODE_START(1, 1, bl);
+      encode(key, bl);
+      encode(value, bl);
+      ENCODE_FINISH(bl);
+    }
+
+    void decode(bufferlist::const_iterator& bl) {
+      DECODE_START(1, bl);
+      decode(key, bl);
+      decode(value, bl);
+      DECODE_FINISH(bl);
+    }
+
+    void dump(ceph::Formatter *f) const;
+    void decode_json(JSONObj *obj);
+
+    bool from_str(const string& s);
+
+    bool operator<(const _tag& t) const {
+      if (key < t.key) {
+        return true;
+      }
+      if (t.key < key) {
+        return false;
+      }
+      return (value < t.value);
+    }
+  };
+
+  std::optional<string> prefix;
+  std::set<_tag> tags;
+
+  void set_prefix(std::optional<std::string> opt_prefix,
+                  bool prefix_rm);
+  void set_tags(std::list<std::string>& tags_add,
+                std::list<std::string>& tags_rm);
+
+  void encode(bufferlist& bl) const;
+  void decode(bufferlist::const_iterator& bl);
+
+  void dump(ceph::Formatter *f) const;
+  void decode_json(JSONObj *obj);
+};
+WRITE_CLASS_ENCODER(rgw_sync_pipe_filter)
+WRITE_CLASS_ENCODER(rgw_sync_pipe_filter::_tag)
+
+struct rgw_sync_pipe_acl_translation {
+  rgw_user owner;
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    encode(owner, bl);
+    ENCODE_FINISH(bl);
+  }
+
+  void decode(bufferlist::const_iterator& bl) {
+    DECODE_START(1, bl);
+    decode(owner, bl);
+    DECODE_FINISH(bl);
+  }
+
+  void dump(ceph::Formatter *f) const;
+  void decode_json(JSONObj *obj);
+};
+WRITE_CLASS_ENCODER(rgw_sync_pipe_acl_translation)
+
+struct rgw_sync_pipe_dest_params { 
+  std::optional<rgw_sync_pipe_acl_translation> acl_translation;
+  std::optional<string> storage_class;
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    encode(acl_translation, bl);
+    encode(storage_class, bl);
+    ENCODE_FINISH(bl);
+  }
+
+  void decode(bufferlist::const_iterator& bl) {
+    DECODE_START(1, bl);
+    decode(acl_translation, bl);
+    decode(storage_class, bl);
+    DECODE_FINISH(bl);
+  }
+
+  void dump(ceph::Formatter *f) const;
+  void decode_json(JSONObj *obj);
+};
+WRITE_CLASS_ENCODER(rgw_sync_pipe_dest_params)
+
+struct rgw_sync_pipe_params { 
+  rgw_sync_pipe_filter filter;
+  rgw_sync_pipe_dest_params dest_params;
+  int32_t priority{0};
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    encode(filter, bl);
+    encode(dest_params, bl);
+    encode(priority, bl);
+    ENCODE_FINISH(bl);
+  }
+
+  void decode(bufferlist::const_iterator& bl) {
+    DECODE_START(1, bl);
+    decode(filter, bl);
+    decode(dest_params, bl);
+    decode(priority, bl);
+    DECODE_FINISH(bl);
+  }
+
+  void dump(ceph::Formatter *f) const;
+  void decode_json(JSONObj *obj);
+};
+WRITE_CLASS_ENCODER(rgw_sync_pipe_params)
+
 struct rgw_sync_bucket_pipe {
-public:
+  string id;
   rgw_sync_bucket_entity source;
   rgw_sync_bucket_entity dest;
 
+  rgw_sync_pipe_params params;
+
   bool specific() const {
     return source.specific() && dest.specific();
   }
 
   void encode(bufferlist& bl) const {
     ENCODE_START(1, 1, bl);
+    encode(id, bl);
     encode(source, bl);
     encode(dest, bl);
+    encode(params, bl);
     ENCODE_FINISH(bl);
   }
 
   void decode(bufferlist::const_iterator& bl) {
     DECODE_START(1, bl);
+    decode(id, bl);
     decode(source, bl);
     decode(dest, bl);
+    decode(params, bl);
     DECODE_FINISH(bl);
   }
 
   const bool operator<(const rgw_sync_bucket_pipe& p) const {
+    if (id < p.id) {
+      return true;
+    }
+    if (id >p.id) {
+      return false;
+    }
     if (source < p.source) {
       return true;
     }
@@ -288,11 +421,14 @@ struct rgw_sync_bucket_pipes {
   rgw_sync_bucket_entities source;
   rgw_sync_bucket_entities dest;
 
+  rgw_sync_pipe_params params;
+
   void encode(bufferlist& bl) const {
     ENCODE_START(1, 1, bl);
     encode(id, bl);
     encode(source, bl);
     encode(dest, bl);
+    encode(params, bl);
     ENCODE_FINISH(bl);
   }
 
@@ -301,6 +437,7 @@ struct rgw_sync_bucket_pipes {
     decode(id, bl);
     decode(source, bl);
     decode(dest, bl);
+    decode(params, bl);
     DECODE_FINISH(bl);
   }