textfile-metrics 0.1.0

// Copyright (c) Ted Kaplan. All Rights Reserved.
// SPDX-License-Identifier: MIT

//! Non-blocking metrics writer with snapshot pattern.
//!
//! This module implements the core pattern: take snapshot under lock, drop
//! lock, then write to file. This ensures minimal lock contention and true
//! non-blocking behavior.

use std::{
    collections::BTreeMap,
    fs,
    path::{Path, PathBuf},
    sync::Arc,
};

use dashmap::DashMap;
use parking_lot::RwLock;
use tracing::{debug, warn};

use crate::{
    errors::{MetricsError, Result},
    labels::Labels,
    metric::{MetricType, PrometheusMetric},
};

/// Key for storing metrics in the internal map.
/// Format: `metric_name::{labels_hash}` to distinguish same metric with
/// different labels.
type MetricKey = String;

/// Non-blocking metrics writer for Prometheus textfile format.
///
/// # Thread Safety
///
/// This writer uses `DashMap` for lock-free metric updates and `RwLock` for
/// type tracking. The flush operation follows the non-blocking pattern:
/// snapshot under lock, drop lock, then write to disk.
///
/// # Example
///
/// ```ignore
/// use textfile_metrics::MetricsWriter;
///
/// #[tokio::main]
/// async fn main() -> anyhow::Result<()> {
///     let metrics = MetricsWriter::new()?;
///
///     // Update metrics (non-blocking)
///     metrics.counter("requests_total", vec![], 1.0)?;
///     metrics.gauge("temperature", vec![("location".to_string(), "office".to_string())], 21.5)?;
///
///     // Flush to disk (blocking I/O, but metrics updates remain lock-free)
///     metrics.flush().await?;
///
///     Ok(())
/// }
/// ```
pub struct MetricsWriter {
    /// Metrics storage using DashMap for lock-free reads/writes.
    metrics: Arc<DashMap<MetricKey, PrometheusMetric>>,

    /// Metric type registry using RwLock for type information.
    metric_types: Arc<RwLock<BTreeMap<String, MetricType>>>,

    /// Output directory path.
    output_path: Arc<PathBuf>,

    /// Output filename (default: "metrics.prom").
    filename: Arc<String>,
}

impl Clone for MetricsWriter {
    fn clone(&self) -> Self {
        Self {
            metrics: Arc::clone(&self.metrics),
            metric_types: Arc::clone(&self.metric_types),
            output_path: Arc::clone(&self.output_path),
            filename: Arc::clone(&self.filename),
        }
    }
}

impl MetricsWriter {
    /// Create a new metrics writer.
    ///
    /// Reads path from `METRICS_TEXTFILE_PATH` environment variable.
    /// Defaults to `/var/lib/node_exporter/textfile_collector/` if not set
    /// (the standard Prometheus node_exporter textfile collector path).
    ///
    /// # Errors
    ///
    /// Returns error if the output directory cannot be created or accessed.
    pub fn new() -> Result<Self> {
        Self::with_path(Self::config_path()?)
    }

    /// Create a metrics writer with a custom path.
    ///
    /// # Errors
    ///
    /// Returns error if the directory cannot be created.
    pub fn with_path(path: impl AsRef<Path>) -> Result<Self> {
        Self::with_path_and_filename(path, "metrics.prom")
    }

    /// Create a metrics writer with a custom path and filename.
    ///
    /// # Errors
    ///
    /// Returns error if the directory cannot be created.
    pub fn with_path_and_filename(
        path: impl AsRef<Path>,
        filename: impl Into<String>,
    ) -> Result<Self> {
        let path = path.as_ref();

        // Create directory if it doesn't exist
        fs::create_dir_all(path).map_err(|e| {
            MetricsError::PathError(format!("Failed to create metrics directory: {}", e))
        })?;

        Ok(Self {
            metrics: Arc::new(DashMap::new()),
            metric_types: Arc::new(RwLock::new(BTreeMap::new())),
            output_path: Arc::new(path.to_path_buf()),
            filename: Arc::new(filename.into()),
        })
    }

    /// Get the configured output path.
    pub fn output_path(&self) -> &Path {
        &self.output_path
    }

