Skip to content

Commit

Permalink
rgw: data sync checks empty next_marker for datalog
Browse files Browse the repository at this point in the history 
RGWReadRemoteDataLogShardCR tracks the marker and next_marker
separately, because next_marker will be empty when it reaches the end

this allows RGWDataSyncShardCR to avoid clearing its sync_marker and
restarting datalog listing from the beginning

Fixes: http://tracker.ceph.com/issues/39033

Signed-off-by: Casey Bodley <cbodley@redhat.com>
(cherry picked from commit 2ff6377)

Conflicts:
	src/rgw/rgw_data_sync.cc: no sync tracing
  • Loading branch information
@cbodley
cbodley committed Apr 1, 2019
1 parent e515077 commit a537029
Showing 1 changed file with 30 additions and 20 deletions.
50 changes: 30 additions & 20 deletions src/rgw/rgw_data_sync.cc
View file
Expand Up @@ -308,24 +308,24 @@ struct read_remote_data_log_response {
class RGWReadRemoteDataLogShardCR : public RGWCoroutine {
RGWDataSyncEnv *sync_env;

RGWRESTReadResource *http_op;
RGWRESTReadResource *http_op = nullptr;

int shard_id;
string *pmarker;
const std::string& marker;
string *pnext_marker;
list<rgw_data_change_log_entry> *entries;
bool *truncated;

read_remote_data_log_response response;

public:
RGWReadRemoteDataLogShardCR(RGWDataSyncEnv *_sync_env,
int _shard_id, string *_pmarker, list<rgw_data_change_log_entry> *_entries, bool *_truncated) : RGWCoroutine(_sync_env->cct),
sync_env(_sync_env),
http_op(NULL),
shard_id(_shard_id),
pmarker(_pmarker),
entries(_entries),
truncated(_truncated) {
RGWReadRemoteDataLogShardCR(RGWDataSyncEnv *_sync_env, int _shard_id,
const std::string& marker, string *pnext_marker,
list<rgw_data_change_log_entry> *_entries,
bool *_truncated)
: RGWCoroutine(_sync_env->cct), sync_env(_sync_env),
shard_id(_shard_id), marker(marker), pnext_marker(pnext_marker),
entries(_entries), truncated(_truncated) {
}
~RGWReadRemoteDataLogShardCR() override {
if (http_op) {
Expand All @@ -340,7 +340,7 @@ class RGWReadRemoteDataLogShardCR : public RGWCoroutine {
snprintf(buf, sizeof(buf), "%d", shard_id);
rgw_http_param_pair pairs[] = { { "type" , "data" },
{ "id", buf },
{ "marker", pmarker->c_str() },
{ "marker", marker.c_str() },
{ "extra-info", "true" },
{ NULL, NULL } };

Expand All @@ -366,7 +366,7 @@ class RGWReadRemoteDataLogShardCR : public RGWCoroutine {
}
entries->clear();
entries->swap(response.entries);
*pmarker = response.marker;
*pnext_marker = response.marker;
*truncated = response.truncated;
return set_cr_done();
}
Expand Down Expand Up @@ -1112,6 +1112,7 @@ class RGWDataSyncShardCR : public RGWCoroutine {

RGWDataSyncShardMarkerTrack *marker_tracker;

std::string next_marker;
list<rgw_data_change_log_entry> log_entries;
list<rgw_data_change_log_entry>::iterator log_iter;
bool truncated;
Expand Down Expand Up @@ -1158,7 +1159,7 @@ class RGWDataSyncShardCR : public RGWCoroutine {
public:
RGWDataSyncShardCR(RGWDataSyncEnv *_sync_env,
rgw_pool& _pool,
uint32_t _shard_id, rgw_data_sync_marker& _marker, bool *_reset_backoff) : RGWCoroutine(_sync_env->cct),
uint32_t _shard_id, const rgw_data_sync_marker& _marker, bool *_reset_backoff) : RGWCoroutine(_sync_env->cct),
sync_env(_sync_env),
pool(_pool),
shard_id(_shard_id),
Expand Down Expand Up @@ -1389,7 +1390,8 @@ class RGWDataSyncShardCR : public RGWCoroutine {
#define INCREMENTAL_MAX_ENTRIES 100
ldout(sync_env->cct, 20) << __func__ << ":" << __LINE__ << ": shard_id=" << shard_id << " sync_marker=" << sync_marker.marker << dendl;
spawned_keys.clear();
yield call(new RGWReadRemoteDataLogShardCR(sync_env, shard_id, &sync_marker.marker, &log_entries, &truncated));
yield call(new RGWReadRemoteDataLogShardCR(sync_env, shard_id, sync_marker.marker,
&next_marker, &log_entries, &truncated));
if (retcode < 0) {
ldout(sync_env->cct, 0) << "ERROR: failed to read remote data log info: ret=" << retcode << dendl;
stop_spawned_services();
Expand Down Expand Up @@ -1432,11 +1434,17 @@ class RGWDataSyncShardCR : public RGWCoroutine {
}
/* not waiting for child here */
}
}
ldout(sync_env->cct, 20) << __func__ << ":" << __LINE__ << ": shard_id=" << shard_id << " sync_marker=" << sync_marker.marker << " truncated=" << truncated << dendl;
if (!truncated) {
yield wait(get_idle_interval());
}
}
ldout(sync_env->cct, 20) << __func__ << ":" << __LINE__ << ": shard_id=" << shard_id << " sync_marker=" << sync_marker.marker
<< " next_marker=" << next_marker << " truncated=" << truncated << dendl;
if (!truncated) {
yield wait(get_idle_interval());
}
if (!next_marker.empty()) {
sync_marker.marker = next_marker;
} else if (!log_entries.empty()) {
sync_marker.marker = log_entries.back().log_id;
}
} while (true);
}
return 0;
Expand Down Expand Up @@ -2088,6 +2096,7 @@ class RGWReadPendingBucketShardsCoroutine : public RGWCoroutine {
rgw_data_sync_marker* sync_marker;
int count;

std::string next_marker;
list<rgw_data_change_log_entry> log_entries;
bool truncated;

Expand Down Expand Up @@ -2123,7 +2132,8 @@ int RGWReadPendingBucketShardsCoroutine::operate()
marker = sync_marker->marker;
count = 0;
do{
yield call(new RGWReadRemoteDataLogShardCR(sync_env, shard_id, &marker, &log_entries, &truncated));
yield call(new RGWReadRemoteDataLogShardCR(sync_env, shard_id, marker,
&next_marker, &log_entries, &truncated));

if (retcode == -ENOENT) {
break;
Expand Down

0 comments on commit a537029

Please sign in to comment.