feat: prometheus metrics endpoint

Cargo.toml

@@ -43,6 +43,10 @@ crc32fast = "1"
 # float ordering (used for sorted set scores)
 ordered-float = "5"
 
+# metrics
+metrics = "0.24"
+metrics-exporter-prometheus = { version = "0.16", features = ["http-listener"] }
+
 # internal crates (version required for crates.io publishing)
 emberkv-core = { version = "0.2.1", path = "crates/ember-core" }
 ember-protocol = { version = "0.2.1", path = "crates/ember-protocol" }

crates/ember-core/src/keyspace.rs

@@ -181,6 +181,10 @@ pub struct KeyspaceStats {
     pub used_bytes: usize,
     /// Number of keys with an expiration set.
     pub keys_with_expiry: usize,
+    /// Cumulative count of keys removed by expiration (lazy + active).
+    pub keys_expired: u64,
+    /// Cumulative count of keys removed by eviction.
+    pub keys_evicted: u64,
 }
 
 /// Number of random keys to sample when looking for an eviction candidate.
@@ -205,6 +209,10 @@ pub struct Keyspace {
     config: ShardConfig,
     /// Number of entries that currently have an expiration set.
     expiry_count: usize,
+    /// Cumulative count of keys removed by expiration (lazy + active).
+    expired_total: u64,
+    /// Cumulative count of keys removed by eviction.
+    evicted_total: u64,
 }
 
 impl Keyspace {
@@ -220,6 +228,8 @@ impl Keyspace {
             memory: MemoryTracker::new(),
             config,
             expiry_count: 0,
+            expired_total: 0,
+            evicted_total: 0,
         }
     }
 
@@ -328,6 +338,7 @@ impl Keyspace {
                 if entry.expires_at.is_some() {
                     self.expiry_count = self.expiry_count.saturating_sub(1);
                 }
+                self.evicted_total += 1;
                 return true;
             }
         }
@@ -535,6 +546,8 @@ impl Keyspace {
             key_count: self.memory.key_count(),
             used_bytes: self.memory.used_bytes(),
             keys_with_expiry: self.expiry_count,
+            keys_expired: self.expired_total,
+            keys_evicted: self.evicted_total,
         }
     }
 
@@ -1565,6 +1578,7 @@ impl Keyspace {
                 if entry.expires_at.is_some() {
                     self.expiry_count = self.expiry_count.saturating_sub(1);
                 }
+                self.expired_total += 1;
             }
         }
         expired

crates/ember-protocol/src/command.rs