    /// Update or create a counter metric.
    ///
    /// Counters are monotonically increasing. This method increments by the
    /// given delta.
    ///
    /// # Arguments
    ///
    /// * `name` - Metric name (e.g., "requests_total")
    /// * `labels` - Label pairs for the metric
    /// * `delta` - Amount to increment by
    ///
    /// # Errors
    ///
    /// Returns error if the value is invalid (NaN, Inf) or metric name is
    /// invalid.
    pub fn counter(
        &self,
        name: impl Into<String>,
        labels: impl Into<Labels>,
        delta: f64,
    ) -> Result<()> {
        self.update_metric(name, labels, MetricType::Counter, delta, false)
    }

    /// Update or create a gauge metric.
    ///
    /// Gauges can be set to arbitrary values. This method sets the value
    /// directly.
    ///
    /// # Arguments
    ///
    /// * `name` - Metric name (e.g., "temperature_celsius")
    /// * `labels` - Label pairs for the metric
    /// * `value` - Value to set
    ///
    /// # Errors
    ///
    /// Returns error if the value is invalid or metric name is invalid.
    pub fn gauge(
        &self,
        name: impl Into<String>,
        labels: impl Into<Labels>,
        value: f64,
    ) -> Result<()> {
        self.update_metric(name, labels, MetricType::Gauge, value, true)
    }

    /// Increment a counter metric.
    ///
    /// This is a convenience method that increments by 1.
    pub fn inc(&self, name: impl Into<String>, labels: impl Into<Labels>) -> Result<()> {
        self.counter(name, labels, 1.0)
    }

    /// Set a gauge value.
    ///
    /// This is a convenience method for setting gauge values.
    pub fn set(
        &self,
        name: impl Into<String>,
        labels: impl Into<Labels>,
        value: f64,
    ) -> Result<()> {
        self.gauge(name, labels, value)
    }

    /// Flush all metrics to disk in Prometheus textfile format.
    ///
    /// This operation:
    /// 1. Snapshots all metrics under lock
    /// 2. Releases the lock
    /// 3. Writes the snapshot to disk
    ///
    /// This ensures minimal lock contention.
    ///
    /// # Errors
    ///
    /// Returns error if file write fails.
    pub async fn flush(&self) -> Result<()> {
        // SNAPSHOT PHASE: Take snapshot under minimal lock
        let snapshot = {
            debug!("Taking metrics snapshot");
            let mut metrics_by_name: BTreeMap<String, Vec<PrometheusMetric>> = BTreeMap::new();

            for entry in self.metrics.iter() {
                let metric = entry.value().clone();
                metrics_by_name
                    .entry(metric.name.clone())
                    .or_insert_with(Vec::new)
                    .push(metric);
            }

            metrics_by_name
        };

        // WRITE PHASE: Write to disk without holding lock
        self.write_snapshot(snapshot).await
    }

    /// Get current metric count.
    pub fn len(&self) -> usize {
        self.metrics.len()
    }

    /// Check if there are any metrics.
    pub fn is_empty(&self) -> bool {
        self.metrics.is_empty()
    }

    /// Clear all metrics (for testing).
    #[cfg(test)]
    pub fn clear(&self) {
        self.metrics.clear();
        self.metric_types.write().clear();
    }

    /// Internal: Update or create a metric.
    fn update_metric(
        &self,
        name: impl Into<String>,
        labels: impl Into<Labels>,
        metric_type: MetricType,
        value: f64,
        is_set: bool,
    ) -> Result<()> {
        let name_str = name.into();
        let labels = labels.into();

        // Validate
        Self::validate_name(&name_str)?;
        Self::validate_value(value)?;

        // Build key
        let key = Self::make_key(&name_str, &labels);

        // Register type
        {
            let mut types = self.metric_types.write();
            types.entry(name_str.clone()).or_insert(metric_type);
        }

        // Update metric
        self.metrics
            .entry(key)
            .and_modify(|m| {
                if is_set {
                    m.value = value;
                } else {
                    m.value += value;
                }
            })
            .or_insert_with(|| {
                PrometheusMetric::new(&name_str, metric_type, labels.clone(), value)
            });

        Ok(())
    }

