rgw: time out pending olh entries

remove old pending entries

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

diff --git a/src/common/config_opts.h b/src/common/config_opts.h
index 08739c417205ed546478ce9183a321b229da59ed..74f1b8cfc105ce75166215ef75d966533714a944 100644 (file)
--- a/src/common/config_opts.h
+++ b/src/common/config_opts.h
@@ -951,6 +951,8 @@ OPTION(rgw_user_quota_sync_wait_time, OPT_INT, 3600 * 24) // min time between tw
 
 OPTION(rgw_multipart_min_part_size, OPT_INT, 5 * 1024 * 1024) // min size for each part (except for last one) in multipart upload
 
+OPTION(rgw_olh_pending_timeout_sec, OPT_INT, 3600) // time until we retire a pending olh change
+
 OPTION(mutex_perf_counter, OPT_BOOL, false) // enable/disable mutex perf counter
 OPTION(throttler_perf_counter, OPT_BOOL, true) // enable/disable throttler perf counter
 

diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index ca57d19258bb453585fd3debc19765c82c1ff0e1..e05c5bc35f4b942f09bba7d12d303933ebf68827 100644 (file)
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -5928,11 +5928,80 @@ int RGWRados::get_olh(rgw_obj& obj, RGWOLHInfo *olh)
   return 0;
 }
 
+void RGWRados::check_pending_olh_entries(map<string, bufferlist>& pending_entries, 
+                                         map<string, bufferlist> *rm_pending_entries)
+{
+  map<string, bufferlist>::iterator iter = pending_entries.begin();
+
+  utime_t now = ceph_clock_now(cct);
+
+  while (iter != pending_entries.end()) {
+    bufferlist::iterator biter = iter->second.begin();
+    RGWOLHPendingInfo pending_info;
+    try {
+      ::decode(pending_info, biter);
+    } catch (buffer::error& err) {
+      /* skipping bad entry, we could remove it but it might hide a bug */
+      ldout(cct, 0) << "ERROR: failed to decode pending entry " << iter->first << dendl;
+      ++iter;
+      continue;
+    }
+
+    map<string, bufferlist>::iterator cur_iter = iter;
+    ++iter;
+    if (now - pending_info.time >= cct->_conf->rgw_olh_pending_timeout_sec) {
+      (*rm_pending_entries)[cur_iter->first] = cur_iter->second;
+      pending_entries.erase(cur_iter);
+    }
+  }
+}
+
+int RGWRados::remove_olh_pending_entries(RGWObjState& state, rgw_obj& olh_obj, map<string, bufferlist>& pending_attrs)
+{
+  ObjectWriteOperation op;
+
+  bucket_index_guard_olh_op(state, op);
+
+  for (map<string, bufferlist>::iterator iter = pending_attrs.begin(); iter != pending_attrs.end(); ++iter) {
+    op.rmxattr(iter->first.c_str());
+  }
+
+  rgw_rados_ref ref;
+  rgw_bucket bucket;
+  int r = get_obj_ref(olh_obj, &ref, &bucket);
+  if (r < 0) {
+    return r;
+  }
+
+  /* update olh object */
+  r = ref.ioctx.operate(ref.oid, &op);
+  if (r == -ENOENT || r == -ECANCELED) {
+    /* raced with some other change, shouldn't sweat about it */
+    r = 0;
+  }
+  if (r < 0) {
+    ldout(cct, 0) << "ERROR: could not apply olh update, r=" << r << dendl;
+    return r;
+  }
+
+  return 0;
+}
+
 int RGWRados::follow_olh(RGWObjectCtx& obj_ctx, RGWObjState *state, rgw_obj& olh_obj, rgw_obj *target)
 {
   map<string, bufferlist> pending_entries;
   filter_attrset(state->attrset, RGW_ATTR_OLH_PENDING_PREFIX, &pending_entries);
 
+  map<string, bufferlist> rm_pending_entries;
+  check_pending_olh_entries(pending_entries, &rm_pending_entries);
+
+  if (!rm_pending_entries.empty()) {
+    int ret = remove_olh_pending_entries(*state, olh_obj, rm_pending_entries);
+    if (ret < 0) {
+      ldout(cct, 20) << "ERROR: rm_pending_entries returned ret=" << ret << dendl;
+      return ret;
+    }
+  }
   if (!pending_entries.empty()) {
     ldout(cct, 20) << __func__ << "(): found pending entries, need to update_olh() on bucket=" << olh_obj.bucket << dendl;
 

diff --git a/src/rgw/rgw_rados.h b/src/rgw/rgw_rados.h
index 98da82ce3153e76b96129233428204e65b926f87..a21e4c52e9496e220fc89a25d87354ec42f41071 100644 (file)
--- a/src/rgw/rgw_rados.h
+++ b/src/rgw/rgw_rados.h
@@ -1728,6 +1728,8 @@ public:
   int set_olh(RGWObjectCtx& obj_ctx, const string& bucket_owner, rgw_obj& target_obj, bool delete_marker, rgw_bucket_dir_entry_meta *meta);
   int unlink_obj_instance(RGWObjectCtx& obj_ctx, const string& bucket_owner, rgw_obj& target_obj);
 
+  void check_pending_olh_entries(map<string, bufferlist>& pending_entries, map<string, bufferlist> *rm_pending_entries);
+  int remove_olh_pending_entries(RGWObjState& state, rgw_obj& olh_obj, map<string, bufferlist>& pending_attrs);
   int follow_olh(RGWObjectCtx& ctx, RGWObjState *state, rgw_obj& olh_obj, rgw_obj *target);
   int get_olh(rgw_obj& obj, RGWOLHInfo *olh);