From a45447e2fb6dc1a252dec6fbb55ec03f43278807 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Fri, 6 Feb 2026 13:26:13 -0500 Subject: [PATCH 1/2] feat: enriched INFO command with server, clients, memory, persistence, stats, keyspace sections MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit INFO with no args returns all sections. INFO
returns just that section. follows Redis conventions — # headers, key:value pairs. sections: - server: ember_version, process_id, uptime_in_seconds, shard_count - clients: connected_clients, max_clients - memory: used_memory, used_memory_human, max_memory, max_memory_human - persistence: aof_enabled - stats: total_connections_received, total_commands_processed, expired_keys, evicted_keys - keyspace: db0:keys=N,expires=N,used_bytes=N adds ServerContext struct (shared via Arc) that tracks start_time, version, config, and atomic counters for connections/commands. --- crates/ember-server/src/connection.rs | 192 +++++++++++++++++++++----- crates/ember-server/src/server.rs | 40 +++++- 2 files changed, 197 insertions(+), 35 deletions(-) diff --git a/crates/ember-server/src/connection.rs b/crates/ember-server/src/connection.rs index 66ec4c16..90498520 100644 --- a/crates/ember-server/src/connection.rs +++ b/crates/ember-server/src/connection.rs @@ -4,6 +4,8 @@ //! sharded engine, and writes responses back. Supports pipelining //! by processing multiple frames from a single read. +use std::sync::atomic::Ordering; +use std::sync::Arc; use std::time::{Duration, Instant}; use bytes::{Bytes, BytesMut}; @@ -12,6 +14,8 @@ use ember_protocol::{parse_frame, Command, Frame, SetExpire}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; +use crate::server::ServerContext; + /// Initial read buffer capacity. 4KB covers most commands comfortably /// without over-allocating for simple PING/SET/GET workloads. const BUF_CAPACITY: usize = 4096; @@ -33,6 +37,7 @@ const IDLE_TIMEOUT: Duration = Duration::from_secs(300); // 5 minutes pub async fn handle( mut stream: TcpStream, engine: Engine, + ctx: &Arc, metrics_enabled: bool, ) -> Result<(), Box> { // disable Nagle's algorithm — cache servers need low-latency writes, @@ -68,7 +73,7 @@ pub async fn handle( match parse_frame(&buf) { Ok(Some((frame, consumed))) => { let _ = buf.split_to(consumed); - let response = process(frame, &engine, metrics_enabled).await; + let response = process(frame, &engine, ctx, metrics_enabled).await; response.serialize(&mut out); } Ok(None) => break, // need more data @@ -91,7 +96,12 @@ pub async fn handle( /// /// When `metrics_enabled` is true, records per-command latency and /// error counters in prometheus. -async fn process(frame: Frame, engine: &Engine, metrics_enabled: bool) -> Frame { +async fn process( + frame: Frame, + engine: &Engine, + ctx: &Arc, + metrics_enabled: bool, +) -> Frame { match Command::from_frame(frame) { Ok(cmd) => { let cmd_name = cmd.command_name(); @@ -101,7 +111,8 @@ async fn process(frame: Frame, engine: &Engine, metrics_enabled: bool) -> Frame None }; - let response = execute(cmd, engine).await; + let response = execute(cmd, engine, ctx).await; + ctx.commands_processed.fetch_add(1, Ordering::Relaxed); if let Some(start) = start { let is_error = matches!(&response, Frame::Error(_)); @@ -119,7 +130,7 @@ async fn process(frame: Frame, engine: &Engine, metrics_enabled: bool) -> Frame /// Ping and Echo are handled inline (no shard routing needed). /// Single-key commands route to the owning shard. Multi-key commands /// (DEL, EXISTS) fan out across shards and aggregate results. -async fn execute(cmd: Command, engine: &Engine) -> Frame { +async fn execute(cmd: Command, engine: &Engine, ctx: &Arc) -> Frame { match cmd { // -- no shard needed -- Command::Ping(None) => Frame::Simple("PONG".into()), @@ -330,36 +341,7 @@ async fn execute(cmd: Command, engine: &Engine) -> Frame { }, Command::Info { section } => { - let section_upper = section.as_deref().map(|s| s.to_ascii_uppercase()); - match section_upper.as_deref() { - None | Some("KEYSPACE") => match engine.broadcast(|| ShardRequest::Stats).await { - Ok(responses) => { - let mut total = KeyspaceStats { - key_count: 0, - used_bytes: 0, - keys_with_expiry: 0, - keys_expired: 0, - keys_evicted: 0, - }; - for r in &responses { - if let ShardResponse::Stats(stats) = r { - total.key_count += stats.key_count; - total.used_bytes += stats.used_bytes; - total.keys_with_expiry += stats.keys_with_expiry; - total.keys_expired += stats.keys_expired; - total.keys_evicted += stats.keys_evicted; - } - } - let info = format!( - "# Keyspace\r\ndb0:keys={},expires={},used_bytes={}\r\n", - total.key_count, total.keys_with_expiry, total.used_bytes - ); - Frame::Bulk(Bytes::from(info)) - } - Err(e) => Frame::Error(format!("ERR {e}")), - }, - Some(other) => Frame::Error(format!("ERR unsupported INFO section '{other}'")), - } + render_info(engine, ctx, section.as_deref()).await } Command::BgSave => match engine.broadcast(|| ShardRequest::Snapshot).await { @@ -940,6 +922,148 @@ where } } +/// Renders the INFO response with multiple sections. +/// +/// With no argument, returns all sections. With a section name, +/// returns only that section. Matches Redis convention of `#` headers +/// followed by `key:value` pairs separated by `\r\n`. +async fn render_info( + engine: &Engine, + ctx: &Arc, + section: Option<&str>, +) -> Frame { + let section_upper = section.map(|s| s.to_ascii_uppercase()); + let want_all = section_upper.is_none(); + let want = |name: &str| want_all || section_upper.as_deref() == Some(name); + + // only broadcast to shards if we need keyspace/memory/persistence sections + let stats = if want("KEYSPACE") || want("MEMORY") || want("PERSISTENCE") || want("STATS") { + match engine.broadcast(|| ShardRequest::Stats).await { + Ok(responses) => { + let mut total = KeyspaceStats { + key_count: 0, + used_bytes: 0, + keys_with_expiry: 0, + keys_expired: 0, + keys_evicted: 0, + }; + for r in &responses { + if let ShardResponse::Stats(s) = r { + total.key_count += s.key_count; + total.used_bytes += s.used_bytes; + total.keys_with_expiry += s.keys_with_expiry; + total.keys_expired += s.keys_expired; + total.keys_evicted += s.keys_evicted; + } + } + Some(total) + } + Err(e) => return Frame::Error(format!("ERR {e}")), + } + } else { + None + }; + + let mut out = String::with_capacity(512); + + if want("SERVER") { + let uptime = ctx.start_time.elapsed().as_secs(); + out.push_str("# Server\r\n"); + out.push_str(&format!("ember_version:{}\r\n", ctx.version)); + out.push_str(&format!("process_id:{}\r\n", std::process::id())); + out.push_str(&format!("uptime_in_seconds:{uptime}\r\n")); + out.push_str(&format!("shard_count:{}\r\n", ctx.shard_count)); + out.push_str("\r\n"); + } + + if want("CLIENTS") { + let connected = ctx.connections_accepted.load(Ordering::Relaxed); + out.push_str("# Clients\r\n"); + out.push_str(&format!("connected_clients:{connected}\r\n")); + out.push_str(&format!("max_clients:{}\r\n", ctx.max_connections)); + out.push_str("\r\n"); + } + + if want("MEMORY") { + if let Some(ref stats) = stats { + out.push_str("# Memory\r\n"); + out.push_str(&format!("used_memory:{}\r\n", stats.used_bytes)); + out.push_str(&format!( + "used_memory_human:{}\r\n", + human_bytes(stats.used_bytes) + )); + if let Some(max) = ctx.max_memory { + out.push_str(&format!("max_memory:{max}\r\n")); + out.push_str(&format!("max_memory_human:{}\r\n", human_bytes(max))); + } else { + out.push_str("max_memory:0\r\n"); + out.push_str("max_memory_human:unlimited\r\n"); + } + out.push_str("\r\n"); + } + } + + if want("PERSISTENCE") { + out.push_str("# Persistence\r\n"); + out.push_str(&format!( + "aof_enabled:{}\r\n", + if ctx.aof_enabled { 1 } else { 0 } + )); + out.push_str("\r\n"); + } + + if want("STATS") { + let total_conns = ctx.connections_accepted.load(Ordering::Relaxed); + let total_cmds = ctx.commands_processed.load(Ordering::Relaxed); + out.push_str("# Stats\r\n"); + out.push_str(&format!("total_connections_received:{total_conns}\r\n")); + out.push_str(&format!("total_commands_processed:{total_cmds}\r\n")); + if let Some(ref stats) = stats { + out.push_str(&format!("expired_keys:{}\r\n", stats.keys_expired)); + out.push_str(&format!("evicted_keys:{}\r\n", stats.keys_evicted)); + } + out.push_str("\r\n"); + } + + if want("KEYSPACE") { + if let Some(ref stats) = stats { + out.push_str("# Keyspace\r\n"); + if stats.key_count > 0 { + out.push_str(&format!( + "db0:keys={},expires={},used_bytes={}\r\n", + stats.key_count, stats.keys_with_expiry, stats.used_bytes + )); + } + out.push_str("\r\n"); + } + } + + // trim trailing blank line + if out.ends_with("\r\n\r\n") { + out.truncate(out.len() - 2); + } + + Frame::Bulk(Bytes::from(out)) +} + +/// Formats a byte count as a human-readable string (e.g. "1.23M"). +fn human_bytes(bytes: usize) -> String { + const KB: f64 = 1024.0; + const MB: f64 = KB * 1024.0; + const GB: f64 = MB * 1024.0; + + let b = bytes as f64; + if b >= GB { + format!("{:.2}G", b / GB) + } else if b >= MB { + format!("{:.2}M", b / MB) + } else if b >= KB { + format!("{:.2}K", b / KB) + } else { + format!("{bytes}B") + } +} + /// Returns the standard WRONGTYPE error frame. fn wrongtype_error() -> Frame { Frame::Error("WRONGTYPE Operation against a key holding the wrong kind of value".into()) diff --git a/crates/ember-server/src/server.rs b/crates/ember-server/src/server.rs index 34f5d61e..edd8a888 100644 --- a/crates/ember-server/src/server.rs +++ b/crates/ember-server/src/server.rs @@ -4,7 +4,9 @@ //! connections and waits for in-flight requests to drain before exiting. use std::net::SocketAddr; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; +use std::time::Instant; use ember_core::{Engine, EngineConfig}; use tokio::net::TcpListener; @@ -16,6 +18,22 @@ use crate::connection; /// Default maximum number of concurrent client connections. const DEFAULT_MAX_CONNECTIONS: usize = 10_000; +/// Shared server state for INFO and observability. +/// +/// Created once at startup and shared (via `Arc`) across all connection +/// handlers. Atomic counters avoid any locking on the hot path. +#[derive(Debug)] +pub struct ServerContext { + pub start_time: Instant, + pub version: &'static str, + pub shard_count: usize, + pub max_connections: usize, + pub max_memory: Option, + pub aof_enabled: bool, + pub connections_accepted: AtomicU64, + pub commands_processed: AtomicU64, +} + /// Binds to `addr` and runs the accept loop. /// /// Spawns a sharded engine with the given shard count and config, then @@ -37,6 +55,13 @@ pub async fn run( std::fs::create_dir_all(&pcfg.data_dir)?; } + let aof_enabled = config + .persistence + .as_ref() + .map(|p| p.append_only) + .unwrap_or(false); + let max_memory = config.shard.max_memory.map(|per_shard| per_shard * shard_count); + let engine = Engine::with_config(shard_count, config); if metrics_enabled { @@ -47,6 +72,17 @@ pub async fn run( let max_conn = max_connections.unwrap_or(DEFAULT_MAX_CONNECTIONS); let semaphore = Arc::new(Semaphore::new(max_conn)); + let ctx = Arc::new(ServerContext { + start_time: Instant::now(), + version: env!("CARGO_PKG_VERSION"), + shard_count, + max_connections: max_conn, + max_memory, + aof_enabled, + connections_accepted: AtomicU64::new(0), + commands_processed: AtomicU64::new(0), + }); + info!( "listening on {addr} with {} shards (max {max_conn} connections)", engine.shard_count() @@ -82,12 +118,14 @@ pub async fn run( if metrics_enabled { crate::metrics::on_connection_accepted(); } + ctx.connections_accepted.fetch_add(1, Ordering::Relaxed); let engine = engine.clone(); + let ctx = Arc::clone(&ctx); let metrics = metrics_enabled; tokio::spawn(async move { - if let Err(e) = connection::handle(stream, engine, metrics).await { + if let Err(e) = connection::handle(stream, engine, &ctx, metrics).await { error!("connection error from {peer}: {e}"); } if metrics { From 083fc29d46edb496fb3d7d755cc56885df5afe8a Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Fri, 6 Feb 2026 13:29:47 -0500 Subject: [PATCH 2/2] feat: add SLOWLOG command with ring buffer and configurable threshold MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit implements SLOWLOG GET [count], SLOWLOG LEN, and SLOWLOG RESET with a fixed-size ring buffer (default 128 entries). commands exceeding the threshold are recorded with id, timestamp, duration, and command name. CLI flags: - --slowlog-log-slower-than: microseconds threshold (default 10000 = 10ms, -1 to disable, 0 to log everything) - --slowlog-max-len: ring buffer capacity (default 128) the slow log is shared across connections via Arc. the inner mutex is effectively uncontended since only slow commands (rare by definition) ever acquire it. the threshold check itself is a single duration comparison — ~1ns overhead per command. also simplifies the process() timing: Instant::now() is always taken (needed for slowlog), and metrics recording is gated separately. --- crates/ember-protocol/src/command.rs | 41 +++++ crates/ember-server/src/connection.rs | 57 +++++-- crates/ember-server/src/main.rs | 27 +++- crates/ember-server/src/server.rs | 7 +- crates/ember-server/src/slowlog.rs | 206 ++++++++++++++++++++++++++ 5 files changed, 326 insertions(+), 12 deletions(-) create mode 100644 crates/ember-server/src/slowlog.rs diff --git a/crates/ember-protocol/src/command.rs b/crates/ember-protocol/src/command.rs index b32a926b..7e9ecfa1 100644 --- a/crates/ember-protocol/src/command.rs +++ b/crates/ember-protocol/src/command.rs @@ -263,6 +263,15 @@ pub enum Command { /// ASKING. Signals that the next command is for a migrating slot. Asking, + /// SLOWLOG GET [count]. Returns recent slow log entries. + SlowLogGet { count: Option }, + + /// SLOWLOG LEN. Returns the number of entries in the slow log. + SlowLogLen, + + /// SLOWLOG RESET. Clears the slow log. + SlowLogReset, + /// A command we don't recognize (yet). Unknown(String), } @@ -359,6 +368,9 @@ impl Command { Command::ClusterGetKeysInSlot { .. } => "cluster_getkeysinslot", Command::Migrate { .. } => "migrate", Command::Asking => "asking", + Command::SlowLogGet { .. } => "slowlog", + Command::SlowLogLen => "slowlog", + Command::SlowLogReset => "slowlog", Command::Unknown(_) => "unknown", } } @@ -439,6 +451,7 @@ impl Command { "CLUSTER" => parse_cluster(&frames[1..]), "ASKING" => parse_asking(&frames[1..]), "MIGRATE" => parse_migrate(&frames[1..]), + "SLOWLOG" => parse_slowlog(&frames[1..]), _ => Ok(Command::Unknown(name)), } } @@ -1298,6 +1311,34 @@ fn parse_asking(args: &[Frame]) -> Result { Ok(Command::Asking) } +fn parse_slowlog(args: &[Frame]) -> Result { + if args.is_empty() { + return Err(ProtocolError::WrongArity("SLOWLOG".into())); + } + + let subcmd = extract_string(&args[0])?.to_ascii_uppercase(); + match subcmd.as_str() { + "GET" => { + let count = if args.len() > 1 { + let n: usize = extract_string(&args[1])? + .parse() + .map_err(|_| { + ProtocolError::InvalidCommandFrame("invalid count for SLOWLOG GET".into()) + })?; + Some(n) + } else { + None + }; + Ok(Command::SlowLogGet { count }) + } + "LEN" => Ok(Command::SlowLogLen), + "RESET" => Ok(Command::SlowLogReset), + other => Err(ProtocolError::InvalidCommandFrame(format!( + "unknown SLOWLOG subcommand '{other}'" + ))), + } +} + fn parse_slot_list(args: &[Frame]) -> Result, ProtocolError> { let mut slots = Vec::with_capacity(args.len()); for arg in args { diff --git a/crates/ember-server/src/connection.rs b/crates/ember-server/src/connection.rs index 90498520..276fe8d0 100644 --- a/crates/ember-server/src/connection.rs +++ b/crates/ember-server/src/connection.rs @@ -15,6 +15,7 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; use crate::server::ServerContext; +use crate::slowlog::SlowLog; /// Initial read buffer capacity. 4KB covers most commands comfortably /// without over-allocating for simple PING/SET/GET workloads. @@ -38,6 +39,7 @@ pub async fn handle( mut stream: TcpStream, engine: Engine, ctx: &Arc, + slow_log: &Arc, metrics_enabled: bool, ) -> Result<(), Box> { // disable Nagle's algorithm — cache servers need low-latency writes, @@ -73,7 +75,8 @@ pub async fn handle( match parse_frame(&buf) { Ok(Some((frame, consumed))) => { let _ = buf.split_to(consumed); - let response = process(frame, &engine, ctx, metrics_enabled).await; + let response = + process(frame, &engine, ctx, slow_log, metrics_enabled).await; response.serialize(&mut out); } Ok(None) => break, // need more data @@ -100,23 +103,23 @@ async fn process( frame: Frame, engine: &Engine, ctx: &Arc, + slow_log: &Arc, metrics_enabled: bool, ) -> Frame { match Command::from_frame(frame) { Ok(cmd) => { let cmd_name = cmd.command_name(); - let start = if metrics_enabled { - Some(Instant::now()) - } else { - None - }; + let start = Instant::now(); + + let response = execute(cmd, engine, ctx, slow_log).await; + let elapsed = start.elapsed(); - let response = execute(cmd, engine, ctx).await; ctx.commands_processed.fetch_add(1, Ordering::Relaxed); + slow_log.maybe_record(elapsed, cmd_name); - if let Some(start) = start { + if metrics_enabled { let is_error = matches!(&response, Frame::Error(_)); - crate::metrics::record_command(cmd_name, start.elapsed(), is_error); + crate::metrics::record_command(cmd_name, elapsed, is_error); } response @@ -130,7 +133,12 @@ async fn process( /// Ping and Echo are handled inline (no shard routing needed). /// Single-key commands route to the owning shard. Multi-key commands /// (DEL, EXISTS) fan out across shards and aggregate results. -async fn execute(cmd: Command, engine: &Engine, ctx: &Arc) -> Frame { +async fn execute( + cmd: Command, + engine: &Engine, + ctx: &Arc, + slow_log: &Arc, +) -> Frame { match cmd { // -- no shard needed -- Command::Ping(None) => Frame::Simple("PONG".into()), @@ -897,6 +905,35 @@ async fn execute(cmd: Command, engine: &Engine, ctx: &Arc) -> Fra Frame::Error("ERR This instance has cluster support disabled".into()) } + // -- slow log commands -- + Command::SlowLogGet { count } => { + let entries = slow_log.get(count); + let frames: Vec = entries + .into_iter() + .map(|e| { + let ts = e + .timestamp + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + Frame::Array(vec![ + Frame::Integer(e.id as i64), + Frame::Integer(ts as i64), + Frame::Integer(e.duration.as_micros() as i64), + Frame::Array(vec![Frame::Bulk(Bytes::from(e.command))]), + ]) + }) + .collect(); + Frame::Array(frames) + } + + Command::SlowLogLen => Frame::Integer(slow_log.len() as i64), + + Command::SlowLogReset => { + slow_log.reset(); + Frame::Simple("OK".into()) + } + Command::Unknown(name) => Frame::Error(format!("ERR unknown command '{name}'")), } } diff --git a/crates/ember-server/src/main.rs b/crates/ember-server/src/main.rs index c571e1f2..504667c9 100644 --- a/crates/ember-server/src/main.rs +++ b/crates/ember-server/src/main.rs @@ -2,6 +2,7 @@ mod config; mod connection; mod metrics; mod server; +mod slowlog; use std::net::SocketAddr; use std::path::PathBuf; @@ -48,6 +49,15 @@ struct Args { /// port for prometheus metrics HTTP endpoint. disabled when not set #[arg(long)] metrics_port: Option, + + /// log commands slower than this many microseconds. default: 10000 (10ms). + /// set to -1 to disable, 0 to log every command + #[arg(long, default_value_t = 10_000)] + slowlog_log_slower_than: i64, + + /// maximum number of entries in the slow log ring buffer + #[arg(long, default_value_t = 128)] + slowlog_max_len: usize, } #[tokio::main] @@ -135,9 +145,24 @@ async fn main() { } } + let slowlog_config = slowlog::SlowLogConfig { + slower_than: std::time::Duration::from_micros( + args.slowlog_log_slower_than.max(0) as u64, + ), + max_len: args.slowlog_max_len, + enabled: args.slowlog_log_slower_than >= 0, + }; + info!("ember server starting..."); - if let Err(e) = server::run(addr, shard_count, engine_config, None, args.metrics_port.is_some()).await { + if let Err(e) = server::run( + addr, + shard_count, + engine_config, + None, + args.metrics_port.is_some(), + slowlog_config, + ).await { eprintln!("server error: {e}"); std::process::exit(1); } diff --git a/crates/ember-server/src/server.rs b/crates/ember-server/src/server.rs index edd8a888..d09914f4 100644 --- a/crates/ember-server/src/server.rs +++ b/crates/ember-server/src/server.rs @@ -14,6 +14,7 @@ use tokio::sync::Semaphore; use tracing::{error, info, warn}; use crate::connection; +use crate::slowlog::{SlowLog, SlowLogConfig}; /// Default maximum number of concurrent client connections. const DEFAULT_MAX_CONNECTIONS: usize = 10_000; @@ -49,6 +50,7 @@ pub async fn run( config: EngineConfig, max_connections: Option, metrics_enabled: bool, + slowlog_config: SlowLogConfig, ) -> Result<(), Box> { // ensure data directory exists if persistence is configured if let Some(ref pcfg) = config.persistence { @@ -83,6 +85,8 @@ pub async fn run( commands_processed: AtomicU64::new(0), }); + let slow_log = Arc::new(SlowLog::new(slowlog_config)); + info!( "listening on {addr} with {} shards (max {max_conn} connections)", engine.shard_count() @@ -122,10 +126,11 @@ pub async fn run( let engine = engine.clone(); let ctx = Arc::clone(&ctx); + let slow_log = Arc::clone(&slow_log); let metrics = metrics_enabled; tokio::spawn(async move { - if let Err(e) = connection::handle(stream, engine, &ctx, metrics).await { + if let Err(e) = connection::handle(stream, engine, &ctx, &slow_log, metrics).await { error!("connection error from {peer}: {e}"); } if metrics { diff --git a/crates/ember-server/src/slowlog.rs b/crates/ember-server/src/slowlog.rs new file mode 100644 index 00000000..943bf41f --- /dev/null +++ b/crates/ember-server/src/slowlog.rs @@ -0,0 +1,206 @@ +//! Slow command log. +//! +//! Records commands that exceed a configurable latency threshold into +//! a fixed-size ring buffer. The buffer is protected by a `Mutex`, +//! but contention is negligible since only slow commands (rare by +//! definition) ever acquire it. + +use std::collections::VecDeque; +use std::sync::Mutex; +use std::time::{Duration, SystemTime}; + +/// A single slow log entry. +#[derive(Debug, Clone)] +pub struct SlowLogEntry { + /// Monotonically increasing entry id. + pub id: u64, + /// Wall clock time when the command was executed. + pub timestamp: SystemTime, + /// How long the command took. + pub duration: Duration, + /// Truncated command summary (e.g. "SET key value"). + pub command: String, +} + +/// Configuration for the slow log. +#[derive(Debug, Clone, Copy)] +pub struct SlowLogConfig { + /// Commands slower than this are logged. 0 means log everything. + /// Negative means disabled (matching Redis semantics where -1 disables). + pub slower_than: Duration, + /// Maximum number of entries to keep. Oldest are evicted when full. + pub max_len: usize, + /// Whether the slow log is enabled at all. + pub enabled: bool, +} + +impl Default for SlowLogConfig { + fn default() -> Self { + Self { + slower_than: Duration::from_micros(10_000), // 10ms, matches Redis + max_len: 128, + enabled: true, + } + } +} + +/// Thread-safe slow command log backed by a ring buffer. +pub struct SlowLog { + config: SlowLogConfig, + inner: Mutex, +} + +struct SlowLogInner { + entries: VecDeque, + next_id: u64, +} + +impl SlowLog { + /// Creates a new slow log with the given configuration. + pub fn new(config: SlowLogConfig) -> Self { + Self { + config, + inner: Mutex::new(SlowLogInner { + entries: VecDeque::with_capacity(config.max_len), + next_id: 0, + }), + } + } + + /// Records a command if it exceeded the threshold. + /// + /// Called from the connection handler after each command completes. + /// The mutex is effectively uncontended since slow commands are rare. + pub fn maybe_record(&self, duration: Duration, command: &str) { + if !self.config.enabled || duration < self.config.slower_than { + return; + } + + let mut inner = self.inner.lock().expect("slowlog lock poisoned"); + let id = inner.next_id; + inner.next_id += 1; + + if inner.entries.len() >= self.config.max_len { + inner.entries.pop_front(); + } + + inner.entries.push_back(SlowLogEntry { + id, + timestamp: SystemTime::now(), + duration, + command: command.to_owned(), + }); + } + + /// Returns the most recent entries, newest first. + /// + /// If `count` is `None`, returns all entries. + pub fn get(&self, count: Option) -> Vec { + let inner = self.inner.lock().expect("slowlog lock poisoned"); + let n = count.unwrap_or(inner.entries.len()).min(inner.entries.len()); + inner.entries.iter().rev().take(n).cloned().collect() + } + + /// Returns the number of entries currently in the log. + pub fn len(&self) -> usize { + self.inner.lock().expect("slowlog lock poisoned").entries.len() + } + + /// Clears all entries from the log. + pub fn reset(&self) { + let mut inner = self.inner.lock().expect("slowlog lock poisoned"); + inner.entries.clear(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn records_slow_command() { + let log = SlowLog::new(SlowLogConfig { + slower_than: Duration::from_millis(1), + max_len: 10, + enabled: true, + }); + + // fast command — should not be recorded + log.maybe_record(Duration::from_micros(500), "GET fast"); + assert_eq!(log.len(), 0); + + // slow command — should be recorded + log.maybe_record(Duration::from_millis(5), "SET slow value"); + assert_eq!(log.len(), 1); + + let entries = log.get(None); + assert_eq!(entries.len(), 1); + assert_eq!(entries[0].command, "SET slow value"); + assert_eq!(entries[0].id, 0); + } + + #[test] + fn ring_buffer_evicts_oldest() { + let log = SlowLog::new(SlowLogConfig { + slower_than: Duration::ZERO, + max_len: 3, + enabled: true, + }); + + for i in 0..5 { + log.maybe_record(Duration::from_millis(1), &format!("CMD {i}")); + } + + assert_eq!(log.len(), 3); + let entries = log.get(None); + // newest first + assert_eq!(entries[0].command, "CMD 4"); + assert_eq!(entries[1].command, "CMD 3"); + assert_eq!(entries[2].command, "CMD 2"); + } + + #[test] + fn get_with_count() { + let log = SlowLog::new(SlowLogConfig { + slower_than: Duration::ZERO, + max_len: 10, + enabled: true, + }); + + for i in 0..5 { + log.maybe_record(Duration::from_millis(1), &format!("CMD {i}")); + } + + let entries = log.get(Some(2)); + assert_eq!(entries.len(), 2); + assert_eq!(entries[0].command, "CMD 4"); + assert_eq!(entries[1].command, "CMD 3"); + } + + #[test] + fn reset_clears_entries() { + let log = SlowLog::new(SlowLogConfig { + slower_than: Duration::ZERO, + max_len: 10, + enabled: true, + }); + + log.maybe_record(Duration::from_millis(1), "CMD 1"); + assert_eq!(log.len(), 1); + + log.reset(); + assert_eq!(log.len(), 0); + } + + #[test] + fn disabled_log_records_nothing() { + let log = SlowLog::new(SlowLogConfig { + slower_than: Duration::ZERO, + max_len: 10, + enabled: false, + }); + + log.maybe_record(Duration::from_millis(100), "SLOW CMD"); + assert_eq!(log.len(), 0); + } +}