diff --git a/src/lib/homestore_backend/pg_blob_iterator.cpp b/src/lib/homestore_backend/pg_blob_iterator.cpp index 5d9f3808..926c2a7e 100644 --- a/src/lib/homestore_backend/pg_blob_iterator.cpp +++ b/src/lib/homestore_backend/pg_blob_iterator.cpp @@ -57,13 +57,15 @@ HSHomeObject::PGBlobIterator::PGBlobIterator(HSHomeObject& home_obj, homestore:: : a.info.create_lsn < b.info.create_lsn; }); } + LOGI("Created PGBlobIterator: pg={}, snapshot_lsn={}, shards={}, batch_size={}", pg_id, snp_start_lsn_, + shard_list_.size(), max_batch_size_); } // result represents if the objId is valid and the cursors are updated bool HSHomeObject::PGBlobIterator::update_cursor(const objId& id) { std::lock_guard lock(op_mut_); if (stopped_) { - LOGW("PGBlobIterator already stopped, rejecting request"); + LOGD("Rejecting cursor update on stopped PGBlobIterator: pg={}, requested_obj={}", pg_id, id.to_string()); return false; } @@ -76,7 +78,7 @@ bool HSHomeObject::PGBlobIterator::update_cursor(const objId& id) { // Resend batch if (id.value == cur_obj_id.value) { - LOGT("Resending the same batch, objId={} is the same as cur_obj_id={}", id.to_string(), cur_obj_id.to_string()); + LOGI("Resending resync object: pg={}, obj={}", pg_id, id.to_string()); COUNTER_INCREMENT(*metrics_, snp_dnr_resend_count, 1); return true; } @@ -95,15 +97,19 @@ bool HSHomeObject::PGBlobIterator::update_cursor(const objId& id) { } if (found) { cur_obj_id = id; + LOGI("Resuming resync at shard: pg={}, shard_seq=0x{:x}, shard_index={}", pg_id, id.shard_seq_num, + cur_shard_idx_); } else { - LOGE("Cur_obj_id is 0|0, but requested id {} not found in shard list", id.to_string()); + LOGE("Cannot resume resync: pg={}, requested_obj={} is not in the snapshot shard list", pg_id, + id.to_string()); } return found; } auto next_obj_id = expected_next_obj_id(); if (id.value != next_obj_id.value) { - LOGE("Invalid objId, expected next_id={}, actual={}", next_obj_id.to_string(), id.to_string()); + LOGE("Invalid resync cursor: pg={}, current_obj={}, expected_obj={}, requested_obj={}", pg_id, + cur_obj_id.to_string(), next_obj_id.to_string(), id.to_string()); return false; } // next shard @@ -117,13 +123,15 @@ bool HSHomeObject::PGBlobIterator::update_cursor(const objId& id) { cur_batch_blob_count_ = 0; } cur_obj_id = id; + LOGD("Advanced resync cursor: pg={}, obj={}, shard_index={}, blob_index={}", pg_id, id.to_string(), + cur_shard_idx_, cur_start_blob_idx_); return true; } void HSHomeObject::PGBlobIterator::reset_cursor() { std::lock_guard lock(op_mut_); if (stopped_) { - LOGW("PGBlobIterator already stopped, rejecting request"); + LOGD("Ignoring cursor reset on stopped PGBlobIterator: pg={}", pg_id); return; } @@ -133,6 +141,7 @@ void HSHomeObject::PGBlobIterator::reset_cursor() { cur_start_blob_idx_ = 0; cur_batch_blob_count_ = 0; cur_batch_start_time_ = Clock::time_point{}; + LOGI("Reset resync cursor: pg={}", pg_id); } objId HSHomeObject::PGBlobIterator::expected_next_obj_id() const { @@ -167,19 +176,19 @@ PG* HSHomeObject::PGBlobIterator::get_pg_metadata() const { bool HSHomeObject::PGBlobIterator::create_pg_snapshot_data(sisl::io_blob_safe& meta_blob) { std::lock_guard lock(op_mut_); if (stopped_) { - LOGW("PGBlobIterator already stopped, rejecting request"); + LOGD("Rejecting PG metadata request on stopped PGBlobIterator: pg={}", pg_id); return false; } #ifdef _PRERELEASE if (iomgr_flip::instance()->test_flip("pg_blob_iterator_create_snapshot_data_error")) { - LOGW("Simulating creating pg snapshot data error"); + LOGD("Simulating resync PG metadata creation failure: pg={}", pg_id); return false; } #endif auto pg = home_obj_._pg_map[pg_id].get(); if (pg == nullptr) { - LOGE("PG not found in pg_map, pg={}", pg_id); + LOGE("Cannot create resync PG metadata: pg={} not found", pg_id); return false; } auto pg_info = pg->pg_info_; @@ -206,32 +215,41 @@ bool HSHomeObject::PGBlobIterator::create_pg_snapshot_data(sisl::io_blob_safe& m builder_.FinishSizePrefixed(pg_entry); pack_resync_message(meta_blob, SyncMessageType::PG_META); + LOGI("Created resync PG metadata: pg={}, shards={}, active_blobs={}, occupied_bytes={}", pg_id, + shard_ids.size(), total_blobs, total_bytes); return true; } bool HSHomeObject::PGBlobIterator::generate_shard_blob_list() { std::lock_guard lock(op_mut_); if (stopped_) { - LOGW("PGBlobIterator already stopped, rejecting request"); + LOGD("Rejecting shard blob-list request on stopped PGBlobIterator: pg={}", pg_id); return false; } #ifdef _PRERELEASE if (iomgr_flip::instance()->test_flip("pg_blob_iterator_generate_shard_blob_list_error")) { - LOGW("Simulating generating shard blob list error"); + LOGD("Simulating resync shard blob-list query failure: pg={}, shard_seq=0x{:x}", pg_id, + cur_obj_id.shard_seq_num); return false; } #endif auto r = home_obj_.query_blobs_in_shard(pg_id, cur_obj_id.shard_seq_num, 0, UINT64_MAX); - if (!r) { return false; } + if (!r) { + LOGE("Failed to query resync shard blobs: pg={}, shard_seq=0x{:x}, error={}", pg_id, + cur_obj_id.shard_seq_num, r.error()); + return false; + } cur_blob_list_ = r.value(); + LOGD("Prepared resync shard blob list: pg={}, shard_seq=0x{:x}, blobs={}", pg_id, cur_obj_id.shard_seq_num, + cur_blob_list_.size()); return true; } bool HSHomeObject::PGBlobIterator::create_shard_snapshot_data(sisl::io_blob_safe& meta_blob) { std::lock_guard lock(op_mut_); if (stopped_) { - LOGW("PGBlobIterator already stopped, rejecting request"); + LOGD("Rejecting shard metadata request on stopped PGBlobIterator: pg={}", pg_id); return false; } @@ -245,6 +263,8 @@ bool HSHomeObject::PGBlobIterator::create_shard_snapshot_data(sisl::io_blob_safe builder_.FinishSizePrefixed(shard_entry); pack_resync_message(meta_blob, SyncMessageType::SHARD_META); + LOGI("Created resync shard metadata: pg={}, shard_id=0x{:x}, shard_seq=0x{:x}, state={}, blobs={}", pg_id, + shard.info.id, cur_obj_id.shard_seq_num, static_cast< uint8_t >(shard.info.state), cur_blob_list_.size()); prefetch_blobs_snapshot_data(); return true; } @@ -264,28 +284,31 @@ HSHomeObject::PGBlobIterator::load_blob_data_with_blkid(shard_id_t shard_id, blo sgs.size = total_size; sgs.iovs.emplace_back(iovec{.iov_base = read_buf.bytes(), .iov_len = read_buf.size()}); - LOGD("Blob get request: shardID=0x{:x}, pg={}, shard=0x{:x}, blob_id={}, blkid={}", shard_id, - (shard_id >> homeobject::shard_width), (shard_id & homeobject::shard_mask), blob_id, blkid.to_string()); + LOGT("Reading resync blob: pg={}, shard=0x{:x}, blob={}, blkid={}, bytes={}", + (shard_id >> homeobject::shard_width), (shard_id & homeobject::shard_mask), blob_id, blkid.to_string(), + total_size); return repl_dev_->async_read(blkid, sgs, total_size) .thenValue([this, blob_id, shard_id, blkid, read_buf = std::move(read_buf)]( auto&& result) mutable -> BlobManager::AsyncResult< blob_read_result > { if (result) { - LOGE("Failed to get blob, shardID=0x{:x}, pg={}, shard=0x{:x}, blob_id={}, err={}", shard_id, + LOGE("Failed to read resync blob: pg={}, shard=0x{:x}, blob={}, blkid={}, error={}", (shard_id >> homeobject::shard_width), (shard_id & homeobject::shard_mask), blob_id, - result.value()); + blkid.to_string(), result.value()); return folly::makeUnexpected(BlobError(BlobErrorCode::READ_FAILED)); } if (home_obj_.verify_blob(read_buf.cbytes(), shard_id, blob_id)) { - LOGD("Blob get success: shardID=0x{:x}, pg={}, shard=0x{:x}, blob_id={}", shard_id, - (shard_id >> homeobject::shard_width), (shard_id & homeobject::shard_mask), blob_id); + LOGT("Read resync blob: pg={}, shard=0x{:x}, blob={}", (shard_id >> homeobject::shard_width), + (shard_id & homeobject::shard_mask), blob_id); return blob_read_result(blob_id, std::move(read_buf), ResyncBlobState::NORMAL); } // verify_blob failed — check if GC moved the blob to a new blkid since cur_blob_list_ was captured. auto index_table = home_obj_.get_index_table(pg_id); if (!index_table) { - LOGE("PG not found when checking index for GC race, pg={}, blob={}", pg_id, blob_id); + LOGE( + "Cannot resolve resync blob verification failure: pg={} no longer exists, shard_id=0x{:x}, blob={}", + pg_id, shard_id, blob_id); return folly::makeUnexpected(BlobError(BlobErrorCode::READ_FAILED)); } auto current_pbas = home_obj_.get_blob_from_index_table(index_table, shard_id, blob_id); @@ -293,21 +316,22 @@ HSHomeObject::PGBlobIterator::load_blob_data_with_blkid(shard_id_t shard_id, blo // Blob was deleted concurrently after generate_shard_blob_list captured its pbas. // Do not send stale bytes as CORRUPTED — signal READ_FAILED so the snapshot restarts // and generate_shard_blob_list picks up tombstone_pbas, skipping the blob cleanly. - LOGI("Blob deleted during GC race, shard=0x{:x}, blob={}, triggering snapshot restart", shard_id, - blob_id); + LOGW("Resync blob was deleted during read; restarting snapshot: pg={}, shard_id=0x{:x}, blob={}", + pg_id, shard_id, blob_id); return folly::makeUnexpected(BlobError(BlobErrorCode::READ_FAILED)); } if (current_pbas.value() == blkid) { // blkid unchanged — genuinely corrupted data at this location. // Metrics for corrupted blobs are handled on the follower side. - LOGE("Blob verification failed, shardID=0x{:x}, pg={}, shard=0x{:x}, blob_id={}", shard_id, - (shard_id >> homeobject::shard_width), (shard_id & homeobject::shard_mask), blob_id); + LOGE("Resync blob verification failed: pg={}, shard=0x{:x}, blob={}, blkid={}", + (shard_id >> homeobject::shard_width), (shard_id & homeobject::shard_mask), blob_id, + blkid.to_string()); return blob_read_result(blob_id, std::move(read_buf), ResyncBlobState::CORRUPTED); } // GC moved the blob — retry with the updated blkid. Folly flattens the returned future. - LOGI("GC moved blob detected during snapshot: shard=0x{:x}, blob={}, old_blkid={}, new_blkid={}", shard_id, - blob_id, blkid.to_string(), current_pbas.value().to_string()); + LOGI("Resync blob relocated by GC during read; retrying: pg={}, shard_id=0x{:x}, blob={}, old_blkid={}, new_blkid={}", + pg_id, shard_id, blob_id, blkid.to_string(), current_pbas.value().to_string()); return load_blob_data_with_blkid(shard_id, blob_id, current_pbas.value()); }); } @@ -317,32 +341,30 @@ bool HSHomeObject::PGBlobIterator::prefetch_blobs_snapshot_data() { auto skipped_blobs = 0; // limit inflight prefect data to 2x of max_batch_size. std::vector< BlobInfo > prefetch_list; - LOGD("prefetch_blobs_snapshot_data, inflight={}, idx={}, max_batch_size * 2 = {}", inflight_prefetch_bytes_, - cur_start_blob_idx_, max_batch_size_ * 2); auto idx = cur_start_blob_idx_; // On batch resend, retained look-ahead may already consume part of the 2x budget. Allow missing blobs before the // earliest retained blob to bypass the limit so the current batch can always be rebuilt. const auto prefetch_frontier = prefetched_blobs_.empty() ? blob_id_t{0} : prefetched_blobs_.begin()->first; + LOGT("Prefetching blobs: pg={}, shard_seq=0x{:x}, cursor_blob={}, frontier={}, inflight_bytes={}", + pg_id, cur_obj_id.shard_seq_num, cur_start_blob_idx_, prefetch_frontier, inflight_prefetch_bytes_); while (idx < cur_blob_list_.size() && (inflight_prefetch_bytes_ < max_batch_size_ * 2 || cur_blob_list_[idx].blob_id < prefetch_frontier)) { auto info = cur_blob_list_[idx++]; total_blobs++; // handle deleted object if (info.pbas == tombstone_pbas) { - LOGT("Blob is deleted: shardID=0x{:x}, pg={}, shard=0x{:x}, blob_id={}, blkid={}", info.shard_id, - (info.shard_id >> homeobject::shard_width), (info.shard_id & homeobject::shard_mask), info.blob_id, - info.pbas.to_string()); + LOGT("Skipping deleted resync blob: pg={}, shard=0x{:x}, blob={}", + (info.shard_id >> homeobject::shard_width), (info.shard_id & homeobject::shard_mask), info.blob_id); // ignore skipped_blobs++; continue; } if (prefetched_blobs_.contains(info.blob_id)) { - LOGT("Blob {} has prefetched, skipping", info.blob_id); + LOGT("Retaining prefetched resync blob: pg={}, blob={}", pg_id, info.blob_id); skipped_blobs++; continue; } inflight_prefetch_bytes_ += info.pbas.blk_count() * repl_dev_->get_blk_size(); - LOGD("Will prefetch {}: frontier {}, inflight {}", info.blob_id, prefetch_frontier, inflight_prefetch_bytes_); prefetch_list.emplace_back(info); } // POC: sort the prefetch_list by pbas, trying to let IO submitted to disk more sequential. @@ -351,21 +373,20 @@ bool HSHomeObject::PGBlobIterator::prefetch_blobs_snapshot_data() { for (auto info : prefetch_list) { #ifdef _PRERELEASE if (iomgr_flip::instance()->test_flip("pg_blob_iterator_load_blob_data_error")) { - LOGW("Simulating loading blob data error"); + LOGD("Simulating resync blob prefetch failure: pg={}, blob={}", pg_id, info.blob_id); prefetched_blobs_.emplace(info.blob_id, folly::makeUnexpected(BlobError(BlobErrorCode::READ_FAILED))); continue; } auto delay = iomgr_flip::instance()->get_test_flip< long >("simulate_read_snapshot_load_blob_delay", static_cast< long >(info.blob_id)); - LOGD("simulate_read_snapshot_load_blob_delay flip, triggered={}, blob={}", delay.has_value(), info.blob_id); if (delay) { - LOGI("Simulating pg blob iterator load data with delay, delay={}, blob_id={}", delay.get(), info.blob_id); + LOGD("Simulating resync blob read delay: pg={}, blob={}, delay_ms={}", pg_id, info.blob_id, delay.get()); std::this_thread::sleep_for(std::chrono::milliseconds(delay.get())); } #endif auto blob_start = Clock::now(); - LOGT("submitting io for blob {}", info.blob_id); + LOGT("Submitting resync blob read: pg={}, shard_id=0x{:x}, blob={}", pg_id, info.shard_id, info.blob_id); // Fixme: Re-enable retries uint8_t retries = HS_BACKEND_DYNAMIC_CONFIG(snapshot_blob_load_retry); prefetched_blobs_.emplace( info.blob_id, @@ -374,24 +395,28 @@ bool HSHomeObject::PGBlobIterator::prefetch_blobs_snapshot_data() { .thenValue( [&, info, blob_start](auto&& result) mutable -> BlobManager::AsyncResult< blob_read_result > { if (result.hasError() && result.error().code == BlobErrorCode::READ_FAILED) { - LOGE("Failed to retrieve blob for shardID=0x{:x}, pg={}, shard=0x{:x} blob={} pbas={}", - info.shard_id, (info.shard_id >> homeobject::shard_width), - (info.shard_id & homeobject::shard_mask), info.blob_id, info.pbas.to_string()); + LOGE("Failed to prefetch resync blob: pg={}, shard=0x{:x}, blob={}, blkid={}", + info.shard_id >> homeobject::shard_width, info.shard_id & homeobject::shard_mask, + info.blob_id, info.pbas.to_string()); COUNTER_INCREMENT(*metrics_, snp_dnr_error_count, 1); } else { - LOGT("retrieved blob, blob={} pbas={}", info.blob_id, info.pbas.to_string()); + LOGT("Prefetched resync blob: pg={}, blob={}, blkid={}", pg_id, info.blob_id, + info.pbas.to_string()); HISTOGRAM_OBSERVE(*metrics_, snp_dnr_blob_process_latency, get_elapsed_time_us(blob_start)); } return result; })); } + LOGD("Resync prefetch window: pg={}, shard_seq=0x{:x}, cursor_blob={}, frontier={}, submitted_blobs={}, skipped_blobs={}, inflight_bytes={}, limit_bytes={}", + pg_id, cur_obj_id.shard_seq_num, cur_start_blob_idx_, prefetch_frontier, prefetch_list.size(), skipped_blobs, + inflight_prefetch_bytes_, max_batch_size_ * 2); return true; } bool HSHomeObject::PGBlobIterator::create_blobs_snapshot_data(sisl::io_blob_safe& data_blob) { std::lock_guard lock(op_mut_); if (stopped_) { - LOGW("PGBlobIterator already stopped, rejecting request"); + LOGD("Rejecting blob batch request on stopped PGBlobIterator: pg={}", pg_id); return false; } @@ -414,19 +439,19 @@ bool HSHomeObject::PGBlobIterator::create_blobs_snapshot_data(sisl::io_blob_safe total_blobs++; // handle deleted object if (info.pbas == tombstone_pbas) { - LOGT("Blob is deleted: shardID=0x{:x}, pg={}, shard=0x{:x}, blob_id={}, blkid={}", info.shard_id, - (info.shard_id >> homeobject::shard_width), (info.shard_id & homeobject::shard_mask), info.blob_id, - info.pbas.to_string()); + LOGT("Skipping deleted resync blob: pg={}, shard=0x{:x}, blob={}", + info.shard_id >> homeobject::shard_width, info.shard_id & homeobject::shard_mask, info.blob_id); // ignore skipped_blobs++; continue; } - LOGT("Getting prefetched data for blob {}", info.blob_id); auto it = prefetched_blobs_.find(info.blob_id); if (it == prefetched_blobs_.end()) { hit_error = true; - LOGE("blob {} not found in prefetched blob map", info.blob_id); + LOGE("Resync batch cannot find prefetched blob: pg={}, shard_seq=0x{:x}, batch={}, blob={}, cursor_blob={}, inflight_bytes={}, prefetched_blobs={}", + pg_id, cur_obj_id.shard_seq_num, cur_obj_id.batch_id, info.blob_id, cur_start_blob_idx_, + inflight_prefetch_bytes_, prefetched_blobs_.size()); break; } auto const expect_blob_size = info.pbas.blk_count() * repl_dev_->get_blk_size(); @@ -435,7 +460,8 @@ bool HSHomeObject::PGBlobIterator::create_blobs_snapshot_data(sisl::io_blob_safe inflight_prefetch_bytes_ -= expect_blob_size; if (res.hasError()) { - LOGE("blob {} hit error {}", info.blob_id, res.error()); + LOGE("Resync batch failed to retrieve blob: pg={}, shard_seq=0x{:x}, batch={}, blob={}, error={}", + pg_id, cur_obj_id.shard_seq_num, cur_obj_id.batch_id, info.blob_id, res.error()); hit_error = true; break; } @@ -447,7 +473,8 @@ bool HSHomeObject::PGBlobIterator::create_blobs_snapshot_data(sisl::io_blob_safe } if (skipped_blobs + fetched_blobs != total_blobs) { - LOGE("total {} blobs, skipped {}, expect {} but only get {} blobs", total_blobs, skipped_blobs, + LOGE("Incomplete resync batch: pg={}, shard_seq=0x{:x}, batch={}, examined_blobs={}, skipped_blobs={}, expected_blobs={}, fetched_blobs={}", + pg_id, cur_obj_id.shard_seq_num, cur_obj_id.batch_id, total_blobs, skipped_blobs, total_blobs - skipped_blobs, fetched_blobs); hit_error = true; } @@ -462,9 +489,9 @@ bool HSHomeObject::PGBlobIterator::create_blobs_snapshot_data(sisl::io_blob_safe if (idx == cur_blob_list_.size()) { end_of_shard = true; } builder_.FinishSizePrefixed(CreateResyncBlobDataBatchDirect(builder_, &blob_entries, end_of_shard)); - LOGD("create blobs snapshot data batch: shard_seq_num={}, batch_num={}, total_bytes={}, blob_num={}, " - "end_of_shard={}", - cur_obj_id.shard_seq_num, cur_obj_id.batch_id, total_bytes, blob_entries.size(), end_of_shard); + LOGI("Created resync shard batch: pg={}, shard_seq=0x{:x}, batch={}, blobs={}, skipped_blobs={}, bytes={}, end_of_shard={}, next_blob={}", + pg_id, cur_obj_id.shard_seq_num, cur_obj_id.batch_id, blob_entries.size(), skipped_blobs, total_bytes, + end_of_shard, idx); COUNTER_INCREMENT(*metrics_, snp_dnr_load_blob, blob_entries.size()); COUNTER_INCREMENT(*metrics_, snp_dnr_load_bytes, total_bytes); @@ -479,7 +506,8 @@ void HSHomeObject::PGBlobIterator::pack_resync_message(sisl::io_blob_safe& dest_ header.payload_size = builder_.GetSize(); header.payload_crc = crc32_ieee(init_crc32, builder_.GetBufferPointer(), builder_.GetSize()); header.seal(); - LOGD("Creating resync message in pg={} with header={}", pg_id, header.to_string()); + LOGT("Packed resync message: pg={}, type={}, payload_bytes={}, payload_crc=0x{:x}", pg_id, + static_cast< uint32_t >(type), header.payload_size, header.payload_crc); dest_blob = sisl::io_blob_safe{static_cast< unsigned int >(builder_.GetSize() + sizeof(SyncMessageHeader))}; std::memcpy(dest_blob.bytes(), &header, sizeof(SyncMessageHeader)); @@ -494,13 +522,14 @@ void HSHomeObject::PGBlobIterator::stop() { // Mutex is sufficient here as operations of a PGBlobIterator are called sequentially std::lock_guard lock(op_mut_); - LOGI("PGBlobIterator stopping, pg={}, group_id={}", pg_id, boost::uuids::to_string(group_id)); + LOGI("Stopping PGBlobIterator: pg={}, prefetched_blobs={}, inflight_bytes={}", pg_id, prefetched_blobs_.size(), + inflight_prefetch_bytes_); // Set stopped flag to prevent future operations stopped_ = true; // Wait for all inflight prefetch blobs to finish and drain the data for (auto& blob : prefetched_blobs_) { - LOGD("Waiting Blob {} ready and drain it", blob.first); + LOGT("Draining prefetched resync blob: pg={}, blob={}", pg_id, blob.first); std::move(blob.second).get(); } prefetched_blobs_.clear(); @@ -509,7 +538,7 @@ void HSHomeObject::PGBlobIterator::stop() { // Clear the builder to ensure no partial data remains builder_.Clear(); - LOGI("PGBlobIterator stopped successfully, pg={}, group_id={}", pg_id, boost::uuids::to_string(group_id)); + LOGI("Stopped PGBlobIterator: pg={}", pg_id); } } // namespace homeobject diff --git a/src/lib/homestore_backend/replication_state_machine.cpp b/src/lib/homestore_backend/replication_state_machine.cpp index d2100fb9..a393fd99 100644 --- a/src/lib/homestore_backend/replication_state_machine.cpp +++ b/src/lib/homestore_backend/replication_state_machine.cpp @@ -272,12 +272,13 @@ homestore::AsyncReplResult<> ReplicationStateMachine::create_snapshot(std::shared_ptr< homestore::snapshot_context > context) { std::lock_guard lk(m_snapshot_lock); if (get_snapshot_context() != nullptr && context->get_lsn() < m_snapshot_context->get_lsn()) { - LOGI("Skipping create snapshot, new snapshot lsn={} is less than current snapshot lsn={}", context->get_lsn(), - m_snapshot_context->get_lsn()); + LOGD("Skipping older snapshot context: group={}, requested_lsn={}, current_lsn={}", + boost::uuids::to_string(repl_dev()->group_id()), context->get_lsn(), m_snapshot_context->get_lsn()); return folly::makeSemiFuture< homestore::ReplResult< folly::Unit > >(folly::Unit{}); } - LOGI("create snapshot with lsn={}", context->get_lsn()); + LOGI("Created resync snapshot context: group={}, lsn={}", boost::uuids::to_string(repl_dev()->group_id()), + context->get_lsn()); set_snapshot_context(context); return folly::makeSemiFuture< homestore::ReplResult< folly::Unit > >(folly::Unit{}); } @@ -285,9 +286,9 @@ ReplicationStateMachine::create_snapshot(std::shared_ptr< homestore::snapshot_co bool ReplicationStateMachine::apply_snapshot(std::shared_ptr< homestore::snapshot_context > context) { #ifdef _PRERELEASE auto delay = iomgr_flip::instance()->get_test_flip< long >("simulate_apply_snapshot_delay"); - LOGD("simulate_apply_snapshot_delay flip, triggered={}", delay.has_value()); if (delay) { - LOGI("Simulating apply snapshot with delay, delay={}", delay.get()); + LOGD("Simulating resync snapshot apply delay: group={}, delay_ms={}", + boost::uuids::to_string(repl_dev()->group_id()), delay.get()); std::this_thread::sleep_for(std::chrono::milliseconds(delay.get())); } // Currently, nuraft will pause state machine and resume it after the last snp obj is saved. So we don't need to @@ -297,6 +298,8 @@ bool ReplicationStateMachine::apply_snapshot(std::shared_ptr< homestore::snapsho std::lock_guard lk(m_snapshot_lock); set_snapshot_context(context); + LOGI("Applied resync snapshot: group={}, lsn={}", boost::uuids::to_string(repl_dev()->group_id()), + context->get_lsn()); return true; } @@ -316,7 +319,7 @@ int ReplicationStateMachine::read_snapshot_obj(std::shared_ptr< homestore::snaps context->get_lsn()); auto pg_iter_ptr = new std::shared_ptr< HSHomeObject::PGBlobIterator >(pg_iter); snp_obj->user_ctx = static_cast< void* >(pg_iter_ptr); - LOGD("Allocated new pg blob iterator={}, group={}, lsn={}", snp_obj->user_ctx, + LOGD("Allocated PGBlobIterator: iterator={}, group={}, snapshot_lsn={}", snp_obj->user_ctx, boost::uuids::to_string(repl_dev()->group_id()), context->get_lsn()); } else { auto pg_iter_ptr = static_cast< std::shared_ptr< HSHomeObject::PGBlobIterator >* >(snp_obj->user_ctx); @@ -340,14 +343,14 @@ int ReplicationStateMachine::read_snapshot_obj(std::shared_ptr< homestore::snaps if (snp_obj->offset == LAST_OBJ_ID) { // No more shards to read, baseline resync is finished after this. snp_obj->is_last_obj = true; - LOGD("Read snapshot end, {}", log_str); + LOGI("Completed resync snapshot read: pg={}, {}", pg_iter->pg_id, log_str); return 0; } auto obj_id = objId(snp_obj->offset); log_str = fmt::format("{} shard_seq_num=0x{:x} batch_num={}", log_str, obj_id.shard_seq_num, obj_id.batch_id); - LOGI("Read current snp obj {}", log_str) + LOGI("Reading resync object: {}", log_str) if (!pg_iter->update_cursor(obj_id)) { // There is a known corner case (not sure if it is the only case): If free_user_snp_ctx and read_snapshot_obj // (we enable NuRaft bg snapshot) occur at the same time, and free_user_snp_ctx is called first, pg_iter is @@ -355,7 +358,7 @@ int ReplicationStateMachine::read_snapshot_obj(std::shared_ptr< homestore::snaps // will be x|y which may hit into invalid objId condition. If inconsistency happens, reset the cursor to the // beginning(0|0), send an empty message, and let follower validate (lsn may change) and reset its cursor to the // checkpoint to proceed with snapshot resync. - LOGW("Invalid objId in snapshot read, reset cursor to the beginning, {}", log_str); + LOGW("Invalid resync cursor; resetting donor cursor: {}", log_str); pg_iter->reset_cursor(); return 0; } @@ -399,19 +402,19 @@ void ReplicationStateMachine::write_snapshot_obj(std::shared_ptr< homestore::sna if (!m_snp_rcv_handler) { m_snp_rcv_handler = std::make_unique< HSHomeObject::SnapshotReceiveHandler >(*home_object_, r_dev); if (m_snp_rcv_handler->load_prev_context_and_metrics()) { - LOGI("Reloaded previous snapshot context, lsn={} pg={} next_shard:0x{:x}", context->get_lsn(), + LOGI("Reloaded resync receiver context: lsn={}, pg={}, next_shard=0x{:x}", + m_snp_rcv_handler->get_context_lsn(), m_snp_rcv_handler->get_context_pg_id(), m_snp_rcv_handler->get_next_shard()); } } auto obj_id = objId(snp_obj->offset); auto log_suffix = - fmt::format("group={} lsn={} shard=0x{:x} batch_num={} size={}", uuids::to_string(r_dev->group_id()), + fmt::format("group={}, lsn={}, shard_seq=0x{:x}, batch={}, bytes={}", uuids::to_string(r_dev->group_id()), context->get_lsn(), obj_id.shard_seq_num, obj_id.batch_id, snp_obj->blob.size()); - LOGI("Received snapshot obj, {}", log_suffix); + LOGI("Writing resync object: {}", log_suffix); if (snp_obj->is_last_obj) { - LOGD("Write snapshot reached is_last_obj true {}", log_suffix); set_snapshot_context(context); // Update the snapshot context in case apply_snapshot is not called auto hs_pg = home_object_->get_hs_pg(m_snp_rcv_handler->get_context_pg_id()); hs_pg->pg_state_.clear_state(PGStateMask::BASELINE_RESYNC); @@ -419,32 +422,29 @@ void ReplicationStateMachine::write_snapshot_obj(std::shared_ptr< homestore::sna // since this replica will leave the PG and no later logs will be received, no need to reset this. reset_no_space_left_error_info(); repl_dev()->reset_latch_lsn(); + LOGI("Completed resync snapshot write: {}", log_suffix); return; } // Check message integrity #ifdef _PRERELEASE if (iomgr_flip::instance()->test_flip("state_machine_write_corrupted_data")) { - LOGW("Simulating writing corrupted snapshot data, lsn={}, obj_id={} shard 0x{:x} batch={}", context->get_lsn(), - obj_id.value, obj_id.shard_seq_num, obj_id.batch_id); + LOGD("Simulating corrupted resync object: {}", log_suffix); return; } #endif if (snp_obj->blob.size() < sizeof(SyncMessageHeader)) { - LOGE("invalid snapshot message size {} in write_snapshot_data, lsn={}, obj_id={} shard 0x{:x} batch={}", - snp_obj->blob.size(), context->get_lsn(), obj_id.value, obj_id.shard_seq_num, obj_id.batch_id); + LOGE("Invalid resync object size: minimum_bytes={}, {}", sizeof(SyncMessageHeader), log_suffix); return; } auto header = r_cast< const SyncMessageHeader* >(snp_obj->blob.cbytes()); if (header->corrupted()) { - LOGE("corrupted message in write_snapshot_data, lsn={}, obj_id={} shard 0x{:x} batch={}", context->get_lsn(), - obj_id.value, obj_id.shard_seq_num, obj_id.batch_id); + LOGE("Corrupted resync object header: {}", log_suffix); return; } if (auto payload_size = snp_obj->blob.size() - sizeof(SyncMessageHeader); payload_size != header->payload_size) { - LOGE("payload size mismatch in write_snapshot_data {} != {}, lsn={}, obj_id={} shard 0x{:x} batch={}", - payload_size, header->payload_size, context->get_lsn(), obj_id.value, obj_id.shard_seq_num, - obj_id.batch_id); + LOGE("Resync object payload size mismatch: actual_bytes={}, expected_bytes={}, {}", payload_size, + header->payload_size, log_suffix); return; } auto data_buf = snp_obj->blob.cbytes() + sizeof(SyncMessageHeader); @@ -458,7 +458,7 @@ void ReplicationStateMachine::write_snapshot_obj(std::shared_ptr< homestore::sna if (m_snp_rcv_handler->get_context_lsn() == context->get_lsn() && m_snp_rcv_handler->get_shard_cursor() != 0) { // Request to resume from the beginning of shard snp_obj->offset = snapshot_offset_for_next_shard(m_snp_rcv_handler->get_shard_cursor()); - LOGI("Resume from previous context breakpoint, lsn={} pg={} next_shard:0x{:x}, shard_cursor:0x{:x}", + LOGI("Resuming resync receiver context: lsn={}, pg={}, next_shard=0x{:x}, shard_cursor=0x{:x}", context->get_lsn(), pg_data->pg_id(), m_snp_rcv_handler->get_next_shard(), m_snp_rcv_handler->get_shard_cursor()); return; @@ -467,25 +467,26 @@ void ReplicationStateMachine::write_snapshot_obj(std::shared_ptr< homestore::sna // Init a new transmission // If PG already exists, clean the stale pg resources. Let's resync on a pristine base if (home_object_->pg_exists(pg_data->pg_id())) { - LOGI("pg already exists, clean pg resources before snapshot, pg={} {}", pg_data->pg_id(), log_suffix); + LOGI("Resetting existing PG before resync: pg={}, {}", pg_data->pg_id(), log_suffix); // Need to pause state machine before destroying the PG, if fail, let raft retry. if (!home_object_->pg_destroy(pg_data->pg_id(), true)) { - LOGE("failed to destroy existing pg, let raft retry, pg={} {}", pg_data->pg_id(), log_suffix); + LOGE("Failed to reset existing PG before resync; requesting retry: pg={}, {}", pg_data->pg_id(), + log_suffix); return; } } - LOGI("reset context from lsn={} to lsn={}", m_snp_rcv_handler->get_context_lsn(), context->get_lsn()); + LOGD("Resetting resync receiver context: previous_lsn={}, new_lsn={}", + m_snp_rcv_handler->get_context_lsn(), context->get_lsn()); m_snp_rcv_handler->reset_context_and_metrics(context->get_lsn(), pg_data->pg_id()); auto ret = m_snp_rcv_handler->process_pg_snapshot_data(*pg_data); if (ret) { // Do not proceed, will request for resending the PG data - LOGE("Failed to process PG snapshot data lsn={} obj_id={} shard 0x{:x} batch={}, err={}", - context->get_lsn(), obj_id.value, obj_id.shard_seq_num, obj_id.batch_id, ret); + LOGE("Failed to process resync PG metadata: error={}, {}", ret, log_suffix); return; } snp_obj->offset = snapshot_offset_for_next_shard(m_snp_rcv_handler->get_next_shard()); - LOGI("Write snapshot, processed PG data pg={} {}", pg_data->pg_id(), log_suffix); + LOGD("Dispatched resync PG metadata: pg={}, {}", pg_data->pg_id(), log_suffix); return; } @@ -496,17 +497,17 @@ void ReplicationStateMachine::write_snapshot_obj(std::shared_ptr< homestore::sna if (!m_snp_rcv_handler->is_valid_obj_id(obj_id)) { if (m_snp_rcv_handler->get_shard_cursor() == HSHomeObject::SnapshotReceiveHandler::shard_list_end_marker) { snp_obj->offset = LAST_OBJ_ID; - LOGW("Leader resending last batch , we already done. Setting offset to LAST_OBJ_ID", context->get_lsn(), - m_snp_rcv_handler->get_next_shard(), m_snp_rcv_handler->get_shard_cursor()); + LOGW("Ignoring resync object after shard completion: lsn={}, next_shard=0x{:x}, shard_cursor=0x{:x}", + context->get_lsn(), m_snp_rcv_handler->get_next_shard(), m_snp_rcv_handler->get_shard_cursor()); } else if (m_snp_rcv_handler->get_shard_cursor() == HSHomeObject::SnapshotReceiveHandler::invalid_shard_id) { // Could happen if interrupted before the first shard is done snp_obj->offset = objId(0, 0).value; - LOGW("No shard cursor found, resume from the beginning pg meta. lsn={}", context->get_lsn()); + LOGW("Resync receiver has no shard cursor; requesting PG metadata: lsn={}", context->get_lsn()); } else { snp_obj->offset = objId(HSHomeObject::get_sequence_num_from_shard_id(m_snp_rcv_handler->get_shard_cursor()), 0).value; - LOGW("Obj id not matching with the current shard/blob cursor, resume from previous context breakpoint, " - "lsn={} next_shard:0x{:x}, shard_cursor:0x{:x}", + LOGW("Invalid resync receiver cursor; requesting shard metadata: lsn={}, next_shard=0x{:x}, " + "shard_cursor=0x{:x}", context->get_lsn(), m_snp_rcv_handler->get_next_shard(), m_snp_rcv_handler->get_shard_cursor()); } return; @@ -522,13 +523,12 @@ void ReplicationStateMachine::write_snapshot_obj(std::shared_ptr< homestore::sna auto ret = m_snp_rcv_handler->process_shard_snapshot_data(*shard_data); if (ret) { // Do not proceed, will request for resending the shard data - LOGE("Failed to process shard snapshot data lsn={} obj_id={} shard 0x{:x} batch={}, err={}", - context->get_lsn(), obj_id.value, obj_id.shard_seq_num, obj_id.batch_id, ret); + LOGE("Failed to process resync shard metadata: error={}, {}", ret, log_suffix); return; } // Request for the next batch snp_obj->offset = objId(obj_id.shard_seq_num, 1).value; - LOGD("Write snapshot, processed shard data shard_seq_num:0x{:x} {}", obj_id.shard_seq_num, log_suffix); + LOGD("Dispatched resync shard metadata: {}", log_suffix); return; } @@ -538,8 +538,7 @@ void ReplicationStateMachine::write_snapshot_obj(std::shared_ptr< homestore::sna m_snp_rcv_handler->process_blobs_snapshot_data(*blob_batch, obj_id.batch_id, blob_batch->is_last_batch()); if (ret) { // Do not proceed, will request for resending the current blob batch - LOGE("Failed to process blob snapshot data lsn={} obj_id={} shard 0x{:x} batch={}, err={}", context->get_lsn(), - obj_id.value, obj_id.shard_seq_num, obj_id.batch_id, ret); + LOGE("Failed to process resync shard batch: error={}, {}", ret, log_suffix); return; } // Set next obj_id to fetch @@ -549,8 +548,7 @@ void ReplicationStateMachine::write_snapshot_obj(std::shared_ptr< homestore::sna snp_obj->offset = objId(obj_id.shard_seq_num, obj_id.batch_id + 1).value; } - LOGD("Write snapshot, processed blob data shard_seq_num:0x{:x} batch_num={} {}", obj_id.shard_seq_num, - obj_id.batch_id, log_suffix); + LOGD("Dispatched resync shard batch: {}", log_suffix); } void ReplicationStateMachine::free_user_snp_ctx(void*& user_snp_ctx) { diff --git a/src/lib/homestore_backend/snapshot_receive_handler.cpp b/src/lib/homestore_backend/snapshot_receive_handler.cpp index a4898a07..4ea5cf0d 100644 --- a/src/lib/homestore_backend/snapshot_receive_handler.cpp +++ b/src/lib/homestore_backend/snapshot_receive_handler.cpp @@ -13,8 +13,6 @@ HSHomeObject::SnapshotReceiveHandler::SnapshotReceiveHandler(HSHomeObject& home_ home_obj_(home_obj), repl_dev_(std::move(repl_dev)), cp_fut(folly::makeFuture< bool >(true)) {} int HSHomeObject::SnapshotReceiveHandler::process_pg_snapshot_data(ResyncPGMetaData const& pg_meta) { - LOGI("process_pg_snapshot_data pg={}", pg_meta.pg_id()); - // Init shard list ctx_->shard_list.clear(); const auto ids = pg_meta.shard_ids(); @@ -37,17 +35,18 @@ int HSHomeObject::SnapshotReceiveHandler::process_pg_snapshot_data(ResyncPGMetaD pg_member.priority = member->priority(); pg_info.members.insert(pg_member); } - LOGI("PG expected member num={}, actual members num={}", pg_info.expected_member_num, pg_meta.members()->size()) + LOGD("Resync PG membership: pg={}, expected_members={}, members={}", pg_meta.pg_id(), + pg_info.expected_member_num, pg_meta.members()->size()) #ifdef _PRERELEASE if (iomgr_flip::instance()->test_flip("snapshot_receiver_pg_error")) { - LOGW("Simulating PG snapshot error"); + LOGD("Simulating resync PG metadata failure: pg={}", pg_meta.pg_id()); return CREATE_PG_ERR; } #endif auto ret = home_obj_.local_create_pg(repl_dev_, pg_info); if (ret.hasError()) { - LOGE("Failed to create pg={}, err {}", pg_meta.pg_id(), ret.error()); + LOGE("Failed to process resync PG metadata: pg={}, error={}", pg_meta.pg_id(), ret.error()); return CREATE_PG_ERR; } auto hs_pg = ret.value(); @@ -66,13 +65,12 @@ int HSHomeObject::SnapshotReceiveHandler::process_pg_snapshot_data(ResyncPGMetaD ctx_->progress.total_blobs = pg_meta.total_blobs_to_transfer(); ctx_->progress.total_bytes = pg_meta.total_bytes_to_transfer(); // No need to persist snp info superblock since it's almost meaningless to resume from this point. + LOGI("Processed resync PG metadata: pg={}, shards={}, total_blobs={}, total_bytes={}", pg_meta.pg_id(), + ctx_->shard_list.size(), pg_meta.total_blobs_to_transfer(), pg_meta.total_bytes_to_transfer()); return 0; } int HSHomeObject::SnapshotReceiveHandler::process_shard_snapshot_data(ResyncShardMetaData const& shard_meta) { - LOGI("process_shard_snapshot_data shardID=0x{:x}, pg={}, shard=0x{:x}", shard_meta.shard_id(), - (shard_meta.shard_id() >> homeobject::shard_width), (shard_meta.shard_id() & homeobject::shard_mask)); - // Persist shard meta on chunk data sisl::io_blob_safe aligned_buf(sisl::round_up(sizeof(shard_info_superblk), io_align), io_align); shard_info_superblk* shard_sb = r_cast< shard_info_superblk* >(aligned_buf.bytes()); @@ -107,6 +105,8 @@ int HSHomeObject::SnapshotReceiveHandler::process_shard_snapshot_data(ResyncShar } ctx_->shard_cursor = shard_meta.shard_id(); ctx_->cur_batch_num = 0; + LOGI("Processed resync shard metadata: pg={}, shard_id=0x{:x}, state={}", shard_meta.pg_id(), + shard_meta.shard_id(), shard_meta.state()); return 0; } @@ -162,7 +162,8 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob // Skip deleted blobs if (blob->state() == static_cast< uint8_t >(ResyncBlobState::DELETED)) { - LOGD("Skip deleted blob_id={}", blob->blob_id()); + LOGT("Skipping deleted resync blob: pg={}, shard_id=0x{:x}, blob={}", ctx_->pg_id, ctx_->shard_cursor, + blob->blob_id()); skipped_blobs++; continue; } @@ -176,7 +177,8 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob auto delay = iomgr_flip::instance()->get_test_flip< long >("simulate_write_snapshot_save_blob_delay", static_cast< long >(blob->blob_id())); if (delay) { - LOGI("Simulating pg snapshot receive data with delay, delay={}, blob_id={}", delay.get(), blob->blob_id()); + LOGD("Simulating resync blob write delay: pg={}, shard_id=0x{:x}, blob={}, delay_ms={}", ctx_->pg_id, + ctx_->shard_cursor, blob->blob_id(), delay.get()); std::this_thread::sleep_for(std::chrono::milliseconds(delay.get())); } #endif @@ -198,7 +200,8 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob // corruption. auto blk_id = blob_index.value(); homestore::data_service().commit_blk(blk_id, true /* recommit */); - LOGD("Skip already persisted blob_id={}, re-committed blk_id={}", blob->blob_id(), blk_id.to_string()); + LOGT("Re-committed persisted resync blob: pg={}, shard_id=0x{:x}, blob={}, blkid={}", ctx_->pg_id, + ctx_->shard_cursor, blob->blob_id(), blk_id.to_string()); skipped_blobs++; continue; } @@ -209,15 +212,16 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob if (blob->state() != static_cast< uint8_t >(ResyncBlobState::CORRUPTED)) { // Verify full blob (includes validation, shard_id check, and hash verification) if (!home_obj_.verify_blob(blob_data, ctx_->shard_cursor, blob->blob_id())) { - LOGE("Blob verification failed for blob_id={}", blob->blob_id()); + LOGE("Resync blob verification failed: pg={}, shard_id=0x{:x}, blob={}", ctx_->pg_id, + ctx_->shard_cursor, blob->blob_id()); std::unique_lock< std::shared_mutex > lock(ctx_->progress_lock); ctx_->progress.error_count++; found_unexpected_corruption = true; break; } } else { - LOGW("find corrupted_blobs={} in shardID=0x{:x}, pg={}, shard=0x{:x}", blob->blob_id(), ctx_->shard_cursor, - (ctx_->shard_cursor >> homeobject::shard_width), (ctx_->shard_cursor & homeobject::shard_mask)); + LOGW("Persisting donor-reported corrupted resync blob: pg={}, shard_id=0x{:x}, blob={}", ctx_->pg_id, + ctx_->shard_cursor, blob->blob_id()); std::unique_lock< std::shared_mutex > lock(ctx_->progress_lock); ctx_->progress.corrupted_blobs++; } @@ -233,7 +237,8 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob homestore::BlkAllocStatus status; #ifdef _PRERELEASE if (iomgr_flip::instance()->test_flip("snapshot_receiver_blk_allocation_error")) { - LOGW("Simulating blob snapshot allocation error"); + LOGD("Simulating resync blob allocation failure: pg={}, shard_id=0x{:x}, blob={}", ctx_->pg_id, + ctx_->shard_cursor, blob->blob_id()); status = homestore::BlkAllocStatus::SPACE_FULL; } else { status = homestore::data_service().alloc_blks( @@ -244,9 +249,8 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob sisl::round_up(aligned_buf->size(), homestore::data_service().get_blk_size()), hints, blk_id); #endif if (status != homestore::BlkAllocStatus::SUCCESS) { - LOGE("Failed to allocate blocks for shardID=0x{:x}, pg={}, shard=0x{:x} blob {}", ctx_->shard_cursor, - (ctx_->shard_cursor >> homeobject::shard_width), (ctx_->shard_cursor & homeobject::shard_mask), - blob->blob_id()); + LOGE("Failed to allocate resync blob blocks: pg={}, shard_id=0x{:x}, blob={}", ctx_->pg_id, + ctx_->shard_cursor, blob->blob_id()); std::unique_lock< std::shared_mutex > lock(ctx_->progress_lock); ctx_->progress.error_count++; break; @@ -254,7 +258,8 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob #ifdef _PRERELEASE if (iomgr_flip::instance()->test_flip("snapshot_receiver_blob_write_data_error")) { - LOGW("Simulating blob snapshot write data error"); + LOGD("Simulating resync blob write failure: pg={}, shard_id=0x{:x}, blob={}", ctx_->pg_id, + ctx_->shard_cursor, blob->blob_id()); std::unique_lock< std::shared_mutex > lock(ctx_->progress_lock); ctx_->progress.error_count++; futs.emplace_back(folly::makeFuture< std::error_code >(std::make_error_code(std::errc::invalid_argument))); @@ -262,7 +267,8 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob } #endif auto blob_id = blob->blob_id(); - LOGD("Writing Blob {} to blk_id {}", blob_id, blk_id.to_string()); + LOGT("Writing resync blob: pg={}, shard_id=0x{:x}, blob={}, blkid={}", ctx_->pg_id, ctx_->shard_cursor, + blob_id, blk_id.to_string()); // ToDo: limit the max concurrent? futs.emplace_back( @@ -271,29 +277,34 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob .thenValue([this, blk_id, start, blob_id](auto&& err) -> folly::Future< std::error_code > { // TODO: do we need to update repl_dev metrics? if (err) { - LOGE("Failed to write blob info to blk_id={}, free the blk.", blk_id.to_string()); + LOGE("Failed to write resync blob; freeing block: pg={}, shard_id=0x{:x}, blob={}, blkid={}", + ctx_->pg_id, ctx_->shard_cursor, blob_id, blk_id.to_string()); homestore::data_service().async_free_blk(blk_id).get(); return err; } - LOGD("Blob {} written to blk_id={}", blob_id, blk_id.to_string()); + LOGT("Wrote resync blob: pg={}, shard_id=0x{:x}, blob={}, blkid={}", ctx_->pg_id, + ctx_->shard_cursor, blob_id, blk_id.to_string()); if (homestore::data_service().commit_blk(blk_id) != homestore::BlkAllocStatus::SUCCESS) { - LOGE("Failed to commit blk_id={} for blob_id={}", blk_id.to_string(), blob_id); + LOGE("Failed to commit resync blob block: pg={}, shard_id=0x{:x}, blob={}, blkid={}", + ctx_->pg_id, ctx_->shard_cursor, blob_id, blk_id.to_string()); homestore::data_service().async_free_blk(blk_id).get(); - return err; + return std::make_error_code(std::errc::io_error); } // Add local blob info to index & PG bool success = home_obj_.local_add_blob_info(ctx_->pg_id, BlobInfo{ctx_->shard_cursor, blob_id, blk_id}); if (!success) { - LOGE("Failed to add blob info for blob_id={}", blob_id); + LOGE("Failed to index resync blob: pg={}, shard_id=0x{:x}, blob={}, blkid={}", ctx_->pg_id, + ctx_->shard_cursor, blob_id, blk_id.to_string()); homestore::data_service().async_free_blk(blk_id).get(); - return err; + return std::make_error_code(std::errc::io_error); } auto duration = get_elapsed_time_us(start); HISTOGRAM_OBSERVE(*metrics_, snp_rcvr_blob_process_time, duration); - LOGD("Persisted blob_id={} in {}us", blob_id, duration); + LOGT("Persisted resync blob: pg={}, shard_id=0x{:x}, blob={}, duration_us={}", ctx_->pg_id, + ctx_->shard_cursor, blob_id, duration); return std::error_code{}; })); } @@ -304,10 +315,12 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob if (!all_io_submitted || ec != std::error_code{}) { if (!all_io_submitted) { - LOGE("Errors in submitting the batch, expect {} blobs, submitted {}.", data_blobs.blob_list()->size(), - futs.size()); + LOGE( + "Failed to submit complete resync shard batch: pg={}, shard_id=0x{:x}, batch={}, expected_blobs={}, submitted_blobs={}", + ctx_->pg_id, ctx_->shard_cursor, batch_num, data_blobs.blob_list()->size(), futs.size()); } else { - LOGE("Errors in writing this batch, code={}, message={}", ec.value(), ec.message()); + LOGE("Failed to write resync shard batch: pg={}, shard_id=0x{:x}, batch={}, error_code={}, error={}", + ctx_->pg_id, ctx_->shard_cursor, batch_num, ec.value(), ec.message()); } std::unique_lock< std::shared_mutex > lock(ctx_->progress_lock); ctx_->progress.error_count++; @@ -347,6 +360,8 @@ int HSHomeObject::SnapshotReceiveHandler::process_blobs_snapshot_data(ResyncBlob } HISTOGRAM_OBSERVE(*metrics_, snp_rcvr_batch_process_time, get_elapsed_time_ms(batch_start)); + LOGI("Processed resync shard batch: pg={}, shard_id=0x{:x}, batch={}, blobs={}, bytes={}, end_of_shard={}", + ctx_->pg_id, ctx_->shard_cursor, batch_num, data_blobs.blob_list()->size(), total_bytes, is_last_batch); return 0; }