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 66ec4c16..276fe8d0 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,9 @@ use ember_protocol::{parse_frame, Command, Frame, SetExpire}; 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. const BUF_CAPACITY: usize = 4096; @@ -33,6 +38,8 @@ const IDLE_TIMEOUT: Duration = Duration::from_secs(300); // 5 minutes 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, @@ -68,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, metrics_enabled).await; + let response = + process(frame, &engine, ctx, slow_log, metrics_enabled).await; response.serialize(&mut out); } Ok(None) => break, // need more data @@ -91,21 +99,27 @@ 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, + 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).await; + let response = execute(cmd, engine, ctx, slow_log).await; + let elapsed = start.elapsed(); - if let Some(start) = start { + ctx.commands_processed.fetch_add(1, Ordering::Relaxed); + slow_log.maybe_record(elapsed, cmd_name); + + 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 @@ -119,7 +133,12 @@ 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, + slow_log: &Arc, +) -> Frame { match cmd { // -- no shard needed -- Command::Ping(None) => Frame::Simple("PONG".into()), @@ -330,36 +349,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 { @@ -915,6 +905,35 @@ async fn execute(cmd: Command, engine: &Engine) -> Frame { 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}'")), } } @@ -940,6 +959,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/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 34f5d61e..d09914f4 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; @@ -12,10 +14,27 @@ 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; +/// 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 @@ -31,12 +50,20 @@ 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 { 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 +74,19 @@ 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), + }); + + let slow_log = Arc::new(SlowLog::new(slowlog_config)); + info!( "listening on {addr} with {} shards (max {max_conn} connections)", engine.shard_count() @@ -82,12 +122,15 @@ 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 slow_log = Arc::clone(&slow_log); 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, &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); + } +}