Added the large Key detection function

Author: wuxianrong

conf/pika.conf

@@ -522,6 +522,8 @@ max-rsync-parallel-num : 4
 # The synchronization mode of Pika primary/secondary replication is determined by ReplicationID. ReplicationID in one replication_cluster are the same
 # replication-id :
 
+keys-analysis: no
+
 ###################
 ## Cache Settings
 ###################

include/pika_conf.h

@@ -49,6 +49,10 @@ class PikaConf : public pstd::BaseConf {
     std::shared_lock l(rwlock_);
     return write_binlog_;
   }
+  bool keys_analysis() {
+    std::shared_lock l(rwlock_);
+    return keys_analysis_;
+  } 
   int thread_num() {
     std::shared_lock l(rwlock_);
     return thread_num_;
@@ -498,6 +502,11 @@ class PikaConf : public pstd::BaseConf {
     TryPushDiffCommands("write-binlog", value);
     write_binlog_ = value == "yes";
   }
+  void SetKeysAnalysis(const bool value) {
+    std::lock_guard l(rwlock_);
+    TryPushDiffCommands("keys-analysis", value ? "yes" : "no");
+    keys_analysis_.store(value);
+  }
   void SetMaxCacheStatisticKeys(const int value) {
     std::lock_guard l(rwlock_);
     TryPushDiffCommands("max-cache-statistic-keys", std::to_string(value));
@@ -941,6 +950,7 @@ class PikaConf : public pstd::BaseConf {
   std::atomic<int> slowlog_log_slower_than_;
   std::atomic<bool> slotmigrate_;
   std::atomic<int> binlog_writer_num_;
+  std::atomic<bool> keys_analysis_ = false;
   int slowlog_max_len_ = 0;
   int expire_logs_days_ = 0;
   int expire_logs_nums_ = 0;

src/pika_admin.cc

@@ -1404,9 +1404,10 @@ void InfoCmd::InfoData(std::string& info) {
   uint64_t total_background_errors = 0;
   uint64_t total_memtable_usage = 0;
   uint64_t total_table_reader_usage = 0;
+  uint64_t total_big_key_count = 0;;
   uint64_t memtable_usage = 0;
   uint64_t table_reader_usage = 0;
-  uint64_t total_big_key_count = 0;
+  uint64_t big_key_count = 0; 
   std::shared_lock db_rwl(g_pika_server->dbs_rw_);
   for (const auto& db_item : g_pika_server->dbs_) {
     if (!db_item.second) {
@@ -1418,8 +1419,7 @@ void InfoCmd::InfoData(std::string& info) {
     db_item.second->storage()->GetUsage(storage::PROPERTY_TYPE_ROCKSDB_CUR_SIZE_ALL_MEM_TABLES, &memtable_usage);
     db_item.second->storage()->GetUsage(storage::PROPERTY_TYPE_ROCKSDB_ESTIMATE_TABLE_READER_MEM, &table_reader_usage);
     db_item.second->storage()->GetUsage(storage::PROPERTY_TYPE_ROCKSDB_BACKGROUND_ERRORS, &background_errors);
-
-    uint64_t big_key_count = db_item.second->storage()->GetBigKeyStatistics(db_item.first, "bigkey_property");
+    db_item.second->storage()->GetBigKeyStatistics();
     total_big_key_count += big_key_count;
     db_item.second->DBUnlockShared();
     total_memtable_usage += memtable_usage;
@@ -2293,6 +2293,12 @@ void ConfigCmd::ConfigGet(std::string& ret) {
     g_pika_conf->acl_pubsub_default() ? EncodeString(&config_body, "allchannels")
                                       : EncodeString(&config_body, "resetchannels");
   }
+ 
+  if (pstd::stringmatch(pattern.data(), "keys-analysis", 1)) {
+    elements += 2;
+    EncodeString(&config_body, "keys-analysis");
+    EncodeNumber(&config_body, g_pika_conf->keys_analysis());
+  }
 
   std::stringstream resp;
   resp << "*" << std::to_string(elements) << "\r\n" << config_body;
@@ -2483,6 +2489,17 @@ void ConfigCmd::ConfigSet(std::shared_ptr<DB> db) {
     g_pika_conf->SetSlowCmdPool(SlowCmdPool);
     g_pika_server->SetSlowCmdThreadPoolFlag(SlowCmdPool);
     res_.AppendStringRaw("+OK\r\n");
+  } else if (set_item == "keys-analysis") {
+    bool KeysAnalysis;
+    if (value == "yes") {
+      KeysAnalysis = true;
+    } else if (value == "no") {
+      KeysAnalysis = false;
+    } else {
+      res_.AppendStringRaw( "-ERR Invalid argument \'" + value + "\' for CONFIG SET 'keys-analysis'\r\n");
+      return;
+    }
+    g_pika_conf->SetKeysAnalysis(KeysAnalysis); 
   } else if (set_item == "slowlog-log-slower-than") {
     if ((pstd::string2int(value.data(), value.size(), &ival) == 0) || ival < 0) {
       res_.AppendStringRaw("-ERR Invalid argument \'" + value + "\' for CONFIG SET 'slowlog-log-slower-than'\r\n");

src/pika_conf.cc

@@ -529,6 +529,11 @@ int PikaConf::Load() {
   }
   GetConfStr("pidfile", &pidfile_);
 
+  // keys-analysis
+  std::string ka;
+  GetConfStr("keys-analysis", &ka);
+  keys_analysis_ = wb != "no";  
+
   // db sync
   GetConfStr("db-sync-path", &db_sync_path_);
   db_sync_path_ = db_sync_path_.empty() ? "./dbsync/" : db_sync_path_;
@@ -766,6 +771,7 @@ int PikaConf::ConfigRewrite() {
   SetConfInt("slave-priority", slave_priority_);
   SetConfStr("log-net-activities", log_net_activities_ ? "yes" : "no");
   SetConfStr("write-binlog", write_binlog_ ? "yes" : "no");
+  SetConfStr("keys-analysis", keys_analysis_ ? "yes" : "no");
   SetConfStr("run-id", run_id_);
   SetConfStr("replication-id", replication_id_);
   SetConfInt("max-cache-statistic-keys", max_cache_statistic_keys_);

src/storage/include/storage/storage.h

@@ -1100,7 +1100,7 @@ class Storage {
   Status EnableAutoCompaction(const OptionType& option_type, 
                     const std::string& db_type, const std::unordered_map<std::string, std::string>& options);
   void GetRocksDBInfo(std::string& info);
-  uint64_t GetBigKeyStatistics(const std::string& db_type, const std::string& property);
+  size_t GetBigKeyStatistics();
 
  private:
   std::unique_ptr<RedisStrings> strings_db_;

src/storage/src/redis.cc

@@ -199,4 +199,30 @@ void Redis::SetCompactRangeOptions(const bool is_canceled) {
   } 
 }
 
+void Redis::CheckBigKeyAndLog(const std::string& key, uint64_t size) {
+  thread_local uint64_t last_log_time = 0;
+  auto current_time = pstd::NowMicros();
+
+  static const uint64_t kLogInterval = 60 * 1000 * 1000;  
+  static const uint64_t kBigKeyThreshold = 10000;
+
+  if (current_time - last_log_time >= kLogInterval) {
+    last_log_time = current_time;
+
+    if (size > kBigKeyThreshold) {
+      std::unique_lock<std::shared_mutex> write_lock(big_key_access_mutex_);
+      big_key_access_count_[key]++;
+
+      LOG(INFO) << "[BIGKEY DETECTED] Key: " << key
+                << ", Size: " << size
+                << ", Access Count: " << big_key_access_count_[key];
+    }
+  }
+}
+
+size_t Redis::GetBigKeyStatistics() {
+  std::shared_lock<std::shared_mutex> read_lock(big_key_access_mutex_);
+  return big_key_access_count_.size();
+}
+
 }  // namespace storage

src/storage/src/redis.h

@@ -11,6 +11,8 @@
 #include <vector>
 #include <iostream>
 
+#include <glog/logging.h>
+
 #include "rocksdb/db.h"
 #include "rocksdb/slice.h"
 #include "rocksdb/status.h"
@@ -116,21 +118,8 @@ class Redis {
   Status SetSmallCompactionDurationThreshold(uint64_t small_compaction_duration_threshold);
   std::vector<rocksdb::ColumnFamilyHandle*> GetHandles(){ return handles_;};
   void GetRocksDBInfo(std::string &info, const char *prefix);
-  void CheckBigKeyAndLog(const std::string& key, uint64_t size) {
-    static const uint64_t kBigKeyThreshold = 10000;
-    if (size > kBigKeyThreshold) {
-      std::lock_guard<std::mutex> lock(big_key_access_mutex_);
-      big_key_access_count_[key]++;
-      std::cerr << "[BIGKEY DETECTED] Key: " << key
-                << ", Size: " << size
-                << ", Access Count: " << big_key_access_count_[key] << std::endl;
-    }
-  }
-
-  std::unordered_map<std::string, int> GetBigKeyStatistics() {
-    std::lock_guard<std::mutex> lock(big_key_access_mutex_);
-    return big_key_access_count_;
-  }
+  void CheckBigKeyAndLog(const std::string& key, uint64_t size);
+  size_t GetBigKeyStatistics();
 
  protected:
   Storage* const storage_;
@@ -158,7 +147,7 @@ class Redis {
   Status UpdateSpecificKeyDuration(const std::string& key, uint64_t duration);
   Status AddCompactKeyTaskIfNeeded(const std::string& key, uint64_t count, uint64_t duration);
   std::unordered_map<std::string, int> big_key_access_count_;
-  std::mutex big_key_access_mutex_;
+  std::shared_mutex big_key_access_mutex_;
 };
 
 }  //  namespace storage

src/storage/src/storage.cc

@@ -1855,30 +1855,13 @@ uint64_t Storage::GetProperty(const std::string& db_type, const std::string& pro
   return result;
 }
 
-uint64_t Storage::GetBigKeyStatistics(const std::string& db_type, const std::string& property) {
-  uint64_t out = 0;
-  uint64_t result = 0;
-  if (db_type == ALL_DB || db_type == HASHES_DB) {
-    hashes_db_->GetBigKeyStatistics();
-    result += out;
-  }
-
-  if (db_type == ALL_DB || db_type == LISTS_DB) {
-    lists_db_->GetBigKeyStatistics();
-    result += out;
-  }
-  if (db_type == ALL_DB || db_type == ZSETS_DB) {
-    zsets_db_->GetBigKeyStatistics();
-    result += out;
-  }
-  if (db_type == ALL_DB || db_type == SETS_DB) {
-    sets_db_->GetBigKeyStatistics();
-    result += out;
-  }
-  if (db_type == ALL_DB || db_type == STREAMS_DB) {
-    streams_db_->GetBigKeyStatistics();
-    result += out;
-  }
+size_t Storage::GetBigKeyStatistics() {
+  size_t result = 0;
+  result += hashes_db_->GetBigKeyStatistics();
+  result += lists_db_->GetBigKeyStatistics();
+  result += zsets_db_->GetBigKeyStatistics();
+  result += sets_db_->GetBigKeyStatistics();
+  result += streams_db_->GetBigKeyStatistics();
   return result;
 }