diff --git a/Makefile b/Makefile index f7b36ac..6a446b4 100644 --- a/Makefile +++ b/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 # ============================================================================= diff --git a/anomstack/df/save.py b/anomstack/df/save.py index a119d58..89e520e 100644 --- a/anomstack/df/save.py +++ b/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}") diff --git a/anomstack/external/prometheus/README.md b/anomstack/external/prometheus/README.md new file mode 100644 index 0000000..040ea26 --- /dev/null +++ b/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 diff --git a/anomstack/external/prometheus/__init__.py b/anomstack/external/prometheus/__init__.py new file mode 100644 index 0000000..eaf709d --- /dev/null +++ b/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", +] \ No newline at end of file diff --git a/anomstack/external/prometheus/alert.py b/anomstack/external/prometheus/alert.py new file mode 100644 index 0000000..f92675d --- /dev/null +++ b/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 \ No newline at end of file diff --git a/anomstack/external/prometheus/change.py b/anomstack/external/prometheus/change.py new file mode 100644 index 0000000..5f4f49b --- /dev/null +++ b/anomstack/external/prometheus/change.py @@ -0,0 +1,46 @@ +""" +Prometheus query processing for change detection jobs. + +Handles getting raw metrics for change point detection. +""" + +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_change_query(spec: Dict) -> pd.DataFrame: + """ + Execute Prometheus query equivalent to change.sql. + + Gets raw metrics for change point detection. + + Args: + spec: Metric batch specification + + Returns: + pd.DataFrame: Change detection data + """ + logger = get_dagster_logger() + logger.info("Executing Prometheus change query") + + metric_batch = spec.get("metric_batch", "") + change_metric_timestamp_max_days_ago = spec.get("change_metric_timestamp_max_days_ago", 7) + + promql = "{__name__!~'.*_(anomaly_score|alert|llm_alert|threshold_alert)'}" + + df = read_sql_prometheus( + promql, + query_type="query_range", + start=f"now-{change_metric_timestamp_max_days_ago}d", + end="now", + step="300s" + ) + + if not df.empty: + df["metric_batch"] = metric_batch + df["metric_type"] = "metric" + logger.info(f"Retrieved {len(df)} rows for change detection") + + return df \ No newline at end of file diff --git a/anomstack/external/prometheus/llmalert.py b/anomstack/external/prometheus/llmalert.py new file mode 100644 index 0000000..3b46ca8 --- /dev/null +++ b/anomstack/external/prometheus/llmalert.py @@ -0,0 +1,48 @@ +""" +Prometheus query processing for LLM alert jobs. + +Handles getting anomaly scores for LLM-based alerting. +""" + +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_llmalert_query(spec: Dict) -> pd.DataFrame: + """ + Execute Prometheus query equivalent to llmalert.sql. + + Gets anomaly scores for LLM-based alert evaluation. + + Args: + spec: Metric batch specification + + Returns: + pd.DataFrame: LLM alert data + """ + logger = get_dagster_logger() + logger.info("Executing Prometheus LLM alert query") + + metric_batch = spec.get("metric_batch", "") + llmalert_metric_timestamp_max_days_ago = spec.get("llmalert_metric_timestamp_max_days_ago", 1) + + # Get anomaly scores for LLM alerting + promql = "{__name__=~'.*_anomaly_score'}" + + df = read_sql_prometheus( + promql, + query_type="query_range", + start=f"now-{llmalert_metric_timestamp_max_days_ago}d", + end="now", + step="300s" + ) + + if not df.empty: + df["metric_batch"] = metric_batch + df["metric_type"] = "score" + df["metric_name"] = df["metric_name"].str.replace("_anomaly_score", "", regex=False) + logger.info(f"Retrieved {len(df)} anomaly scores for LLM alerting") + + return df \ No newline at end of file diff --git a/anomstack/external/prometheus/prometheus.py b/anomstack/external/prometheus/prometheus.py new file mode 100644 index 0000000..411eb5d --- /dev/null +++ b/anomstack/external/prometheus/prometheus.py @@ -0,0 +1,510 @@ +""" +Prometheus integration for Anomstack. + +This module provides functions for reading from and writing to Prometheus instances, +supporting both PromQL queries for reading and remote write API for sending data. +""" + +import os +import struct +import time +from typing import Dict, List, Optional +from urllib.parse import urljoin + +from dagster import get_dagster_logger +import pandas as pd +import requests + +# Add imports for Protocol Buffers and Prometheus client +try: + from prometheus_client import parser + from prometheus_client.core import CollectorRegistry, Gauge + from prometheus_client.exposition import generate_latest + import snappy + + PROTOBUF_AVAILABLE = True + IMPORT_ERROR = None +except ImportError as e: + PROTOBUF_AVAILABLE = False + IMPORT_ERROR = str(e) + + +def get_prometheus_config() -> Dict[str, Optional[str]]: + """ + Get Prometheus connection configuration from environment variables. + + Returns: + Dict containing Prometheus connection parameters + """ + return { + "host": os.environ.get("ANOMSTACK_PROMETHEUS_HOST", "localhost"), + "port": os.environ.get("ANOMSTACK_PROMETHEUS_PORT", "9090"), + "protocol": os.environ.get("ANOMSTACK_PROMETHEUS_PROTOCOL", "http"), + "username": os.environ.get("ANOMSTACK_PROMETHEUS_USERNAME"), + "password": os.environ.get("ANOMSTACK_PROMETHEUS_PASSWORD"), + "remote_write_url": os.environ.get("ANOMSTACK_PROMETHEUS_REMOTE_WRITE_URL"), + } + + +def build_prometheus_url(config: Dict[str, Optional[str]], endpoint: str = "") -> str: + """ + Build Prometheus URL from configuration. + + Args: + config: Prometheus configuration dict + endpoint: API endpoint to append + + Returns: + Complete Prometheus URL + """ + base_url = f"{config['protocol']}://{config['host']}:{config['port']}" + return urljoin(base_url, endpoint) + + +def get_prometheus_auth(config: Dict[str, Optional[str]]) -> Optional[tuple]: + """ + Get authentication tuple for Prometheus requests. + + Args: + config: Prometheus configuration dict + + Returns: + Authentication tuple or None + """ + if config.get("username") and config.get("password"): + return (config["username"], config["password"]) + return None + + +def read_sql_prometheus(promql: str, **kwargs) -> pd.DataFrame: + """ + Read data from Prometheus using PromQL query. + + Note: This function is named read_sql_prometheus to match the pattern used by other + database integrations, but it actually executes PromQL queries, not SQL. + + Args: + promql: PromQL query string + **kwargs: Additional parameters like time range, step, etc. + + Returns: + pd.DataFrame: Query results with columns: metric_timestamp, metric_name, metric_value + """ + logger = get_dagster_logger() + config = get_prometheus_config() + + # Default to instant query, but support range queries + query_type = kwargs.get("query_type", "query") # "query" or "query_range" + + if query_type == "query_range": + endpoint = "/api/v1/query_range" + params = { + "query": promql, + "start": kwargs.get("start", "now-1h"), + "end": kwargs.get("end", "now"), + "step": kwargs.get("step", "60s"), + } + else: + endpoint = "/api/v1/query" + params = {"query": promql} + + url = build_prometheus_url(config, endpoint) + auth = get_prometheus_auth(config) + + logger.info(f"Executing PromQL query: {promql}") + logger.debug(f"Prometheus URL: {url}") + + try: + response = requests.get(url, params=params, auth=auth, timeout=30) + response.raise_for_status() + except requests.RequestException as e: + logger.error(f"Failed to query Prometheus at {url}: {e}") + raise + + result = response.json() + if result.get("status") != "success": + logger.error(f"Prometheus query failed: {result}") + raise RuntimeError(f"Prometheus query failed: {result.get('error', 'Unknown error')}") + + # Convert Prometheus response to standard DataFrame format + all_rows = [] + data = result.get("data", {}) + + if query_type == "query_range": + # Handle range query results + for series in data.get("result", []): + metric_labels = series.get("metric", {}) + metric_name = _build_metric_name(promql, metric_labels) + + for timestamp_val, value_str in series.get("values", []): + try: + ts = pd.to_datetime(float(timestamp_val), unit="s").floor("S") + metric_value = float(value_str) + + all_rows.append( + { + "metric_timestamp": ts, + "metric_name": metric_name, + "metric_value": metric_value, + } + ) + except (ValueError, TypeError) as e: + logger.warning(f"Error parsing value {value_str}: {e}") + continue + else: + # Handle instant query results + for series in data.get("result", []): + metric_labels = series.get("metric", {}) + metric_name = _build_metric_name(promql, metric_labels) + + value_arr = series.get("value", []) + if len(value_arr) == 2: + timestamp_val, value_str = value_arr + try: + ts = pd.to_datetime(float(timestamp_val), unit="s").floor("S") + metric_value = float(value_str) + + all_rows.append( + { + "metric_timestamp": ts, + "metric_name": metric_name, + "metric_value": metric_value, + } + ) + except (ValueError, TypeError) as e: + logger.warning(f"Error parsing value {value_str}: {e}") + continue + + df = pd.DataFrame(all_rows) + df = df.drop_duplicates(subset=["metric_timestamp", "metric_name"]) + + logger.info(f"Retrieved {len(df)} rows from Prometheus") + return df + + +def _build_metric_name(promql: str, metric_labels: Dict) -> str: + """ + Build a readable metric name from PromQL query and labels. + + Args: + promql: Original PromQL query + metric_labels: Metric labels dict + + Returns: + String metric name + """ + # Try to extract base metric name from PromQL + base_name = promql.split("(")[0].split("{")[0].strip() + + # Add instance info if available + instance = metric_labels.get("instance", "") + if instance: + # Clean up instance name + instance = instance.replace(":", "_").replace(".", "_") + return f"{base_name}.{instance}" + + # Add other relevant labels + job = metric_labels.get("job", "") + if job: + return f"{base_name}.{job}" + + return base_name + + +def save_df_prometheus(df: pd.DataFrame, table_key: str) -> pd.DataFrame: + """ + Save DataFrame to Prometheus using remote write API with proper Protocol Buffers format. + + Args: + df: DataFrame with columns: metric_timestamp, metric_name, metric_value + table_key: Not used for Prometheus, but kept for interface consistency + + Returns: + pd.DataFrame: The input DataFrame + """ + logger = get_dagster_logger() + config = get_prometheus_config() + + remote_write_url = config.get("remote_write_url") + if not remote_write_url: + logger.warning( + "No ANOMSTACK_PROMETHEUS_REMOTE_WRITE_URL configured, skipping Prometheus write" + ) + return df + + if not PROTOBUF_AVAILABLE: + logger.error(f"Missing dependencies: {IMPORT_ERROR}") + logger.error("Required: pip install protobuf python-snappy") + logger.warning("Skipping Prometheus write - install required packages") + return df + + logger.info(f"Saving {len(df)} rows to Prometheus via remote write") + logger.debug(f"PROTOBUF_AVAILABLE: {PROTOBUF_AVAILABLE}") + logger.debug(f"Remote write URL: {remote_write_url}") + + # Convert DataFrame to Prometheus remote write format + time_series = _convert_df_to_prometheus_format(df) + + if not time_series: + logger.warning("No valid time series data to write to Prometheus") + return df + + try: + # Create proper Protocol Buffers message + logger.debug(f"Creating protobuf message for {len(time_series)} time series") + write_request = _create_protobuf_write_request(time_series) + logger.debug(f"Protobuf message size: {len(write_request)} bytes") + + # Compress with Snappy + compressed_data = snappy.compress(write_request) + logger.debug(f"Compressed size: {len(compressed_data)} bytes") + + # Send to Prometheus remote write endpoint + auth = get_prometheus_auth(config) + headers = { + "Content-Type": "application/x-protobuf", + "Content-Encoding": "snappy", + "X-Prometheus-Remote-Write-Version": "0.1.0", + } + logger.debug(f"Headers: {headers}") + logger.debug(f"Auth: {auth is not None}") + + # Debug: Log sample timestamps to check for "out of bounds" issues + if time_series: + sample_ts = time_series[0] + if sample_ts.get("samples"): + sample_timestamp = sample_ts["samples"][0]["timestamp"] + import time as time_module + + current_time_ms = int(time_module.time() * 1000) + logger.debug( + f"Sample timestamp: {sample_timestamp}, Current time: {current_time_ms}, Diff: {current_time_ms - sample_timestamp}ms" + ) + + response = requests.post( + remote_write_url, data=compressed_data, headers=headers, auth=auth, timeout=30 + ) + logger.debug(f"Response status: {response.status_code}") + logger.debug(f"Response headers: {dict(response.headers)}") + logger.debug(f"Response text: {response.text}") + response.raise_for_status() + logger.info(f"โœ… Successfully wrote {len(time_series)} time series to Prometheus") + + except Exception as e: + logger.error(f"Failed to write to Prometheus remote write endpoint: {e}") + logger.warning("Continuing without writing to Prometheus") + + return df + + +def _convert_df_to_prometheus_format(df: pd.DataFrame) -> List[Dict]: + """ + Convert DataFrame to Prometheus remote write time series format. + + Args: + df: DataFrame with metric data + + Returns: + List of time series dictionaries + """ + logger = get_dagster_logger() + time_series = [] + + # Group by metric name to create separate time series + for metric_name, group in df.groupby("metric_name"): + try: + # Parse metric name and add suffix based on metric_type + metric_name_str = str(metric_name) + + # Add suffix based on metric_type if present + if "metric_type" in group.columns: + metric_type = group["metric_type"].iloc[0] # All rows in group should have same type + if metric_type == "score": + metric_name_str = f"{metric_name_str}_anomaly_score" + elif metric_type == "alert": + metric_name_str = f"{metric_name_str}_alert" + elif metric_type == "llmalert": + metric_name_str = f"{metric_name_str}_llm_alert" + elif metric_type == "tholdalert": + metric_name_str = f"{metric_name_str}_threshold_alert" + # metric_type == "metric" keeps original name (no suffix) + + labels = {"__name__": metric_name_str} + + # If metric name contains dots, use the parts as labels + if "." in metric_name_str: + parts = metric_name_str.split(".") + labels["__name__"] = parts[0] + if len(parts) > 1: + labels["instance"] = ".".join(parts[1:]) + + # Create samples from the group + samples = [] + # Use current timestamp to avoid "out of bounds" errors + # Prometheus typically rejects samples older than 1 hour or in the future + current_time_ms = int(time.time() * 1000) + + # Sort group by timestamp to ensure chronological order + sorted_group = group.sort_values('metric_timestamp') + + for _, row in sorted_group.iterrows(): + try: + value = float(row["metric_value"]) + + # Use the actual metric timestamp, not current time + timestamp_ms = current_time_ms # Default fallback + try: + metric_ts = row.get("metric_timestamp") + if metric_ts is not None and not pd.isna(metric_ts): + # Convert to timestamp milliseconds + if hasattr(metric_ts, 'timestamp'): + # Already a datetime object + timestamp_ms = int(metric_ts.timestamp() * 1000) + else: + # Convert to datetime first + dt = pd.to_datetime(metric_ts) + timestamp_ms = int(dt.timestamp() * 1000) + except (ValueError, TypeError, AttributeError): + # Use fallback timestamp if conversion fails + pass + + samples.append({"timestamp": timestamp_ms, "value": value}) + except (ValueError, TypeError, AttributeError) as e: + logger.warning(f"Error processing row for metric {metric_name_str}: {e}") + continue + + if samples: + time_series.append( + { + "labels": [{"name": k, "value": v} for k, v in labels.items()], + "samples": samples, + } + ) + + except Exception as e: + logger.warning(f"Error processing metric {metric_name}: {e}") + continue + + return time_series + + +def _create_protobuf_write_request(time_series: List[Dict]) -> bytes: + """ + Create a Protocol Buffers WriteRequest message for Prometheus remote write. + + This manually creates the protobuf binary format without requiring .proto compilation. + Based on the Prometheus remote.proto specification. + """ + logger = get_dagster_logger() + + # Build WriteRequest protobuf message manually + # WriteRequest { repeated TimeSeries timeseries = 1; } + write_request_data = b"" + + for ts in time_series: + # Create TimeSeries message + timeseries_data = b"" + + # Add labels (field 1, repeated Label) + for label in ts.get("labels", []): + label_data = b"" + # Label { string name = 1; string value = 2; } + label_data += _encode_protobuf_string(1, str(label["name"])) + label_data += _encode_protobuf_string(2, str(label["value"])) + timeseries_data += _encode_protobuf_message(1, label_data) + + # Add samples (field 2, repeated Sample) + for sample in ts.get("samples", []): + sample_data = b"" + # Sample { double value = 1; int64 timestamp = 2; } + sample_data += _encode_protobuf_double(1, float(sample["value"])) + sample_data += _encode_protobuf_int64(2, int(sample["timestamp"])) + timeseries_data += _encode_protobuf_message(2, sample_data) + + # Add TimeSeries to WriteRequest (field 1, repeated) + write_request_data += _encode_protobuf_message(1, timeseries_data) + + logger.debug(f"Created protobuf WriteRequest with {len(time_series)} time series") + return write_request_data + + +def _encode_protobuf_string(field_number: int, value: str) -> bytes: + """Encode a protobuf string field.""" + value_bytes = value.encode("utf-8") + length = len(value_bytes) + field_header = (field_number << 3) | 2 # Wire type 2 for length-delimited + return _encode_varint(field_header) + _encode_varint(length) + value_bytes + + +def _encode_protobuf_double(field_number: int, value: float) -> bytes: + """Encode a protobuf double field.""" + field_header = (field_number << 3) | 1 # Wire type 1 for 64-bit + return _encode_varint(field_header) + struct.pack(" bytes: + """Encode a protobuf int64 field.""" + field_header = (field_number << 3) | 0 # Wire type 0 for varint + return _encode_varint(field_header) + _encode_varint(value) + + +def _encode_protobuf_message(field_number: int, message_data: bytes) -> bytes: + """Encode a protobuf message field.""" + field_header = (field_number << 3) | 2 # Wire type 2 for length-delimited + return _encode_varint(field_header) + _encode_varint(len(message_data)) + message_data + + +def _encode_varint(value: int) -> bytes: + """Encode a variable-length integer.""" + result = b"" + while value >= 0x80: + result += bytes([value & 0x7F | 0x80]) + value >>= 7 + result += bytes([value & 0x7F]) + return result + + +def execute_promql_queries(queries: List[Dict[str, str]], **kwargs) -> pd.DataFrame: + """ + Execute multiple PromQL queries and return combined results. + + This function is designed to be used from YAML-configured metric batches. + + Args: + queries: List of dicts with "name" and "query" keys + **kwargs: Additional query parameters + + Returns: + pd.DataFrame: Combined results from all queries + """ + logger = get_dagster_logger() + all_rows = [] + + for query_config in queries: + query_name = query_config.get("name", "unnamed_query") + promql = query_config.get("query", "") + + if not promql: + logger.warning(f"Empty PromQL query for {query_name}, skipping") + continue + + try: + logger.info(f"Executing query '{query_name}': {promql}") + df = read_sql_prometheus(promql, **kwargs) + + # Prefix metric names with query name for clarity + df["metric_name"] = df["metric_name"].apply(lambda x: f"{query_name}.{x}") + + all_rows.append(df) + + except Exception as e: + logger.error(f"Error executing query '{query_name}': {e}") + continue + + if all_rows: + combined_df = pd.concat(all_rows, ignore_index=True) + combined_df = combined_df.drop_duplicates(subset=["metric_timestamp", "metric_name"]) + return combined_df + else: + logger.warning("No successful queries, returning empty DataFrame") + return pd.DataFrame(columns=["metric_timestamp", "metric_name", "metric_value"]) diff --git a/anomstack/external/prometheus/score.py b/anomstack/external/prometheus/score.py new file mode 100644 index 0000000..b93c35d --- /dev/null +++ b/anomstack/external/prometheus/score.py @@ -0,0 +1,50 @@ +""" +Prometheus query processing for score jobs. + +Handles getting raw metrics for anomaly scoring. +""" + +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_score_query(spec: Dict) -> pd.DataFrame: + """ + Execute Prometheus query equivalent to score.sql. + + Gets raw metrics for scoring (recent data that needs anomaly scores). + + Args: + spec: Metric batch specification + + Returns: + pd.DataFrame: Scoring data in expected format + """ + logger = get_dagster_logger() + logger.info("Executing Prometheus score query") + + metric_batch = spec.get("metric_batch", "") + score_metric_timestamp_max_days_ago = spec.get("score_metric_timestamp_max_days_ago", 1) + + # Get raw metrics for scoring + promql = "{__name__!~'.*_(anomaly_score|alert|llm_alert|threshold_alert)'}" + + df = read_sql_prometheus( + promql, + query_type="query_range", + start=f"now-{score_metric_timestamp_max_days_ago}d", + end="now", + step="300s" + ) + + if not df.empty: + df["metric_batch"] = metric_batch + df["metric_type"] = "metric" + + # For score job, we typically want recent data that doesn't have scores yet + # This is simplified - in practice you'd check for missing scores + logger.info(f"Retrieved {len(df)} rows for scoring") + + return df \ No newline at end of file diff --git a/anomstack/external/prometheus/train.py b/anomstack/external/prometheus/train.py new file mode 100644 index 0000000..cdd7ee7 --- /dev/null +++ b/anomstack/external/prometheus/train.py @@ -0,0 +1,128 @@ +""" +Prometheus query processing for train jobs. + +Handles getting raw metrics for model training with SQL-equivalent operations +like filtering, grouping, ranking, and row limiting. +""" + +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_train_query(spec: Dict) -> pd.DataFrame: + """ + Execute Prometheus query equivalent to train.sql. + + Handles: time filtering, metric filtering, grouping, ranking, row limiting + + Args: + spec: Metric batch specification with all config parameters + + Returns: + pd.DataFrame: Training data in expected format + """ + logger = get_dagster_logger() + logger.info("Executing Prometheus train query") + + # Extract parameters from spec + metric_batch = spec.get("metric_batch", "") + train_metric_timestamp_max_days_ago = spec.get("train_metric_timestamp_max_days_ago", 7) + train_exclude_metrics = spec.get("train_exclude_metrics", []) + train_max_n = spec.get("train_max_n", 1000) + train_min_n = spec.get("train_min_n", 1) + + # Get metric names from prometheus_queries configuration in spec + prometheus_queries = spec.get("prometheus_queries", []) + if not prometheus_queries: + logger.warning("No prometheus_queries found in spec") + return pd.DataFrame() + + # Extract metric names from the configuration + # Note: ingest function adds "anomstack_" prefix, so we need to match what's in Prometheus + metric_names = [] + for query_config in prometheus_queries: + name = query_config.get("name", "") + if name: + # Add the prefix that the ingest function uses + prefixed_name = f"anomstack_{name}" + metric_names.append(prefixed_name) + + logger.info(f"Querying {len(metric_names)} metrics from prometheus_queries config: {metric_names}") + + # Combine results from all metrics + all_dfs = [] + for metric_name in metric_names: + try: + df_single = read_sql_prometheus( + metric_name, + query_type="query_range", + start=f"now-{train_metric_timestamp_max_days_ago}d", + end="now", + step="300s" # 5 minute intervals + ) + if not df_single.empty: + all_dfs.append(df_single) + except Exception as e: + logger.warning(f"Failed to query {metric_name}: {e}") + continue + + # Concatenate all results + if all_dfs: + df = pd.concat(all_dfs, ignore_index=True) + else: + df = pd.DataFrame() # Empty DataFrame + + if df.empty: + logger.warning("No data returned from Prometheus train query") + return df + + logger.info(f"Retrieved {len(df)} raw rows from Prometheus") + + # Filter out metrics with anomaly detection suffixes + suffix_pattern = r'_(anomaly_score|alert|llm_alert|threshold_alert)$' + before_count = len(df) + df = df[~df["metric_name"].str.contains(suffix_pattern, regex=True, na=False)] + after_count = len(df) + logger.info(f"Filtered out {before_count - after_count} rows with anomaly detection suffixes") + + # Post-process to match SQL template behavior + + # 1. Add required columns + df["metric_batch"] = metric_batch + df["metric_type"] = "metric" + + # 2. Filter by metric_batch if needed (though Prometheus doesn't store this metadata) + # For now, assume all metrics belong to the current batch + + # 3. Exclude specific metrics if configured + if train_exclude_metrics: + excluded_pattern = "|".join(train_exclude_metrics) + logger.info(f"Excluding metrics matching: {excluded_pattern}") + df = df[~df["metric_name"].str.contains(excluded_pattern, regex=True, na=False)] + + # 4. Group by timestamp and metric_name, then average (like SQL template) + logger.info("Grouping and averaging metric values") + grouped_df = df.groupby(["metric_timestamp", "metric_name"], as_index=False)["metric_value"].mean() + + # 5. Add metric_recency_rank (like SQL window function) + logger.info("Adding metric recency ranking") + grouped_df["metric_recency_rank"] = grouped_df.groupby("metric_name")["metric_timestamp"].rank( + method="dense", ascending=False + ).astype(int) + + # 6. Apply row limits per metric (like SQL BETWEEN clause) + logger.info(f"Filtering to ranks between {train_min_n} and {train_max_n}") + filtered_df = grouped_df[ + (grouped_df["metric_recency_rank"] >= train_min_n) & + (grouped_df["metric_recency_rank"] <= train_max_n) + ].copy() + + # 7. Add required columns + filtered_df["metric_batch"] = metric_batch + filtered_df["metric_type"] = "metric" + + unique_metrics = len(filtered_df["metric_name"].unique()) if len(filtered_df) > 0 else 0 + logger.info(f"Final train dataset: {len(filtered_df)} rows, {unique_metrics} unique metrics") + return filtered_df \ No newline at end of file diff --git a/anomstack/sql/read.py b/anomstack/sql/read.py index eab4900..abcdd6c 100644 --- a/anomstack/sql/read.py +++ b/anomstack/sql/read.py @@ -13,13 +13,103 @@ ) from anomstack.external.duckdb.duckdb import read_sql_duckdb, run_sql_duckdb from anomstack.external.gcp.bigquery import read_sql_bigquery +from anomstack.external.prometheus.prometheus import read_sql_prometheus from anomstack.external.snowflake.snowflake import read_sql_snowflake from anomstack.external.sqlite.sqlite import read_sql_sqlite, run_sql_sqlite from anomstack.sql.translate import db_translate +import re +from jinja2 import Template pd.options.display.max_columns = 10 +def _read_sql_prometheus_converted(sql: str, logger) -> pd.DataFrame: + """ + Convert SQL patterns to PromQL queries using dedicated Prometheus query functions. + + This detects the job type and calls the appropriate Prometheus query function + which handles PromQL construction and SQL-equivalent post-processing. + + Args: + sql: SQL query (typically from Jinja template) + logger: Dagster logger + + Returns: + pd.DataFrame: Query results in expected format + """ + from anomstack.config import get_specs + from anomstack.external.prometheus import ( + execute_prometheus_train_query, + execute_prometheus_score_query, + execute_prometheus_alert_query, + execute_prometheus_change_query, + execute_prometheus_llmalert_query + ) + + logger.info("Converting SQL to Prometheus query using dedicated functions") + + # Detect job type from SQL patterns + job_type = None + if "metric_type = 'metric'" in sql: + if "metric_recency_rank" in sql: + job_type = "train" + else: + job_type = "score" + elif "metric_type = 'score'" in sql: + job_type = "alerts" + else: + # Try to detect from common SQL patterns + if "train_max_n" in sql or "train_min_n" in sql: + job_type = "train" + elif "score_" in sql: + job_type = "score" + elif "alert_" in sql: + job_type = "alerts" + elif "change_" in sql: + job_type = "change" + elif "llmalert_" in sql: + job_type = "llmalert" + else: + job_type = "train" # Default fallback + + logger.info(f"Detected job type: {job_type}") + + # Get the current spec context + try: + specs = get_specs() + spec = specs[0] if specs else {} # Use first spec as fallback + except Exception as e: + logger.warning(f"Failed to get specs: {e}") + spec = {} + + # Call appropriate Prometheus query function + try: + if job_type == "train": + return execute_prometheus_train_query(spec) + elif job_type == "score": + return execute_prometheus_score_query(spec) + elif job_type == "alerts": + return execute_prometheus_alert_query(spec) + elif job_type == "change": + return execute_prometheus_change_query(spec) + elif job_type == "llmalert": + return execute_prometheus_llmalert_query(spec) + else: + logger.warning(f"Unknown job type: {job_type}, falling back to train query") + return execute_prometheus_train_query(spec) + + except Exception as e: + logger.error(f"Failed to execute Prometheus {job_type} query: {e}") + # Return empty DataFrame with expected columns + empty_df = pd.DataFrame() + empty_df["metric_timestamp"] = pd.Series(dtype="datetime64[ns]") + empty_df["metric_name"] = pd.Series(dtype="string") + empty_df["metric_value"] = pd.Series(dtype="float64") + empty_df["metric_batch"] = pd.Series(dtype="string") + empty_df["metric_type"] = pd.Series(dtype="string") + return empty_df + + def read_sql(sql: str, db: str, returns_df: bool = True) -> pd.DataFrame: """ Read data from SQL. @@ -35,10 +125,16 @@ def read_sql(sql: str, db: str, returns_df: bool = True) -> pd.DataFrame: logger = get_dagster_logger() + # For Prometheus, convert SQL patterns to PromQL instead of translating + if db == "prometheus": + return _read_sql_prometheus_converted(sql, logger) + sql = db_translate(sql, db) logger.debug(f"-- read_sql() is about to read this qry:\n{sql}") + df = pd.DataFrame() # Initialize df to avoid unbound variable error + if db == "bigquery": if returns_df: df = read_sql_bigquery(sql) @@ -67,6 +163,13 @@ def read_sql(sql: str, db: str, returns_df: bool = True) -> pd.DataFrame: elif not returns_df: run_sql_clickhouse(sql) df = pd.DataFrame() + elif db == "prometheus": + if returns_df: + # For Prometheus, sql parameter actually contains PromQL + df = read_sql_prometheus(sql) + elif not returns_df: + # Prometheus doesn't support non-returning queries in this context + raise NotImplementedError("Prometheus not yet implemented for non-returns_df queries.") else: raise ValueError(f"Unknown db: {db}") diff --git a/dagster_docker.yaml b/dagster_docker.yaml index 2d93648..d879eb6 100644 --- a/dagster_docker.yaml +++ b/dagster_docker.yaml @@ -6,19 +6,20 @@ run_coordinator: module: dagster.core.run_coordinator class: QueuedRunCoordinator config: - max_concurrent_runs: 10 # Reduced from 25 to prevent excessive storage + max_concurrent_runs: 10 tag_concurrency_limits: - key: "dagster/concurrency_key" value: "database" - limit: 2 # Reduced from 3 + limit: 2 - key: "dagster/concurrency_key" value: "ml_training" - limit: 1 # Reduced from 2 + limit: 1 run_launcher: module: dagster_docker class: DockerRunLauncher config: + image: anomstack_code_image:latest env_vars: - DAGSTER_POSTGRES_USER - DAGSTER_POSTGRES_PASSWORD @@ -29,16 +30,21 @@ run_launcher: - ANOMSTACK_IGNORE_EXAMPLES - DAGSTER_HOME - ANOMSTACK_HOME + - ANOMSTACK_PROMETHEUS_HOST + - ANOMSTACK_PROMETHEUS_PORT + - ANOMSTACK_PROMETHEUS_PROTOCOL + - ANOMSTACK_PROMETHEUS_REMOTE_WRITE_URL network: anomstack_network container_kwargs: volumes: # Make docker client accessible to any launched containers as well - /var/run/docker.sock:/var/run/docker.sock - /tmp/io_manager_storage:/tmp/io_manager_storage - - ${ANOMSTACK_HOME}/tmp:/opt/dagster/app/tmp - - ${ANOMSTACK_HOME}/dagster_home:/opt/dagster/dagster_home - - ${ANOMSTACK_HOME}/metrics:/opt/dagster/app/metrics + - /Users/andrewmaguire/Documents/GitHub/anomstack/tmp:/opt/dagster/app/tmp + - /Users/andrewmaguire/Documents/GitHub/anomstack/dagster_home:/opt/dagster/dagster_home + - /Users/andrewmaguire/Documents/GitHub/anomstack/metrics:/opt/dagster/app/metrics - anomstack_metrics_duckdb:/data +# Keep PostgreSQL storage since we have it running run_storage: module: dagster_postgres.run_storage class: PostgresRunStorage @@ -57,26 +63,25 @@ run_retries: enabled: true max_retries: 1 -# Aggressive retention policies to prevent disk space issues +# Retention policies retention: schedule: - purge_after_days: 3 # Keep schedule runs for 3 days + purge_after_days: 3 sensor: purge_after_days: - skipped: 1 # Keep only 1 day of skipped sensor runs - failure: 3 # Keep 3 days of failed runs for debugging - success: 1 # Keep only 1 day of successful sensor runs + skipped: 1 + failure: 3 + success: 1 -# Run monitoring to detect and clean up stuck runs +# Run monitoring run_monitoring: enabled: true - start_timeout_seconds: 300 # 5 minutes to start - cancel_timeout_seconds: 180 # 3 minutes to cancel - max_runtime_seconds: 900 # 15 minutes max runtime per run - max_resume_run_attempts: 2 # Resume runs after worker crashes (DockerRunLauncher only) - poll_interval_seconds: 60 # Check every minute + start_timeout_seconds: 300 + cancel_timeout_seconds: 180 + max_runtime_seconds: 900 + poll_interval_seconds: 60 -# Disable telemetry to reduce disk writes +# Disable telemetry telemetry: enabled: false diff --git a/dagster_docker.yaml.backup b/dagster_docker.yaml.backup new file mode 100644 index 0000000..23db4d0 --- /dev/null +++ b/dagster_docker.yaml.backup @@ -0,0 +1,122 @@ +scheduler: + module: dagster.core.scheduler + class: DagsterDaemonScheduler + +run_coordinator: + module: dagster.core.run_coordinator + class: QueuedRunCoordinator + config: + max_concurrent_runs: 10 # Reduced from 25 to prevent excessive storage + tag_concurrency_limits: + - key: "dagster/concurrency_key" + value: "database" + limit: 2 # Reduced from 3 + - key: "dagster/concurrency_key" + value: "ml_training" + limit: 1 # Reduced from 2 + +run_launcher: + module: dagster_docker + class: DockerRunLauncher + config: + image: anomstack_code_image:latest + env_vars: + - DAGSTER_POSTGRES_USER + - DAGSTER_POSTGRES_PASSWORD + - DAGSTER_POSTGRES_DB + - ANOMSTACK_DUCKDB_PATH + - ANOMSTACK_TABLE_KEY + - ANOMSTACK_MODEL_PATH + - ANOMSTACK_IGNORE_EXAMPLES + - DAGSTER_HOME + - ANOMSTACK_HOME + - ANOMSTACK_PROMETHEUS_HOST + - ANOMSTACK_PROMETHEUS_PORT + - ANOMSTACK_PROMETHEUS_PROTOCOL + - ANOMSTACK_PROMETHEUS_REMOTE_WRITE_URL + network: anomstack_network + container_kwargs: + volumes: # Make docker client accessible to any launched containers as well + - /var/run/docker.sock:/var/run/docker.sock + - /tmp/io_manager_storage:/tmp/io_manager_storage + - ${ANOMSTACK_HOME}/tmp:/opt/dagster/app/tmp + - ${ANOMSTACK_HOME}/dagster_home:/opt/dagster/dagster_home + - ${ANOMSTACK_HOME}/metrics:/opt/dagster/app/metrics + - anomstack_metrics_duckdb:/data + +run_storage: + module: dagster_postgres.run_storage + class: PostgresRunStorage + config: + postgres_db: + hostname: anomstack_postgresql + username: + env: DAGSTER_POSTGRES_USER + password: + env: DAGSTER_POSTGRES_PASSWORD + db_name: + env: DAGSTER_POSTGRES_DB + port: 5432 + +run_retries: + enabled: true + max_retries: 1 + +# Aggressive retention policies to prevent disk space issues +retention: + schedule: + purge_after_days: 3 # Keep schedule runs for 3 days + sensor: + purge_after_days: + skipped: 1 # Keep only 1 day of skipped sensor runs + failure: 3 # Keep 3 days of failed runs for debugging + success: 1 # Keep only 1 day of successful sensor runs + +# Run monitoring to detect and clean up stuck runs +run_monitoring: + enabled: true + start_timeout_seconds: 300 # 5 minutes to start + cancel_timeout_seconds: 180 # 3 minutes to cancel + max_runtime_seconds: 900 # 15 minutes max runtime per run + max_resume_run_attempts: 2 # Resume runs after worker crashes (DockerRunLauncher only) + poll_interval_seconds: 60 # Check every minute + +# Disable telemetry to reduce disk writes +telemetry: + enabled: false + +schedules: + use_threads: true + num_workers: 8 + +sensors: + use_threads: true + num_workers: 4 + +schedule_storage: + module: dagster_postgres.schedule_storage + class: PostgresScheduleStorage + config: + postgres_db: + hostname: anomstack_postgresql + username: + env: DAGSTER_POSTGRES_USER + password: + env: DAGSTER_POSTGRES_PASSWORD + db_name: + env: DAGSTER_POSTGRES_DB + port: 5432 + +event_log_storage: + module: dagster_postgres.event_log + class: PostgresEventLogStorage + config: + postgres_db: + hostname: anomstack_postgresql + username: + env: DAGSTER_POSTGRES_USER + password: + env: DAGSTER_POSTGRES_PASSWORD + db_name: + env: DAGSTER_POSTGRES_DB + port: 5432 diff --git a/dagster_local.yaml b/dagster_local.yaml new file mode 100644 index 0000000..8455cbb --- /dev/null +++ b/dagster_local.yaml @@ -0,0 +1,98 @@ +scheduler: + module: dagster.core.scheduler + class: DagsterDaemonScheduler + +run_coordinator: + module: dagster.core.run_coordinator + class: QueuedRunCoordinator + config: + max_concurrent_runs: 10 + tag_concurrency_limits: + - key: "dagster/concurrency_key" + value: "database" + limit: 2 + - key: "dagster/concurrency_key" + value: "ml_training" + limit: 1 + +# Use DefaultRunLauncher instead of DockerRunLauncher for simplicity +run_launcher: + module: dagster.core.launcher.default_run_launcher + class: DefaultRunLauncher + +# Keep PostgreSQL storage since we have it running +run_storage: + module: dagster_postgres.run_storage + class: PostgresRunStorage + config: + postgres_db: + hostname: anomstack_postgresql + username: + env: DAGSTER_POSTGRES_USER + password: + env: DAGSTER_POSTGRES_PASSWORD + db_name: + env: DAGSTER_POSTGRES_DB + port: 5432 + +run_retries: + enabled: true + max_retries: 1 + +# Retention policies +retention: + schedule: + purge_after_days: 3 + sensor: + purge_after_days: + skipped: 1 + failure: 3 + success: 1 + +# Run monitoring +run_monitoring: + enabled: true + start_timeout_seconds: 300 + cancel_timeout_seconds: 180 + max_runtime_seconds: 900 + poll_interval_seconds: 60 + +# Disable telemetry +telemetry: + enabled: false + +schedules: + use_threads: true + num_workers: 8 + +sensors: + use_threads: true + num_workers: 4 + +schedule_storage: + module: dagster_postgres.schedule_storage + class: PostgresScheduleStorage + config: + postgres_db: + hostname: anomstack_postgresql + username: + env: DAGSTER_POSTGRES_USER + password: + env: DAGSTER_POSTGRES_PASSWORD + db_name: + env: DAGSTER_POSTGRES_DB + port: 5432 + +event_log_storage: + module: dagster_postgres.event_log + class: PostgresEventLogStorage + config: + postgres_db: + hostname: anomstack_postgresql + username: + env: DAGSTER_POSTGRES_USER + password: + env: DAGSTER_POSTGRES_PASSWORD + db_name: + env: DAGSTER_POSTGRES_DB + port: 5432 diff --git a/docker-compose.dev.yaml b/docker-compose.dev.yaml index 24710fc..5a6f32a 100644 --- a/docker-compose.dev.yaml +++ b/docker-compose.dev.yaml @@ -24,12 +24,16 @@ services: context: . dockerfile: docker/Dockerfile.dagster image: anomstack_dagster_image + volumes: + - ./dagster_local.yaml:/opt/dagster/dagster_home/dagster.yaml anomstack_daemon: build: context: . dockerfile: docker/Dockerfile.dagster image: anomstack_dagster_image + volumes: + - ./dagster_local.yaml:/opt/dagster/dagster_home/dagster.yaml anomstack_dashboard: build: diff --git a/docker-compose.prometheus.yaml b/docker-compose.prometheus.yaml new file mode 100644 index 0000000..3202126 --- /dev/null +++ b/docker-compose.prometheus.yaml @@ -0,0 +1,63 @@ +services: + # Prometheus server + prometheus: + image: prom/prometheus:latest + container_name: anomstack_prometheus + ports: + - "9090:9090" + volumes: + - ./prometheus/prometheus.yml:/etc/prometheus/prometheus.yml + - ./prometheus/rules.yml:/etc/prometheus/rules.yml + - prometheus_data:/prometheus + command: + - '--config.file=/etc/prometheus/prometheus.yml' + - '--storage.tsdb.path=/prometheus' + - '--web.console.libraries=/etc/prometheus/console_libraries' + - '--web.console.templates=/etc/prometheus/consoles' + - '--storage.tsdb.retention.time=200h' + - '--web.enable-lifecycle' + - '--web.enable-admin-api' + - '--web.enable-remote-write-receiver' + networks: + - anomstack_network + + # Node Exporter for system metrics + node-exporter: + image: prom/node-exporter:latest + container_name: anomstack_node_exporter + ports: + - "9100:9100" + volumes: + - /proc:/host/proc:ro + - /sys:/host/sys:ro + - /:/rootfs:ro + command: + - '--path.procfs=/host/proc' + - '--path.rootfs=/rootfs' + - '--path.sysfs=/host/sys' + - '--collector.filesystem.mount-points-exclude=^/(sys|proc|dev|host|etc)($$|/)' + networks: + - anomstack_network + + # Sample application with metrics (optional) - removed for now + + # Grafana for visualization (optional) + grafana: + image: grafana/grafana:latest + container_name: anomstack_grafana + ports: + - "3001:3000" # Use 3001 to avoid conflict with Dagster + environment: + - GF_SECURITY_ADMIN_PASSWORD=admin + volumes: + - grafana_data:/var/lib/grafana + networks: + - anomstack_network + +volumes: + prometheus_data: + grafana_data: + +networks: + anomstack_network: + external: true diff --git a/docker-compose.yaml b/docker-compose.yaml index eb7c4b4..57ed70d 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -55,7 +55,7 @@ services: DAGSTER_POSTGRES_PASSWORD: "${ANOMSTACK_POSTGRES_PASSWORD:-postgres_password}" DAGSTER_POSTGRES_DB: "${ANOMSTACK_POSTGRES_DB:-postgres_db}" DAGSTER_CODE_SERVER_HOST: "${DAGSTER_CODE_SERVER_HOST:-anomstack_code}" - DAGSTER_CURRENT_IMAGE: "andrewm4894/anomstack_code:latest" + DAGSTER_CURRENT_IMAGE: "anomstack_code_image:latest" ANOMSTACK_DUCKDB_PATH: "/data/anomstack.db" DAGSTER_HOME: "/opt/dagster/dagster_home" networks: @@ -79,7 +79,7 @@ services: # Since our instance uses the QueuedRunCoordinator, any runs submitted from the webserver will be put on # a queue and later dequeued and launched by dagster-daemon. anomstack_webserver: - image: andrewm4894/anomstack_dagster:latest + image: anomstack_dagster_image:latest entrypoint: - dagster-webserver - -h @@ -132,7 +132,7 @@ services: # This service runs the dagster-daemon process, which is responsible for taking runs # off of the queue and launching them, as well as creating runs from schedules or sensors. anomstack_daemon: - image: andrewm4894/anomstack_dagster:latest + image: anomstack_dagster_image:latest entrypoint: - dagster-daemon - run @@ -175,7 +175,7 @@ services: # This service runs the dashboard anomstack_dashboard: - image: andrewm4894/anomstack_dashboard:latest + image: anomstack_dashboard_image:latest container_name: anomstack_dashboard restart: on-failure ports: diff --git a/metrics/examples/coindesk/coindesk.py b/metrics/examples/coindesk/coindesk.py index a40e014..58fc1b0 100644 --- a/metrics/examples/coindesk/coindesk.py +++ b/metrics/examples/coindesk/coindesk.py @@ -5,8 +5,8 @@ def ingest(): metric_timestamp, metric_name, metric_value. The timestamp is derived from VALUE_LAST_UPDATE_TS, aggregated to the second, and duplicates are removed. """ - import pandas as pd from dagster import get_dagster_logger + import pandas as pd import requests logger = get_dagster_logger() diff --git a/metrics/examples/currency/currency.py b/metrics/examples/currency/currency.py index 15d4e18..215dab7 100644 --- a/metrics/examples/currency/currency.py +++ b/metrics/examples/currency/currency.py @@ -5,10 +5,10 @@ def ingest(): Filters the response to include only a subset of target currencies (e.g. USD, GBP, JPY, CHF, AUD, CAD). Returns a DataFrame with columns: metric_timestamp, metric_name, and metric_value. """ - import pandas as pd from datetime import datetime from dagster import get_dagster_logger + import pandas as pd import requests logger = get_dagster_logger() diff --git a/metrics/examples/github/github.py b/metrics/examples/github/github.py index cb47667..024b312 100644 --- a/metrics/examples/github/github.py +++ b/metrics/examples/github/github.py @@ -14,8 +14,8 @@ def ingest(): - subscribers_count - size """ - import pandas as pd from dagster import get_dagster_logger + import pandas as pd import requests logger = get_dagster_logger() diff --git a/metrics/examples/prometheus/prometheus.py b/metrics/examples/prometheus/prometheus.py index af825aa..0afde8e 100644 --- a/metrics/examples/prometheus/prometheus.py +++ b/metrics/examples/prometheus/prometheus.py @@ -5,8 +5,8 @@ def ingest(): metric_timestamp, metric_name, metric_value. The metric_timestamp is aggregated to the second level and duplicates are removed. """ - import pandas as pd from dagster import get_dagster_logger + import pandas as pd import requests logger = get_dagster_logger() diff --git a/metrics/examples/prometheus/prometheus_parameterized.py b/metrics/examples/prometheus/prometheus_parameterized.py new file mode 100644 index 0000000..fa07624 --- /dev/null +++ b/metrics/examples/prometheus/prometheus_parameterized.py @@ -0,0 +1,75 @@ +""" +Parameterized Prometheus ingest function for Anomstack. + +This version reads configuration from YAML instead of hardcoded values, +making it more flexible and reusable. +""" + +from dagster import get_dagster_logger + +from anomstack.external.prometheus.prometheus import execute_promql_queries + + +def ingest( + prometheus_queries=None, + prometheus_query_type="query", + prometheus_start="now-1h", + prometheus_end="now", + prometheus_step="60s", + **kwargs, +): + """ + Ingest data from Prometheus using configurable PromQL queries. + + Args: + prometheus_queries: List of query configurations with 'name' and 'query' fields + prometheus_query_type: "query" for instant queries or "query_range" for range queries + prometheus_start: Start time for range queries (e.g., "now-1h", "2023-01-01T00:00:00Z") + prometheus_end: End time for range queries (e.g., "now", "2023-01-01T01:00:00Z") + prometheus_step: Step size for range queries (e.g., "60s", "5m") + **kwargs: Additional parameters + + Returns: + pd.DataFrame: DataFrame with columns: metric_timestamp, metric_name, metric_value + """ + logger = get_dagster_logger() + + # Default queries if none provided (for backward compatibility) + if not prometheus_queries: + logger.info("No prometheus_queries provided, using default demo queries") + prometheus_queries = [ + { + "name": "demo_api_http_requests_in_progress", + "query": "demo_api_http_requests_in_progress", + }, + { + "name": "avg_api_request_duration", + "query": "rate(demo_api_request_duration_seconds_sum[5m]) / rate(demo_api_request_duration_seconds_count[5m])", + }, + {"name": "cpu_usage_rate", "query": "rate(demo_cpu_usage_seconds_total[5m])"}, + {"name": "demo_memory_usage_bytes", "query": "demo_memory_usage_bytes"}, + { + "name": "demo_batch_last_run_duration_seconds", + "query": "demo_batch_last_run_duration_seconds", + }, + {"name": "demo_disk_usage_bytes", "query": "demo_disk_usage_bytes"}, + {"name": "demo_disk_total_bytes", "query": "demo_disk_total_bytes"}, + {"name": "items_shipped_rate", "query": "rate(demo_items_shipped_total[5m])"}, + {"name": "demo_intermittent_metric", "query": "demo_intermittent_metric"}, + {"name": "demo_is_holiday", "query": "demo_is_holiday"}, + ] + + logger.info(f"Executing {len(prometheus_queries)} Prometheus queries") + + # Execute queries using the centralized function + df = execute_promql_queries( + queries=prometheus_queries, + query_type=prometheus_query_type, + start=prometheus_start, + end=prometheus_end, + step=prometheus_step, + ) + + logger.info(f"Retrieved {len(df)} total rows from Prometheus") + + return df diff --git a/metrics/examples/prometheus/prometheus_parameterized.yaml b/metrics/examples/prometheus/prometheus_parameterized.yaml new file mode 100644 index 0000000..76e6fe1 --- /dev/null +++ b/metrics/examples/prometheus/prometheus_parameterized.yaml @@ -0,0 +1,51 @@ +metric_batch: "prometheus_parameterized" +table_key: "metrics_prometheus_parameterized" +db: "prometheus" # Use Prometheus as the database destination + +# Schedule configuration +ingest_cron_schedule: "*/5 * * * *" +train_cron_schedule: "*/180 * * * *" +score_cron_schedule: "*/10 * * * *" +alert_cron_schedule: "*/15 * * * *" +change_cron_schedule: "*/15 * * * *" +llmalert_cron_schedule: "*/60 * * * *" +plot_cron_schedule: "*/30 * * * *" + +# Alert configuration +alert_always: False +alert_metric_timestamp_max_days_ago: 3 +disable_llmalert: False +alert_methods: "email,slack" + +# Prometheus-specific configuration +prometheus_query_type: "query_range" # "query" for instant, "query_range" for time series +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 + +# Configure PromQL queries +prometheus_queries: + - name: "http_requests_in_progress" + query: "demo_api_http_requests_in_progress" + + - name: "avg_request_duration" + query: "rate(demo_api_request_duration_seconds_sum[5m]) / rate(demo_api_request_duration_seconds_count[5m])" + + - name: "cpu_usage_rate" + query: "rate(demo_cpu_usage_seconds_total[5m])" + + - name: "memory_usage" + query: "demo_memory_usage_bytes" + + - name: "disk_usage" + query: "demo_disk_usage_bytes" + + - name: "disk_total" + query: "demo_disk_total_bytes" + + - name: "items_shipped_rate" + query: "rate(demo_items_shipped_total[5m])" + +# Use the parameterized ingest function +ingest_fn: > + {% include "./examples/prometheus/prometheus_parameterized.py" %} diff --git a/metrics/examples/prometheus/prometheus_production.yaml b/metrics/examples/prometheus/prometheus_production.yaml new file mode 100644 index 0000000..477e005 --- /dev/null +++ b/metrics/examples/prometheus/prometheus_production.yaml @@ -0,0 +1,88 @@ +metric_batch: "prometheus_production" +table_key: "metrics_prometheus_production" +db: "prometheus" # Use Prometheus as the database destination + +# Production schedule configuration +ingest_cron_schedule: "*/1 * * * *" # Every minute for real-time monitoring +train_cron_schedule: "0 */4 * * *" # Every 4 hours +score_cron_schedule: "*/5 * * * *" # Every 5 minutes +alert_cron_schedule: "*/5 * * * *" # Every 5 minutes +change_cron_schedule: "*/10 * * * *" # Every 10 minutes +llmalert_cron_schedule: "*/30 * * * *" # Every 30 minutes +plot_cron_schedule: "*/15 * * * *" # Every 15 minutes + +# Alert configuration +alert_always: False +alert_metric_timestamp_max_days_ago: 1 +disable_llmalert: False +alert_methods: "email,slack" + +# Prometheus configuration for production monitoring +prometheus_query_type: "query_range" # Use range queries for better ML training +prometheus_start: "now-10m" # 10 minutes of recent data +prometheus_end: "now" +prometheus_step: "30s" # 30-second resolution + +# Production-ready PromQL queries for system monitoring +prometheus_queries: + # CPU Metrics + - name: "cpu_usage_percent" + query: "100 - (avg by (instance) (rate(node_cpu_seconds_total{mode=\"idle\"}[5m])) * 100)" + + - name: "cpu_load_1m" + query: "node_load1" + + - name: "cpu_load_5m" + query: "node_load5" + + # Memory Metrics + - name: "memory_usage_percent" + query: "(1 - (node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes)) * 100" + + - name: "memory_usage_bytes" + query: "node_memory_MemTotal_bytes - node_memory_MemAvailable_bytes" + + # Disk Metrics + - name: "disk_usage_percent_root" + query: "100 - (node_filesystem_avail_bytes{mountpoint=\"/\"} / node_filesystem_size_bytes{mountpoint=\"/\"}) * 100" + + - name: "disk_io_read_bytes" + query: "rate(node_disk_read_bytes_total[5m])" + + - name: "disk_io_write_bytes" + query: "rate(node_disk_written_bytes_total[5m])" + + # Network Metrics + - name: "network_receive_bytes" + query: "rate(node_network_receive_bytes_total{device!=\"lo\"}[5m])" + + - name: "network_transmit_bytes" + query: "rate(node_network_transmit_bytes_total{device!=\"lo\"}[5m])" + + # Application Metrics (if available) + - 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: "http_error_rate_percent" + query: "rate(http_requests_total{status=~\"5..\"}[5m]) / rate(http_requests_total[5m]) * 100" + + # Database Metrics (if available) + - name: "mysql_connections_active" + query: "mysql_global_status_threads_connected" + + - name: "mysql_queries_per_second" + query: "rate(mysql_global_status_queries[5m])" + + # Container Metrics (if using Docker/Kubernetes) + - name: "container_cpu_usage_percent" + query: "rate(container_cpu_usage_seconds_total{name!=\"\"}[5m]) * 100" + + - name: "container_memory_usage_bytes" + query: "container_memory_usage_bytes{name!=\"\"}" + +# Use the parameterized ingest function +ingest_fn: > + {% include "./examples/prometheus/prometheus_parameterized.py" %} diff --git a/metrics/examples/prometheus_db/README.md b/metrics/examples/prometheus_db/README.md new file mode 100644 index 0000000..7122631 --- /dev/null +++ b/metrics/examples/prometheus_db/README.md @@ -0,0 +1,93 @@ +# Prometheus DB Example + +This example demonstrates using **Prometheus as both data source and database destination** for anomaly detection with Anomstack. + +## Overview + +The `prometheus_db` example: +- **Reads** system metrics from Prometheus using PromQL queries +- **Writes** anomaly detection results back to Prometheus +- Uses node_exporter metrics for monitoring system performance + +This is different from the general `prometheus` example which only reads from Prometheus but can write to any database backend. + +## Prerequisites + +1. **Prometheus server** running and accessible +2. **node_exporter** running to provide system metrics +3. **Prometheus configured** to scrape node_exporter metrics + +## Configuration + +### Environment Variables + +Set these environment variables for Prometheus connection: + +```bash +# Prometheus connection settings +export ANOMSTACK_PROMETHEUS_HOST="localhost" +export ANOMSTACK_PROMETHEUS_PORT="9090" +export ANOMSTACK_PROMETHEUS_PROTOCOL="http" + +# Optional authentication +export ANOMSTACK_PROMETHEUS_USERNAME="your-username" +export ANOMSTACK_PROMETHEUS_PASSWORD="your-password" +``` + +### Metric Configuration + +The example monitors these system metrics: +- **CPU usage percentage** - Calculated from idle time +- **Memory usage percentage** - Available vs total memory +- **Memory total/available bytes** - Raw memory values +- **Load averages** - 1m and 5m system load +- **Prometheus health** - Up/down status + +## Usage + +1. **Start the metric batch:** + ```bash + # Copy to your metrics directory + cp -r metrics/examples/prometheus_db/ metrics/ + + # Run with Anomstack + make local + ``` + +2. **Monitor in Dashboard:** + - View ingested metrics and anomaly scores + - Check alert status and trends + - Review system performance patterns + +3. **Customize queries:** + - Edit the `prometheus_queries` in the YAML file + - Add your own PromQL queries for specific metrics + - Adjust time ranges and step sizes as needed + +## Metrics Generated + +Each PromQL query generates metrics with names like: +- `cpu_usage_percent.100.instance_name` +- `memory_usage_percent.1.instance_name` +- `load_1m.node_load1.instance_name` + +## Schedule Configuration + +- **Ingest**: Every 5 minutes (`*/5 * * * *`) +- **Training**: Every 3 hours (`*/180 * * * *`) +- **Scoring**: Every 10 minutes (`*/10 * * * *`) +- **Alerts**: Every 15 minutes (`*/15 * * * *`) + +Adjust these schedules based on your monitoring requirements and data volume. + +## Troubleshooting + +1. **No data ingested**: Check Prometheus connectivity and node_exporter status +2. **Authentication errors**: Verify username/password if using Prometheus auth +3. **Query failures**: Test PromQL queries directly in Prometheus UI +4. **Performance issues**: Reduce query frequency or time ranges + +## Related Examples + +- `metrics/examples/prometheus/` - Read from Prometheus, write to any DB +- `metrics/examples/netdata/` - Similar system monitoring with Netdata \ No newline at end of file diff --git a/metrics/examples/prometheus_db/prometheus_db.py b/metrics/examples/prometheus_db/prometheus_db.py new file mode 100644 index 0000000..f9cce22 --- /dev/null +++ b/metrics/examples/prometheus_db/prometheus_db.py @@ -0,0 +1,213 @@ +def ingest(): + """ + Ingest data from Prometheus using node_exporter metrics. + + This example demonstrates using Prometheus as both data source and database destination. + It reads system metrics from Prometheus and stores them back to Prometheus for anomaly detection. + + Returns: + pd.DataFrame: DataFrame with columns: metric_timestamp, metric_name, metric_value + """ + import pandas as pd + import requests + import os + from dagster import get_dagster_logger + + logger = get_dagster_logger() + + # Use real node_exporter metrics instead of demo queries + prometheus_queries = [ + { + "name": "anomstack_cpu_usage_percent", + "query": "100 - (avg by (instance) (rate(node_cpu_seconds_total{mode=\"idle\"}[5m])) * 100)" + }, + { + "name": "anomstack_memory_usage_percent", + "query": "(1 - (node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes)) * 100" + }, + { + "name": "anomstack_memory_total_bytes", + "query": "node_memory_MemTotal_bytes" + }, + { + "name": "anomstack_memory_available_bytes", + "query": "node_memory_MemAvailable_bytes" + }, + { + "name": "anomstack_load_1m", + "query": "node_load1" + }, + { + "name": "anomstack_load_5m", + "query": "node_load5" + }, + { + "name": "anomstack_prometheus_up", + "query": "up" + } + ] + + # Configuration from environment variables + prometheus_query_type = "query" # Use instant queries to get current values + prometheus_start = "now-5m" # Not used for instant queries + prometheus_end = "now" + prometheus_step = "60s" + + logger.info(f"Executing {len(prometheus_queries)} Prometheus queries") + + # Get Prometheus configuration + config = { + "host": os.environ.get("ANOMSTACK_PROMETHEUS_HOST", "localhost"), + "port": os.environ.get("ANOMSTACK_PROMETHEUS_PORT", "9090"), + "protocol": os.environ.get("ANOMSTACK_PROMETHEUS_PROTOCOL", "http"), + "username": os.environ.get("ANOMSTACK_PROMETHEUS_USERNAME"), + "password": os.environ.get("ANOMSTACK_PROMETHEUS_PASSWORD"), + } + + # Build base URL + base_url = f"{config['protocol']}://{config['host']}:{config['port']}" + + # Setup authentication if provided + auth = None + if config.get("username") and config.get("password"): + auth = (config["username"], config["password"]) + + all_rows = [] + + for query_config in prometheus_queries: + query_name = query_config.get("name", "unnamed_query") + promql = query_config.get("query", "") + + if not promql: + logger.warning(f"Empty PromQL query for {query_name}, skipping") + continue + + try: + logger.info(f"Executing query '{query_name}': {promql}") + + # Setup endpoint and parameters based on query type + if prometheus_query_type == "query_range": + endpoint = "/api/v1/query_range" + + # Convert relative time to Unix timestamps + import time + now = int(time.time()) + + # Parse start time (e.g., "now-1h" -> 1 hour ago) + if prometheus_start.startswith("now-"): + time_str = prometheus_start.replace("now-", "") + if time_str.endswith("h"): + hours = int(time_str.replace("h", "")) + start_timestamp = now - (hours * 3600) + elif time_str.endswith("m"): + minutes = int(time_str.replace("m", "")) + start_timestamp = now - (minutes * 60) + else: + start_timestamp = now - 3600 # default 1 hour + else: + start_timestamp = now - 3600 # default 1 hour + + # Parse end time + if prometheus_end == "now": + end_timestamp = now + else: + end_timestamp = now + + params = { + "query": promql, + "start": str(start_timestamp), + "end": str(end_timestamp), + "step": prometheus_step + } + else: + endpoint = "/api/v1/query" + params = {"query": promql} + + url = base_url + endpoint + + # Execute the query + response = requests.get(url, params=params, auth=auth, timeout=30) + response.raise_for_status() + + result = response.json() + if result.get("status") != "success": + logger.error(f"Prometheus query failed: {result}") + continue + + # Process results + data = result.get("data", {}) + + if prometheus_query_type == "query_range": + # Handle range query results + for series in data.get("result", []): + metric_labels = series.get("metric", {}) + + # Build metric name + base_name = promql.split('(')[0].split('{')[0].strip() + instance = metric_labels.get("instance", "") + if instance: + instance = instance.replace(":", "_").replace(".", "_") + metric_name = f"{query_name}.{base_name}.{instance}" + else: + metric_name = f"{query_name}.{base_name}" + + for timestamp_val, value_str in series.get("values", []): + try: + ts = pd.to_datetime(float(timestamp_val), unit="s").floor("S") + metric_value = float(value_str) + + all_rows.append({ + "metric_timestamp": ts, + "metric_name": metric_name, + "metric_value": metric_value + }) + except (ValueError, TypeError) as e: + logger.warning(f"Error parsing value {value_str}: {e}") + continue + else: + # Handle instant query results + for series in data.get("result", []): + metric_labels = series.get("metric", {}) + + # Build metric name + base_name = promql.split('(')[0].split('{')[0].strip() + instance = metric_labels.get("instance", "") + if instance: + instance = instance.replace(":", "_").replace(".", "_") + metric_name = f"{query_name}.{base_name}.{instance}" + else: + metric_name = f"{query_name}.{base_name}" + + value_arr = series.get("value", []) + if len(value_arr) == 2: + timestamp_val, value_str = value_arr + try: + ts = pd.to_datetime(float(timestamp_val), unit="s").floor("S") + metric_value = float(value_str) + + all_rows.append({ + "metric_timestamp": ts, + "metric_name": metric_name, + "metric_value": metric_value + }) + except (ValueError, TypeError) as e: + logger.warning(f"Error parsing value {value_str}: {e}") + continue + + except Exception as e: + logger.error(f"Error executing query '{query_name}': {e}") + continue + + # Create DataFrame from results + if all_rows: + df = pd.DataFrame(all_rows) + df = df.drop_duplicates(subset=["metric_timestamp", "metric_name"]) + logger.info(f"Retrieved {len(df)} total rows from Prometheus") + return df + else: + logger.warning("No successful queries, returning empty DataFrame") + empty_df = pd.DataFrame() + empty_df["metric_timestamp"] = pd.Series(dtype="datetime64[ns]") + empty_df["metric_name"] = pd.Series(dtype="string") + empty_df["metric_value"] = pd.Series(dtype="float64") + return empty_df \ No newline at end of file diff --git a/metrics/examples/prometheus_db/prometheus_db.yaml b/metrics/examples/prometheus_db/prometheus_db.yaml new file mode 100644 index 0000000..3edf9d0 --- /dev/null +++ b/metrics/examples/prometheus_db/prometheus_db.yaml @@ -0,0 +1,51 @@ +metric_batch: "prometheus_db" +table_key: "metrics_prometheus_db" +db: "prometheus" # Use Prometheus as the database destination + +# Schedule configuration +ingest_cron_schedule: "*/5 * * * *" +train_cron_schedule: "*/180 * * * *" +score_cron_schedule: "*/10 * * * *" +alert_cron_schedule: "*/15 * * * *" +change_cron_schedule: "*/15 * * * *" +llmalert_cron_schedule: "*/60 * * * *" +plot_cron_schedule: "*/30 * * * *" + +# Alert configuration +alert_always: False +alert_metric_timestamp_max_days_ago: 3 +disable_llmalert: False +alert_methods: "email,slack" + +# Prometheus-specific configuration +prometheus_query_type: "query_range" # "query" for instant, "query_range" for time series +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 + +# Configure PromQL queries (using node_exporter metrics) +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: "memory_total_bytes" + query: "node_memory_MemTotal_bytes" + + - name: "memory_available_bytes" + query: "node_memory_MemAvailable_bytes" + + - name: "load_1m" + query: "node_load1" + + - name: "load_5m" + query: "node_load5" + + - name: "prometheus_up" + query: "up" + +# Use the parameterized ingest function +ingest_fn: > + {% include "./examples/prometheus_db/prometheus_db.py" %} \ No newline at end of file diff --git a/metrics/prometheus_test.py b/metrics/prometheus_test.py new file mode 100644 index 0000000..87e8e1e --- /dev/null +++ b/metrics/prometheus_test.py @@ -0,0 +1,197 @@ +def ingest(): + """ + Ingest data from Prometheus using node_exporter metrics. + + Returns: + pd.DataFrame: DataFrame with columns: metric_timestamp, metric_name, metric_value + """ + import os + + from dagster import get_dagster_logger + import pandas as pd + import requests + + logger = get_dagster_logger() + + # Use real node_exporter metrics instead of demo queries + 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": "memory_total_bytes", "query": "node_memory_MemTotal_bytes"}, + {"name": "memory_available_bytes", "query": "node_memory_MemAvailable_bytes"}, + {"name": "load_1m", "query": "node_load1"}, + {"name": "load_5m", "query": "node_load5"}, + {"name": "prometheus_up", "query": "up"}, + ] + + # Configuration from environment variables + prometheus_query_type = "query_range" + prometheus_start = "now-1h" + prometheus_end = "now" + prometheus_step = "60s" + + logger.info(f"Executing {len(prometheus_queries)} Prometheus queries") + + # Get Prometheus configuration + config = { + "host": os.environ.get("ANOMSTACK_PROMETHEUS_HOST", "localhost"), + "port": os.environ.get("ANOMSTACK_PROMETHEUS_PORT", "9090"), + "protocol": os.environ.get("ANOMSTACK_PROMETHEUS_PROTOCOL", "http"), + "username": os.environ.get("ANOMSTACK_PROMETHEUS_USERNAME"), + "password": os.environ.get("ANOMSTACK_PROMETHEUS_PASSWORD"), + } + + # Build base URL + base_url = f"{config['protocol']}://{config['host']}:{config['port']}" + + # Setup authentication if provided + auth = None + if config.get("username") and config.get("password"): + auth = (config["username"], config["password"]) + + all_rows = [] + + for query_config in prometheus_queries: + query_name = query_config.get("name", "unnamed_query") + promql = query_config.get("query", "") + + if not promql: + logger.warning(f"Empty PromQL query for {query_name}, skipping") + continue + + try: + logger.info(f"Executing query '{query_name}': {promql}") + + # Setup endpoint and parameters based on query type + if prometheus_query_type == "query_range": + endpoint = "/api/v1/query_range" + + # Convert relative time to Unix timestamps + import time + + now = int(time.time()) + + # Parse start time (e.g., "now-1h" -> 1 hour ago) + if prometheus_start.startswith("now-"): + time_str = prometheus_start.replace("now-", "") + if time_str.endswith("h"): + hours = int(time_str.replace("h", "")) + start_timestamp = now - (hours * 3600) + elif time_str.endswith("m"): + minutes = int(time_str.replace("m", "")) + start_timestamp = now - (minutes * 60) + else: + start_timestamp = now - 3600 # default 1 hour + else: + start_timestamp = now - 3600 # default 1 hour + + # Parse end time + if prometheus_end == "now": + end_timestamp = now + else: + end_timestamp = now + + params = { + "query": promql, + "start": str(start_timestamp), + "end": str(end_timestamp), + "step": prometheus_step, + } + else: + endpoint = "/api/v1/query" + params = {"query": promql} + + url = base_url + endpoint + + # Execute the query + response = requests.get(url, params=params, auth=auth, timeout=30) + response.raise_for_status() + + result = response.json() + if result.get("status") != "success": + logger.error(f"Prometheus query failed: {result}") + continue + + # Process results + data = result.get("data", {}) + + if prometheus_query_type == "query_range": + # Handle range query results + for series in data.get("result", []): + metric_labels = series.get("metric", {}) + + # Build metric name + base_name = promql.split("(")[0].split("{")[0].strip() + instance = metric_labels.get("instance", "") + if instance: + instance = instance.replace(":", "_").replace(".", "_") + metric_name = f"{query_name}.{base_name}.{instance}" + else: + metric_name = f"{query_name}.{base_name}" + + for timestamp_val, value_str in series.get("values", []): + try: + ts = pd.to_datetime(float(timestamp_val), unit="s").floor("S") + metric_value = float(value_str) + + all_rows.append( + { + "metric_timestamp": ts, + "metric_name": metric_name, + "metric_value": metric_value, + } + ) + except (ValueError, TypeError) as e: + logger.warning(f"Error parsing value {value_str}: {e}") + continue + else: + # Handle instant query results + for series in data.get("result", []): + metric_labels = series.get("metric", {}) + + # Build metric name + base_name = promql.split("(")[0].split("{")[0].strip() + instance = metric_labels.get("instance", "") + if instance: + instance = instance.replace(":", "_").replace(".", "_") + metric_name = f"{query_name}.{base_name}.{instance}" + else: + metric_name = f"{query_name}.{base_name}" + + value_arr = series.get("value", []) + if len(value_arr) == 2: + timestamp_val, value_str = value_arr + try: + ts = pd.to_datetime(float(timestamp_val), unit="s").floor("S") + metric_value = float(value_str) + + all_rows.append( + { + "metric_timestamp": ts, + "metric_name": metric_name, + "metric_value": metric_value, + } + ) + except (ValueError, TypeError) as e: + logger.warning(f"Error parsing value {value_str}: {e}") + continue + + except Exception as e: + logger.error(f"Error executing query '{query_name}': {e}") + continue + + # Create DataFrame from results + if all_rows: + df = pd.DataFrame(all_rows) + df = df.drop_duplicates(subset=["metric_timestamp", "metric_name"]) + logger.info(f"Retrieved {len(df)} total rows from Prometheus") + return df + else: + logger.warning("No successful queries, returning empty DataFrame") + return pd.DataFrame(columns=["metric_timestamp", "metric_name", "metric_value"]) diff --git a/metrics/prometheus_test.yaml b/metrics/prometheus_test.yaml new file mode 100644 index 0000000..e647057 --- /dev/null +++ b/metrics/prometheus_test.yaml @@ -0,0 +1,51 @@ +metric_batch: "prometheus_test" +table_key: "metrics_prometheus_test" +db: "prometheus" # Use Prometheus as the database destination + +# Schedule configuration +ingest_cron_schedule: "*/5 * * * *" +train_cron_schedule: "*/180 * * * *" +score_cron_schedule: "*/10 * * * *" +alert_cron_schedule: "*/15 * * * *" +change_cron_schedule: "*/15 * * * *" +llmalert_cron_schedule: "*/60 * * * *" +plot_cron_schedule: "*/30 * * * *" + +# Alert configuration +alert_always: False +alert_metric_timestamp_max_days_ago: 3 +disable_llmalert: False +alert_methods: "email,slack" + +# Prometheus-specific configuration +prometheus_query_type: "query_range" # "query" for instant, "query_range" for time series +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 + +# Configure PromQL queries (using node_exporter metrics) +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: "memory_total_bytes" + query: "node_memory_MemTotal_bytes" + + - name: "memory_available_bytes" + query: "node_memory_MemAvailable_bytes" + + - name: "load_1m" + query: "node_load1" + + - name: "load_5m" + query: "node_load5" + + - name: "prometheus_up" + query: "up" + +# Use the parameterized ingest function +ingest_fn: > + {% include "./prometheus_test.py" %} diff --git a/profiles/demo.env b/profiles/demo.env index 2f2f1bc..2dc1ef5 100644 --- a/profiles/demo.env +++ b/profiles/demo.env @@ -134,4 +134,4 @@ ANOMSTACK__ISS_LOCATION__SCORE_DEFAULT_SCHEDULE_STATUS=RUNNING ANOMSTACK__ISS_LOCATION__ALERT_DEFAULT_SCHEDULE_STATUS=RUNNING ANOMSTACK__ISS_LOCATION__CHANGE_DEFAULT_SCHEDULE_STATUS=RUNNING ANOMSTACK__ISS_LOCATION__DELETE_DEFAULT_SCHEDULE_STATUS=RUNNING -ANOMSTACK__ISS_LOCATION__LLMALERT_DEFAULT_SCHEDULE_STATUS=STOPPED \ No newline at end of file +ANOMSTACK__ISS_LOCATION__LLMALERT_DEFAULT_SCHEDULE_STATUS=STOPPED diff --git a/prometheus/prometheus.yml b/prometheus/prometheus.yml new file mode 100644 index 0000000..eeaeedd --- /dev/null +++ b/prometheus/prometheus.yml @@ -0,0 +1,33 @@ +global: + scrape_interval: 15s + evaluation_interval: 15s + +rule_files: + - "rules.yml" + +scrape_configs: + # Prometheus itself + - job_name: 'prometheus' + static_configs: + - targets: ['localhost:9090'] + + # Node Exporter for system metrics + - job_name: 'node-exporter' + static_configs: + - targets: ['node-exporter:9100'] + + # Demo application - removed for now + # - job_name: 'demo-app' + # static_configs: + # - targets: ['demo-app:8080'] + + # Add more targets as needed + # - job_name: 'your-app' + # static_configs: + # - targets: ['your-app:8080'] + +# Remote write configuration (for testing Anomstack writes) +remote_write: + - url: "http://localhost:9090/api/v1/write" + # This allows Prometheus to receive data via remote write + # In production, you'd use a different endpoint like VictoriaMetrics or Cortex diff --git a/prometheus/rules.yml b/prometheus/rules.yml new file mode 100644 index 0000000..7f6eb92 --- /dev/null +++ b/prometheus/rules.yml @@ -0,0 +1,32 @@ +groups: + - name: anomstack_test_rules + rules: + # High CPU usage + - alert: HighCPUUsage + expr: 100 - (avg by (instance) (rate(node_cpu_seconds_total{mode="idle"}[5m])) * 100) > 80 + for: 2m + labels: + severity: warning + annotations: + summary: "High CPU usage detected" + description: "CPU usage is above 80% for instance {{ $labels.instance }}" + + # High memory usage + - alert: HighMemoryUsage + expr: (1 - (node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes)) * 100 > 85 + for: 2m + labels: + severity: warning + annotations: + summary: "High memory usage detected" + description: "Memory usage is above 85% for instance {{ $labels.instance }}" + + # Disk space usage + - alert: HighDiskUsage + expr: (1 - (node_filesystem_avail_bytes{fstype!="tmpfs"} / node_filesystem_size_bytes{fstype!="tmpfs"})) * 100 > 90 + for: 2m + labels: + severity: critical + annotations: + summary: "High disk usage detected" + description: "Disk usage is above 90% for {{ $labels.mountpoint }} on {{ $labels.instance }}" diff --git a/requirements.txt b/requirements.txt index 4437ef3..580a323 100644 --- a/requirements.txt +++ b/requirements.txt @@ -31,3 +31,7 @@ slack_sdk snowflake-connector-python[pandas] sqlglot streamlit +protobuf +python-snappy +requests +prometheus-client diff --git a/tests/test_examples.py b/tests/test_examples.py index 63f4ff5..16fdef0 100644 --- a/tests/test_examples.py +++ b/tests/test_examples.py @@ -8,232 +8,261 @@ import importlib.util import os import sys +from unittest.mock import MagicMock, patch + import pandas as pd import pytest -from unittest.mock import patch, MagicMock -import requests class TestExampleIngests: """Test all example ingest functions.""" - + def load_ingest_function(self, example_path: str, module_name: str = "ingest"): """Load an ingest function from a Python file.""" spec = importlib.util.spec_from_file_location(module_name, example_path) if spec is None or spec.loader is None: raise ImportError(f"Could not load module from {example_path}") - + module = importlib.util.module_from_spec(spec) sys.modules[module_name] = module spec.loader.exec_module(module) - + return getattr(module, "ingest") - + def validate_dataframe(self, df: pd.DataFrame, example_name: str): """Validate that a DataFrame has the expected structure.""" assert isinstance(df, pd.DataFrame), f"{example_name}: Expected DataFrame, got {type(df)}" assert not df.empty, f"{example_name}: DataFrame should not be empty" - + # Check required columns required_columns = {"metric_timestamp", "metric_name", "metric_value"} actual_columns = set(df.columns) missing_columns = required_columns - actual_columns assert not missing_columns, f"{example_name}: Missing columns: {missing_columns}" - + # Check data types - be flexible with timestamp types as examples might return strings timestamp_col = df["metric_timestamp"] is_datetime = pd.api.types.is_datetime64_any_dtype(timestamp_col) is_string = timestamp_col.dtype == "object" - assert is_datetime or is_string, f"{example_name}: metric_timestamp should be datetime or string (got {timestamp_col.dtype})" - - assert df["metric_name"].dtype == "object", f"{example_name}: metric_name should be string/object" - assert pd.api.types.is_numeric_dtype(df["metric_value"]), f"{example_name}: metric_value should be numeric" - + assert ( + is_datetime or is_string + ), f"{example_name}: metric_timestamp should be datetime or string (got {timestamp_col.dtype})" + + assert ( + df["metric_name"].dtype == "object" + ), f"{example_name}: metric_name should be string/object" + assert pd.api.types.is_numeric_dtype( + df["metric_value"] + ), f"{example_name}: metric_value should be numeric" + # Check no null values in required columns for col in required_columns: - null_count = df[col].isnull().sum() + null_count = df[col].isnull().sum() assert null_count == 0, f"{example_name}: Column '{col}' has {null_count} null values" - - - @patch('requests.get') + @patch("requests.get") def test_bitcoin_price_ingest_mocked(self, mock_get): """Test bitcoin_price with mocked API call.""" # Mock the API response mock_response = MagicMock() mock_response.json.return_value = { "bpi": {"USD": {"rate_float": 45000.0}}, - "time": {"updatedISO": "2024-01-01T12:00:00+00:00"} + "time": {"updatedISO": "2024-01-01T12:00:00+00:00"}, } mock_get.return_value = mock_response - + ingest_fn = self.load_ingest_function("metrics/examples/bitcoin_price/bitcoin_price.py") df = ingest_fn() self.validate_dataframe(df, "bitcoin_price_mocked") - + @pytest.mark.skipif("CI" in os.environ, reason="Requires internet access") def test_hackernews_ingest(self): """Test hackernews example ingest function.""" ingest_fn = self.load_ingest_function("metrics/examples/hackernews/hackernews.py") df = ingest_fn(top_n=5) # Use smaller number for faster testing self.validate_dataframe(df, "hackernews") - + # Check that we get expected metrics - expected_metrics = {"hn_top_5_min_score", "hn_top_5_max_score", "hn_top_5_avg_score", "hn_top_5_total_score"} + expected_metrics = { + "hn_top_5_min_score", + "hn_top_5_max_score", + "hn_top_5_avg_score", + "hn_top_5_total_score", + } actual_metrics = set(df["metric_name"]) - assert expected_metrics == actual_metrics, f"Expected {expected_metrics}, got {actual_metrics}" - - @patch('requests.get') + assert ( + expected_metrics == actual_metrics + ), f"Expected {expected_metrics}, got {actual_metrics}" + + @patch("requests.get") def test_hackernews_ingest_mocked(self, mock_get): """Test hackernews with mocked API calls.""" + # Mock the top stories response def mock_requests_get(url): mock_response = MagicMock() mock_response.status_code = 200 - + if "topstories.json" in url: mock_response.json.return_value = [1, 2, 3] else: # individual story mock_response.json.return_value = {"score": 100} - + return mock_response - + mock_get.side_effect = mock_requests_get - - ingest_fn = self.load_ingest_function("metrics/examples/hackernews/hackernews.py") + + ingest_fn = self.load_ingest_function("metrics/examples/hackernews/hackernews.py") df = ingest_fn(top_n=3) self.validate_dataframe(df, "hackernews_mocked") - + @pytest.mark.skipif("CI" in os.environ, reason="Requires internet access") def test_weather_ingest(self): - """Test weather example ingest function.""" + """Test weather example ingest function.""" ingest_fn = self.load_ingest_function("metrics/examples/weather/ingest_weather.py") df = ingest_fn() self.validate_dataframe(df, "weather") - + # Should have metrics for multiple cities and weather parameters assert len(df) > 10, "Weather should return metrics for multiple cities and parameters" - + # Check that metric names start with city names - city_prefixes = ["dublin", "athens", "london", "berlin", "paris", "madrid", "new_york", "los_angeles", "sydney", "tokyo", "beijing", "cape_town"] + city_prefixes = [ + "dublin", + "athens", + "london", + "berlin", + "paris", + "madrid", + "new_york", + "los_angeles", + "sydney", + "tokyo", + "beijing", + "cape_town", + ] metric_names = df["metric_name"].tolist() - has_expected_cities = any(any(name.startswith(f"temp_{city}") for city in city_prefixes) for name in metric_names) + has_expected_cities = any( + any(name.startswith(f"temp_{city}") for city in city_prefixes) for name in metric_names + ) assert has_expected_cities, "Should have metrics for expected cities" - + @pytest.mark.skipif("CI" in os.environ, reason="Requires internet access") def test_coindesk_ingest(self): """Test coindesk example ingest function.""" ingest_fn = self.load_ingest_function("metrics/examples/coindesk/coindesk.py") - df = ingest_fn() + df = ingest_fn() self.validate_dataframe(df, "coindesk") - + # Check that all metric names contain CURRENT_HOUR (as per the filtering logic) - assert all("CURRENT_HOUR" in name for name in df["metric_name"]), "All metrics should contain CURRENT_HOUR" - - @pytest.mark.skipif("CI" in os.environ, reason="Requires internet access") + assert all( + "CURRENT_HOUR" in name for name in df["metric_name"] + ), "All metrics should contain CURRENT_HOUR" + + @pytest.mark.skipif("CI" in os.environ, reason="Requires internet access") def test_iss_location_ingest(self): """Test ISS location example ingest function.""" ingest_fn = self.load_ingest_function("metrics/examples/iss_location/iss_location.py") df = ingest_fn() self.validate_dataframe(df, "iss_location") - + # Should have latitude and longitude metrics metric_names = set(df["metric_name"]) assert "iss_latitude" in metric_names assert "iss_longitude" in metric_names - + @pytest.mark.skipif("CI" in os.environ, reason="Requires internet access") def test_earthquake_ingest(self): """Test earthquake example ingest function.""" ingest_fn = self.load_ingest_function("metrics/examples/earthquake/earthquake.py") df = ingest_fn() self.validate_dataframe(df, "earthquake") - + @pytest.mark.skipif("CI" in os.environ, reason="Requires internet access") def test_currency_ingest(self): """Test currency example ingest function.""" ingest_fn = self.load_ingest_function("metrics/examples/currency/currency.py") df = ingest_fn() self.validate_dataframe(df, "currency") - + def test_python_simple_ingest(self): """Test the simple Python example ingest function.""" - ingest_fn = self.load_ingest_function("metrics/examples/python/python_ingest_simple/ingest.py") + ingest_fn = self.load_ingest_function( + "metrics/examples/python/python_ingest_simple/ingest.py" + ) df = ingest_fn() self.validate_dataframe(df, "python_simple") - + # This should be a simple example that doesn't require external APIs # Check for expected simple metrics assert len(df) >= 1, "Simple Python example should return at least 1 metric" - + @pytest.mark.skipif("CI" in os.environ, reason="May require API keys or external services") def test_github_ingest(self): """Test GitHub example ingest function.""" ingest_fn = self.load_ingest_function("metrics/examples/github/github.py") df = ingest_fn() self.validate_dataframe(df, "github") - + @pytest.mark.skipif("CI" in os.environ, reason="May require API keys or external services") def test_yfinance_ingest(self): """Test yfinance example ingest function.""" ingest_fn = self.load_ingest_function("metrics/examples/yfinance/yfinance.py") df = ingest_fn() self.validate_dataframe(df, "yfinance") - + @pytest.mark.skipif("CI" in os.environ, reason="May require API keys or external services") def test_prometheus_ingest(self): """Test Prometheus example ingest function.""" ingest_fn = self.load_ingest_function("metrics/examples/prometheus/prometheus.py") df = ingest_fn() self.validate_dataframe(df, "prometheus") - + @pytest.mark.skipif("CI" in os.environ, reason="May require external services") def test_netdata_ingest(self): """Test Netdata example ingest function.""" ingest_fn = self.load_ingest_function("metrics/examples/netdata/netdata.py") df = ingest_fn() self.validate_dataframe(df, "netdata") - + @pytest.mark.skipif("CI" in os.environ, reason="May require external services") def test_netdata_httpcheck_ingest(self): """Test Netdata HTTP check example ingest function.""" - ingest_fn = self.load_ingest_function("metrics/examples/netdata_httpcheck/netdata_httpcheck.py") + ingest_fn = self.load_ingest_function( + "metrics/examples/netdata_httpcheck/netdata_httpcheck.py" + ) df = ingest_fn() self.validate_dataframe(df, "netdata_httpcheck") - - @pytest.mark.skipif("CI" in os.environ, reason="May require API keys") def test_eirgrid_ingest(self): """Test EirGrid example ingest function.""" ingest_fn = self.load_ingest_function("metrics/examples/eirgrid/eirgrid.py") df = ingest_fn() self.validate_dataframe(df, "eirgrid") - - class TestExampleIntegration: """Integration tests using the run_example.py script.""" - + def test_run_example_script_exists(self): """Test that the run_example.py script exists and can be imported.""" script_path = "scripts/examples/run_example.py" assert os.path.exists(script_path), f"run_example.py script should exist at {script_path}" - + # Test that we can import the main functions spec = importlib.util.spec_from_file_location("run_example", script_path) assert spec is not None and spec.loader is not None - + module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) - + # Check that required functions exist assert hasattr(module, "run_example"), "Should have run_example function" assert hasattr(module, "list_examples"), "Should have list_examples function" assert hasattr(module, "main"), "Should have main function" - - @patch('anomstack.config.get_specs') + + @patch("anomstack.config.get_specs") def test_list_examples_function(self, mock_get_specs): """Test the list_examples function with mocked specs.""" # Mock some example specs @@ -243,15 +272,15 @@ def test_list_examples_function(self, mock_get_specs): "example_sql": {"ingest_sql": "SELECT 1 as value"}, } mock_get_specs.return_value = mock_specs - - # Import and test the function + + # Import and test the function script_path = "scripts/examples/run_example.py" spec = importlib.util.spec_from_file_location("run_example", script_path) assert spec is not None and spec.loader is not None - + module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) - + # Should return 0 for success result = module.list_examples() - assert result == 0, "list_examples should return 0 for success" \ No newline at end of file + assert result == 0, "list_examples should return 0 for success"