fix: add metrics instrumentation to all modules

Universal auto-instrumentation via global singleton pattern:
- tiered-sink: spool messages/bytes, circuit state, hot/cold path,
  disk available, circuit trips
- spool: queue depth gauge on push/pop/recv/clear
- dlq: entries total, entries written, write errors
- cache: hits/misses by source, entry count
- http-client: request count by method/status, duration histogram
- secrets: fetch total, cache hits/misses

All metrics #[cfg(feature = "metrics")] gated — zero cost when off.
Apps with MetricsManager get full observability across every rustlib
feature with zero additional wiring.

Document core pillar design decision in CLAUDE.md: config, logging,
and metrics are non-negotiable for every module.

Author: catinspace-au

src/cache/mod.rs

@@ -91,7 +91,17 @@ impl Cache {
     /// Returns `None` if not found or expired.
     pub async fn get<T: serde::de::DeserializeOwned>(&self, source: &str, key: &str) -> Option<T> {
         let full_key = format!("{source}:{key}");
-        let bytes = self.inner.get(&full_key).await?;
+        let bytes = self.inner.get(&full_key).await;
+
+        #[cfg(feature = "metrics")]
+        if bytes.is_some() {
+            metrics::counter!("dfe_cache_hits_total", "source" => source.to_string()).increment(1);
+        } else {
+            metrics::counter!("dfe_cache_misses_total", "source" => source.to_string())
+                .increment(1);
+        }
+
+        let bytes = bytes?;
         serde_json::from_slice(&bytes).ok()
     }
 
@@ -112,6 +122,9 @@ impl Cache {
 
         self.inner.insert(full_key.clone(), bytes).await;
 
+        #[cfg(feature = "metrics")]
+        metrics::gauge!("dfe_cache_entries").set(self.inner.entry_count() as f64);
+
         // Track key for source-level invalidation
         if let Ok(mut keys) = self.source_keys.lock() {
             keys.entry(source.to_string()).or_default().push(full_key);
@@ -130,6 +143,9 @@ impl Cache {
         for key in keys {
             self.inner.invalidate(&key).await;
         }
+
+        #[cfg(feature = "metrics")]
+        metrics::gauge!("dfe_cache_entries").set(self.inner.entry_count() as f64);
     }
 
     /// Invalidate a single entry.

src/dlq/file.rs

@@ -94,7 +94,17 @@ impl DlqBackend for FileDlq {
             .map_err(|e| DlqError::Serialization(format!("failed to serialise DLQ entry: {e}")))?;
         line.push(b'\n');
 
-        self.writer.write_line(&line)?;
+        if let Err(e) = self.writer.write_line(&line) {
+            #[cfg(feature = "metrics")]
+            metrics::counter!("dfe_dlq_write_errors_total").increment(1);
+            return Err(e.into());
+        }
+
+        #[cfg(feature = "metrics")]
+        {
+            metrics::counter!("dfe_dlq_entries_total").increment(1);
+            metrics::gauge!("dfe_dlq_entries_written").set(self.writer.lines_written() as f64);
+        }
 
         debug!(
             service = %self.service_name,

src/http_client/mod.rs

@@ -81,7 +81,20 @@ impl HttpClient {
 
     /// Send a GET request.
     pub async fn get(&self, url: &str) -> Result<Response, reqwest_middleware::Error> {
-        self.inner.get(url).send().await
+        #[cfg(feature = "metrics")]
+        let start = std::time::Instant::now();
+
+        let result = self.inner.get(url).send().await;
+
+        #[cfg(feature = "metrics")]
+        {
+            let status = if result.is_ok() { "success" } else { "error" };
+            metrics::counter!("dfe_http_client_requests_total", "method" => "GET", "status" => status).increment(1);
+            metrics::histogram!("dfe_http_client_duration_seconds", "method" => "GET")
+                .record(start.elapsed().as_secs_f64());
+        }
+
+        result
     }
 
     /// Send a POST request with a JSON body.
@@ -90,12 +103,26 @@ impl HttpClient {
         url: &str,
         body: &T,
     ) -> Result<Response, reqwest_middleware::Error> {
-        self.inner
+        #[cfg(feature = "metrics")]
+        let start = std::time::Instant::now();
+
+        let result = self
+            .inner
             .post(url)
             .header("content-type", "application/json")
             .body(serde_json::to_vec(body).unwrap_or_default())
             .send()
-            .await
+            .await;
+
+        #[cfg(feature = "metrics")]
+        {
+            let status = if result.is_ok() { "success" } else { "error" };
+            metrics::counter!("dfe_http_client_requests_total", "method" => "POST", "status" => status).increment(1);
+            metrics::histogram!("dfe_http_client_duration_seconds", "method" => "POST")
+                .record(start.elapsed().as_secs_f64());
+        }
+
+        result
     }
 
     /// Send a PUT request with a JSON body.
@@ -104,17 +131,44 @@ impl HttpClient {
         url: &str,
         body: &T,
     ) -> Result<Response, reqwest_middleware::Error> {
-        self.inner
+        #[cfg(feature = "metrics")]
+        let start = std::time::Instant::now();
+
+        let result = self
+            .inner
             .put(url)
             .header("content-type", "application/json")
             .body(serde_json::to_vec(body).unwrap_or_default())
             .send()
-            .await
+            .await;
+
+        #[cfg(feature = "metrics")]
+        {
+            let status = if result.is_ok() { "success" } else { "error" };
+            metrics::counter!("dfe_http_client_requests_total", "method" => "PUT", "status" => status).increment(1);
+            metrics::histogram!("dfe_http_client_duration_seconds", "method" => "PUT")
+                .record(start.elapsed().as_secs_f64());
+        }
+
+        result
     }
 
     /// Send a DELETE request.
     pub async fn delete(&self, url: &str) -> Result<Response, reqwest_middleware::Error> {
-        self.inner.delete(url).send().await
+        #[cfg(feature = "metrics")]
+        let start = std::time::Instant::now();
+
+        let result = self.inner.delete(url).send().await;
+
+        #[cfg(feature = "metrics")]
+        {
+            let status = if result.is_ok() { "success" } else { "error" };
+            metrics::counter!("dfe_http_client_requests_total", "method" => "DELETE", "status" => status).increment(1);
+            metrics::histogram!("dfe_http_client_duration_seconds", "method" => "DELETE")
+                .record(start.elapsed().as_secs_f64());
+        }
+
+        result
     }
 
     /// Access the underlying middleware client for custom requests.

src/secrets/mod.rs

@@ -158,12 +158,20 @@ impl SecretsManager {
         // Check cache first
         if let Some(cached) = self.cache.read().get(&cache_key) {
             debug!(path = %path, "Secret loaded from cache");
+            #[cfg(feature = "metrics")]
+            metrics::counter!("dfe_secrets_cache_hits_total").increment(1);
             return Ok(cached);
         }
 
+        #[cfg(feature = "metrics")]
+        metrics::counter!("dfe_secrets_cache_misses_total").increment(1);
+
         // Fetch from file
         let value = self.file_provider.get(path).await?;
 
+        #[cfg(feature = "metrics")]
+        metrics::counter!("dfe_secrets_fetch_total").increment(1);
+
         // Update cache
         if let Err(e) = self.cache.write().set(&cache_key, &value) {
             warn!(error = %e, "Failed to cache secret");
@@ -181,9 +189,14 @@ impl SecretsManager {
         // Check cache first
         if let Some(cached) = self.cache.read().get(cache_key) {
             debug!(key = %cache_key, "Secret loaded from cache");
+            #[cfg(feature = "metrics")]
+            metrics::counter!("dfe_secrets_cache_hits_total").increment(1);
             return Ok(cached);
         }
 
+        #[cfg(feature = "metrics")]
+        metrics::counter!("dfe_secrets_cache_misses_total").increment(1);
+
         // Fetch from provider
         let result = match source {
             SecretSource::File { path } => self.file_provider.get(path).await,
@@ -221,6 +234,9 @@ impl SecretsManager {
             }
         };
 
+        #[cfg(feature = "metrics")]
+        metrics::counter!("dfe_secrets_fetch_total").increment(1);
+
         match result {
             Ok(value) => {
                 // Update cache

src/spool/queue.rs

@@ -109,6 +109,8 @@ impl Spool {
             .map_err(|e| SpoolError::Queue(e.to_string()))?;
 
         self.len += 1;
+        #[cfg(feature = "metrics")]
+        ::metrics::gauge!("dfe_spool_queue_depth").set(self.len as f64);
         Ok(())
     }
 
@@ -181,6 +183,8 @@ impl Spool {
                     .commit()
                     .map_err(|e| SpoolError::Queue(e.to_string()))?;
                 self.len = self.len.saturating_sub(1);
+                #[cfg(feature = "metrics")]
+                ::metrics::gauge!("dfe_spool_queue_depth").set(self.len as f64);
                 Ok(Some(data))
             }
             Err(yaque::TryRecvError::Io(e)) => Err(SpoolError::Io(e)),
@@ -215,6 +219,8 @@ impl Spool {
             .commit()
             .map_err(|e| SpoolError::Queue(e.to_string()))?;
         self.len = self.len.saturating_sub(1);
+        #[cfg(feature = "metrics")]
+        ::metrics::gauge!("dfe_spool_queue_depth").set(self.len as f64);
         Ok(data)
     }
 
@@ -255,6 +261,8 @@ impl Spool {
             }
         }
         self.len = 0;
+        #[cfg(feature = "metrics")]
+        ::metrics::gauge!("dfe_spool_queue_depth").set(0.0);
         Ok(())
     }
 

src/tiered_sink/tiered.rs

@@ -146,6 +146,8 @@ impl<S: Sink> TieredSink<S> {
             match self.try_hot_path(data).await {
                 Ok(()) => {
                     self.hot_path_count.fetch_add(1, AtomicOrdering::Relaxed);
+                    #[cfg(feature = "metrics")]
+                    ::metrics::counter!("dfe_spool_hot_path_total").increment(1);
                     return Ok(());
                 }
                 Err(TieredSinkError::Sink(_)) => {
@@ -161,13 +163,22 @@ impl<S: Sink> TieredSink<S> {
         // Cold path: spool to disk
         self.spool_message(data).await?;
         self.cold_path_count.fetch_add(1, AtomicOrdering::Relaxed);
+        #[cfg(feature = "metrics")]
+        ::metrics::counter!("dfe_spool_cold_path_total").increment(1);
         Ok(())
     }
 
     /// Determine if we should attempt the hot path.
     async fn should_use_hot_path(&self) -> bool {
         let circuit_state = self.circuit.state().await;
 
+        #[cfg(feature = "metrics")]
+        ::metrics::gauge!("dfe_spool_circuit_state").set(match circuit_state {
+            CircuitState::Closed => 0.0,
+            CircuitState::HalfOpen => 1.0,
+            CircuitState::Open => 2.0,
+        });
+
         match circuit_state {
             CircuitState::Open => false,
             CircuitState::Closed | CircuitState::HalfOpen => match self.config.ordering {
@@ -195,6 +206,8 @@ impl<S: Sink> TieredSink<S> {
             }
             Ok(Err(SinkError::Unavailable)) => {
                 self.circuit.record_failure().await;
+                #[cfg(feature = "metrics")]
+                ::metrics::counter!("dfe_spool_circuit_trips_total").increment(1);
                 Err(TieredSinkError::Spool("sink unavailable".into()))
             }
             Ok(Err(SinkError::Fatal(e))) => {
@@ -203,6 +216,8 @@ impl<S: Sink> TieredSink<S> {
             }
             Err(_timeout) => {
                 self.circuit.record_failure().await;
+                #[cfg(feature = "metrics")]
+                ::metrics::counter!("dfe_spool_circuit_trips_total").increment(1);
                 Err(TieredSinkError::Spool("send timeout".into()))
             }
         }
@@ -250,6 +265,14 @@ impl<S: Sink> TieredSink<S> {
         self.spool_bytes
             .fetch_add(compressed_len, AtomicOrdering::Relaxed);
 
+        #[cfg(feature = "metrics")]
+        {
+            ::metrics::gauge!("dfe_spool_messages")
+                .set(self.spool_count.load(AtomicOrdering::Relaxed) as f64);
+            ::metrics::gauge!("dfe_spool_bytes")
+                .set(self.spool_bytes.load(AtomicOrdering::Relaxed) as f64);
+        }
+
         #[cfg(feature = "logger")]
         tracing::debug!(
             spool_items = self.spool_count.load(AtomicOrdering::Relaxed),
@@ -453,7 +476,15 @@ async fn disk_capacity_poller(
             () = tokio::time::sleep(poll_interval) => {}
         }
 
-        let available = check_disk_space(&spool_path).is_none_or(|(total, avail)| {
+        let disk_space = check_disk_space(&spool_path);
+
+        #[cfg(feature = "metrics")]
+        if let Some((total, avail)) = disk_space {
+            ::metrics::gauge!("dfe_spool_disk_available_bytes").set(avail as f64);
+            ::metrics::gauge!("dfe_spool_disk_total_bytes").set(total as f64);
+        }
+
+        let available = disk_space.is_none_or(|(total, avail)| {
             if total == 0 {
                 return true;
             }