feat: comprehensive Prometheus integration with full pipeline support

Makefile

@@ -26,6 +26,43 @@ ps-locald:
 dev:
 	pre-commit install
 
+# =============================================================================
+# PROMETHEUS TESTING
+# =============================================================================
+
+.PHONY: prometheus-up prometheus-down prometheus-logs prometheus-test
+
+# start prometheus testing stack
+prometheus-up:
+	@echo "🚀 Starting Prometheus testing stack..."
+	docker network create anomstack_network 2>/dev/null || true
+	docker compose -f docker-compose.prometheus.yaml up -d
+	@echo "✅ Prometheus stack started!"
+	@echo "📊 Prometheus UI: http://localhost:9090"
+	@echo "📈 Grafana UI: http://localhost:3001 (admin/admin)"
+	@echo "🔧 Node Exporter: http://localhost:9100"
+
+# stop prometheus testing stack
+prometheus-down:
+	@echo "🛑 Stopping Prometheus testing stack..."
+	docker compose -f docker-compose.prometheus.yaml down -v
+	@echo "✅ Prometheus stack stopped!"
+
+# show prometheus logs
+prometheus-logs:
+	docker compose -f docker-compose.prometheus.yaml logs -f
+
+# run prometheus integration tests
+prometheus-test: prometheus-up
+	@echo "🧪 Testing Prometheus integration..."
+	@echo "⏰ Waiting for services to start..."
+	@sleep 10
+	@echo "🔍 Testing Prometheus API..."
+	@curl -s http://localhost:9090/api/v1/label/__name__/values | grep -q "prometheus_build_info" && echo "✅ Prometheus API working" || echo "❌ Prometheus API failed"
+	@echo "🔍 Testing Node Exporter metrics..."
+	@curl -s http://localhost:9100/metrics | grep -q "node_cpu_seconds_total" && echo "✅ Node Exporter working" || echo "❌ Node Exporter failed"
+	@echo "🎉 Prometheus integration test complete!"
+
 # =============================================================================
 # DOCKER OPERATIONS
 # =============================================================================

anomstack/df/save.py

@@ -7,6 +7,7 @@
 from anomstack.external.clickhouse.clickhouse import save_df_clickhouse
 from anomstack.external.duckdb.duckdb import save_df_duckdb
 from anomstack.external.gcp.bigquery import save_df_bigquery
+from anomstack.external.prometheus.prometheus import save_df_prometheus
 from anomstack.external.snowflake.snowflake import save_df_snowflake
 from anomstack.external.sqlite.sqlite import save_df_sqlite
 
