shadowhunter_metrics.py

#!/usr/bin/env python3
"""
ShadowHunter — Prometheus Metrics for Monitoring

Exposes counters and histograms for:
- HTTP requests (total, by path, status)
- Scans, enrichments, ingest operations
- Blockchain analyses, Neo4j operations
- Alert counts by severity

Scrape endpoint: GET /metrics (Prometheus text format).
"""

import time
import logging
from typing import Optional

logger = logging.getLogger(__name__)

try:
    from prometheus_client import (
        Counter,
        Histogram,
        Gauge,
        generate_latest,
        CONTENT_TYPE_LATEST,
        REGISTRY,
    )
    PROMETHEUS_AVAILABLE = True
except ImportError:
    PROMETHEUS_AVAILABLE = False
    Counter = None
    Histogram = None
    Gauge = None
    generate_latest = None
    CONTENT_TYPE_LATEST = None
    REGISTRY = None


# ---------------------------------------------------------------------------
# HTTP / API
# ---------------------------------------------------------------------------
REQUEST_COUNT: Optional[Counter] = None
REQUEST_LATENCY: Optional[Histogram] = None

# ---------------------------------------------------------------------------
# Operations
# ---------------------------------------------------------------------------
SCAN_COUNT: Optional[Counter] = None
ENRICHMENT_COUNT: Optional[Counter] = None
ENRICHMENT_LATENCY: Optional[Histogram] = None
INGEST_FETCH_COUNT: Optional[Counter] = None
INGEST_STIX_COUNT: Optional[Counter] = None
INGEST_DB_WRITE_COUNT: Optional[Counter] = None
BLOCKCHAIN_ANALYSIS_COUNT: Optional[Counter] = None
BLOCKCHAIN_ANALYSIS_LATENCY: Optional[Histogram] = None
NEO4J_OPERATIONS_COUNT: Optional[Counter] = None
ALERTS_CREATED_COUNT: Optional[Counter] = None

# ---------------------------------------------------------------------------
# Gauges (current state)
# ---------------------------------------------------------------------------
ACTIVE_ALERTS_GAUGE: Optional[Gauge] = None
MONITORED_DOMAINS_GAUGE: Optional[Gauge] = None


def _init_metrics() -> None:
    """Initialize Prometheus metrics (idempotent)."""
    global REQUEST_COUNT, REQUEST_LATENCY, SCAN_COUNT, ENRICHMENT_COUNT
    global ENRICHMENT_LATENCY, INGEST_FETCH_COUNT, INGEST_STIX_COUNT, INGEST_DB_WRITE_COUNT
    global BLOCKCHAIN_ANALYSIS_COUNT, BLOCKCHAIN_ANALYSIS_LATENCY, NEO4J_OPERATIONS_COUNT
    global ALERTS_CREATED_COUNT, ACTIVE_ALERTS_GAUGE, MONITORED_DOMAINS_GAUGE
    if not PROMETHEUS_AVAILABLE:
        return
    if REQUEST_COUNT is not None:
        return

    REQUEST_COUNT = Counter(
        "shadowhunter_http_requests_total",
        "Total HTTP requests",
        ["method", "path", "status"],
    )
    REQUEST_LATENCY = Histogram(
        "shadowhunter_http_request_duration_seconds",
        "HTTP request latency",
        ["method", "path"],
        buckets=(0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0),
    )
    SCAN_COUNT = Counter(
        "shadowhunter_scans_total",
        "Total scan operations",
        ["module"],
    )
    ENRICHMENT_COUNT = Counter(
        "shadowhunter_enrichments_total",
        "Total IOC enrichments",
        ["source", "ioc_type"],
    )
    ENRICHMENT_LATENCY = Histogram(
        "shadowhunter_enrichment_duration_seconds",
        "Enrichment API latency",
        ["source"],
        buckets=(0.1, 0.25, 0.5, 1.0, 2.5, 5.0),
    )
    INGEST_FETCH_COUNT = Counter(
        "shadowhunter_ingest_fetches_total",
        "Ingestion pipeline fetches",
        ["source", "status"],
    )
    INGEST_STIX_COUNT = Counter(
        "shadowhunter_ingest_stix_objects_total",
        "STIX objects produced by ingestion",
        ["type"],
    )
    INGEST_DB_WRITE_COUNT = Counter(
        "shadowhunter_ingest_db_writes_total",
        "Neo4j writes from ingestion",
        ["label"],
    )
    BLOCKCHAIN_ANALYSIS_COUNT = Counter(
        "shadowhunter_blockchain_analyses_total",
        "Blockchain wallet analyses",
        ["chain", "status"],
    )
    BLOCKCHAIN_ANALYSIS_LATENCY = Histogram(
        "shadowhunter_blockchain_analysis_duration_seconds",
        "Blockchain analysis latency",
        ["chain"],
        buckets=(0.5, 1.0, 2.0, 5.0, 10.0),
    )
    NEO4J_OPERATIONS_COUNT = Counter(
        "shadowhunter_neo4j_operations_total",
        "Neo4j graph operations",
        ["operation", "label"],
    )
    ALERTS_CREATED_COUNT = Counter(
        "shadowhunter_alerts_created_total",
        "Alerts created",
        ["severity"],
    )
    ACTIVE_ALERTS_GAUGE = Gauge(
        "shadowhunter_alerts_active",
        "Current number of active alerts",
    )
    MONITORED_DOMAINS_GAUGE = Gauge(
        "shadowhunter_domains_monitored",
        "Current number of monitored domains",
    )


