flowfish/docs/change-detection-worker.md

Change Detection Worker

Overview

The Change Detection Worker is a standalone, scalable microservice that detects infrastructure and behavioral changes in Kubernetes clusters using a hybrid detection approach:

• K8s API Detector: Polls Kubernetes API for infrastructure state changes (replicas, configs, images, resources, env, services, network policies, ingresses, routes)
• eBPF Event Detector: Analyzes eBPF events from ClickHouse for behavioral changes (connections, ports) and anomalies (traffic, DNS, process, error)

ClickHouse-only Architecture: All change events are stored exclusively in ClickHouse. PostgreSQL is used only for analysis metadata and workflow state.

Architecture

┌─────────────────────────────────────────────────────────────────────────────┐
│                         Change Detection Worker                              │
│                        (Separate Pod/Deployment)                            │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│  ┌─────────────────────┐          ┌─────────────────────┐                  │
│  │   K8s Detector      │          │   eBPF Detector     │                  │
│  │  (Infrastructure)   │          │   (Behavioral)      │                  │
│  │                     │          │                     │                  │
│  │  • replica_changed  │          │  • connection_added │                  │
│  │  • config_changed   │          │  • connection_removed│                 │
│  │  • image_changed    │          │  • port_changed     │                  │
│  │  • resource_changed │          │  • traffic_anomaly  │                  │
│  │  • env_changed      │          │  • dns_anomaly      │                  │
│  │  • spec_changed     │          │  • process_anomaly  │                  │
│  │  • label_changed    │          │  • error_anomaly    │                  │
│  │  • service_*        │          │                     │                  │
│  │  • network_policy_* │          │                     │                  │
│  │  • ingress_*        │          │                     │                  │
│  │  • route_*          │          │                     │                  │
│  └──────────┬──────────┘          └──────────┬──────────┘                  │
│             │                                │                              │
│             └───────────┬────────────────────┘                              │
│                         │                                                   │
│                         ▼                                                   │
│                ┌─────────────────┐                                         │
│                │  Change Merger  │                                         │
│                │  (All Changes)  │                                         │
│                └────────┬────────┘                                         │
│                         │                                                   │
└─────────────────────────┼───────────────────────────────────────────────────┘
                          │
        ┌─────────────────┼─────────────────┐
        │                 │                 │
        ▼                 ▼                 ▼
┌──────────────┐   ┌──────────────┐   ┌──────────────┐
│   K8s API    │   │  ClickHouse  │   │    Redis     │
│   (Source)   │   │  (Storage)   │   │   (Leader    │
│              │   │              │   │   Election)  │
└──────────────┘   └──────────────┘   └──────────────┘

Detection Methods

K8s API Detector (Infrastructure Changes)

The K8s Detector polls the Kubernetes API every detection cycle (default: 60 seconds) and compares the current state with the stored state in PostgreSQL.

Change Type	Description	Detection Method
replica_changed	Deployment/StatefulSet replica count changed	Compare K8s API vs PostgreSQL
config_changed	ConfigMap/Secret content changed	.data hash comparison (OpenShift-safe)
image_changed	Container image updated	Layered spec hash per container
resource_changed	CPU/memory requests/limits changed	Per-container resource diff
env_changed	Environment variables changed	Env hash (values not logged)
spec_changed	Pod spec catch-all (probes, volumes)	spec_hash comparison
label_changed	Pod/Service labels modified	Label set comparison
service_port_changed	Service port/targetPort/protocol	Port tuple comparison
service_selector_changed	Service selector changed	Selector map diff
service_type_changed	Service type changed	Direct comparison
service_added / removed	Service lifecycle	K8s vs PostgreSQL
network_policy_*	NetworkPolicy lifecycle + spec	Spec hash comparison
ingress_*	Ingress lifecycle + spec	Spec hash (status ignored)
route_*	OpenShift Route lifecycle + spec	Spec hash (graceful fail)

eBPF Event Detector (Behavioral Changes + Anomalies)

The eBPF Detector queries ClickHouse tables (network_flows, dns_queries, process_events) for behavioral changes and statistical anomalies.

Change Type	Description	Detection Method
connection_added	New connection appeared	Compare baseline vs current window
connection_removed	Connection disappeared	Compare baseline vs current window
port_changed	Same source/dest, different port	Track port changes per pair
traffic_anomaly	Volume 3x+ or latency 2.5x+	Statistical threshold
dns_anomaly	New DNS domains, NXDOMAIN spikes	Baseline domain diff
process_anomaly	New/suspicious process execution	Process allowlist
error_anomaly	New error types or 2x+ rate	Error rate threshold

Detection Strategies

Users can select a detection strategy when creating an analysis:

Baseline Strategy (Default)

Captures connections during the first N minutes as baseline, then detects deviations.

|-------- baseline (first 10 min) --------|
                                          |---- current (last 5 min) ----|

Best for: Long-running analyses, drift detection

Rolling Window Strategy

Continuously compares recent time window vs previous window for real-time detection.

|---- previous (5 min ago) ----|---- current (last 5 min) ----|

Best for: Continuous monitoring, alerting

Run Comparison Strategy

Compares current run against previous run. Ideal for deployment validation.

|-------- Run N-1 --------|
                          |-------- Run N (current) --------|

Best for: Deployment validation, canary deployments, A/B testing

Configuration

Environment Variables

