diff --git a/README.md b/README.md index 38596e635..a20b7e5c4 100644 --- a/README.md +++ b/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, diff --git a/cluster/lib/redis/cluster.rb b/cluster/lib/redis/cluster.rb index 0a5a3492f..bca79f66f 100644 --- a/cluster/lib/redis/cluster.rb +++ b/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 diff --git a/cluster/lib/redis/cluster/client.rb b/cluster/lib/redis/cluster/client.rb index 906d8c406..016a466a7 100644 --- a/cluster/lib/redis/cluster/client.rb +++ b/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) diff --git a/cluster/test/commands_on_server_test.rb b/cluster/test/commands_on_server_test.rb index b1c1a8fa4..ee5f25e94 100644 --- a/cluster/test/commands_on_server_test.rb +++ b/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 diff --git a/lib/redis/commands.rb b/lib/redis/commands.rb index 2d8d98306..c7d1e3601 100644 --- a/lib/redis/commands.rb +++ b/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| diff --git a/lib/redis/commands/server.rb b/lib/redis/commands/server.rb index e2551653e..0311ac454 100644 --- a/lib/redis/commands/server.rb +++ b/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] 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, 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 diff --git a/lib/redis/distributed.rb b/lib/redis/distributed.rb index 3bb37d1b7..d65ddb035 100644 --- a/lib/redis/distributed.rb +++ b/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 diff --git a/test/distributed/remote_server_control_commands_test.rb b/test/distributed/remote_server_control_commands_test.rb index 1e3bd0873..859d64974 100644 --- a/test/distributed/remote_server_control_commands_test.rb +++ b/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 diff --git a/test/redis/remote_server_control_commands_test.rb b/test/redis/remote_server_control_commands_test.rb index 43d9efd0d..71a88aabf 100644 --- a/test/redis/remote_server_control_commands_test.rb +++ b/test/redis/remote_server_control_commands_test.rb @@ -112,6 +112,133 @@ def test_waitaof_in_pipeline end end + def test_hotkeys_start_get_stop_reset + target_version "8.6.0" do + with_hotkeys_session_cleanup do + assert_equal "OK", r.hotkeys_start(metrics: %i[cpu net], count: 10) + r.set("hotkeys:a", "x") + r.set("hotkeys:b", "x") + 20.times { r.get("hotkeys:a") } + r.get("hotkeys:b") + + result = r.hotkeys_get + + assert_equal 1, result["tracking-active"] + assert_equal [[0, 16_383]], result["selected-slots"] + %w[sample-ratio all-commands-all-slots-us net-bytes-all-commands-all-slots + collection-start-time-unix-ms collection-duration-ms total-cpu-time-user-ms + total-cpu-time-sys-ms total-net-bytes].each do |field| + assert_kind_of Integer, result[field], field + end + %w[by-cpu-time-us by-net-bytes].each do |field| + assert_includes result[field].map(&:first), "hotkeys:a", field + result[field].each do |key, value| + assert_kind_of String, key + assert_kind_of Integer, value + end + end + # Byte counts are deterministic; CPU time is per-command wall-clock, so one + # stalled command on "hotkeys:b" can outrank 20 GETs of "hotkeys:a". + assert_equal "hotkeys:a", result["by-net-bytes"].first.first + + assert_equal "OK", r.hotkeys_stop + assert_equal 0, r.hotkeys_get["tracking-active"] + assert_equal "OK", r.hotkeys_reset + assert_nil r.hotkeys_get + assert_nil r.hotkeys_stop + end + end + end + + def test_hotkeys_start_with_a_single_metric_omits_the_other_fields + target_version "8.6.0" do + with_hotkeys_session_cleanup do + assert_equal "OK", r.hotkeys_start(metrics: :cpu) + r.get("hotkeys:a") + + result = r.hotkeys_get + + assert result.key?("total-cpu-time-user-ms") + assert_kind_of Array, result["by-cpu-time-us"] + refute result.key?("total-net-bytes") + refute result.key?("by-net-bytes") + end + end + end + + def test_hotkeys_start_twice_raises + target_version "8.6.0" do + with_hotkeys_session_cleanup do + r.hotkeys_start(metrics: :cpu) + + error = assert_raises(Redis::CommandError) { r.hotkeys_start(metrics: :net) } + assert_match(/already in progress/, error.message) + end + end + end + + def test_hotkeys_reset_while_active_raises + target_version "8.6.0" do + with_hotkeys_session_cleanup do + r.hotkeys_start(metrics: :cpu) + + error = assert_raises(Redis::CommandError) { r.hotkeys_reset } + assert_match(/stop tracking first/, error.message) + end + end + end + + def test_hotkeys_start_with_slots_outside_cluster_mode_raises + target_version "8.6.0" do + with_hotkeys_session_cleanup do + error = assert_raises(Redis::CommandError) { r.hotkeys_start(metrics: :cpu, slots: [0]) } + assert_match(/non-cluster mode/, error.message) + end + end + end + + def test_hotkeys_start_sends_the_arguments_in_order + received = nil + commands = { hotkeys: ->(*args) { received = args; "+OK" } } + + redis_mock(commands) do |redis| + assert_equal "OK", redis.hotkeys_start(metrics: %w[cpu net], count: 10, duration: 60, sample: 5, slots: [0, 1]) + end + + assert_equal %w[START METRICS 2 CPU NET COUNT 10 DURATION 60 SAMPLE 5 SLOTS 2 0 1], received + end + + def test_hotkeys_start_omits_empty_slots + received = nil + commands = { hotkeys: ->(*args) { received = args; "+OK" } } + + redis_mock(commands) do |redis| + redis.hotkeys_start(metrics: :net, slots: []) + end + + assert_equal %w[START METRICS 1 NET], received + end + + def test_hotkeys_start_requires_a_metric + assert_raises(ArgumentError) { r.hotkeys_start(metrics: []) } + end + + def test_hotkeys_get_in_pipeline + target_version "8.6.0" do + with_hotkeys_session_cleanup do + r.hotkeys_start(metrics: :net) + r.get("hotkeys:a") + + result = r.pipelined do |pipe| + pipe.hotkeys_get + end + + assert_equal 1, result.first["tracking-active"] + assert_kind_of Array, result.first["by-net-bytes"] + end + end + end + def test_info_with_a_single_section info = r.info(:server) @@ -272,6 +399,20 @@ def test_cluster_slot_stats_sends_the_subcommand_and_arguments end end + def with_hotkeys_session_cleanup + clear_hotkeys_session + yield + ensure + clear_hotkeys_session + end + + def clear_hotkeys_session + r.hotkeys_stop + r.hotkeys_reset + rescue Redis::CommandError + nil + end + def test_cluster_slot_stats_hashifies_the_per_slot_metrics target_version "8.2.0" do # CLUSTER SLOT-STATS replies with [[slot, metrics], ...] where `metrics` is a flat