rgw: bucket-granularity sync: define policy structure

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

src/rgw/CMakeLists.txt

@@ -56,6 +56,7 @@ set(librgw_common_srcs
   rgw_arn.cc
   rgw_basic_types.cc
   rgw_bucket.cc
+  rgw_bucket_sync.cc
   rgw_cache.cc
   rgw_common.cc
   rgw_compression.cc

src/rgw/rgw_bucket_sync.cc

@@ -0,0 +1,17 @@
+
+
+#include "rgw_common.h"
+#include "rgw_bucket_sync.h"
+
+
+
+void RGWBucketSyncPolicy::post_init()
+{
+  source_zones.clear();
+  for (auto& t : targets) {
+    for (auto& r : t.second.rules) {
+      source_zones.insert(r.zone_id);
+    }
+  }
+}
+

src/rgw/rgw_bucket_sync.h

@@ -0,0 +1,116 @@
+
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2018 Red Hat, Inc.
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation. See file COPYING.
+ *
+ */
+
+#pragma once
+
+#include "rgw_common.h"
+
+class JSONObj;
+
+class RGWBucketSyncPolicy {
+public:
+  struct target;
+
+private:
+  rgw_bucket bucket; /* source bucket */
+  std::map<string, target> targets; /* map: target zone_id -> target rules */
+
+  /* in-memory only */
+  std::set<string> source_zones;
+
+  void post_init();
+
+public:
+
+  struct rule {
+    std::string zone_id;
+    std::string dest_bucket;
+    std::string source_obj_prefix;
+    std::string dest_obj_prefix;
+
+    void encode(bufferlist& bl) const {
+      ENCODE_START(1, 1, bl);
+      encode(zone_id, bl);
+      encode(dest_bucket, bl);
+      encode(source_obj_prefix, bl);
+      encode(dest_obj_prefix, bl);
+      ENCODE_FINISH(bl);
+    }
+
+    void decode(bufferlist::const_iterator& bl) {
+      DECODE_START(1, bl);
+      decode(zone_id, bl);
+      decode(dest_bucket, bl);
+      decode(source_obj_prefix, bl);
+      decode(dest_obj_prefix, bl);
+      DECODE_FINISH(bl);
+    }
+
+    void dump(ceph::Formatter *f) const;
+    void decode_json(JSONObj *obj);
+  };
+
+  struct target {
+    std::string target_zone_id;
+    std::vector<rule> rules;
+
+    void encode(bufferlist& bl) const {
+      ENCODE_START(1, 1, bl);
+      encode(target_zone_id, bl);
+      encode(rules, bl);
+      ENCODE_FINISH(bl);
+    }
+
+    void decode(bufferlist::const_iterator& bl) {
+      DECODE_START(1, bl);
+      decode(target_zone_id, bl);
+      decode(rules, bl);
+      DECODE_FINISH(bl);
+    }
+
+    void dump(ceph::Formatter *f) const;
+    void decode_json(JSONObj *obj);
+  };
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    encode(bucket, bl);
+    encode(targets, bl);
+    ENCODE_FINISH(bl);
+  }
+
+  void decode(bufferlist::const_iterator& bl) {
+    DECODE_START(1, bl);
+    decode(bucket, bl);
+    decode(targets, bl);
+    post_init();
+    DECODE_FINISH(bl);
+  }
+
+  void dump(ceph::Formatter *f) const;
+  void decode_json(JSONObj *obj);
+
+  bool empty() const {
+    return targets.empty();
+  }
+
+  bool zone_is_source(const string& zone_id) const {
+    return source_zones.find(zone_id) != source_zones.end();
+  }
+};
+WRITE_CLASS_ENCODER(RGWBucketSyncPolicy::rule)
+WRITE_CLASS_ENCODER(RGWBucketSyncPolicy::target)
+WRITE_CLASS_ENCODER(RGWBucketSyncPolicy)

src/rgw/rgw_common.cc

@@ -27,6 +27,7 @@
 #include "common/strtol.h"
 #include "include/str_list.h"
 #include "rgw_crypt_sanitize.h"
+#include "rgw_bucket_sync.h"
 
 #include <sstream>
 
@@ -1938,3 +1939,148 @@ string camelcase_dash_http_attr(const string& orig)
   }
   return string(buf);
 }