def init_metrics() -> None:
    """Public init: ensure metrics are registered."""
    _init_metrics()


def get_metrics_output() -> tuple[bytes, str]:
    """
    Return (body, content_type) for Prometheus scrape.
    If prometheus_client not installed, returns (b"", "text/plain").
    """
    _init_metrics()
    if not PROMETHEUS_AVAILABLE or generate_latest is None:
        return b"# Prometheus client not installed\n", "text/plain; charset=utf-8"
    return generate_latest(REGISTRY), CONTENT_TYPE_LATEST


# ---------------------------------------------------------------------------
# Recording helpers
# ---------------------------------------------------------------------------

def record_request(method: str, path: str, status: int, duration_seconds: float) -> None:
    """Record an HTTP request."""
    _init_metrics()
    if REQUEST_COUNT is not None:
        REQUEST_COUNT.labels(method=method, path=path, status=status).inc()
    if REQUEST_LATENCY is not None:
        REQUEST_LATENCY.labels(method=method, path=path).observe(duration_seconds)


def record_scan(module: str) -> None:
    """Record a scan operation."""
    _init_metrics()
    if SCAN_COUNT is not None:
        SCAN_COUNT.labels(module=module).inc()


def record_enrichment(source: str, ioc_type: str, duration_seconds: Optional[float] = None) -> None:
    """Record an enrichment call."""
    _init_metrics()
    if ENRICHMENT_COUNT is not None:
        ENRICHMENT_COUNT.labels(source=source, ioc_type=ioc_type).inc()
    if duration_seconds is not None and ENRICHMENT_LATENCY is not None:
        ENRICHMENT_LATENCY.labels(source=source).observe(duration_seconds)


def record_ingest_fetch(source: str, status: str) -> None:
    """Record an ingestion fetch (e.g. I2P, Telegram)."""
    _init_metrics()
    if INGEST_FETCH_COUNT is not None:
        INGEST_FETCH_COUNT.labels(source=source, status=status).inc()


def record_ingest_stix(obj_type: str, count: int = 1) -> None:
    """Record STIX objects produced."""
    _init_metrics()
    if INGEST_STIX_COUNT is not None:
        INGEST_STIX_COUNT.labels(type=obj_type).inc(count)


def record_ingest_db_write(label: str, count: int = 1) -> None:
    """Record Neo4j writes from ingestion."""
    _init_metrics()
    if INGEST_DB_WRITE_COUNT is not None:
        INGEST_DB_WRITE_COUNT.labels(label=label).inc(count)


def record_blockchain_analysis(chain: str, status: str, duration_seconds: Optional[float] = None) -> None:
    """Record a blockchain wallet analysis."""
    _init_metrics()
    if BLOCKCHAIN_ANALYSIS_COUNT is not None:
        BLOCKCHAIN_ANALYSIS_COUNT.labels(chain=chain, status=status).inc()
    if duration_seconds is not None and BLOCKCHAIN_ANALYSIS_LATENCY is not None:
        BLOCKCHAIN_ANALYSIS_LATENCY.labels(chain=chain).observe(duration_seconds)


def record_neo4j_operation(operation: str, label: str) -> None:
    """Record a Neo4j operation (e.g. merge, create)."""
    _init_metrics()
    if NEO4J_OPERATIONS_COUNT is not None:
        NEO4J_OPERATIONS_COUNT.labels(operation=operation, label=label).inc()


def record_alert_created(severity: str) -> None:
    """Record an alert creation."""
    _init_metrics()
    if ALERTS_CREATED_COUNT is not None:
        ALERTS_CREATED_COUNT.labels(severity=severity).inc()


def set_active_alerts(value: int) -> None:
    """Set current active alerts gauge."""
    _init_metrics()
    if ACTIVE_ALERTS_GAUGE is not None:
        ACTIVE_ALERTS_GAUGE.set(value)


def set_monitored_domains(value: int) -> None:
    """Set current monitored domains gauge."""
    _init_metrics()
    if MONITORED_DOMAINS_GAUGE is not None:
        MONITORED_DOMAINS_GAUGE.set(value)