@@ -285,6 +285,84 @@ pub struct ZAddFlags {
 impl Eq for ZAddFlags {}
 
 impl Command {
+    /// Returns the lowercase command name as a static string.
+    ///
+    /// Used for metrics labels and slow log entries. Zero allocation —
+    /// returns a `&'static str` for every known variant.
+    pub fn command_name(&self) -> &'static str {
+        match self {
+            Command::Ping(_) => "ping",
+            Command::Echo(_) => "echo",
+            Command::Get { .. } => "get",
+            Command::Set { .. } => "set",
+            Command::Incr { .. } => "incr",
+            Command::Decr { .. } => "decr",
+            Command::Del { .. } => "del",
+            Command::Exists { .. } => "exists",
+            Command::MGet { .. } => "mget",
+            Command::MSet { .. } => "mset",
+            Command::Expire { .. } => "expire",
+            Command::Ttl { .. } => "ttl",
+            Command::Persist { .. } => "persist",
+            Command::Pttl { .. } => "pttl",
+            Command::Pexpire { .. } => "pexpire",
+            Command::DbSize => "dbsize",
+            Command::Info { .. } => "info",
+            Command::BgSave => "bgsave",
+            Command::BgRewriteAof => "bgrewriteaof",
+            Command::FlushDb => "flushdb",
+            Command::Scan { .. } => "scan",
+            Command::LPush { .. } => "lpush",
+            Command::RPush { .. } => "rpush",
+            Command::LPop { .. } => "lpop",
+            Command::RPop { .. } => "rpop",
+            Command::LRange { .. } => "lrange",
+            Command::LLen { .. } => "llen",
+            Command::Type { .. } => "type",
+            Command::ZAdd { .. } => "zadd",
+            Command::ZRem { .. } => "zrem",
+            Command::ZScore { .. } => "zscore",
+            Command::ZRank { .. } => "zrank",
+            Command::ZCard { .. } => "zcard",
+            Command::ZRange { .. } => "zrange",
+            Command::HSet { .. } => "hset",
+            Command::HGet { .. } => "hget",
+            Command::HGetAll { .. } => "hgetall",
+            Command::HDel { .. } => "hdel",
+            Command::HExists { .. } => "hexists",
+            Command::HLen { .. } => "hlen",
+            Command::HIncrBy { .. } => "hincrby",
+            Command::HKeys { .. } => "hkeys",
+            Command::HVals { .. } => "hvals",
+            Command::HMGet { .. } => "hmget",
+            Command::SAdd { .. } => "sadd",
+            Command::SRem { .. } => "srem",
+            Command::SMembers { .. } => "smembers",
+            Command::SIsMember { .. } => "sismember",
+            Command::SCard { .. } => "scard",
+            Command::ClusterInfo => "cluster_info",
+            Command::ClusterNodes => "cluster_nodes",
+            Command::ClusterSlots => "cluster_slots",
+            Command::ClusterKeySlot { .. } => "cluster_keyslot",
+            Command::ClusterMyId => "cluster_myid",
+            Command::ClusterSetSlotImporting { .. } => "cluster_setslot",
+            Command::ClusterSetSlotMigrating { .. } => "cluster_setslot",
+            Command::ClusterSetSlotNode { .. } => "cluster_setslot",
+            Command::ClusterSetSlotStable { .. } => "cluster_setslot",
+            Command::ClusterMeet { .. } => "cluster_meet",
+            Command::ClusterAddSlots { .. } => "cluster_addslots",
+            Command::ClusterDelSlots { .. } => "cluster_delslots",
+            Command::ClusterForget { .. } => "cluster_forget",
+            Command::ClusterReplicate { .. } => "cluster_replicate",
+            Command::ClusterFailover { .. } => "cluster_failover",
+            Command::ClusterCountKeysInSlot { .. } => "cluster_countkeysinslot",
+            Command::ClusterGetKeysInSlot { .. } => "cluster_getkeysinslot",
+            Command::Migrate { .. } => "migrate",
+            Command::Asking => "asking",
+            Command::Unknown(_) => "unknown",
+        }
+    }
+
     /// Parses a [`Frame`] into a [`Command`].
     ///
     /// Expects an array frame where the first element is the command name

crates/ember-server/Cargo.toml

@@ -19,3 +19,5 @@ tokio = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
 clap = { workspace = true }
+metrics = { workspace = true }
+metrics-exporter-prometheus = { workspace = true }

crates/ember-server/src/connection.rs

@@ -4,7 +4,7 @@
 //! sharded engine, and writes responses back. Supports pipelining
 //! by processing multiple frames from a single read.
 
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use bytes::{Bytes, BytesMut};
 use ember_core::{Engine, KeyspaceStats, ShardRequest, ShardResponse, TtlResult, Value};
@@ -33,6 +33,7 @@ const IDLE_TIMEOUT: Duration = Duration::from_secs(300); // 5 minutes
 pub async fn handle(
     mut stream: TcpStream,
     engine: Engine,
+    metrics_enabled: bool,
 ) -> Result<(), Box<dyn std::error::Error>> {
     // disable Nagle's algorithm — cache servers need low-latency writes,
     // and we already batch responses from pipelining into a single write
@@ -67,7 +68,7 @@ pub async fn handle(
             match parse_frame(&buf) {
                 Ok(Some((frame, consumed))) => {
                     let _ = buf.split_to(consumed);
-                    let response = process(frame, &engine).await;
+                    let response = process(frame, &engine, metrics_enabled).await;
                     response.serialize(&mut out);
                 }
                 Ok(None) => break, // need more data
@@ -87,9 +88,28 @@ pub async fn handle(
 }
 
 /// Converts a raw frame into a command and executes it.
-async fn process(frame: Frame, engine: &Engine) -> Frame {
+///
+/// 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 {
     match Command::from_frame(frame) {
-        Ok(cmd) => execute(cmd, engine).await,
+        Ok(cmd) => {
+            let cmd_name = cmd.command_name();
+            let start = if metrics_enabled {
+                Some(Instant::now())
+            } else {
+                None
+            };
+
+            let response = execute(cmd, engine).await;
+
+            if let Some(start) = start {
+                let is_error = matches!(&response, Frame::Error(_));
+                crate::metrics::record_command(cmd_name, start.elapsed(), is_error);
+            }
+
+            response
+        }
         Err(e) => Frame::Error(format!("ERR {e}")),
     }
 }
@@ -318,12 +338,16 @@ async fn execute(cmd: Command, engine: &Engine) -> Frame {
                             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!(

crates/ember-server/src/main.rs

@@ -1,5 +1,6 @@
 mod config;
 mod connection;
+mod metrics;
 mod server;
 
 use std::net::SocketAddr;
@@ -43,6 +44,10 @@ struct Args {
     /// fsync policy for the AOF: always, everysec, or no
     #[arg(long, default_value = "everysec")]
     appendfsync: String,
+
+    /// port for prometheus metrics HTTP endpoint. disabled when not set
+    #[arg(long)]
+    metrics_port: Option<u16>,
 }
 
 #[tokio::main]
@@ -119,9 +124,20 @@ async fn main() {
         );
     }
 
+    // install prometheus metrics exporter if --metrics-port is set
+    if let Some(metrics_port) = args.metrics_port {
+        let metrics_addr: std::net::SocketAddr = format!("{}:{}", args.host, metrics_port)
+            .parse()
+            .expect("invalid metrics bind address");
+        if let Err(e) = metrics::install_exporter(metrics_addr) {
+            eprintln!("failed to start metrics exporter: {e}");
+            std::process::exit(1);
+        }
+    }
+
     info!("ember server starting...");
 
-    if let Err(e) = server::run(addr, shard_count, engine_config, None).await {
+    if let Err(e) = server::run(addr, shard_count, engine_config, None, args.metrics_port.is_some()).await {
         eprintln!("server error: {e}");
         std::process::exit(1);
     }

crates/ember-server/src/metrics.rs

@@ -0,0 +1,127 @@
+//! Prometheus metrics exposition.
+//!
+//! When `--metrics-port` is set, installs a prometheus exporter that
+//! serves `/metrics` on a separate HTTP port. The hot path records
+//! per-command counters and latency histograms through the `metrics`
+//! crate's global recorder. A background stats poller periodically
+//! broadcasts `ShardRequest::Stats` and updates gauge values.
+
+use std::net::SocketAddr;
+use std::time::Duration;
+
+use ember_core::{Engine, KeyspaceStats, ShardRequest, ShardResponse};
+use metrics::{counter, gauge, histogram};
+use metrics_exporter_prometheus::PrometheusBuilder;
+use tracing::{info, warn};
+
+/// Histogram buckets tuned for cache latency (10µs to 100ms).
+const HISTOGRAM_BUCKETS: &[f64] = &[
+    0.000_01,  // 10µs
+    0.000_025, // 25µs
+    0.000_05,  // 50µs
+    0.000_1,   // 100µs
+    0.000_25,  // 250µs
+    0.000_5,   // 500µs
+    0.001,     // 1ms
+    0.002_5,   // 2.5ms
+    0.005,     // 5ms
+    0.01,      // 10ms
+    0.025,     // 25ms
+    0.05,      // 50ms
+    0.1,       // 100ms
+];
+
+/// How often the stats poller queries shards and updates gauges.
+const STATS_POLL_INTERVAL: Duration = Duration::from_secs(5);
+
+/// Installs the prometheus exporter and starts the HTTP listener.
+///
+/// Returns an error if the listener can't bind to the address.
+pub fn install_exporter(addr: SocketAddr) -> Result<(), Box<dyn std::error::Error>> {
+    PrometheusBuilder::new()
+        .with_http_listener(addr)
+        .set_buckets(HISTOGRAM_BUCKETS)
+        .map_err(|e| format!("failed to set histogram buckets: {e}"))?
+        .install()
+        .map_err(|e| format!("failed to install prometheus exporter: {e}"))?;
+
+    info!("prometheus metrics available on http://{addr}/metrics");
+    Ok(())
+}
+
+/// Spawns a background task that polls shard stats every 5 seconds
+/// and publishes them as prometheus gauges.
+///
+/// Keeps `ember-core` free of metrics dependencies — the poller
+/// pulls stats through the existing `ShardRequest::Stats` broadcast.
+pub fn spawn_stats_poller(engine: Engine) {
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(STATS_POLL_INTERVAL);
+        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+
+        loop {
+            interval.tick().await;
+
+            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;
+                        }
+                    }
+
+                    gauge!("ember_keys_total").set(total.key_count as f64);
+                    gauge!("ember_memory_used_bytes").set(total.used_bytes as f64);
+                    gauge!("ember_keys_expired_total").set(total.keys_expired as f64);
+                    gauge!("ember_keys_evicted_total").set(total.keys_evicted as f64);
+                }
+                Err(e) => {
+                    warn!("stats poller broadcast failed: {e}");
+                }
+            }
+        }
+    });
+}
+
+/// Records a command execution in prometheus metrics.
+///
+/// Called from the connection handler after each command completes.
+#[inline]
+pub fn record_command(cmd_name: &'static str, duration: Duration, is_error: bool) {
+    let labels = [("cmd", cmd_name)];
+    counter!("ember_commands_total", &labels).increment(1);
+    histogram!("ember_commands_duration_seconds", &labels).record(duration.as_secs_f64());
+    if is_error {
+        counter!("ember_commands_errors_total", &labels).increment(1);
+    }
+}
+
+/// Increments the active connection gauge and total counter.
+#[inline]
+pub fn on_connection_accepted() {
+    gauge!("ember_connections_active").increment(1.0);
+    counter!("ember_connections_total").increment(1);
+}
+
+/// Decrements the active connection gauge.
+#[inline]
+pub fn on_connection_closed() {
+    gauge!("ember_connections_active").decrement(1.0);
+}
+
+/// Records a rejected connection (limit reached).
+#[inline]
+pub fn on_connection_rejected() {
+    counter!("ember_connections_rejected").increment(1);
+}