feat(server): implement LATENCY command set (Redis 7.0+)

Implement LATENCY HISTOGRAM, LATENCY RESET and LATENCY HELP subcommands
compatible with Redis 7.0 protocol. Skip commands with zero calls in
histogram output. Add Go integration tests covering all subcommands,
error handling, histogram output structure, filtering and zero-calls
exclusion.

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>

Author: gongna-au

src/commands/cmd_server.cc

@@ -1627,6 +1627,102 @@ class CommandFlushBlockCache : public Commander {
   }
 };
 
+class CommandLatency : public Commander {
+ public:
+  Status Execute([[maybe_unused]] engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
+    if (args_.size() < 2) {
+      return {Status::RedisParseErr, errWrongNumOfArguments};
+    }
+
+    std::string subcommand = util::ToLower(args_[1]);
+    if (subcommand == "histogram") {
+      return GetHistogram(srv, conn, output);
+    } else if (subcommand == "reset") {
+      *output = redis::Integer(0);
+      return Status::OK();
+    } else if (subcommand == "help") {
+      std::vector<std::string> help = {
+          "HELP",
+          "    Print this help message.",
+          "HISTOGRAM [command ...]",
+          "    Return a cumulative distribution of latencies in the format of a histogram for the specified "
+          "command(s).",
+          "    If no commands are specified, all commands with latency statistics are returned.",
+          "RESET [event ...]",
+          "    Return the number of reset event classes (Kvrocks has no spike-sampling infrastructure, always returns "
+          "0)."};
+      *output = ArrayOfBulkStrings(help);
+      return Status::OK();
+    }
+
+    return {Status::RedisParseErr, "Unknown LATENCY subcommand or wrong number of arguments"};
+  }
+
+ private:
+  Status GetHistogram(Server *srv, Connection *conn, std::string *output) {
+    if (srv->stats.bucket_boundaries.empty()) {
+      *output = conn->HeaderOfMap(0);
+      return Status::OK();
+    }
+
+    std::vector<const std::pair<const std::string, CommandHistogram> *> target_histograms;
+    if (args_.size() > 2) {
+      for (size_t i = 2; i < args_.size(); i++) {
+        auto it = srv->stats.commands_histogram.find(util::ToLower(args_[i]));
+        if (it != srv->stats.commands_histogram.end() && it->second.calls > 0) {
+          target_histograms.push_back(&(*it));
+        }
+      }
+    } else {
+      for (const auto &iter : srv->stats.commands_histogram) {
+        if (iter.second.calls > 0) {
+          target_histograms.push_back(&iter);
+        }
+      }
+    }
+
+    *output = conn->HeaderOfMap(target_histograms.size());
+    for (const auto *pair_ptr : target_histograms) {
+      const auto &cmd_name = pair_ptr->first;
+      const auto &hist = pair_ptr->second;
+
+      std::vector<std::pair<int64_t, uint64_t>> cumulative_buckets;
+      uint64_t cumulative = 0;
+      for (size_t i = 0; i < hist.buckets.size(); i++) {
+        cumulative += hist.buckets[i]->load(std::memory_order_relaxed);
+        if (cumulative == 0) continue;
+
+        int64_t boundary;
+        if (i < srv->stats.bucket_boundaries.size()) {
+          boundary = static_cast<int64_t>(srv->stats.bucket_boundaries[i]);
+        } else {
+          boundary = -1;
+        }
+        cumulative_buckets.emplace_back(boundary, cumulative);
+      }
+
+      *output += redis::BulkString(cmd_name);
+      *output += conn->HeaderOfMap(2);
+
+      *output += redis::BulkString("calls");
+      *output += redis::Integer(hist.calls.load(std::memory_order_relaxed));
+
+      *output += redis::BulkString("histogram_usec");
+      *output += conn->HeaderOfMap(cumulative_buckets.size());
+      for (const auto &[boundary, count] : cumulative_buckets) {
+        if (boundary < 0) {
+          *output += redis::BulkString("inf");
+        } else {
+          *output += redis::Integer(boundary);
+        }
+        *output += redis::Integer(count);
+      }
+    }
+
+    return Status::OK();
+  }
+};
+
 REDIS_REGISTER_COMMANDS(
     Server, MakeCmdAttr<CommandAuth>("auth", 2, "read-only ok-loading auth", NO_KEY),
     MakeCmdAttr<CommandPing>("ping", -1, "read-only", NO_KEY),
@@ -1670,5 +1766,6 @@ REDIS_REGISTER_COMMANDS(
     MakeCmdAttr<CommandPollUpdates>("pollupdates", -2, "read-only admin", NO_KEY),
     MakeCmdAttr<CommandSST>("sst", -3, "write exclusive admin", 1, 1, 1),
     MakeCmdAttr<CommandFlushMemTable>("flushmemtable", -1, "exclusive write", NO_KEY),
-    MakeCmdAttr<CommandFlushBlockCache>("flushblockcache", 1, "exclusive write", NO_KEY), )
+    MakeCmdAttr<CommandFlushBlockCache>("flushblockcache", 1, "exclusive write", NO_KEY),
+    MakeCmdAttr<CommandLatency>("latency", -2, "read-only admin", NO_KEY), )
 }  // namespace redis

