rgw: resend data log entry if took too long

If took too long, we want to resend. Also, fix issue with
renew.

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

1 files changed, 29 insertions, 12 deletions

diff --git a/src/rgw/rgw_bucket.cc b/src/rgw/rgw_bucket.cc
index f1a7b08de30..308678d42d8 100644
--- a/src/rgw/rgw_bucket.cc
+++ b/src/rgw/rgw_bucket.cc
@@ -984,6 +984,7 @@ int RGWDataChangesLog::renew_entries()
 
     store->time_log_prepare_entry(entry, ut, section, bucket.name, bl);
 
+    m[index].first.push_back(bucket.name);
     m[index].second.push_back(entry);
   }
 
@@ -1035,6 +1036,7 @@ void RGWDataChangesLog::update_renewed(string& bucket_name, utime_t& expiration)
   ChangeStatusPtr status;
   _get_change(bucket_name, status);
 
+  ldout(cct, 20) << "RGWDataChangesLog::update_renewd() bucket_name=" << bucket_name << " expiration=" << expiration << dendl;
   status->cur_expiration = expiration;
 }
 
@@ -1050,6 +1052,8 @@ int RGWDataChangesLog::add_entry(rgw_bucket& bucket) {
 
   status->lock->Lock();
 
+  ldout(cct, 20) << "RGWDataChangesLog::add_entry() bucket.name=" << bucket.name << " now=" << now << " cur_expiration=" << status->cur_expiration << dendl;
+
   if (now < status->cur_expiration) {
     /* no need to send, recently completed */
     status->lock->Unlock();
@@ -1079,22 +1083,35 @@ int RGWDataChangesLog::add_entry(rgw_bucket& bucket) {
   status->cond = new RefCountedCond;
   status->pending = true;
 
-  status->cur_sent = now;
+  string& oid = oids[choose_oid(bucket)];
+  utime_t expiration;
 
-  status->lock->Unlock();
+  int ret;
+
+  do {
+    status->cur_sent = now;
+
+    expiration = now;
+    expiration += utime_t(cct->_conf->rgw_data_log_window, 0);
+
+    status->lock->Unlock();
   
-  string& oid = oids[choose_oid(bucket)];
+    bufferlist bl;
+    rgw_data_change change;
+    change.entity_type = ENTITY_TYPE_BUCKET;
+    change.key = bucket.name;
+    ::encode(change, bl);
+    string section;
 
-  utime_t ut = ceph_clock_now(cct);
-  bufferlist bl;
-  rgw_data_change change;
-  change.entity_type = ENTITY_TYPE_BUCKET;
-  change.key = bucket.name;
-  ::encode(change, bl);
-  string section;
-  int ret = store->time_log_add(oid, ut, section, change.key, bl);
+    ldout(cct, 20) << "RGWDataChangesLog::add_entry() sending update with now=" << now << " cur_expiration=" << expiration << dendl;
 
-  status->lock->Lock();
+    ret = store->time_log_add(oid, now, section, change.key, bl);
+
+    now = ceph_clock_now(cct);
+
+    status->lock->Lock();
+
+  } while (!ret && ceph_clock_now(cct) > expiration);
 
   cond = status->cond;