    /// Write snapshot to disk.
    async fn write_snapshot(
        &self,
        snapshot: BTreeMap<String, Vec<PrometheusMetric>>,
    ) -> Result<()> {
        let output_path = self.output_path.as_ref();

        // Build the output content
        let mut content = String::new();
        let types = self.metric_types.read().clone();

        for (metric_name, metric_type) in types {
            // Write TYPE declaration
            if let Some(metrics) = snapshot.get(&metric_name) {
                if !metrics.is_empty() {
                    content.push_str(&format!("# TYPE {} {}\n", metric_name, metric_type));

                    // Write all instances of this metric
                    let mut sorted_metrics = metrics.clone();
                    sorted_metrics.sort();

                    for metric in sorted_metrics {
                        if metric.is_valid() {
                            content.push_str(&metric.to_prometheus_line());
                            content.push('\n');
                        } else {
                            warn!("Skipping invalid metric: {}", metric.name);
                        }
                    }
                }
            }
        }

        // Write to file
        let file_path = output_path.join(self.filename.as_str());
        fs::write(&file_path, content).map_err(MetricsError::IoError)?;

        debug!("Wrote metrics to {}", file_path.display());

        Ok(())
    }

    /// Get configuration path from environment or default.
    fn config_path() -> Result<PathBuf> {
        let path = std::env::var("METRICS_TEXTFILE_PATH")
            .unwrap_or_else(|_| "/var/lib/node_exporter/textfile_collector/".to_string());

        Ok(PathBuf::from(path))
    }

    /// Validate metric name (must not be empty, contain valid characters).
    fn validate_name(name: &str) -> Result<()> {
        if name.is_empty() {
            return Err(MetricsError::InvalidName(
                "Metric name cannot be empty".to_string(),
            ));
        }

        // Check for valid Prometheus metric name: [a-zA-Z_:][a-zA-Z0-9_:]*
        if !name.chars().next().unwrap().is_alphabetic() && name.chars().next() != Some('_') {
            return Err(MetricsError::InvalidName(format!(
                "Metric name must start with letter or underscore: {}",
                name
            )));
        }

        for ch in name.chars() {
            if !ch.is_alphanumeric() && ch != '_' && ch != ':' {
                return Err(MetricsError::InvalidName(format!(
                    "Invalid character in metric name '{}': {}",
                    name, ch
                )));
            }
        }

        Ok(())
    }

    /// Validate metric value (must be finite).
    fn validate_value(value: f64) -> Result<()> {
        if !value.is_finite() {
            return Err(MetricsError::InvalidValue(format!(
                "Metric value must be finite, got: {}",
                value
            )));
        }
        Ok(())
    }

