Added support for HOTKEYS commands

README.md

@@ -413,6 +413,37 @@ reply per node; `himport_set` routes by key. With `Redis::Cluster`, the same
 commands fan out to every master node and return a single aggregated reply,
 matching the standalone API.
 
+## Hot key tracking (HOTKEYS)
+
+Redis 8.6 adds the `HOTKEYS` command family, which ranks the keys that consume
+the most CPU time and/or network bytes during a tracking session:
+
+```ruby
+redis.hotkeys_start(metrics: [:cpu, :net], count: 10) # => "OK"
+# ... let traffic flow ...
+result = redis.hotkeys_get
+result["by-cpu-time-us"] # => [["product:1", 1070], ["product:2", 67]]
+result["by-net-bytes"]   # => [["product:1", 411], ["product:2", 37]]
+redis.hotkeys_stop       # => "OK" (nil when no session is active)
+redis.hotkeys_reset      # => "OK"; hotkeys_get now returns nil
+```
+
+`hotkeys_start` also accepts `duration:` (seconds after which the session stops
+by itself), `sample:` (track one command in N) and, in cluster mode, `slots:`.
+`hotkeys_get` returns a Hash keyed by the server's field names: the
+`total-cpu-time-*-ms` and `by-cpu-time-us` fields are present only when `:cpu`
+was requested, `total-net-bytes` and `by-net-bytes` only when `:net` was. The
+`by-*` lists are `[key, value]` pairs, hottest first, with at most `count`
+entries.
+
+A tracking session is state of the server, shared by every connection, and only
+one can run at a time. Because the commands can be neither routed by key nor
+aggregated across nodes, they are only available on a single-node client:
+`Redis::Cluster` and `Redis::Distributed` raise `NotImplementedError`. To
+profile one node of a cluster or ring, connect to it directly with
+`Redis.new(host:, port:)`. The same applies inside `pipelined`/`multi` blocks
+on `Redis::Cluster`.
+
 ## Compare-and-set (CAS/CAD)
 
 Redis 8.4 adds atomic compare-and-set to `SET` and a new compare-and-delete command,

cluster/lib/redis/cluster.rb

@@ -204,6 +204,27 @@ def himport_discard_all
       reply.is_a?(Array) ? reply.max : reply
     end
 
+    HOTKEYS_NOT_SUPPORTED = 'HOTKEYS commands are not supported in cluster mode: the tracking ' \
+                            'session is state of a single node. Connect to that node with ' \
+                            'Redis.new(host:, port:) instead.'
+    private_constant :HOTKEYS_NOT_SUPPORTED
+
+    def hotkeys_start(**)
+      raise NotImplementedError, HOTKEYS_NOT_SUPPORTED
+    end
+
+    def hotkeys_get
+      raise NotImplementedError, HOTKEYS_NOT_SUPPORTED
+    end
+
+    def hotkeys_stop
+      raise NotImplementedError, HOTKEYS_NOT_SUPPORTED
+    end
+
+    def hotkeys_reset
+      raise NotImplementedError, HOTKEYS_NOT_SUPPORTED
+    end
+
     # WAITAOF on cluster: the guarantee is per connection, so the command must reach every
     # primary over the connection that carried this client's writes. The driver leaves it on
     # single-node routing (no aggregation for its `agg_min` tip over array replies), so

cluster/lib/redis/cluster/client.rb

@@ -1,5 +1,6 @@
 # frozen_string_literal: true
 
+require 'delegate'
 require 'redis-cluster-client'
 require 'redis/cluster/transaction_adapter'
 
@@ -91,12 +92,33 @@ def blocking_call_v(timeout, command, &block)
         handle_errors { super(timeout, command, &block) }
       end
 
-      def pipelined(exception: true, &block)
-        handle_errors { super(exception: exception, &block) }
+      # Inside a pipeline or transaction the driver routes the keyless HOTKEYS verbs to an
+      # arbitrary node, so they are rejected there just as Redis::Cluster rejects direct calls.
+      class HotkeysGuard < SimpleDelegator
+        def call_v(command, &block)
+          reject_hotkeys(command)
+          super
+        end
+
+        def blocking_call_v(timeout, command, &block)
+          reject_hotkeys(command)
+          super
+        end
+
+        private
+
+        def reject_hotkeys(command)
+          raise NotImplementedError, HOTKEYS_NOT_SUPPORTED if command.first.to_s.casecmp?('hotkeys')
+        end
+      end
+      private_constant :HotkeysGuard
+
+      def pipelined(exception: true)
+        handle_errors { super(exception: exception) { |pipeline| yield HotkeysGuard.new(pipeline) } }
       end
 
