chore: add more logs for takeover (#5765)

Author: BorysTheDev

src/server/dflycmd.cc

@@ -85,6 +85,7 @@ bool WaitReplicaFlowToCatchup(absl::Time end_time, const DflyCmd::ReplicaInfo* r
   shard->journal()->RecordEntry(0, journal::Op::PING, 0, 0, nullopt, {});
 
   const FlowInfo* flow = &replica->flows[shard->shard_id()];
+
   while (flow->last_acked_lsn < shard->journal()->GetLsn()) {
     if (absl::Now() > end_time) {
       LOG(WARNING) << "Couldn't synchronize with replica for takeover in time: " << replica->address
@@ -95,8 +96,9 @@ bool WaitReplicaFlowToCatchup(absl::Time end_time, const DflyCmd::ReplicaInfo* r
     if (!replica->exec_st.IsRunning()) {
       return false;
     }
-    VLOG(1) << "Replica lsn:" << flow->last_acked_lsn
-            << " master lsn:" << shard->journal()->GetLsn();
+    LOG_EVERY_T(INFO, 1) << "Replica lsn:" << flow->last_acked_lsn
+                         << " master lsn:" << shard->journal()->GetLsn()
+                         << "; Journal streamer state: " << flow->streamer->FormatInternalState();
     ThisFiber::SleepFor(1ms);
   }
 
@@ -512,7 +514,7 @@ void DflyCmd::TakeOver(CmdArgList args, RedisReplyBuilder* rb, ConnectionContext
   atomic_bool catchup_success = true;
   if (*status == OpStatus::OK) {
     dfly::SharedLock lk{replica_ptr->shared_mu};
-    auto cb = [replica_ptr, end_time, &catchup_success](EngineShard* shard) {
+    auto cb = [replica_ptr = replica_ptr, end_time, &catchup_success](EngineShard* shard) {
       if (!WaitReplicaFlowToCatchup(end_time, replica_ptr.get(), shard)) {
         catchup_success.store(false);
       }

src/server/journal/streamer.cc

@@ -164,6 +164,15 @@ size_t JournalStreamer::UsedBytes() const {
   return pending_buf_.Size();
 }
 
+std::string JournalStreamer::FormatInternalState() const {
+  return absl::StrCat(
+      "pending_buf_size:", pending_buf_.Size(), " in_flight_bytes:", in_flight_bytes_,
+      " total_sent:", total_sent_, " throttle_count:", throttle_count_,
+      " total_throttle_wait_usec:", total_throttle_wait_usec_,
+      " throttle_waiters:", throttle_waiters_, " last_async_write_time_ms:", last_async_write_time_,
+      " last_lsn_time_s:", last_lsn_time_, " last_lsn_writen_:", last_lsn_writen_);
+}
+
 void JournalStreamer::Write(std::string str) {
   DCHECK(!str.empty());
   DVLOG(3) << "Writing " << str.size() << " bytes";

src/server/journal/streamer.h

@@ -36,6 +36,9 @@ class JournalStreamer : public journal::JournalConsumerInterface {
 
   size_t UsedBytes() const;
 
+  // For debugging purposes. Return string with formatted internal state.
+  std::string FormatInternalState() const;
+
  protected:
   // TODO: we copy the string on each write because JournalItem may be passed to multiple
   // streamers so we can not move it. However, if we would either wrap JournalItem in shared_ptr