+
+RGWBucketInfo::RGWBucketInfo()
+{
+}
+
+RGWBucketInfo::~RGWBucketInfo()
+{
+}
+
+void RGWBucketInfo::encode(bufferlist& bl) const {
+  ENCODE_START(21, 4, bl);
+  encode(bucket, bl);
+  encode(owner.id, bl);
+  encode(flags, bl);
+  encode(zonegroup, bl);
+  uint64_t ct = real_clock::to_time_t(creation_time);
+  encode(ct, bl);
+  encode(placement_rule, bl);
+  encode(has_instance_obj, bl);
+  encode(quota, bl);
+  encode(num_shards, bl);
+  encode(bucket_index_shard_hash_type, bl);
+  encode(requester_pays, bl);
+  encode(owner.tenant, bl);
+  encode(has_website, bl);
+  if (has_website) {
+    encode(website_conf, bl);
+  }
+  encode((uint32_t)index_type, bl);
+  encode(swift_versioning, bl);
+  if (swift_versioning) {
+    encode(swift_ver_location, bl);
+  }
+  encode(creation_time, bl);
+  encode(mdsearch_config, bl);
+  encode(reshard_status, bl);
+  encode(new_bucket_instance_id, bl);
+  if (obj_lock_enabled()) {
+    encode(obj_lock, bl);
+  }
+  bool has_sync_policy = !empty_sync_policy();
+  encode(has_sync_policy, bl);
+  if (has_sync_policy) {
+    encode(*sync_policy, bl);
+  }
+  ENCODE_FINISH(bl);
+}
+
+void RGWBucketInfo::decode(bufferlist::const_iterator& bl) {
+  DECODE_START_LEGACY_COMPAT_LEN_32(21, 4, 4, bl);
+  decode(bucket, bl);
+  if (struct_v >= 2) {
+    string s;
+    decode(s, bl);
+    owner.from_str(s);
+  }
+  if (struct_v >= 3)
+    decode(flags, bl);
+  if (struct_v >= 5)
+    decode(zonegroup, bl);
+  if (struct_v >= 6) {
+    uint64_t ct;
+    decode(ct, bl);
+    if (struct_v < 17)
+      creation_time = ceph::real_clock::from_time_t((time_t)ct);
+  }
+  if (struct_v >= 7)
+    decode(placement_rule, bl);
+  if (struct_v >= 8)
+    decode(has_instance_obj, bl);
+  if (struct_v >= 9)
+    decode(quota, bl);
+  if (struct_v >= 10)
+    decode(num_shards, bl);
+  if (struct_v >= 11)
+    decode(bucket_index_shard_hash_type, bl);
+  if (struct_v >= 12)
+    decode(requester_pays, bl);
+  if (struct_v >= 13)
+    decode(owner.tenant, bl);
+  if (struct_v >= 14) {
+    decode(has_website, bl);
+    if (has_website) {
+      decode(website_conf, bl);
+    } else {
+      website_conf = RGWBucketWebsiteConf();
+    }
+  }
+  if (struct_v >= 15) {
+    uint32_t it;
+    decode(it, bl);
+    index_type = (RGWBucketIndexType)it;
+  } else {
+    index_type = RGWBIType_Normal;
+  }
+  swift_versioning = false;
+  swift_ver_location.clear();
+  if (struct_v >= 16) {
+    decode(swift_versioning, bl);
+    if (swift_versioning) {
+      decode(swift_ver_location, bl);
+    }
+  }
+  if (struct_v >= 17) {
+    decode(creation_time, bl);
+  }
+  if (struct_v >= 18) {
+    decode(mdsearch_config, bl);
+  }
+  if (struct_v >= 19) {
+    decode(reshard_status, bl);
+    decode(new_bucket_instance_id, bl);
+  }
+  if (struct_v >= 20 && obj_lock_enabled()) {
+    decode(obj_lock, bl);
+  }
+  if (struct_v >= 21) {
+    bool has_sync_policy;
+    decode(has_sync_policy, bl);
+    if (has_sync_policy) {
+      auto policy = make_shared<RGWBucketSyncPolicy>();
+      decode(*policy, bl);
+      sync_policy = std::const_pointer_cast<const RGWBucketSyncPolicy>(policy);
+    } else {
+      sync_policy.reset();
+    }
+  }
+  
+  DECODE_FINISH(bl);
+}
+
+void RGWBucketInfo::set_sync_policy(RGWBucketSyncPolicy&& policy)
+{
+  auto shared_policy = make_shared<RGWBucketSyncPolicy>(policy);
+  sync_policy = std::const_pointer_cast<const RGWBucketSyncPolicy>(shared_policy);
+}
+
+bool RGWBucketInfo::empty_sync_policy() const
+{
+  if (!sync_policy) {
+    return true;
+  }
+
+  return sync_policy->empty();
+}