-      def multi(watch: nil, &block)
-        handle_errors { super(watch: watch, &block) }
+      def multi(watch: nil)
+        handle_errors { super(watch: watch) { |transaction| yield HotkeysGuard.new(transaction) } }
       end
 
       def watch(*keys, &block)

cluster/test/commands_on_server_test.rb

@@ -282,4 +282,18 @@ def test_waitaof_in_pipeline_returns_a_single_node_pair
       assert_kind_of Integer, replicas
     end
   end
+
+  def test_hotkeys_commands_are_not_supported
+    assert_raises(NotImplementedError) { redis.hotkeys_start(metrics: :cpu) }
+    assert_raises(NotImplementedError) { redis.hotkeys_get }
+    assert_raises(NotImplementedError) { redis.hotkeys_stop }
+    assert_raises(NotImplementedError) { redis.hotkeys_reset }
+  end
+
+  def test_hotkeys_commands_are_not_supported_in_pipelines_and_transactions
+    assert_raises(NotImplementedError) { redis.pipelined(&:hotkeys_get) }
+    assert_raises(NotImplementedError) { redis.pipelined { |pipe| pipe.hotkeys_start(metrics: :cpu) } }
+    assert_raises(NotImplementedError) { redis.multi(&:hotkeys_stop) }
+    assert_raises(NotImplementedError) { redis.multi(&:hotkeys_reset) }
+  end
 end

lib/redis/commands.rb

@@ -322,6 +322,18 @@ module Commands
       end
     }
 
