show synchronization delay information in master (#273)

OpenAtomFoundation · May 14, 2018 · 3b86e4a · 3b86e4a
1 parent a44f878
commit 3b86e4a
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 5 deletions.
diff --git a/include/pika_binlog_sender_thread.h b/include/pika_binlog_sender_thread.h
@@ -29,6 +29,9 @@ class PikaBinlogSenderThread : public pink::Thread {
  uint32_t filenum() {
  return filenum_;
  }
+ uint64_t con_offset() {
+ return con_offset_;
+ }
 
  int trim();
 
@@ -38,9 +41,8 @@ class PikaBinlogSenderThread : public pink::Thread {
  Status Consume(std::string &scratch);
  unsigned int ReadPhysicalRecord(slash::Slice *fragment);
 
- uint64_t con_offset_;
  uint32_t filenum_;
-
+ uint64_t con_offset_;
  uint64_t last_record_offset_;
 
  slash::SequentialFile* queue_;

diff --git a/src/pika_binlog_sender_thread.cc b/src/pika_binlog_sender_thread.cc
@@ -21,8 +21,8 @@ PikaBinlogSenderThread::PikaBinlogSenderThread(const std::string &ip, int port,
  slash::SequentialFile *queue,
  uint32_t filenum,
  uint64_t con_offset)
- : con_offset_(con_offset),
- filenum_(filenum),
+ : filenum_(filenum),
+ con_offset_(con_offset),
  queue_(queue),
  backing_store_(new char[kBlockSize]),
  buffer_(),

diff --git a/src/pika_server.cc b/src/pika_server.cc
@@ -537,12 +537,24 @@ int32_t PikaServer::GetSlaveListString(std::string& slave_list_str) {
  // Binlog Sender has not yet created
  continue;
  }
+
+ uint32_t master_filenum, slave_filenum;
+ uint64_t master_offset, slave_offset;
+ logger_->GetProducerStatus(&master_filenum, &master_offset);
+ PikaBinlogSenderThread* ptr_sender = static_cast<PikaBinlogSenderThread*>(iter->sender);
+ slave_filenum = ptr_sender->filenum();
+ slave_offset = ptr_sender->con_offset();
+
+ uint64_t lag = (master_filenum - slave_filenum) * logger_->file_size()
+ + (master_offset - slave_offset);
+
  slave_ip_port =(*iter).ip_port;
  tmp_stream << "slave" << index++
  << ":ip=" << slave_ip_port.substr(0, slave_ip_port.find(":"))
- << ",port=" << slave_ip_port.substr(slave_ip_port.find(":")+1)
+ << ",port=" << slave_ip_port.substr(slave_ip_port.find(":") + 1)
  << ",state=" << ((*iter).stage == SLAVE_ITEM_STAGE_TWO ? "online" : "offline")
  << ",sid=" << (*iter).sid
+ << ",lag=" << lag
  << "\r\n";
  }
  slave_list_str.assign(tmp_stream.str());