feat: Add Keyspace hits metrics for set (OpenAtomFoundation#2579)

* 1.add keyspace_hits and keyspace_misses
Co-authored-by: Mixficsol <[email protected]>

include/pika_command.h

@@ -332,13 +332,15 @@ class CmdRes {
     kInvalidTransaction,
     kTxnQueued,
     kTxnAbort,
-    kMultiKey
+    kMultiKey,
+    kNoExists,
   };
 
   CmdRes() = default;
 
   bool none() const { return ret_ == kNone && message_.empty(); }
-  bool ok() const { return ret_ == kOk || ret_ == kNone; }
+  bool noexist() const { return ret_ == kNoExists; }
+  bool ok() const { return ret_ == kOk || ret_ == kNone || ret_ == kNoExists; }
   CmdRet ret() const { return ret_; }
   void clear() {
     message_.clear();
@@ -365,7 +367,6 @@ class CmdRes {
         return "-ERR bit offset is not an integer or out of range\r\n";
       case kWrongBitOpNotNum:
         return "-ERR BITOP NOT must be called with a single source key.\r\n";
-
       case kInvalidBitPosArgument:
         return "-ERR The bit argument must be 1 or 0.\r\n";
       case kInvalidFloat:
@@ -431,6 +432,8 @@ class CmdRes {
         result = "-WRONGTYPE Operation against a key holding the wrong kind of value";
         result.append(kNewLine);
         break;
+      case kNoExists:
+        return message_;
       default:
         break;
     }
@@ -542,7 +545,7 @@ class Cmd : public std::enable_shared_from_this<Cmd> {
 
   bool IsLocal() const;
   bool IsSuspend() const;
-  bool IsAdminRequire() const;
+  bool IsAdmin() const;
   bool HasSubCommand() const;                   // The command is there a sub command
   std::vector<std::string> SubCommand() const;  // Get command is there a sub command
   bool IsNeedUpdateCache() const;

include/pika_server.h

@@ -249,8 +249,12 @@ class PikaServer : public pstd::noncopyable {
   uint64_t ServerQueryNum();
   uint64_t ServerCurrentQps();
   uint64_t accumulative_connections();
+  long long ServerKeyspaceHits();
+  long long ServerKeyspaceMisses();
   void ResetStat();
   void incr_accumulative_connections();
+  void incr_server_keyspace_hits();
+  void incr_server_keyspace_misses();
   void ResetLastSecQuerynum();
   void UpdateQueryNumAndExecCountDB(const std::string& db_name, const std::string& command, bool is_write);
   std::unordered_map<std::string, uint64_t> ServerExecCountDB();

include/pika_statistic.h

@@ -37,6 +37,8 @@ struct ServerStatistic {
 
   std::atomic<uint64_t> accumulative_connections;
   std::unordered_map<std::string, std::atomic<uint64_t>> exec_count_db;
+  std::atomic<long long> keyspace_hits;
+  std::atomic<long long> keyspace_misses;
   QpsStatistic qps;
 };
 

src/pika_admin.cc

@@ -1070,6 +1070,8 @@ void InfoCmd::InfoStats(std::string& info) {
   tmp_stream << "total_connections_received:" << g_pika_server->accumulative_connections() << "\r\n";
   tmp_stream << "instantaneous_ops_per_sec:" << g_pika_server->ServerCurrentQps() << "\r\n";
   tmp_stream << "total_commands_processed:" << g_pika_server->ServerQueryNum() << "\r\n";
+  tmp_stream << "keyspace_hits:" << g_pika_server->ServerKeyspaceHits() << "\r\n";
+  tmp_stream << "keyspace_misses:" << g_pika_server->ServerKeyspaceMisses() << "\r\n";
 
   // Network stats
   tmp_stream << "total_net_input_bytes:" << g_pika_server->NetInputBytes() + g_pika_server->NetReplInputBytes()

src/pika_command.cc

@@ -910,6 +910,13 @@ void Cmd::DoCommand(const HintKeys& hint_keys) {
   } else {
     Do();
   }
+  if (!IsAdmin() && res().ok()) {
+    if (res().noexist()) {
+      g_pika_server->incr_server_keyspace_misses();
+    } else {
+      g_pika_server->incr_server_keyspace_hits();
+    }
+  }
 }
 
 bool Cmd::DoReadCommandInCache() {
@@ -983,7 +990,7 @@ bool Cmd::IsSuspend() const { return (flag_ & kCmdFlagsSuspend); }
 // std::string Cmd::CurrentSubCommand() const { return ""; };
 bool Cmd::HasSubCommand() const { return subCmdName_.size() > 0; };
 std::vector<std::string> Cmd::SubCommand() const { return subCmdName_; };
-bool Cmd::IsAdminRequire() const { return (flag_ & kCmdFlagsAdminRequire); }
+bool Cmd::IsAdmin() const { return (flag_ & kCmdFlagsAdmin); }
 bool Cmd::IsNeedUpdateCache() const { return (flag_ & kCmdFlagsUpdateCache); }
 bool Cmd::IsNeedCacheDo() const {
   if (g_pika_conf->IsCacheDisabledTemporarily()) {

src/pika_server.cc

@@ -998,7 +998,12 @@ uint64_t PikaServer::ServerCurrentQps() { return statistic_.server_stat.qps.last
 
 uint64_t PikaServer::accumulative_connections() { return statistic_.server_stat.accumulative_connections.load(); }
 
+long long PikaServer::ServerKeyspaceHits() { return statistic_.server_stat.keyspace_hits.load(); } 
+long long PikaServer::ServerKeyspaceMisses() { return statistic_.server_stat.keyspace_misses.load(); }
+
 void PikaServer::incr_accumulative_connections() { ++(statistic_.server_stat.accumulative_connections); }
+void PikaServer::incr_server_keyspace_hits() { ++(statistic_.server_stat.keyspace_hits); }
+void PikaServer::incr_server_keyspace_misses() { ++(statistic_.server_stat.keyspace_misses); }
 
 // only one thread invoke this right now
 void PikaServer::ResetLastSecQuerynum() {

src/pika_set.cc

@@ -77,6 +77,7 @@ void SPopCmd::Do() {
     }
   } else if (s_.IsNotFound()) {
     res_.AppendContent("$-1");
+    res_.SetRes(CmdRes::kNoExists);
   } else if (s_.IsInvalidArgument()) {
     res_.SetRes(CmdRes::kMultiKey);
   } else {
@@ -123,8 +124,11 @@ void SCardCmd::DoInitial() {
 void SCardCmd::Do() {
   int32_t card = 0;
   s_ = db_->storage()->SCard(key_, &card);
-  if (s_.ok() || s_.IsNotFound()) {
+  if (s_.ok()) {
     res_.AppendInteger(card);
+  } else if (s_.IsNotFound()) {
+    res_.AppendInteger(card);
+    res_.SetRes(CmdRes::kNoExists);
   } else if (s_.IsInvalidArgument()) {
     res_.SetRes(CmdRes::kMultiKey);
   } else {
@@ -166,12 +170,15 @@ void SMembersCmd::DoInitial() {
 void SMembersCmd::Do() {
   std::vector<std::string> members;
   s_ = db_->storage()->SMembers(key_, &members);
-  if (s_.ok() || s_.IsNotFound()) {
+  if (s_.ok()) {
     res_.AppendArrayLenUint64(members.size());
     for (const auto& member : members) {
       res_.AppendStringLenUint64(member.size());
       res_.AppendContent(member);
     }
+  } else if (s_.IsNotFound()) {
+    res_.SetRes(CmdRes::kNoExists);
+    res_.AppendArrayLenUint64(members.size());
   } else if (s_.IsInvalidArgument()) {
     res_.SetRes(CmdRes::kMultiKey);
   } else {
@@ -249,7 +256,7 @@ void SScanCmd::Do() {
   std::vector<std::string> members;
   rocksdb::Status s = db_->storage()->SScan(key_, cursor_, pattern_, count_, &members, &next_cursor);
 
-  if (s.ok() || s.IsNotFound()) {
+  if (s.ok()) {
     res_.AppendContent("*2");
     char buf[32];
     int64_t len = pstd::ll2string(buf, sizeof(buf), next_cursor);
@@ -260,6 +267,15 @@ void SScanCmd::Do() {
     for (const auto& member : members) {
       res_.AppendString(member);
     }
+  } else if (s.IsNotFound()) {
+    res_.AppendContent("*2");
+    char buf[32];
+    int64_t len = pstd::ll2string(buf, sizeof(buf), next_cursor);
+    res_.AppendStringLen(len);
+    res_.AppendContent(buf);
+
+    res_.AppendArrayLenUint64(members.size());
+    res_.SetRes(CmdRes::kNoExists);
   } else if (s_.IsInvalidArgument()) {
     res_.SetRes(CmdRes::kMultiKey);
   } else {
@@ -280,7 +296,10 @@ void SRemCmd::DoInitial() {
 
 void SRemCmd::Do() {
   s_ = db_->storage()->SRem(key_, members_, &deleted_);
-  if (s_.ok() || s_.IsNotFound()) {
+  if (s_.ok()) {
+    res_.AppendInteger(deleted_);
+  } else if (s_.IsNotFound()) {
+    res_.SetRes(CmdRes::kNoExists);
     res_.AppendInteger(deleted_);
   } else if (s_.IsInvalidArgument()) {
     res_.SetRes(CmdRes::kMultiKey);
@@ -311,7 +330,7 @@ void SUnionCmd::DoInitial() {
 void SUnionCmd::Do() {
   std::vector<std::string> members;
   s_ = db_->storage()->SUnion(keys_, &members);
-  if (s_.ok() || s_.IsNotFound()) {
+  if (s_.ok()) {
     res_.AppendArrayLenUint64(members.size());
     for (const auto& member : members) {
       res_.AppendStringLenUint64(member.size());
@@ -478,6 +497,9 @@ void SIsmemberCmd::Do() {
   } else {
     res_.AppendContent(":0");
   }
+  if (s_.IsNotFound()) {
+    res_.SetRes(CmdRes::kNoExists);
+  }
 }
 
 void SIsmemberCmd::ReadCache() {
@@ -576,9 +598,13 @@ void SMoveCmd::DoInitial() {
 void SMoveCmd::Do() {
   int32_t res = 0;
   s_ = db_->storage()->SMove(src_key_, dest_key_, member_, &res);
-  if (s_.ok() || s_.IsNotFound()) {
+  if (s_.ok()) {
+    res_.AppendInteger(res);
+    move_success_ = res;
+  } else if (s_.IsNotFound()) {
     res_.AppendInteger(res);
     move_success_ = res;
+    res_.SetRes(CmdRes::kNoExists);
   } else if (s_.IsInvalidArgument()) {
     res_.SetRes(CmdRes::kMultiKey);
   } else {
@@ -648,7 +674,7 @@ void SRandmemberCmd::DoInitial() {
 void SRandmemberCmd::Do() {
   std::vector<std::string> members;
   s_ = db_->storage()->SRandmember(key_, static_cast<int32_t>(count_), &members);
-  if (s_.ok() || s_.IsNotFound()) {
+  if (s_.ok()) {
     if (!reply_arr && (static_cast<unsigned int>(!members.empty()) != 0U)) {
       res_.AppendStringLenUint64(members[0].size());
       res_.AppendContent(members[0]);
@@ -659,6 +685,9 @@ void SRandmemberCmd::Do() {
         res_.AppendContent(member);
       }
     }
+  } else if (s_.IsNotFound()) {
+    res_.SetRes(CmdRes::kNoExists);
+    res_.AppendArrayLenUint64(members.size());
   } else if (s_.IsInvalidArgument()) {
     res_.SetRes(CmdRes::kMultiKey);
   } else {

tests/integration/integrate_test.sh

@@ -4,5 +4,16 @@
 # of patent rights can be found in the PATENTS file in the same directory.
 
 go mod tidy
-go test -run=TestPikaWithCache -timeout 30m
-go test -run=TestPikaWithoutCache -timeout 30m
+
+echo $PATH
+echo $GOBIN
+
+# install ginkgo
+go get github.com/onsi/ginkgo/v2/ginkgo
+go install github.com/onsi/ginkgo/v2/ginkgo
+go get github.com/onsi/gomega/...
+
+ginkgo --dry-run -v |grep -E -v "\[[0-9]+\.[0-9]+ seconds]"
+
+go test -run=TestPikaWithCache -timeout 60m
+go test -run=TestPikaWithoutCache -timeout 60m

tests/integration/server_test.go

@@ -2,6 +2,8 @@ package pika_integration
 
 import (
 	"context"
+	"strconv"
+	"strings"
 	"time"
 	"os"
 	"path/filepath"
@@ -10,6 +12,20 @@ import (
 	"github.com/redis/go-redis/v9"
 )
 
+
+func extractKeyspaceHits(infoVal string, kWords string) string {
+	lines := strings.Split(infoVal, "\n")
+	for _, line := range lines {
+		if strings.Contains(line, kWords+":") {
+			parts := strings.Split(line, ":")
+			if len(parts) == 2 {
+				return strings.TrimSpace(parts[1])
+			}
+		}
+	}
+	return "0"
+}
+
 var _ = Describe("Server", func() {
 	ctx := context.TODO()
 	var client *redis.Client
@@ -538,6 +554,45 @@ var _ = Describe("Server", func() {
 			Expect(info.Val()).To(ContainSubstring(`used_cpu_sys`))
 		})
 
+		It("should Info keyspace hits", func() {
+			sRem := client.SRem(ctx, "keyspace_hits", "one")
+			Expect(sRem.Err()).NotTo(HaveOccurred())
+			sAdd := client.SAdd(ctx, "keyspace_hits", "one")
+			Expect(sAdd.Err()).NotTo(HaveOccurred())
+
+			info := client.Info(ctx, "stats")
+			Expect(info.Err()).NotTo(HaveOccurred())
+			Expect(info.Val()).NotTo(Equal(""))
+			Expect(info.Val()).To(ContainSubstring("keyspace_hits"))
+			Expect(info.Val()).To(ContainSubstring("keyspace_misses"))
+			oldInfoKeyspaceHitsStr := extractKeyspaceHits(info.Val(), "keyspace_hits")
+			oldInfoKeyspaceHits, err := strconv.ParseInt(oldInfoKeyspaceHitsStr, 10, 64)
+			Expect(err).NotTo(HaveOccurred())
+			oldInfoKeyspaceMissesStr := extractKeyspaceHits(info.Val(), "keyspace_misses")
+			oldInfoKeyspaceMisses, err := strconv.ParseInt(oldInfoKeyspaceMissesStr, 10, 64)
+			Expect(err).NotTo(HaveOccurred())
+
+			Expect(client.SMembers(ctx, "keyspace_hits").Err()).NotTo(HaveOccurred())
+			Expect(client.SMembers(ctx, "keyspace_misses").Err()).NotTo(HaveOccurred())
+
+			newInfo := client.Info(ctx, "stats")
+			Expect(newInfo.Err()).NotTo(HaveOccurred())
+			Expect(newInfo.Val()).NotTo(Equal(""))
+			Expect(newInfo.Val()).To(ContainSubstring("keyspace_hits"))
+			Expect(newInfo.Val()).To(ContainSubstring("keyspace_misses"))
+			newInfoKeyspaceHitsStr := extractKeyspaceHits(newInfo.Val(), "keyspace_hits")
+			newInfoKeyspaceHits, err := strconv.ParseInt(newInfoKeyspaceHitsStr, 10, 64)
+			Expect(err).NotTo(HaveOccurred())
+			newInfoKeyspaceMissesStr := extractKeyspaceHits(newInfo.Val(), "keyspace_misses")
+			newInfoKeyspaceMisses, err := strconv.ParseInt(newInfoKeyspaceMissesStr, 10, 64)
+			Expect(err).NotTo(HaveOccurred())
+
+			Expect(newInfoKeyspaceHits - oldInfoKeyspaceHits).To(Equal(int64(1)))
+			Expect(newInfoKeyspaceMisses - oldInfoKeyspaceMisses).To(Equal(int64(1)))
+
+			Expect(client.SRem(ctx, "keyspace_hits", "one").Err()).NotTo(HaveOccurred())
+		})
+
 		It("should Info after second", func() {
 			info := client.Info(ctx)
 			time.Sleep(1 * time.Second)