    /// Create a unique key for a metric+labels combination.
    fn make_key(name: &str, labels: &Labels) -> MetricKey {
        // Use a simple hash of the labels string to distinguish same metric with
        // different labels
        let labels_str = labels.to_string();
        if labels_str.is_empty() {
            name.to_string()
        } else {
            format!("{}:{}", name, labels_str)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_validate_name_valid() {
        assert!(MetricsWriter::validate_name("requests_total").is_ok());
        assert!(MetricsWriter::validate_name("_internal_metric").is_ok());
        assert!(MetricsWriter::validate_name("metric123").is_ok());
        assert!(MetricsWriter::validate_name("metric:subsystem").is_ok());
    }

    #[test]
    fn test_validate_name_invalid() {
        assert!(MetricsWriter::validate_name("").is_err());
        assert!(MetricsWriter::validate_name("123invalid").is_err());
        assert!(MetricsWriter::validate_name("invalid-name").is_err());
    }

    #[test]
    fn test_validate_value_valid() {
        assert!(MetricsWriter::validate_value(0.0).is_ok());
        assert!(MetricsWriter::validate_value(42.5).is_ok());
        assert!(MetricsWriter::validate_value(-100.0).is_ok());
    }

    #[test]
    fn test_validate_value_invalid() {
        assert!(MetricsWriter::validate_value(f64::NAN).is_err());
        assert!(MetricsWriter::validate_value(f64::INFINITY).is_err());
        assert!(MetricsWriter::validate_value(f64::NEG_INFINITY).is_err());
    }

    #[tokio::test]
    async fn test_writer_creation() {
        let writer = MetricsWriter::new();
        assert!(writer.is_ok());
    }

    #[tokio::test]
    async fn test_counter_update() -> Result<()> {
        let writer = MetricsWriter::with_path("/tmp/metrics_test")?;
        writer.clear();

        writer.counter("test_counter", Vec::<(String, String)>::new(), 5.0)?;
        assert_eq!(writer.len(), 1);

        writer.counter("test_counter", Vec::<(String, String)>::new(), 3.0)?;
        assert_eq!(writer.len(), 1); // Same metric, same labels

        Ok(())
    }

    #[tokio::test]
    async fn test_gauge_update() -> Result<()> {
        let writer = MetricsWriter::with_path("/tmp/metrics_test")?;
        writer.clear();

        writer.gauge("test_gauge", Vec::<(String, String)>::new(), 42.0)?;
        assert_eq!(writer.len(), 1);

        writer.gauge("test_gauge", Vec::<(String, String)>::new(), 99.0)?;
        assert_eq!(writer.len(), 1); // Same metric, updated value

        Ok(())
    }

    #[tokio::test]
    async fn test_multiple_labels() -> Result<()> {
        let writer = MetricsWriter::with_path("/tmp/metrics_test")?;
        writer.clear();

        let labels1: Vec<(String, String)> = vec![("method".to_string(), "GET".to_string())];
        let labels2: Vec<(String, String)> = vec![("method".to_string(), "POST".to_string())];

        writer.counter("requests_total", labels1, 10.0)?;
        writer.counter("requests_total", labels2, 5.0)?;

        assert_eq!(writer.len(), 2); // Different labels = different metrics

        Ok(())
    }

    #[tokio::test]
    async fn test_flush() -> Result<()> {
        let temp_dir = tempfile::tempdir()?;
        let writer = MetricsWriter::with_path(temp_dir.path())?;
        writer.clear();

        writer.counter("test_counter", Vec::<(String, String)>::new(), 42.0)?;
        writer.gauge("test_gauge", Vec::<(String, String)>::new(), 21.5)?;

        writer.flush().await?;

        let content = fs::read_to_string(temp_dir.path().join("metrics.prom"))?;
        assert!(content.contains("test_counter"));
        assert!(content.contains("test_gauge"));
        assert!(content.contains("42"));
        assert!(content.contains("21.5"));

        Ok(())
    }

    #[tokio::test]
    async fn test_flush_with_labels() -> Result<()> {
        let temp_dir = tempfile::tempdir()?;
        let writer = MetricsWriter::with_path(temp_dir.path())?;
        writer.clear();

        let labels: Vec<(String, String)> = vec![("method".to_string(), "GET".to_string())];
        writer.counter("requests_total", labels, 100.0)?;

        writer.flush().await?;

        let content = fs::read_to_string(temp_dir.path().join("metrics.prom"))?;
        assert!(content.contains("requests_total{method=\"GET\"}"));
        assert!(content.contains("100"));

        Ok(())
    }

    #[test]
    fn test_make_key_no_labels() {
        let key = MetricsWriter::make_key("metric_name", &Labels::new());
        assert_eq!(key, "metric_name");
    }

    #[test]
    fn test_make_key_with_labels() {
        let labels = Labels::from(vec![("method".to_string(), "GET".to_string())]);
        let key1 = MetricsWriter::make_key("requests", &labels);
        let key2 = MetricsWriter::make_key("requests", &Labels::new());

        assert_ne!(key1, key2);
    }

    #[tokio::test]
    async fn test_counter_increment() -> Result<()> {
        let writer = MetricsWriter::with_path("/tmp/metrics_test")?;
        writer.clear();

        writer.inc("test_metric", Vec::<(String, String)>::new())?;
        writer.inc("test_metric", Vec::<(String, String)>::new())?;
        writer.inc("test_metric", Vec::<(String, String)>::new())?;

        // All go to same metric (no labels), so count is 1
        assert_eq!(writer.len(), 1);

        Ok(())
    }
}