feat: add log-net-activities configuration

conf/pika.conf

@@ -3,8 +3,8 @@
 ###########################
 
 # Pika port, the default value is 9221.
-# [NOTICE] Port Magic offsets of port+1000 / port+2000 are used by Pika at present.
-# Port 10221 is used for Rsync, and port 11221 is used for Replication, while the listening port is 9221.
+# [NOTICE] Port Magic offsets of port+1000 / port+10000 are used by Pika at present.
+# Port 9221+10000 is used for Rsync, and port 9221+1000 is used for incr Replication, while the listening port is 9221.
 port : 9221
 
 db-instance-num : 3
@@ -74,6 +74,13 @@ log-path : ./log/
 # The unit of serverlogs is in [days] and the default value is 7(days).
 log-retention-time : 7
 
+# logging-mode can be config as normal or debug, if an invalid value is given, normal will be auto set.
+# when logging-mode is normal: only important info will be logged.
+# when logging-mode is debug: more info will be logged (eg. connection activities)
+# Default logging-mode is normal.
+# [NOTICE] you can use config set command to change logging-mode dynamically.
+logging-mode : normal
+
 # Directory to store the data of Pika.
 db-path : ./db/
 

include/pika_conf.h

@@ -86,9 +86,8 @@ class PikaConf : public pstd::BaseConf {
     std::shared_lock l(rwlock_);
     return log_retention_time_;
   }
