gsadmin/flowfish
1
0
Fork 0
mirror of https://github.com/taylanbakircioglu/flowfish.git synced 2026-09-25 11:52:05 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
flowfish/docs/change-detection-worker.md
T
taylanbakircioglu d7ca50b387 feat: Flowfish - eBPF-powered multi-cluster Kubernetes observability platform 
Multi-cluster dependency mapping, real-time network monitoring,
impact analysis, and CI/CD integration capabilities.

Made-with: Cursor
2026-03-29 21:43:20 +03:00

354 lines
17 KiB
Markdown
Raw Permalink Blame History

# 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
```sql
-- 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
```sql
-- 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
```bash
GET /health
GET /healthz
```
Returns:
```json
{
"status": "healthy",
"instance_id": "abc123",
"is_leader": true,
"detection_cycles": 42,
"last_detection": "2025-01-08T10:30:00Z"
}
```
### Readiness Check
```bash
GET /ready
GET /readyz
```
### Prometheus Metrics
```bash
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
```bash
POST /trigger/{analysis_id}
```
Manually trigger change detection for a specific analysis.
## Deployment
### Single Instance (Simple Mode)
```yaml
spec:
replicas: 1
env:
- name: LEADER_ELECTION_ENABLED
value: "false"
```
### Multiple Instances (High Availability)
```yaml
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
Reference in New Issue View Git Blame Copy Permalink