@@ -36,6 +37,8 @@ def save_df(df: pd.DataFrame, db: str, table_key: str, if_exists: str = "append"
         df = save_df_sqlite(df, table_key)
     elif db == "clickhouse":
         df = save_df_clickhouse(df, table_key)
+    elif db == "prometheus":
+        df = save_df_prometheus(df, table_key)
     else:
         raise ValueError(f"Unknown db: {db}")
 

anomstack/external/prometheus/README.md

@@ -0,0 +1,215 @@
+# Prometheus Database Integration
+
+This module provides Prometheus integration for Anomstack, enabling both reading metrics from Prometheus via PromQL queries and writing processed results back to Prometheus via the remote write API.
+
+## Features
+
+- **Read from Prometheus**: Execute PromQL queries to retrieve time series data
+- **Write to Prometheus**: Send processed metrics back via remote write API
+- **Flexible Configuration**: Parameterizable via environment variables and YAML
+- **Query Types**: Support for both instant queries and range queries
+- **Authentication**: Support for basic authentication
+- **Error Handling**: Robust error handling with logging
+
+## Configuration
+
+### Environment Variables
+
+Set these environment variables to configure Prometheus connection:
+
+```bash
+# Prometheus server connection
+export ANOMSTACK_PROMETHEUS_HOST="localhost"
+export ANOMSTACK_PROMETHEUS_PORT="9090"
+export ANOMSTACK_PROMETHEUS_PROTOCOL="http"
+
+# Authentication (optional)
+export ANOMSTACK_PROMETHEUS_USERNAME="your_username"
+export ANOMSTACK_PROMETHEUS_PASSWORD="your_password"
+
+# Remote write endpoint (for saving data back to Prometheus)
+export ANOMSTACK_PROMETHEUS_REMOTE_WRITE_URL="http://localhost:9090/api/v1/write"
+```
+
+### YAML Configuration
+
+In your metric batch YAML file, specify `db: "prometheus"` to use Prometheus as the database:
+
+```yaml
+metric_batch: "my_prometheus_metrics"
+db: "prometheus"  # Use Prometheus as database destination
+
+# Prometheus-specific configuration
+prometheus_query_type: "query_range"  # or "query" for instant queries
+prometheus_start: "now-1h"           # Start time for range queries
+prometheus_end: "now"                # End time for range queries
+prometheus_step: "60s"               # Step size for range queries
+
+# Define your PromQL queries
+prometheus_queries:
+  - name: "cpu_usage"
+    query: "100 - (avg by (instance) (rate(node_cpu_seconds_total{mode=\"idle\"}[5m])) * 100)"
+
+  - name: "memory_usage"
+    query: "(1 - (node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes)) * 100"
+
+  - name: "disk_usage"
+    query: "100 - (node_filesystem_avail_bytes / node_filesystem_size_bytes) * 100"
+```
+
+## Usage Examples
+
+### Reading Data from Prometheus
+
+```python
+from anomstack.external.prometheus.prometheus import read_sql_prometheus
+
+# Execute a PromQL query (instant query)
+df = read_sql_prometheus("up")
+
+# Execute a range query
+df = read_sql_prometheus(
+    "rate(http_requests_total[5m])",
+    query_type="query_range",
+    start="now-1h",
+    end="now",
+    step="60s"
+)
+```
+
+### Writing Data to Prometheus
+
+```python
+from anomstack.external.prometheus.prometheus import save_df_prometheus
+import pandas as pd
+
+# Create sample data
+df = pd.DataFrame({
+    'metric_timestamp': [pd.Timestamp.now()],
+    'metric_name': ['anomaly_score'],
+    'metric_value': [0.85]
+})
+
+# Save to Prometheus via remote write
+save_df_prometheus(df, "anomaly_metrics")
+```
+
+### Using with Anomstack Pipeline
+
+```python
+from anomstack.external.prometheus.prometheus import execute_promql_queries
+
+def ingest(**kwargs):
+    queries = [
+        {"name": "cpu_usage", "query": "rate(node_cpu_seconds_total[5m])"},
+        {"name": "memory_usage", "query": "node_memory_MemUsed_bytes"}
+    ]
+
+    return execute_promql_queries(
+        queries=queries,
+        query_type="query_range",
+        start="now-1h",
+        end="now",
+        step="60s"
+    )
+```
+
+## Query Types
+
+### Instant Queries (`query_type: "query"`)
+- Returns the current value of metrics at a single point in time
+- Best for real-time monitoring and alerting
+- Faster execution
+
+### Range Queries (`query_type: "query_range"`)
+- Returns time series data over a specified time range
+- Better for historical analysis and anomaly detection
+- Provides more data points for ML training
+
+## Data Format
+
+All functions return data in Anomstack's standard format:
+
+```python
+pd.DataFrame({
+    'metric_timestamp': [timestamp],  # pandas.Timestamp
+    'metric_name': [name],           # string
+    'metric_value': [value]          # float
+})
+```
+
+## Testing with Local Prometheus
+
+Use the provided Docker Compose setup for local testing:
+
+```bash
+# Start Prometheus stack
+docker-compose -f docker-compose.prometheus.yaml up -d
+
+# Set environment variables for local testing
+export ANOMSTACK_PROMETHEUS_HOST="localhost"
+export ANOMSTACK_PROMETHEUS_PORT="9090"
+export ANOMSTACK_PROMETHEUS_REMOTE_WRITE_URL="http://localhost:9090/api/v1/write"
+
+# Run your Anomstack pipeline
+```
+
+The test setup includes:
+- Prometheus server (http://localhost:9090)
+- Node Exporter for system metrics (http://localhost:9100)
+- Grafana for visualization (http://localhost:3001, admin/admin)
+- Demo application with sample metrics (http://localhost:8080)
+
+## Common PromQL Queries
+
+### System Metrics
+```yaml
+prometheus_queries:
+  - name: "cpu_usage_percent"
+    query: "100 - (avg by (instance) (rate(node_cpu_seconds_total{mode=\"idle\"}[5m])) * 100)"
+
+  - name: "memory_usage_percent"
+    query: "(1 - (node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes)) * 100"
+
+  - name: "disk_usage_percent"
+    query: "100 - (node_filesystem_avail_bytes{fstype!=\"tmpfs\"} / node_filesystem_size_bytes{fstype!=\"tmpfs\"}) * 100"
+
+  - name: "network_received_bytes"
+    query: "rate(node_network_receive_bytes_total[5m])"
+```
+
+### Application Metrics
+```yaml
+prometheus_queries:
+  - name: "http_requests_per_second"
+    query: "rate(http_requests_total[5m])"
+
+  - name: "http_request_duration_95th"
+    query: "histogram_quantile(0.95, rate(http_request_duration_seconds_bucket[5m]))"
+
+  - name: "error_rate_percent"
+    query: "rate(http_requests_total{status=~\"5..\"}[5m]) / rate(http_requests_total[5m]) * 100"
+```
+
+## Limitations
+
+1. **Remote Write Format**: Currently uses JSON format instead of the official Prometheus protobuf format for simplicity
+2. **Authentication**: Only basic authentication is supported
+3. **Error Handling**: Write failures don't stop the pipeline (by design)
+
+## Troubleshooting
+
+### Connection Issues
+- Verify Prometheus is running: `curl http://localhost:9090/api/v1/label/__name__/values`
+- Check environment variables: `env | grep ANOMSTACK_PROMETHEUS`
+- Verify network connectivity and firewall settings
+
+### Query Issues
+- Test PromQL queries in Prometheus UI: http://localhost:9090/graph
+- Check query syntax and available metrics: http://localhost:9090/api/v1/label/__name__/values
+- Verify time ranges are reasonable (not too large)
+
+### Remote Write Issues
+- Ensure remote write is enabled in Prometheus configuration
+- Check if the endpoint accepts the format you're sending
+- Monitor Prometheus logs for write errors

anomstack/external/prometheus/__init__.py

@@ -0,0 +1,40 @@
+"""
+Prometheus integration for Anomstack.
+
+This module provides comprehensive Prometheus support including:
+- Reading metrics via PromQL queries  
+- Writing metrics via remote write API
+- SQL-equivalent query processing for all job types
+
+Job-specific query functions:
+"""
+
+from .prometheus import (
+    read_sql_prometheus,
+    save_df_prometheus,
+    get_prometheus_config,
+    get_prometheus_auth,
+    execute_promql_queries
+)
+
+from .train import execute_prometheus_train_query
+from .score import execute_prometheus_score_query  
+from .alert import execute_prometheus_alert_query
+from .change import execute_prometheus_change_query
+from .llmalert import execute_prometheus_llmalert_query
+
+__all__ = [
+    # Core Prometheus functions
+    "read_sql_prometheus",
+    "save_df_prometheus", 
+    "get_prometheus_config",
+    "get_prometheus_auth",
+    "execute_promql_queries",
+
+    # Job-specific query functions
+    "execute_prometheus_train_query",
+    "execute_prometheus_score_query",
+    "execute_prometheus_alert_query", 
+    "execute_prometheus_change_query",
+    "execute_prometheus_llmalert_query",
+] 

anomstack/external/prometheus/alert.py

@@ -0,0 +1,51 @@
+"""
+Prometheus query processing for alert jobs.
+
+Handles getting anomaly scores for alert evaluation.
+"""
+
+import pandas as pd
+from dagster import get_dagster_logger
+from typing import Dict
+from anomstack.external.prometheus.prometheus import read_sql_prometheus
+
+
+def execute_prometheus_alert_query(spec: Dict) -> pd.DataFrame:
+    """
+    Execute Prometheus query equivalent to alerts.sql.
+
+    Gets anomaly scores for alert evaluation.
+
+    Args:
+        spec: Metric batch specification
+
+    Returns:
+        pd.DataFrame: Alert data in expected format
+    """
+    logger = get_dagster_logger()
+    logger.info("Executing Prometheus alert query")
+
+    metric_batch = spec.get("metric_batch", "")
+    alert_metric_timestamp_max_days_ago = spec.get("alert_metric_timestamp_max_days_ago", 1)
+
+    # Get anomaly scores for alerting
+    promql = "{__name__=~'.*_anomaly_score'}"
+
+    df = read_sql_prometheus(
+        promql,
+        query_type="query_range",
+        start=f"now-{alert_metric_timestamp_max_days_ago}d",
+        end="now",
+        step="300s"
+    )
+
+    if not df.empty:
+        df["metric_batch"] = metric_batch
+        df["metric_type"] = "score"
+
+        # Remove the _anomaly_score suffix from metric names to match expected format
+        df["metric_name"] = df["metric_name"].str.replace("_anomaly_score", "", regex=False)
+
+        logger.info(f"Retrieved {len(df)} anomaly scores for alerting")
+
+    return df