-  std::string log_level() {
-    std::shared_lock l(rwlock_);
-    return log_level_;
+  net::LogMode logging_mode() {
+    return logging_mode_.load(std::memory_order::memory_order_relaxed);
   }
   std::string db_path() {
     std::shared_lock l(rwlock_);
@@ -826,10 +825,13 @@ class PikaConf : public pstd::BaseConf {
     max_compaction_bytes_ = value;
   }
 
-  void SetLogLevel(const std::string& value) {
-    std::lock_guard l(rwlock_);
-    TryPushDiffCommands("loglevel", value);
-    log_level_ = value;
+  void SetLoggingMode(std::string& value) {
+    TryPushDiffCommands("logging-mode", value);
+    if (value == "debug") {
+      logging_mode_.store(net::LogMode::DEBUG, std::memory_order::memory_order_relaxed);
+    } else {
+      logging_mode_.store(net::LogMode::NORMAL, std::memory_order::memory_order_relaxed);
+    }
   }
 
   // Rsync Rate limiting configuration
@@ -952,7 +954,6 @@ class PikaConf : public pstd::BaseConf {
   std::string slaveof_;
   std::string log_path_;
   int log_retention_time_;
-  std::string log_level_;
   std::string db_path_;
   int db_instance_num_ = 0;
   std::string db_sync_path_;
@@ -1093,6 +1094,8 @@ class PikaConf : public pstd::BaseConf {
   std::atomic_int cache_maxmemory_policy_ = 1;
   std::atomic_int cache_maxmemory_samples_ = 5;
   std::atomic_int cache_lfu_decay_time_ = 1;
+  std::atomic<net::LogMode> logging_mode_ = net::LogMode::NORMAL;
+
 
   // rocksdb blob
   bool enable_blob_files_ = false;

include/pika_dispatch_thread.h

@@ -19,7 +19,7 @@ class PikaDispatchThread {
 
   bool ClientKill(const std::string& ip_port);
   void ClientKillAll();
-
+  void SetLogLevel(net::LogMode new_mode);
   void SetQueueLimit(int queue_limit) { thread_rep_->SetQueueLimit(queue_limit); }
 
   void UnAuthUserAndKillClient(const std::set<std::string> &users, const std::shared_ptr<User>& defaultUser);

include/pika_server.h

@@ -492,7 +492,7 @@ class PikaServer : public pstd::noncopyable {
   void CacheConfigInit(cache::CacheConfig &cache_cfg);
   void ProcessCronTask();
   double HitRatio();
-
+  void SetLogLevel(net::LogMode new_mode);
   /*
   * disable compact
   */

src/net/include/net_define.h

@@ -26,6 +26,11 @@ const int kProtoMaxMessage = 512 * 1024 * 1024;  // 512MB
 
 const int kCommandHeaderLength = 4;
 
+enum LogMode {
+  DEBUG = 0,
+  NORMAL = 1
+};
+
 /*
  * The socket block type
  */

src/net/include/server_thread.h

@@ -128,6 +128,8 @@ class ServerThread : public Thread {
 
   int SetTcpNoDelay(int connfd);
 
+  void SetLogLevel(net::LogMode new_mode);
+
   /*
    * StartThread will return the error code as pthread_create
    * Return 0 if success
@@ -167,6 +169,8 @@ class ServerThread : public Thread {
    */
   std::unique_ptr<NetMultiplexer> net_multiplexer_;
 
+  std::atomic<net::LogMode> logging_mode_{net::LogMode::NORMAL};
+
  private:
   friend class HolyThread;
   friend class DispatchThread;

src/net/src/dispatch_thread.cc

@@ -148,15 +148,19 @@ void DispatchThread::HandleNewConn(const int connfd, const std::string& ip_port)
   // Slow workers may consume many fds.
   // We simply loop to find next legal worker.
   NetItem ti(connfd, ip_port);
-  LOG(INFO) << "accept new conn " << ti.String();
+  if (logging_mode_.load(std::memory_order::memory_order_relaxed) == net::LogMode::DEBUG) {
+    LOG(INFO) << "accept new conn " << ti.String();
+  }
   int next_thread = last_thread_;
   bool find = false;
   for (int cnt = 0; cnt < work_num_; cnt++) {
     std::unique_ptr<WorkerThread>& worker_thread = worker_thread_[next_thread];
     find = worker_thread->MoveConnIn(ti, false);
     if (find) {
       last_thread_ = (next_thread + 1) % work_num_;
-      LOG(INFO) << "find worker(" << next_thread << "), refresh the last_thread_ to " << last_thread_;
+      if (logging_mode_.load(std::memory_order::memory_order_relaxed) == net::LogMode::DEBUG) {
+        LOG(INFO) << "find worker(" << next_thread << "), refresh the last_thread_ to " << last_thread_;
+      }
       break;
     }
     next_thread = (next_thread + 1) % work_num_;
@@ -189,7 +193,7 @@ void DispatchThread::CleanWaitNodeOfUnBlockedBlrConn(std::shared_ptr<net::RedisC
   // removed all the waiting info of this conn/ doing cleaning work
   auto pair = blocked_conn_to_keys_.find(conn_unblocked->fd());
   if (pair == blocked_conn_to_keys_.end()) {
-    LOG(WARNING) << "blocking info of blpop/brpop went wrong, blpop/brpop can't working correctly";
+    LOG(ERROR) << "blocking info of blpop/brpop went wrong, blpop/brpop can't working correctly";
     return;
   }
   auto& blpop_keys_list = pair->second;

src/net/src/server_thread.cc

@@ -264,6 +264,10 @@ void* ServerThread::ThreadMain() {
   return nullptr;
 }
 
+void ServerThread::SetLogLevel(net::LogMode new_mode) {
+  logging_mode_.store(new_mode, std::memory_order::memory_order_relaxed);
+}
+
 #ifdef __ENABLE_SSL
 static std::vector<std::unique_ptr<pstd::Mutex>> ssl_mutex_;
 

src/pika_admin.cc

@@ -1590,6 +1590,13 @@ void ConfigCmd::ConfigGet(std::string& ret) {
     EncodeNumber(&config_body, g_pika_conf->port());
   }
 
+  if (pstd::stringmatch(pattern.data(), "logging-mode", 1) != 0) {
+    elements += 2;
+    EncodeString(&config_body, "logging-mode");
+    auto output_str = g_pika_conf->logging_mode() == net::LogMode::DEBUG ? "debug" : "normal";
+    EncodeString(&config_body, output_str);
+  }
+
   if (pstd::stringmatch(pattern.data(), "thread-num", 1) != 0) {
     elements += 2;
     EncodeString(&config_body, "thread-num");
@@ -2154,12 +2161,6 @@ void ConfigCmd::ConfigGet(std::string& ret) {
     EncodeString(&config_body, g_pika_conf->enable_blob_garbage_collection() ? "yes" : "no");
   }
 
-  if (pstd::stringmatch(pattern.data(), "loglevel", 1) != 0) {
-    elements += 2;
-    EncodeString(&config_body, "loglevel");
-    EncodeString(&config_body, g_pika_conf->log_level());
-  }
-
   if (pstd::stringmatch(pattern.data(), "min-blob-size", 1) != 0) {
     elements += 2;
     EncodeString(&config_body, "min-blob-size");
@@ -2472,6 +2473,15 @@ void ConfigCmd::ConfigSet(std::shared_ptr<DB> db) {
     g_pika_conf->SetSlowlogMaxLen(static_cast<int>(ival));
     g_pika_server->SlowlogTrim();
     res_.AppendStringRaw("+OK\r\n");
+  } else if (set_item == "logging-mode") {
+    if (value != "debug" && value != "normal") {
+      res_.AppendStringRaw("-ERR Invalid argument \'" + value +
+                           "\' for CONFIG SET 'logging-mode', only debug or normal is valid\r\n");
+      return;
+    }
+    g_pika_conf->SetLoggingMode(value);
+    g_pika_server->SetLogLevel(value == "debug" ? net::LogMode::DEBUG : net::LogMode::NORMAL);
+    res_.AppendStringRaw("+OK\r\n");
   } else if (set_item == "max-cache-statistic-keys") {
     if ((pstd::string2int(value.data(), value.size(), &ival) == 0) || ival < 0) {
       res_.AppendStringRaw("-ERR Invalid argument \'" + value + "\' for CONFIG SET 'max-cache-statistic-keys'\r\n");
@@ -3312,7 +3322,9 @@ void QuitCmd::DoInitial() {
 
 void QuitCmd::Do() {
   res_.SetRes(CmdRes::kOk);
-  LOG(INFO) << "QutCmd will close connection " << GetConn()->String();
+  if (g_pika_conf->logging_mode() == net::LogMode::DEBUG) {
+    LOG(INFO) << "QutCmd will close connection " << GetConn()->String();
+  }
   GetConn()->SetClose(true);
 }
 

src/pika_conf.cc

@@ -132,7 +132,18 @@ int PikaConf::Load() {
   if(log_retention_time_ < 0){
     LOG(FATAL) << "log-retention-time invalid";
   }
-  GetConfStr("loglevel", &log_level_);
+
+  std::string logging_mode;
+  GetConfStr("logging-mode", &logging_mode);
+  if (logging_mode == "debug") {
+    logging_mode_.store(net::LogMode::DEBUG, std::memory_order::memory_order_relaxed);
+  } else if (logging_mode == "normal") {
+    logging_mode_.store(net::LogMode::NORMAL, std::memory_order::memory_order_relaxed);
+  } else {
+    LOG(ERROR) << "logging-mode loaded from pika.conf is invalid, auto change it to normal";
+    logging_mode_.store(net::LogMode::NORMAL, std::memory_order::memory_order_relaxed);
+  };
+
   GetConfStr("db-path", &db_path_);
   GetConfInt("db-instance-num", &db_instance_num_);
   if (db_instance_num_ <= 0) {

src/pika_dispatch_thread.cc

@@ -62,6 +62,9 @@ void PikaDispatchThread::UnAuthUserAndKillClient(const std::set<std::string>& us
 void PikaDispatchThread::StopThread() {
   thread_rep_->StopThread();
 }
+void PikaDispatchThread::SetLogLevel(net::LogMode new_mode) {
+  thread_rep_->SetLogLevel(new_mode);
+}
 
 bool PikaDispatchThread::Handles::AccessHandle(std::string& ip) const {
   if (ip == "127.0.0.1") {

src/pika_server.cc

@@ -197,6 +197,7 @@ void PikaServer::Start() {
                << (ret == net::kBindError ? ": bind port " + std::to_string(port_) + " conflict" : ": other error")
                << ", Listen on this port to handle the connected redis client";
   }
+  pika_dispatch_thread_->SetLogLevel(g_pika_conf->logging_mode());
   ret = pika_pubsub_thread_->StartThread();
   if (ret != net::kSuccess) {
     dbs_.clear();
@@ -1919,3 +1920,4 @@ void PikaServer::CacheConfigInit(cache::CacheConfig& cache_cfg) {
   cache_cfg.maxmemory_samples = g_pika_conf->cache_maxmemory_samples();
   cache_cfg.lfu_decay_time = g_pika_conf->cache_lfu_decay_time();
 }
+void PikaServer::SetLogLevel(net::LogMode new_mode) { pika_dispatch_thread_->SetLogLevel(new_mode); }