+    # HOTKEYS GET: the server wraps the reply map (RESP3) or flat [field, value, ...] array (RESP2)
+    # in a one-element array, and the `by-*` lists are flat [key, value, ...] under both protocols.
+    HashifyHotkeys = lambda { |reply|
+      return reply unless reply.is_a?(Array)
+
+      hash = Hashify.call(reply.first)
+      %w[by-cpu-time-us by-net-bytes].each do |field|
+        hash[field] = Pairify.call(hash[field]) if hash.key?(field)
+      end
+      hash
+    }
+
     # FUNCTION LIST: RESP2 replies with a flat [k, v, ...] array per library (and per
     # function within it); RESP3 already returns maps. Converge on nested Hashes.
     HashifyFunctionList = lambda { |reply|

lib/redis/commands/server.rb

@@ -182,6 +182,84 @@ def waitaof(numlocal, numreplicas, timeout)
         send_blocking_command(command, timeout / 1000.0)
       end
 
+      # Start a hot-key tracking session (Redis 8.6).
+      #
+      # Keys are ranked by the share of CPU time and/or network bytes they consume while the
+      # session runs. The session is server-wide state shared by every connection, and only one
+      # can run at a time, so the command family is only available on a single-node client:
+      # `Redis::Cluster` and `Redis::Distributed` raise `NotImplementedError`.
+      #
+      # @example Track the top 10 keys by CPU time and network bytes
+      #   redis.hotkeys_start(metrics: [:cpu, :net], count: 10)
+      #     # => "OK"
+      # @example Track CPU time for one minute, sampling one command in five
+      #   redis.hotkeys_start(metrics: :cpu, duration: 60, sample: 5)
+      #     # => "OK"
+      #
+      # @param [Symbol, String, Array<Symbol, String>] metrics one or both of `:cpu` and `:net`
+      # @param [Integer, nil] count keys to report per metric, `1` to `64` (server default `10`)
+      # @param [Integer, nil] duration stop the session after this many seconds, `1` to `1_000_000`;
+      #   without it the session runs until #hotkeys_stop
+      # @param [Integer, nil] sample track one command in `sample` (server default `1`: every command)
+      # @param [Array<Integer>, nil] slots cluster mode only: hash slots to track, all of them owned
+      #   by the receiving node; the server rejects the option outside cluster mode
+      # @return [String] `"OK"`
+      # @raise [Redis::CommandError] when a session is already in progress or a value is out of range
+      def hotkeys_start(metrics:, count: nil, duration: nil, sample: nil, slots: nil)
+        metrics = Array(metrics).map { |metric| metric.to_s.upcase }
+        raise ArgumentError, "metrics must not be empty" if metrics.empty?
+
+        args = [:hotkeys, "START", "METRICS", metrics.size].concat(metrics)
+        args << "COUNT" << Integer(count) if count
+        args << "DURATION" << Integer(duration) if duration
+        args << "SAMPLE" << Integer(sample) if sample
+        slots = Array(slots)
+        args.push("SLOTS", slots.size, *slots.map { |slot| Integer(slot) }) unless slots.empty?
+        send_command(args)
+      end
+
+      # Return the results of the running or most recent hot-key tracking session (Redis 8.6).
+      #
+      # @example
+      #   redis.hotkeys_get
+      #     # => { "tracking-active" => 1, "sample-ratio" => 1, "selected-slots" => [[0, 16383]],
+      #     #      "all-commands-all-slots-us" => 73, "net-bytes-all-commands-all-slots" => 311,
+      #     #      "collection-start-time-unix-ms" => 1793782936885, "collection-duration-ms" => 4,
+      #     #      "total-cpu-time-user-ms" => 0, "total-cpu-time-sys-ms" => 0, "total-net-bytes" => 311,
+      #     #      "by-cpu-time-us" => [["product:1", 69], ["product:2", 4]],
+      #     #      "by-net-bytes" => [["product:1", 285], ["product:2", 26]] }
+      #
+      # @return [Hash{String => Object}] the reply fields, keyed by the server's names. Always
+      #   present: `tracking-active` (`1` or `0`), `sample-ratio`, `selected-slots` (`[first, last]`
+      #   slot ranges), `all-commands-all-slots-us`, `net-bytes-all-commands-all-slots`,
+      #   `collection-start-time-unix-ms` and `collection-duration-ms`. With the `:cpu` metric:
+      #   `total-cpu-time-user-ms`, `total-cpu-time-sys-ms` and `by-cpu-time-us`; with `:net`:
+      #   `total-net-bytes` and `by-net-bytes`. The `by-*` lists are `[key, Integer]` pairs,
+      #   hottest first, with at most `count` entries (possibly none). When `slots:` was given to
+      #   #hotkeys_start the per-slot counters `all-commands-selected-slots-us`,
+      #   `net-bytes-all-commands-selected-slots` and, with sampling,
+      #   `sampled-commands-selected-slots-us`, `net-bytes-sampled-commands-selected-slots` are added.
+      # @return [nil] when no session was started, or after #hotkeys_reset
+      def hotkeys_get
+        send_command([:hotkeys, "GET"], &HashifyHotkeys)
+      end
+
+      # Stop the running hot-key tracking session, keeping its results for #hotkeys_get (Redis 8.6).
+      #
+      # @return [String] `"OK"`
+      # @return [nil] when no session is active
+      def hotkeys_stop
+        send_command([:hotkeys, "STOP"])
+      end
+
+      # Release the resources of a stopped hot-key tracking session (Redis 8.6).
+      #
+      # @return [String] `"OK"`
+      # @raise [Redis::CommandError] while a session is active; call #hotkeys_stop first
+      def hotkeys_reset
+        send_command([:hotkeys, "RESET"])
+      end
+
       # Synchronously save the dataset to disk and then shut down the server.
       #
       # @example Default: save if a save point is configured, then exit

lib/redis/distributed.rb

@@ -114,6 +114,27 @@ def monitor
       raise NotImplementedError
     end
 
+    HOTKEYS_NOT_SUPPORTED = "HOTKEYS commands are not supported on Redis::Distributed: the tracking " \
+                            "session is state of a single server. Call them on one node, " \
+                            "e.g. distributed.nodes.first.hotkeys_start(...)."
+    private_constant :HOTKEYS_NOT_SUPPORTED
+
+    def hotkeys_start(**)
+      raise NotImplementedError, HOTKEYS_NOT_SUPPORTED
+    end
+
+    def hotkeys_get
+      raise NotImplementedError, HOTKEYS_NOT_SUPPORTED
+    end
+
+    def hotkeys_stop
+      raise NotImplementedError, HOTKEYS_NOT_SUPPORTED
+    end
+
+    def hotkeys_reset
+      raise NotImplementedError, HOTKEYS_NOT_SUPPORTED
+    end
+
     # Synchronously save the dataset to disk.
     def save
       on_each_node :save

test/distributed/remote_server_control_commands_test.rb

@@ -69,6 +69,13 @@ def test_monitor
     assert ex.is_a?(NotImplementedError)
   end
 
+  def test_hotkeys_commands_are_not_supported
+    assert_raises(NotImplementedError) { r.hotkeys_start(metrics: :cpu) }
+    assert_raises(NotImplementedError) { r.hotkeys_get }
+    assert_raises(NotImplementedError) { r.hotkeys_stop }
+    assert_raises(NotImplementedError) { r.hotkeys_reset }
+  end
+
   def test_echo
     assert_equal ["foo bar baz\n"], r.echo("foo bar baz\n")
   end