src/rgw/rgw_common.h

@@ -1392,6 +1392,8 @@ inline ostream& operator<<(ostream& out, const RGWBucketIndexType &index_type)
   }
 }
 
+struct RGWBucketSyncPolicy;
+
 struct RGWBucketInfo {
   enum BIShardsHashType {
     MOD = 0
@@ -1399,11 +1401,11 @@ struct RGWBucketInfo {
 
   rgw_bucket bucket;
   rgw_user owner;
-  uint32_t flags;
+  uint32_t flags{0};
   string zonegroup;
   ceph::real_time creation_time;
   rgw_placement_rule placement_rule;
-  bool has_instance_obj;
+  bool has_instance_obj{false};
   RGWObjVersionTracker objv_tracker; /* we don't need to serialize this, for runtime tracking */
   RGWQuotaInfo quota;
 
@@ -1411,22 +1413,22 @@ struct RGWBucketInfo {
   //   - value of 0 indicates there is no sharding (this is by default before this
   //     feature is implemented).
   //   - value of UINT32_T::MAX indicates this is a blind bucket.
-  uint32_t num_shards;
+  uint32_t num_shards{0};
 
   // Represents the bucket index shard hash type.
-  uint8_t bucket_index_shard_hash_type;
+  uint8_t bucket_index_shard_hash_type{MOD};
 
   // Represents the shard number for blind bucket.
   const static uint32_t NUM_SHARDS_BLIND_BUCKET;
 
-  bool requester_pays;
+  bool requester_pays{false};
 
-  bool has_website;
+  bool has_website{false};
   RGWBucketWebsiteConf website_conf;
 
   RGWBucketIndexType index_type = RGWBIType_Normal;
 
-  bool swift_versioning;
+  bool swift_versioning{false};
   string swift_ver_location;
 
   map<string, uint32_t> mdsearch_config;
@@ -1434,114 +1436,17 @@ struct RGWBucketInfo {
 
 
   /* resharding */
-  uint8_t reshard_status;
+  uint8_t reshard_status{0};
   string new_bucket_instance_id;
 
   RGWObjectLock obj_lock;
 
-  void encode(bufferlist& bl) const {
-     ENCODE_START(20, 4, bl);
-     encode(bucket, bl);
-     encode(owner.id, bl);
-     encode(flags, bl);
-     encode(zonegroup, bl);
-     uint64_t ct = real_clock::to_time_t(creation_time);
-     encode(ct, bl);
-     encode(placement_rule, bl);
-     encode(has_instance_obj, bl);
-     encode(quota, bl);
-     encode(num_shards, bl);
-     encode(bucket_index_shard_hash_type, bl);
-     encode(requester_pays, bl);
-     encode(owner.tenant, bl);
-     encode(has_website, bl);
-     if (has_website) {
-       encode(website_conf, bl);
-     }
-     encode((uint32_t)index_type, bl);
-     encode(swift_versioning, bl);
-     if (swift_versioning) {
-       encode(swift_ver_location, bl);
-     }
-     encode(creation_time, bl);
-     encode(mdsearch_config, bl);
-     encode(reshard_status, bl);
-     encode(new_bucket_instance_id, bl);
-     if (obj_lock_enabled()) {
-       encode(obj_lock, bl);
-     }
-     ENCODE_FINISH(bl);
-  }
-  void decode(bufferlist::const_iterator& bl) {
-    DECODE_START_LEGACY_COMPAT_LEN_32(20, 4, 4, bl);
-     decode(bucket, bl);
-     if (struct_v >= 2) {
-       string s;
-       decode(s, bl);
-       owner.from_str(s);
-     }
-     if (struct_v >= 3)
-       decode(flags, bl);
-     if (struct_v >= 5)
-       decode(zonegroup, bl);
-     if (struct_v >= 6) {
-       uint64_t ct;
-       decode(ct, bl);
-       if (struct_v < 17)
-	 creation_time = ceph::real_clock::from_time_t((time_t)ct);
-     }
-     if (struct_v >= 7)
-       decode(placement_rule, bl);
-     if (struct_v >= 8)
-       decode(has_instance_obj, bl);
-     if (struct_v >= 9)
-       decode(quota, bl);
-     if (struct_v >= 10)
-       decode(num_shards, bl);
-     if (struct_v >= 11)
-       decode(bucket_index_shard_hash_type, bl);
-     if (struct_v >= 12)
-       decode(requester_pays, bl);
-     if (struct_v >= 13)
-       decode(owner.tenant, bl);
-     if (struct_v >= 14) {
-       decode(has_website, bl);
-       if (has_website) {
-         decode(website_conf, bl);
-       } else {
-         website_conf = RGWBucketWebsiteConf();
-       }
-     }
-     if (struct_v >= 15) {
-       uint32_t it;
-       decode(it, bl);
-       index_type = (RGWBucketIndexType)it;
-     } else {
-       index_type = RGWBIType_Normal;
-     }
-     swift_versioning = false;
-     swift_ver_location.clear();
-     if (struct_v >= 16) {
-       decode(swift_versioning, bl);
-       if (swift_versioning) {
-         decode(swift_ver_location, bl);
-       }
-     }
-     if (struct_v >= 17) {
-       decode(creation_time, bl);
-     }
-     if (struct_v >= 18) {
-       decode(mdsearch_config, bl);
-     }
-     if (struct_v >= 19) {
-       decode(reshard_status, bl);
-       decode(new_bucket_instance_id, bl);
-     }
-     if (struct_v >= 20 && obj_lock_enabled()) {
-       decode(obj_lock, bl);
-     }
-     DECODE_FINISH(bl);
-  }
+  std::shared_ptr<const RGWBucketSyncPolicy> sync_policy;
+
+
+  void encode(bufferlist& bl) const;
+  void decode(bufferlist::const_iterator& bl);
+
   void dump(Formatter *f) const;
   static void generate_test_instances(list<RGWBucketInfo*>& o);
 
@@ -1559,8 +1464,12 @@ struct RGWBucketInfo {
     return swift_versioning && !versioned();
   }
 
-  RGWBucketInfo() : flags(0), has_instance_obj(false), num_shards(0), bucket_index_shard_hash_type(MOD), requester_pays(false),
-                    has_website(false), swift_versioning(false), reshard_status(0) {}
+  void set_sync_policy(RGWBucketSyncPolicy&& policy);
+
+  bool empty_sync_policy() const;
+
+  RGWBucketInfo();
+  ~RGWBucketInfo();
 };
 WRITE_CLASS_ENCODER(RGWBucketInfo)
 

src/rgw/rgw_json_enc.cc

@@ -15,6 +15,7 @@
 #include "rgw_data_sync.h"
 #include "rgw_sync.h"
 #include "rgw_orphan.h"
+#include "rgw_bucket_sync.h"
 
 #include "common/ceph_json.h"
 #include "common/Formatter.h"
@@ -829,6 +830,46 @@ void RGWBucketInfo::decode_json(JSONObj *obj) {
   reshard_status = (cls_rgw_reshard_status)rs;
 }
 
+void RGWBucketSyncPolicy::rule::dump(Formatter *f) const
+{
+  encode_json("zone_id", zone_id, f);
+  encode_json("dest_bucket", dest_bucket, f);
+  encode_json("source_obj_prefix", source_obj_prefix, f);
+  encode_json("dest_obj_prefix", dest_obj_prefix, f);
+}
+
+void RGWBucketSyncPolicy::rule::decode_json(JSONObj *obj)
+{
+  JSONDecoder::decode_json("zone_id", zone_id, obj);
+  JSONDecoder::decode_json("dest_bucket", dest_bucket, obj);
+  JSONDecoder::decode_json("source_obj_prefix", source_obj_prefix, obj);
+  JSONDecoder::decode_json("dest_obj_prefix", dest_obj_prefix, obj);
+}
+
+void RGWBucketSyncPolicy::target::dump(Formatter *f) const
+{
+  encode_json("target_zone_id", target_zone_id, f);
+  encode_json("rules", rules, f);
+}
+
+void RGWBucketSyncPolicy::target::decode_json(JSONObj *obj)
+{
+  JSONDecoder::decode_json("target_zone_id", target_zone_id, obj);
+  JSONDecoder::decode_json("rules", rules, obj);
+}
+
+void RGWBucketSyncPolicy::dump(Formatter *f) const
+{
+  encode_json("bucket", bucket, f);
+  encode_json("targets", targets, f);
+}
+
+void RGWBucketSyncPolicy::decode_json(JSONObj *obj)
+{
+  JSONDecoder::decode_json("bucket", bucket, obj);
+  JSONDecoder::decode_json("targets", targets, obj);
+}
+
 void rgw_obj_key::dump(Formatter *f) const
 {
   encode_json("name", name, f);