tests/gocase/unit/latency/latency_test.go

@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package latency
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/apache/kvrocks/tests/gocase/util"
+	"github.com/stretchr/testify/require"
+)
+
+func TestLatencyHelp(t *testing.T) {
+	srv := util.StartServer(t, map[string]string{})
+	defer srv.Close()
+
+	ctx := context.Background()
+	rdb := srv.NewClient()
+	defer func() { require.NoError(t, rdb.Close()) }()
+
+	t.Run("LATENCY HELP returns help text", func(t *testing.T) {
+		result, err := rdb.Do(ctx, "LATENCY", "HELP").StringSlice()
+		require.NoError(t, err)
+		require.NotEmpty(t, result)
+		require.Contains(t, result[0], "HELP")
+	})
+
+	t.Run("LATENCY HELP is case-insensitive", func(t *testing.T) {
+		result, err := rdb.Do(ctx, "latency", "help").StringSlice()
+		require.NoError(t, err)
+		require.NotEmpty(t, result)
+	})
+}
+
+func TestLatencyReset(t *testing.T) {
+	srv := util.StartServer(t, map[string]string{})
+	defer srv.Close()
+
+	ctx := context.Background()
+	rdb := srv.NewClient()
+	defer func() { require.NoError(t, rdb.Close()) }()
+
+	t.Run("LATENCY RESET always returns 0", func(t *testing.T) {
+		val, err := rdb.Do(ctx, "LATENCY", "RESET").Int64()
+		require.NoError(t, err)
+		require.EqualValues(t, 0, val)
+	})
+
+	t.Run("LATENCY RESET with event arguments still returns 0", func(t *testing.T) {
+		val, err := rdb.Do(ctx, "LATENCY", "RESET", "event1", "event2").Int64()
+		require.NoError(t, err)
+		require.EqualValues(t, 0, val)
+	})
+}
+
+func TestLatencyErrors(t *testing.T) {
+	srv := util.StartServer(t, map[string]string{})
+	defer srv.Close()
+
+	ctx := context.Background()
+	rdb := srv.NewClient()
+	defer func() { require.NoError(t, rdb.Close()) }()
+
+	t.Run("LATENCY without subcommand returns error", func(t *testing.T) {
+		err := rdb.Do(ctx, "LATENCY").Err()
+		require.Error(t, err)
+	})
+
+	t.Run("LATENCY with unknown subcommand returns error", func(t *testing.T) {
+		err := rdb.Do(ctx, "LATENCY", "UNKNOWN").Err()
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "Unknown LATENCY subcommand")
+	})
+}
+
+func TestLatencyHistogramWithoutBuckets(t *testing.T) {
+	srv := util.StartServer(t, map[string]string{})
+	defer srv.Close()
+
+	ctx := context.Background()
+	rdb := srv.NewClient()
+	defer func() { require.NoError(t, rdb.Close()) }()
+
+	t.Run("LATENCY HISTOGRAM returns empty when histogram-bucket-boundaries is not set", func(t *testing.T) {
+		result, err := rdb.Do(ctx, "LATENCY", "HISTOGRAM").Slice()
+		require.NoError(t, err)
+		require.Empty(t, result)
+	})
+}
+
+func TestLatencyHistogram(t *testing.T) {
+	srv := util.StartServer(t, map[string]string{
+		"histogram-bucket-boundaries": "10,20,30,50,100,200,500,1000",
+	})
+	defer srv.Close()
+
+	ctx := context.Background()
+	rdb := srv.NewClient()
+	defer func() { require.NoError(t, rdb.Close()) }()
+
+	t.Run("LATENCY HISTOGRAM returns data after running commands", func(t *testing.T) {
+		for i := 0; i < 10; i++ {
+			require.NoError(t, rdb.Set(ctx, fmt.Sprintf("key-%d", i), "value", 0).Err())
+		}
+		for i := 0; i < 10; i++ {
+			require.NoError(t, rdb.Get(ctx, fmt.Sprintf("key-%d", i)).Err())
+		}
+
+		result, err := rdb.Do(ctx, "LATENCY", "HISTOGRAM").Slice()
+		require.NoError(t, err)
+		require.NotEmpty(t, result)
+	})
+
+	t.Run("LATENCY HISTOGRAM filters by command name", func(t *testing.T) {
+		for i := 0; i < 5; i++ {
+			require.NoError(t, rdb.Set(ctx, fmt.Sprintf("filter-key-%d", i), "value", 0).Err())
+		}
+
+		result, err := rdb.Do(ctx, "LATENCY", "HISTOGRAM", "set").Slice()
+		require.NoError(t, err)
+		require.NotEmpty(t, result)
+
+		cmdName, ok := result[0].(string)
+		require.True(t, ok)
+		require.Equal(t, "set", cmdName)
+	})
+
+	t.Run("LATENCY HISTOGRAM for nonexistent command returns empty", func(t *testing.T) {
+		result, err := rdb.Do(ctx, "LATENCY", "HISTOGRAM", "nonexistent_command").Slice()
+		require.NoError(t, err)
+		require.Empty(t, result)
+	})
+
+	t.Run("LATENCY HISTOGRAM skips commands with zero calls", func(t *testing.T) {
+		// hset has never been called in this test, so it should have calls==0
+		result, err := rdb.Do(ctx, "LATENCY", "HISTOGRAM", "hset").Slice()
+		require.NoError(t, err)
+		require.Empty(t, result)
+	})
+
+	t.Run("LATENCY HISTOGRAM is case-insensitive for command names", func(t *testing.T) {
+		resultLower, err := rdb.Do(ctx, "LATENCY", "HISTOGRAM", "set").Slice()
+		require.NoError(t, err)
+
+		resultUpper, err := rdb.Do(ctx, "LATENCY", "HISTOGRAM", "SET").Slice()
+		require.NoError(t, err)
+
+		require.Equal(t, len(resultLower), len(resultUpper))
+	})
+
+	t.Run("LATENCY HISTOGRAM with multiple command filters", func(t *testing.T) {
+		require.NoError(t, rdb.Set(ctx, "multi-key", "value", 0).Err())
+		_, err := rdb.Get(ctx, "multi-key").Result()
+		require.NoError(t, err)
+
+		result, err := rdb.Do(ctx, "LATENCY", "HISTOGRAM", "set", "get").Slice()
+		require.NoError(t, err)
+		// RESP2 flattened map: [cmd1, data1, cmd2, data2, ...]
+		require.GreaterOrEqual(t, len(result), 4)
+	})
+
+	t.Run("LATENCY HISTOGRAM contains calls and histogram_usec fields", func(t *testing.T) {
+		require.NoError(t, rdb.Set(ctx, "field-check-key", "value", 0).Err())
+
+		result, err := rdb.Do(ctx, "LATENCY", "HISTOGRAM", "set").Slice()
+		require.NoError(t, err)
+		require.GreaterOrEqual(t, len(result), 2)
+
+		innerSlice, ok := result[1].([]interface{})
+		require.True(t, ok, "expected inner result to be a slice")
+		require.GreaterOrEqual(t, len(innerSlice), 4)
+
+		require.Equal(t, "calls", innerSlice[0])
+		calls, ok := innerSlice[1].(int64)
+		require.True(t, ok)
+		require.Greater(t, calls, int64(0))
+
+		require.Equal(t, "histogram_usec", innerSlice[2])
+		histSlice, ok := innerSlice[3].([]interface{})
+		require.True(t, ok, "expected histogram_usec value to be a slice")
+		require.NotEmpty(t, histSlice)
+	})
+}