rgw: metadata, data logs can be turned on / off

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

src/cls/rgw/cls_rgw.cc

@@ -469,9 +469,11 @@ int rgw_bucket_prepare_op(cls_method_context_t hctx, bufferlist *in, bufferlist
     return -EINVAL;
   }
 
-  rc = log_index_operation(hctx, op.name, op.op, op.tag, entry.meta.mtime, entry.ver, info.state, header.ver);
+  if (op.log_op) {
-  if (rc < 0)
+    rc = log_index_operation(hctx, op.name, op.op, op.tag, entry.meta.mtime, entry.ver, info.state, header.ver);
-    return rc;
+    if (rc < 0)
+      return rc;
+  }
 
   // write out new key to disk
   bufferlist info_bl;
@@ -581,9 +583,11 @@ int rgw_bucket_complete_op(cls_method_context_t hctx, bufferlist *in, bufferlist
 
   bufferlist op_bl;
   if (cancel) {
-    rc = log_index_operation(hctx, op.name, op.op, op.tag, entry.meta.mtime, entry.ver, CLS_RGW_STATE_COMPLETE, header.ver);
+    if (op.log_op) {
-    if (rc < 0)
+      rc = log_index_operation(hctx, op.name, op.op, op.tag, entry.meta.mtime, entry.ver, CLS_RGW_STATE_COMPLETE, header.ver);
-      return rc;
+      if (rc < 0)
+        return rc;
+    }
 
     if (op.tag.size()) {
       bufferlist new_key_bl;
@@ -638,9 +642,11 @@ int rgw_bucket_complete_op(cls_method_context_t hctx, bufferlist *in, bufferlist
     break;
   }
 
-  rc = log_index_operation(hctx, op.name, op.op, op.tag, entry.meta.mtime, entry.ver, CLS_RGW_STATE_COMPLETE, header.ver);
+  if (op.log_op) {
-  if (rc < 0)
+    rc = log_index_operation(hctx, op.name, op.op, op.tag, entry.meta.mtime, entry.ver, CLS_RGW_STATE_COMPLETE, header.ver);
-    return rc;
+    if (rc < 0)
+      return rc;
+  }
 
   list<string>::iterator remove_iter;
   CLS_LOG(0, "rgw_bucket_complete_op(): remove_objs.size()=%d\n", (int)op.remove_objs.size());
@@ -656,10 +662,12 @@ int rgw_bucket_complete_op(cls_method_context_t hctx, bufferlist *in, bufferlist
     CLS_LOG(0, "rgw_bucket_complete_op(): entry.name=%s entry.meta.category=%d\n", remove_entry.name.c_str(), remove_entry.meta.category);
     unaccount_entry(header, remove_entry);
 
-    rc = log_index_operation(hctx, op.name, CLS_RGW_OP_DEL, op.tag, remove_entry.meta.mtime,
+    if (op.log_op) {
-                             remove_entry.ver, CLS_RGW_STATE_COMPLETE, header.ver);
+      rc = log_index_operation(hctx, op.name, CLS_RGW_OP_DEL, op.tag, remove_entry.meta.mtime,
-    if (rc < 0)
+                               remove_entry.ver, CLS_RGW_STATE_COMPLETE, header.ver);
-      continue;
+      if (rc < 0)
+        continue;
+    }
 
     ret = cls_cxx_map_remove_key(hctx, remove_oid_name);
     if (ret < 0) {

src/cls/rgw/cls_rgw_client.cc

@@ -24,13 +24,14 @@ void cls_rgw_bucket_set_tag_timeout(ObjectWriteOperation& o, uint64_t tag_timeou
 }
 
 void cls_rgw_bucket_prepare_op(ObjectWriteOperation& o, RGWModifyOp op, string& tag,
-                               string& name, string& locator)
+                               string& name, string& locator, bool log_op)
 {
   struct rgw_cls_obj_prepare_op call;
   call.op = op;
   call.tag = tag;
   call.name = name;
   call.locator = locator;
+  call.log_op = log_op;
   bufferlist in;
   ::encode(call, in);
   o.exec("rgw", "bucket_prepare_op", in);
@@ -38,7 +39,7 @@ void cls_rgw_bucket_prepare_op(ObjectWriteOperation& o, RGWModifyOp op, string&
 
 void cls_rgw_bucket_complete_op(ObjectWriteOperation& o, RGWModifyOp op, string& tag,
                                 rgw_bucket_entry_ver& ver, string& name, rgw_bucket_dir_entry_meta& dir_meta,
-				list<string> *remove_objs)
+				list<string> *remove_objs, bool log_op)
 {
 
   bufferlist in;
@@ -48,6 +49,7 @@ void cls_rgw_bucket_complete_op(ObjectWriteOperation& o, RGWModifyOp op, string&
   call.name = name;
   call.ver = ver;
   call.meta = dir_meta;
+  call.log_op = log_op;
   if (remove_objs)
     call.remove_objs = *remove_objs;
   ::encode(call, in);

src/cls/rgw/cls_rgw_client.h

@@ -11,11 +11,11 @@ void cls_rgw_bucket_init(librados::ObjectWriteOperation& o);
 void cls_rgw_bucket_set_tag_timeout(librados::ObjectWriteOperation& o, uint64_t tag_timeout);
 
 void cls_rgw_bucket_prepare_op(librados::ObjectWriteOperation& o, RGWModifyOp op, string& tag,
-                               string& name, string& locator);
+                               string& name, string& locator, bool log_op);
 
 void cls_rgw_bucket_complete_op(librados::ObjectWriteOperation& o, RGWModifyOp op, string& tag,
                                 rgw_bucket_entry_ver& ver, string& name, rgw_bucket_dir_entry_meta& dir_meta,
-				list<string> *remove_objs);
+				list<string> *remove_objs, bool log_op);
 
 int cls_rgw_list_op(librados::IoCtx& io_ctx, string& oid, string& start_obj,
                     string& filter_prefix, uint32_t num_entries,

src/cls/rgw/cls_rgw_ops.h

@@ -31,20 +31,22 @@ struct rgw_cls_obj_prepare_op
   string name;
   string tag;
   string locator;
+  bool log_op;
 
-  rgw_cls_obj_prepare_op() : op(CLS_RGW_OP_UNKNOWN) {}
+  rgw_cls_obj_prepare_op() : op(CLS_RGW_OP_UNKNOWN), log_op(false) {}
 
   void encode(bufferlist &bl) const {
-    ENCODE_START(3, 3, bl);
+    ENCODE_START(4, 3, bl);
     uint8_t c = (uint8_t)op;
     ::encode(c, bl);
     ::encode(name, bl);
     ::encode(tag, bl);
     ::encode(locator, bl);
+    ::encode(log_op, bl);
     ENCODE_FINISH(bl);
   }
   void decode(bufferlist::iterator &bl) {
-    DECODE_START_LEGACY_COMPAT_LEN(3, 3, 3, bl);
+    DECODE_START_LEGACY_COMPAT_LEN(4, 3, 3, bl);
     uint8_t c;
     ::decode(c, bl);
     op = (RGWModifyOp)c;
@@ -53,6 +55,9 @@ struct rgw_cls_obj_prepare_op
     if (struct_v >= 2) {
       ::decode(locator, bl);
     }
+    if (struct_v >= 4) {
+      ::decode(log_op, bl);
+    }
     DECODE_FINISH(bl);
   }
   void dump(Formatter *f) const;
@@ -68,11 +73,14 @@ struct rgw_cls_obj_complete_op
   rgw_bucket_entry_ver ver;
   struct rgw_bucket_dir_entry_meta meta;
   string tag;
+  bool log_op;
 
   list<string> remove_objs;
 
+  rgw_cls_obj_complete_op() : log_op(false) {}
+
   void encode(bufferlist &bl) const {
-    ENCODE_START(5, 3, bl);
+    ENCODE_START(6, 3, bl);
     uint8_t c = (uint8_t)op;
     ::encode(c, bl);
     ::encode(name, bl);
@@ -81,10 +89,11 @@ struct rgw_cls_obj_complete_op
     ::encode(tag, bl);
     ::encode(locator, bl);
     ::encode(remove_objs, bl);
+    ::encode(log_op, bl);
     ENCODE_FINISH(bl);
  }
   void decode(bufferlist::iterator &bl) {
-    DECODE_START_LEGACY_COMPAT_LEN(5, 3, 3, bl);
+    DECODE_START_LEGACY_COMPAT_LEN(6, 3, 3, bl);
     uint8_t c;
     ::decode(c, bl);
     op = (RGWModifyOp)c;
@@ -103,6 +112,9 @@ struct rgw_cls_obj_complete_op
     if (struct_v >= 4) {
       ::decode(remove_objs, bl);
     }
+    if (struct_v >= 6) {
+      ::decode(log_op, bl);
+    }
     DECODE_FINISH(bl);
   }
   void dump(Formatter *f) const;

src/rgw/rgw_bucket.cc

@@ -990,6 +990,9 @@ int RGWDataChangesLog::choose_oid(rgw_bucket& bucket) {
 
 int RGWDataChangesLog::renew_entries()
 {
+  if (!store->need_to_log_data())
+    return 0;
+
   /* we can't keep the bucket name as part of the cls_log_entry, and we need
    * it later, so we keep two lists under the map */
   map<int, pair<list<string>, list<cls_log_entry> > > m;
@@ -1074,6 +1077,9 @@ void RGWDataChangesLog::update_renewed(string& bucket_name, utime_t& expiration)
 }
 
 int RGWDataChangesLog::add_entry(rgw_bucket& bucket) {
+  if (!store->need_to_log_data())
+    return 0;
+
   lock.Lock();
 
   ChangeStatusPtr status;

src/rgw/rgw_json_enc.cc

@@ -576,12 +576,16 @@ void RGWZone::dump(Formatter *f) const
 {
   encode_json("name", name, f);
   encode_json("endpoints", endpoints, f);
+  encode_json("log_meta", log_meta, f);
+  encode_json("log_data", log_data, f);
 }
 
 void RGWZone::decode_json(JSONObj *obj)
 {
   JSONDecoder::decode_json("name", name, obj);
   JSONDecoder::decode_json("endpoints", endpoints, obj);
+  JSONDecoder::decode_json("log_meta", log_meta, obj);
+  JSONDecoder::decode_json("log_data", log_data, obj);
 }
 
 void RGWRegionPlacementTarget::dump(Formatter *f) const

src/rgw/rgw_metadata.cc

@@ -75,6 +75,9 @@ WRITE_CLASS_ENCODER(RGWMetadataLogData);
 
 
 int RGWMetadataLog::add_entry(RGWRados *store, RGWMetadataHandler *handler, const string& section, const string& key, bufferlist& bl) {
+  if (!store->need_to_log_metadata())
+    return 0;
+
   string oid;
 
   string hash_key;

src/rgw/rgw_rados.cc

@@ -895,10 +895,12 @@ int RGWRados::init_complete()
   map<string, RGWZone>::iterator ziter;
   for (ziter = region.zones.begin(); ziter != region.zones.end(); ++ziter) {
     const string& name = ziter->first;
+    RGWZone& z = ziter->second;
     if (name != zone.name) {
-      RGWZone& z = ziter->second;
       ldout(cct, 20) << "generating connection object for zone " << name << dendl;
       zone_conn_map[name] = new RGWRESTConn(cct, this, z.endpoints);
+    } else {
+      zone_public_config = z;
     }
   }
 
@@ -4951,7 +4953,7 @@ int RGWRados::cls_obj_prepare_op(rgw_bucket& bucket, RGWModifyOp op, string& tag
     return r;
 
   ObjectWriteOperation o;
-  cls_rgw_bucket_prepare_op(o, op, tag, name, locator);
+  cls_rgw_bucket_prepare_op(o, op, tag, name, locator, zone_public_config.log_data);
   r = index_ctx.operate(oid, &o);
   return r;
 }
@@ -4981,7 +4983,7 @@ int RGWRados::cls_obj_complete_op(rgw_bucket& bucket, RGWModifyOp op, string& ta
   rgw_bucket_entry_ver ver;
   ver.pool = pool;
   ver.epoch = epoch;
-  cls_rgw_bucket_complete_op(o, op, tag, ver, ent.name, dir_meta, remove_objs);
+  cls_rgw_bucket_complete_op(o, op, tag, ver, ent.name, dir_meta, remove_objs, zone_public_config.log_data);
 
   AioCompletion *c = librados::Rados::aio_create_completion(NULL, NULL, NULL);
   r = index_ctx.aio_operate(oid, c, &o);

src/rgw/rgw_rados.h

@@ -484,18 +484,28 @@ WRITE_CLASS_ENCODER(RGWZoneParams);
 struct RGWZone {
   string name;
   list<string> endpoints;
+  bool log_meta;
+  bool log_data;
+
+  RGWZone() : log_meta(false), log_data(false) {}
 
   void encode(bufferlist& bl) const {
-    ENCODE_START(1, 1, bl);
+    ENCODE_START(2, 1, bl);
     ::encode(name, bl);
     ::encode(endpoints, bl);
+    ::encode(log_meta, bl);
+    ::encode(log_data, bl);
     ENCODE_FINISH(bl);
   }
 
   void decode(bufferlist::iterator& bl) {
-    DECODE_START(1, bl);
+    DECODE_START(2, bl);
     ::decode(name, bl);
     ::decode(endpoints, bl);
+    if (struct_v >= 2) {
+      ::decode(log_meta, bl);
+      ::decode(log_data, bl);
+    }
     DECODE_FINISH(bl);
   }
   void dump(Formatter *f) const;
@@ -872,7 +882,8 @@ public:
   }
 
   RGWRegion region;
-  RGWZoneParams zone;
+  RGWZoneParams zone; /* internal zone params, e.g., rados pools */
+  RGWZone zone_public_config; /* external zone params, e.g., entrypoints, log flags, etc. */
   RGWRegionMap region_map;
   RGWRESTConn *rest_master_conn;
   map<string, RGWRESTConn *> zone_conn_map;
@@ -1363,6 +1374,14 @@ public:
     name = zone.log_pool.name;
   }
 
+  bool need_to_log_data() {
+    return zone_public_config.log_data;
+  }
+
+  bool need_to_log_metadata() {
+    return zone_public_config.log_meta;
+  }
+
  private:
   int process_intent_log(rgw_bucket& bucket, string& oid,
 			 time_t epoch, int flags, bool purge);