Variable	Default	Description
CHANGE_DETECTION_ENABLED	true	Enable/disable detection
CHANGE_DETECTION_INTERVAL	60	Detection interval (seconds)
CHANGE_DETECTION_LOOKBACK_MINUTES	5	How far back to look for changes
LEADER_ELECTION_ENABLED	false	Enable leader election for HA
WORKER_INSTANCE_ID	auto	Unique instance identifier
CIRCUIT_BREAKER_THRESHOLD	3	Failures before circuit opens
CIRCUIT_BREAKER_RESET	300	Seconds before circuit resets

Per-Analysis Configuration

Each analysis can have individual change detection settings:

Field	Default	Description
change_detection_enabled	true	Enable/disable for this analysis
change_detection_strategy	baseline	Detection strategy to use
change_detection_types	["all"]	Types of changes to track

Data Flow

┌────────────────────────────────────────────────────────────────────────┐
│                        Detection Cycle (60s)                            │
└────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌────────────────────────────────────────────────────────────────────────┐
│                    1. Get Active Analyses                               │
│                       (PostgreSQL)                                      │
│                                                                        │
│    SELECT id, cluster_ids, change_detection_strategy,                  │
│           change_detection_types FROM analyses                          │
│    WHERE status = 'running' AND change_detection_enabled = true        │
└────────────────────────────────────────────────────────────────────────┘
                                    │
                    ┌───────────────┴───────────────┐
                    │                               │
                    ▼                               ▼
┌──────────────────────────────┐    ┌──────────────────────────────┐
│   2a. K8s API Detection      │    │   2b. eBPF Event Detection   │
│       (Per Cluster)          │    │       (Per Analysis)         │
│                              │    │                              │
│   - Poll K8s deployments     │    │   - Query ClickHouse         │
│   - Compare with PostgreSQL  │    │     network_flows            │
│   - Detect replica changes   │    │   - Apply strategy           │
│   - Update stored state      │    │   - Compare time windows     │
└──────────────────────────────┘    └──────────────────────────────┘
                    │                               │
                    └───────────────┬───────────────┘
                                    │
                                    ▼
┌────────────────────────────────────────────────────────────────────────┐
│                     3. Merge All Changes                                │
│                                                                        │
│    - Filter by enabled types                                           │
│    - Assess risk levels                                                │
│    - Enrich with metadata                                              │
└────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌────────────────────────────────────────────────────────────────────────┐
│                   4. Write to ClickHouse                                │
│                                                                        │
│    Via RabbitMQ → Timeseries Writer → ClickHouse change_events         │
│    (Fallback: Direct ClickHouse write)                                 │
└────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌────────────────────────────────────────────────────────────────────────┐
│                  5. Notify Critical Changes                             │
│                                                                        │
│    POST to Backend /ws/broadcast for WebSocket notification            │
└────────────────────────────────────────────────────────────────────────┘

ClickHouse Queries

Baseline Connections Query

-- Get baseline connections (first 10 minutes of analysis)
SELECT DISTINCT 
    source_pod, dest_pod, dest_port, protocol
FROM network_flows
WHERE analysis_id = {analysis_id}
  AND timestamp BETWEEN {analysis_start} AND {analysis_start} + INTERVAL 10 MINUTE

Current Connections Query

-- Get current connections (last 5 minutes)
SELECT DISTINCT 
    source_pod, dest_pod, dest_port, protocol
FROM network_flows
WHERE analysis_id = {analysis_id}
  AND timestamp > now() - INTERVAL 5 MINUTE

Change Detection Result

added = current_connections - baseline_connections    → connection_added
removed = baseline_connections - current_connections  → connection_removed

API Endpoints

Health Check

GET /health
GET /healthz

Returns:

{
  "status": "healthy",
  "instance_id": "abc123",
  "is_leader": true,
  "detection_cycles": 42,
  "last_detection": "2025-01-08T10:30:00Z"
}

Readiness Check

GET /ready
GET /readyz

Prometheus Metrics

GET /metrics

Metrics:

• flowfish_change_worker_detection_cycles_total
• flowfish_change_worker_errors_total
• flowfish_change_worker_is_leader
• flowfish_change_worker_running
• flowfish_change_worker_circuits_open
• flowfish_change_worker_k8s_detections_total
• flowfish_change_worker_ebpf_detections_total

Manual Trigger

POST /trigger/{analysis_id}

Manually trigger change detection for a specific analysis.

Deployment

Single Instance (Simple Mode)

spec:
  replicas: 1
  env:
  - name: LEADER_ELECTION_ENABLED
    value: "false"

Multiple Instances (High Availability)

spec:
  replicas: 3
  env:
  - name: LEADER_ELECTION_ENABLED
    value: "true"

Troubleshooting

No changes detected

1. Check analysis settings: Ensure change_detection_enabled = true
2. Check enabled types: Verify change_detection_types includes expected types
3. Check eBPF data: Query ClickHouse network_flows for the analysis
4. Check K8s connectivity: Verify cluster API is accessible

K8s detection not working

1. Verify cluster configuration in PostgreSQL clusters table
2. Check API server URL and credentials
3. Review worker logs for K8s API errors

eBPF detection not working

1. Ensure ClickHouse is accessible
2. Verify network_flows table has data for the analysis
3. Check strategy time windows (baseline may not be populated yet)

Changes not appearing in UI

1. Check RabbitMQ connectivity
2. Verify timeseries-writer is consuming messages
3. Query ClickHouse change_events table directly

Data Retention

Change events follow analysis lifecycle-based retention:

• No TTL: Change events are never automatically deleted based on time
• Analysis Cascade: When an analysis is deleted, all change events are removed:
  • ClickHouse: DELETE FROM change_events WHERE analysis_id = ?

---

Version: 3.0.0
Last Updated: January 2026
Architecture: Hybrid K8s API + eBPF Detection, ClickHouse-only Storage