mirror of
https://github.com/taylanbakircioglu/flowfish.git
synced 2026-09-21 01:53:27 +00:00
taylanbakircioglu
6e503368f7
- Grafana Beyla DaemonSet for kernel-level HTTP/gRPC/DNS capture (passive, zero application changes, W3C traceparent header propagation) - flowfish-l7-collector in-cluster bridge: OTLP receiver + buffered pull API - L7 Ingestion Service: K8s service-proxy poll → enrich → RabbitMQ - ClickHouse l7_http_flows / l7_grpc_flows / l7_dns_flows + APM RED MVs - Neo4j L7Workload nodes + SAME_WORKLOAD cross-cluster bridges - New pages: Service Map, Trace Explorer, APM Services List, APM Service Detail - Analysis Wizard now supports L4 / L7 / Both modes with HTTP/gRPC/DNS picks - Integration Hub gains L7 dependency summary + tree-summary integrations - Multi-Cluster Management: dual-agent install (Inspector Gadget L4 + Beyla L7), runtime OpenShift detection so SCCs auto-install with kubectl too - ServiceMap edge → Trace Explorer drill-down with virtual_trace_id correlation - Docs: new L7 architecture diagram, README L7 sections, 3 new screenshots
1919 lines
81 KiB
Python
1919 lines
81 KiB
Python
"""
|
|
Analysis management API endpoints
|
|
Sprint 5-6: Analysis Wizard & Communication Discovery
|
|
"""
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status, Request
|
|
from typing import List, Literal, Optional
|
|
from pydantic import BaseModel, Field
|
|
from datetime import datetime
|
|
import structlog
|
|
import json
|
|
import asyncio
|
|
|
|
from database.postgresql import database
|
|
from utils.jwt_utils import get_current_user, require_permissions
|
|
from grpc_clients.analysis_orchestrator_client import analysis_orchestrator_client
|
|
from services.activity_service import activity_service, ActivityService
|
|
|
|
logger = structlog.get_logger()
|
|
router = APIRouter()
|
|
|
|
|
|
def check_analysis_permission(current_user: dict, required_action: str) -> bool:
|
|
"""
|
|
Check if user has permission for analysis actions.
|
|
|
|
Permission mapping:
|
|
- analysis.create: Create new analyses
|
|
- analysis.start: Start/execute analyses
|
|
- analysis.stop: Stop running analyses
|
|
- analysis.delete: Delete analyses
|
|
- analysis.view: View analyses (read-only)
|
|
|
|
Super Admin, Admin, Platform Admin bypass all checks.
|
|
Viewer role should only have analysis.view permission.
|
|
"""
|
|
user_roles = current_user.get("roles", [])
|
|
|
|
# Super Admin, Admin, Platform Admin bypass all permission checks
|
|
admin_roles = ["super admin", "admin", "platform admin"]
|
|
if any(r.lower() in admin_roles for r in user_roles):
|
|
return True
|
|
|
|
# For non-admin users, check against role-based permissions
|
|
# Viewer role should NOT have start/stop/create/delete permissions
|
|
viewer_roles = ["viewer", "read-only", "readonly"]
|
|
is_viewer = any(r.lower() in viewer_roles for r in user_roles)
|
|
|
|
# Define what viewers CAN do (only view)
|
|
viewer_allowed_actions = ["view"]
|
|
|
|
if is_viewer and required_action not in viewer_allowed_actions:
|
|
return False
|
|
|
|
# For other roles (Operator, Analyst, etc.), allow the action
|
|
# In production, this should check the permissions table
|
|
return True
|
|
|
|
|
|
def require_analysis_permission(required_action: str):
|
|
"""Decorator to require specific analysis permission"""
|
|
def check(current_user: dict):
|
|
if not check_analysis_permission(current_user, required_action):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=f"Permission denied. Required permission: analysis.{required_action}"
|
|
)
|
|
return current_user
|
|
return check
|
|
|
|
|
|
def get_client_ip(request: Request) -> str:
|
|
"""Extract client IP from request headers"""
|
|
# Priority: X-Client-IP (from frontend) > X-Forwarded-For > X-Real-IP > client.host
|
|
if request.headers.get("X-Client-IP"):
|
|
return request.headers.get("X-Client-IP")
|
|
elif request.headers.get("X-Forwarded-For"):
|
|
return request.headers.get("X-Forwarded-For").split(",")[0].strip()
|
|
elif request.headers.get("X-Real-IP"):
|
|
return request.headers.get("X-Real-IP")
|
|
elif request.client:
|
|
return request.client.host
|
|
return "0.0.0.0"
|
|
|
|
# Request/Response Models
|
|
class ScopeConfig(BaseModel):
|
|
"""
|
|
Analysis scope configuration
|
|
|
|
Supports both single-cluster and multi-cluster analysis:
|
|
- cluster_id: Primary cluster (required)
|
|
- cluster_ids: List of all cluster IDs for multi-cluster (optional)
|
|
"""
|
|
cluster_id: int = Field(..., description="Primary cluster ID")
|
|
cluster_ids: Optional[List[int]] = Field(None, description="List of cluster IDs for multi-cluster analysis")
|
|
scope_type: str = Field(..., description="cluster, namespace, deployment, pod, label")
|
|
namespaces: Optional[List[str]] = None
|
|
deployments: Optional[List[str]] = None
|
|
pods: Optional[List[str]] = None
|
|
labels: Optional[dict] = None
|
|
|
|
# Per-cluster scope configuration for multi-cluster
|
|
# Format: { "cluster_id": { "namespaces": [...], "deployments": [...] } }
|
|
per_cluster_scope: Optional[dict] = Field(None, description="Per-cluster scope configuration")
|
|
|
|
# Exclusion filters (opt-in, empty = no exclusion)
|
|
# Uses fnmatch glob patterns: openshift-* matches openshift-monitoring, openshift-ingress, etc.
|
|
exclude_namespaces: Optional[List[str]] = Field(None, description="Namespace patterns to exclude (e.g. 'openshift-*', 'kube-system')")
|
|
exclude_pod_patterns: Optional[List[str]] = Field(None, description="Pod name patterns to exclude (e.g. 'calico-node-*')")
|
|
exclude_strategy: Optional[str] = Field('aggressive', description="'aggressive' drops if EITHER endpoint matches, 'conservative' drops only if BOTH match")
|
|
|
|
@staticmethod
|
|
def _strip_cluster_suffix(values: Optional[List[str]]) -> Optional[List[str]]:
|
|
"""Strip @clusterId suffix from values (e.g., 'namespace@4' -> 'namespace')"""
|
|
if not values:
|
|
return values
|
|
cleaned = []
|
|
for val in values:
|
|
if '@' in val:
|
|
# Use lastIndexOf in case value itself contains @
|
|
clean_val = val[:val.rfind('@')]
|
|
cleaned.append(clean_val)
|
|
else:
|
|
cleaned.append(val)
|
|
# Remove duplicates while preserving order
|
|
return list(dict.fromkeys(cleaned))
|
|
|
|
def __init__(self, **data):
|
|
# Clean values before validation
|
|
if 'namespaces' in data and data['namespaces']:
|
|
data['namespaces'] = self._strip_cluster_suffix(data['namespaces'])
|
|
if 'deployments' in data and data['deployments']:
|
|
data['deployments'] = self._strip_cluster_suffix(data['deployments'])
|
|
if 'pods' in data and data['pods']:
|
|
data['pods'] = self._strip_cluster_suffix(data['pods'])
|
|
# Also clean per_cluster_scope values
|
|
if 'per_cluster_scope' in data and data['per_cluster_scope']:
|
|
for cluster_id, scope in data['per_cluster_scope'].items():
|
|
if isinstance(scope, dict):
|
|
if 'namespaces' in scope:
|
|
scope['namespaces'] = self._strip_cluster_suffix(scope['namespaces'])
|
|
if 'deployments' in scope:
|
|
scope['deployments'] = self._strip_cluster_suffix(scope['deployments'])
|
|
if 'pods' in scope:
|
|
scope['pods'] = self._strip_cluster_suffix(scope['pods'])
|
|
super().__init__(**data)
|
|
|
|
class GadgetConfig(BaseModel):
|
|
"""eBPF Gadget module configuration"""
|
|
enabled_gadgets: List[str] = Field(
|
|
...,
|
|
description="List of gadget modules",
|
|
example=["network_traffic", "dns_queries", "tcp_connections"]
|
|
)
|
|
network_traffic: Optional[dict] = Field(None, description="Network traffic gadget config")
|
|
dns_queries: Optional[dict] = Field(None, description="DNS queries gadget config")
|
|
tcp_connections: Optional[dict] = Field(None, description="TCP connections gadget config")
|
|
process_events: Optional[dict] = Field(None, description="Process events gadget config")
|
|
syscall_tracking: Optional[dict] = Field(None, description="Syscall tracking gadget config")
|
|
file_access: Optional[dict] = Field(None, description="File access gadget config")
|
|
|
|
class TimeConfig(BaseModel):
|
|
"""Analysis time and profiling configuration with data sizing support"""
|
|
mode: str = Field(..., description="continuous, timed, time_range, periodic, baseline, recurring")
|
|
start_time: Optional[datetime] = None
|
|
end_time: Optional[datetime] = None
|
|
duration_seconds: Optional[int] = Field(None, ge=60, le=604800, description="Fixed duration in seconds (min 60s, max 7 days)")
|
|
periodic_interval: Optional[int] = Field(None, ge=60, le=86400, description="Interval in seconds for periodic mode (min 60s, max 24h)")
|
|
baseline_name: Optional[str] = Field(None, description="Baseline profile name")
|
|
|
|
# Recurring schedule fields (stored here for reference; authoritative values are in analyses columns)
|
|
schedule_expression: Optional[str] = Field(None, max_length=100, description="Cron expression for recurring mode")
|
|
schedule_duration_seconds: Optional[int] = Field(None, ge=60, le=86400, description="Per-run duration for recurring mode")
|
|
|
|
# Data sizing configuration
|
|
data_retention_policy: Optional[str] = Field(
|
|
'unlimited',
|
|
description="unlimited, stop_on_limit, rolling_window"
|
|
)
|
|
max_data_size_mb: Optional[int] = Field(
|
|
None,
|
|
ge=10, # Minimum 10 MB
|
|
le=10240, # Maximum 10 GB
|
|
description="Maximum data size in MB (10 MB - 10 GB, required for stop_on_limit and rolling_window)"
|
|
)
|
|
|
|
@classmethod
|
|
def validate_retention_config(cls, values):
|
|
"""Validate that max_data_size_mb is provided when retention policy requires it"""
|
|
policy = values.get('data_retention_policy', 'unlimited')
|
|
max_size = values.get('max_data_size_mb')
|
|
|
|
# Validate policy value
|
|
valid_policies = ['unlimited', 'stop_on_limit', 'rolling_window']
|
|
if policy and policy not in valid_policies:
|
|
# Don't raise error, just default to unlimited for backward compatibility
|
|
values['data_retention_policy'] = 'unlimited'
|
|
return values
|
|
|
|
# If policy requires size limit, ensure max_data_size_mb is provided
|
|
if policy in ['stop_on_limit', 'rolling_window'] and not max_size:
|
|
# Default to 500 MB if not provided
|
|
values['max_data_size_mb'] = 500
|
|
|
|
return values
|
|
|
|
def __init__(self, **data):
|
|
# Apply validation before pydantic validation
|
|
data = self.validate_retention_config(data)
|
|
super().__init__(**data)
|
|
|
|
class OutputConfig(BaseModel):
|
|
"""Analysis output and integration configuration"""
|
|
enable_dashboard: bool = True
|
|
enable_llm_analysis: bool = False
|
|
llm_provider: Optional[str] = Field(None, description="openai, anthropic, azure")
|
|
llm_model: Optional[str] = None
|
|
enable_alarms: bool = False
|
|
alarm_thresholds: Optional[dict] = None
|
|
enable_webhooks: bool = False
|
|
webhook_urls: Optional[List[str]] = None
|
|
export_format: Optional[List[str]] = Field(None, description="csv, json, graph_json")
|
|
|
|
class AnalysisCreateRequest(BaseModel):
|
|
"""Complete analysis creation request"""
|
|
name: str = Field(..., min_length=3, max_length=100)
|
|
description: Optional[str] = None
|
|
scope: ScopeConfig
|
|
gadgets: GadgetConfig
|
|
time_config: TimeConfig
|
|
output: OutputConfig
|
|
change_detection_enabled: bool = Field(
|
|
default=True,
|
|
description="Enable change detection for this analysis. When enabled, infrastructure changes are tracked."
|
|
)
|
|
change_detection_strategy: str = Field(
|
|
default="baseline",
|
|
description="Detection strategy: 'baseline' (compare against initial state), 'rolling_window' (compare recent periods), 'run_comparison' (compare between runs)"
|
|
)
|
|
change_detection_types: List[str] = Field(
|
|
default=["all"],
|
|
description="Change types to track: ['all'] or specific types like ['replica_changed', 'connection_added', 'port_changed']"
|
|
)
|
|
analysis_level: Literal["l4", "l7", "both"] = Field(
|
|
default="l4",
|
|
description="Network capture depth: 'l4' (Inspector Gadget), 'l7' (Beyla/Kubeshark), or 'both'",
|
|
)
|
|
l7_config: Optional[dict] = Field(
|
|
default=None,
|
|
description="Optional L7 capture configuration (protocols, sampling, filters) when analysis_level is l7 or both",
|
|
)
|
|
|
|
class AnalysisResponse(BaseModel):
|
|
"""Analysis response model with multi-cluster support"""
|
|
id: int
|
|
name: str = "unnamed"
|
|
description: Optional[str] = None
|
|
cluster_id: int # Primary cluster
|
|
cluster_name: Optional[str] = None # Primary cluster name
|
|
cluster_ids: Optional[List[int]] = None # All clusters for multi-cluster
|
|
is_multi_cluster: bool = False
|
|
status: str = "unknown"
|
|
scope_type: str = "cluster"
|
|
scope_config: dict = {}
|
|
gadget_config: dict = {}
|
|
time_config: dict = {}
|
|
output_config: dict = {}
|
|
change_detection_enabled: bool = True
|
|
change_detection_strategy: str = "baseline"
|
|
change_detection_types: List[str] = ["all"]
|
|
created_at: datetime
|
|
updated_at: Optional[datetime]
|
|
started_at: Optional[datetime] = None
|
|
stopped_at: Optional[datetime] = None
|
|
created_by: int
|
|
is_scheduled: bool = False
|
|
schedule_expression: Optional[str] = None
|
|
schedule_duration_seconds: Optional[int] = None
|
|
next_run_at: Optional[datetime] = None
|
|
last_run_at: Optional[datetime] = None
|
|
schedule_run_count: int = 0
|
|
max_scheduled_runs: Optional[int] = None
|
|
analysis_level: str = "l4"
|
|
l7_config: Optional[dict] = None
|
|
|
|
class AnalysisRunResponse(BaseModel):
|
|
"""Analysis run response"""
|
|
id: int
|
|
analysis_id: int
|
|
status: str
|
|
start_time: Optional[datetime]
|
|
end_time: Optional[datetime]
|
|
events_collected: int
|
|
communications_discovered: int
|
|
error_message: Optional[str]
|
|
|
|
|
|
# API Endpoints
|
|
|
|
@router.post("", response_model=AnalysisResponse, status_code=status.HTTP_201_CREATED)
|
|
async def create_analysis(
|
|
analysis: AnalysisCreateRequest,
|
|
request: Request,
|
|
current_user: dict = Depends(get_current_user)
|
|
):
|
|
"""
|
|
Create a new analysis configuration via wizard
|
|
|
|
**Sprint 5-6 Feature - Multi-Cluster Support**
|
|
|
|
This endpoint creates a complete analysis configuration after the user
|
|
completes the 4-step wizard:
|
|
1. Scope Selection (supports single or multiple clusters)
|
|
2. Gadget Modules
|
|
3. Time & Profile
|
|
4. Output & Integration
|
|
|
|
**Multi-Cluster Support:**
|
|
- Set cluster_ids in scope to analyze multiple clusters simultaneously
|
|
- Each cluster can have its own scope configuration via per_cluster_scope
|
|
|
|
**Permissions Required:** analysis.create (Viewer role cannot create)
|
|
"""
|
|
# Check permission - Viewer role cannot create analyses
|
|
if not check_analysis_permission(current_user, "create"):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Permission denied. You do not have permission to create analyses. Required role: Admin, Operator, or Analyst."
|
|
)
|
|
|
|
# Determine if this is a multi-cluster analysis
|
|
cluster_ids = analysis.scope.cluster_ids or [analysis.scope.cluster_id]
|
|
is_multi_cluster = len(cluster_ids) > 1
|
|
|
|
# Validate all clusters exist
|
|
for cid in cluster_ids:
|
|
cluster_query = "SELECT id, name FROM clusters WHERE id = :cluster_id AND status = 'active'"
|
|
cluster = await database.fetch_one(cluster_query, {"cluster_id": cid})
|
|
|
|
if not cluster:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Cluster with id {cid} not found or inactive"
|
|
)
|
|
|
|
logger.info("Creating analysis",
|
|
name=analysis.name,
|
|
is_multi_cluster=is_multi_cluster,
|
|
cluster_ids=cluster_ids)
|
|
|
|
# Insert analysis with multi-cluster support
|
|
# NOTE: namespaces column is populated separately to maintain backward compatibility
|
|
# if migration hasn't run yet
|
|
insert_query = """
|
|
INSERT INTO analyses (
|
|
name, description, cluster_id, cluster_ids, is_multi_cluster,
|
|
status, scope_type, scope_config,
|
|
gadget_config, time_config, output_config,
|
|
change_detection_enabled, change_detection_strategy, change_detection_types,
|
|
analysis_level, l7_config,
|
|
created_by
|
|
) VALUES (
|
|
:name, :description, :cluster_id, :cluster_ids, :is_multi_cluster,
|
|
'draft', :scope_type, :scope_config,
|
|
:gadget_config, :time_config, :output_config,
|
|
:change_detection_enabled, :change_detection_strategy, :change_detection_types,
|
|
:analysis_level, CAST(:l7_config AS jsonb),
|
|
:created_by
|
|
) RETURNING *
|
|
"""
|
|
|
|
l7_json = json.dumps(analysis.l7_config) if analysis.l7_config else None
|
|
|
|
values = {
|
|
"name": analysis.name,
|
|
"description": analysis.description,
|
|
"cluster_id": analysis.scope.cluster_id, # Primary cluster
|
|
"cluster_ids": json.dumps(cluster_ids), # All clusters as JSON array
|
|
"is_multi_cluster": is_multi_cluster,
|
|
"scope_type": analysis.scope.scope_type,
|
|
"scope_config": json.dumps(analysis.scope.dict()),
|
|
"gadget_config": json.dumps(analysis.gadgets.dict()),
|
|
"time_config": json.dumps(analysis.time_config.dict(), default=str),
|
|
"output_config": json.dumps(analysis.output.dict()),
|
|
"change_detection_enabled": analysis.change_detection_enabled,
|
|
"change_detection_strategy": analysis.change_detection_strategy,
|
|
"change_detection_types": json.dumps(analysis.change_detection_types),
|
|
"analysis_level": (analysis.analysis_level or "l4").strip().lower(),
|
|
"l7_config": l7_json,
|
|
"created_by": current_user.get('user_id', 1)
|
|
}
|
|
|
|
result = await database.fetch_one(insert_query, values)
|
|
analysis_id = result['id']
|
|
|
|
# Extract namespaces from scope config for change detection filtering
|
|
# This is done as a separate UPDATE to maintain backward compatibility
|
|
# if the namespaces column doesn't exist yet (migration not run)
|
|
namespaces_list = None
|
|
if analysis.scope.namespaces:
|
|
namespaces_list = analysis.scope.namespaces
|
|
elif analysis.scope.per_cluster_scope:
|
|
# Collect namespaces from all clusters in multi-cluster setup
|
|
all_namespaces = set()
|
|
for cluster_scope in analysis.scope.per_cluster_scope.values():
|
|
if isinstance(cluster_scope, dict) and cluster_scope.get("namespaces"):
|
|
all_namespaces.update(cluster_scope["namespaces"])
|
|
if all_namespaces:
|
|
namespaces_list = list(all_namespaces)
|
|
|
|
# Try to populate namespaces column (fails gracefully if column doesn't exist)
|
|
if namespaces_list:
|
|
try:
|
|
update_ns_query = """
|
|
UPDATE analyses SET namespaces = :namespaces WHERE id = :id
|
|
"""
|
|
await database.execute(update_ns_query, {
|
|
"id": analysis_id,
|
|
"namespaces": json.dumps(namespaces_list)
|
|
})
|
|
logger.debug("Namespaces column updated", analysis_id=analysis_id, namespaces=namespaces_list)
|
|
except Exception as e:
|
|
# Column might not exist if migration hasn't run - this is OK
|
|
# Change detection will fall back to scope_config
|
|
logger.debug("Could not update namespaces column (migration may not have run)",
|
|
analysis_id=analysis_id, error=str(e))
|
|
|
|
# Get cluster name for response
|
|
cluster_query = "SELECT name FROM clusters WHERE id = :cluster_id"
|
|
cluster = await database.fetch_one(cluster_query, {"cluster_id": analysis.scope.cluster_id})
|
|
cluster_name = cluster["name"] if cluster else None
|
|
|
|
# Parse cluster_ids and change_detection_types back from JSON for response
|
|
response_data = dict(result)
|
|
response_data['cluster_name'] = cluster_name
|
|
if response_data.get('cluster_ids'):
|
|
response_data['cluster_ids'] = json.loads(response_data['cluster_ids']) if isinstance(response_data['cluster_ids'], str) else response_data['cluster_ids']
|
|
if response_data.get('change_detection_types'):
|
|
response_data['change_detection_types'] = json.loads(response_data['change_detection_types']) if isinstance(response_data['change_detection_types'], str) else response_data['change_detection_types']
|
|
if response_data.get('l7_config') and isinstance(response_data['l7_config'], str):
|
|
response_data['l7_config'] = json.loads(response_data['l7_config'])
|
|
|
|
# Log activity
|
|
await activity_service.log_activity(
|
|
user_id=current_user.get('user_id'),
|
|
username=current_user.get('username', 'system'),
|
|
action=ActivityService.ACTION_CREATE,
|
|
resource_type=ActivityService.RESOURCE_ANALYSIS,
|
|
resource_id=str(result['id']),
|
|
resource_name=analysis.name,
|
|
ip_address=get_client_ip(request),
|
|
details={
|
|
"cluster_id": analysis.scope.cluster_id,
|
|
"cluster_name": cluster_name,
|
|
"is_multi_cluster": is_multi_cluster,
|
|
"scope_type": analysis.scope.scope_type
|
|
}
|
|
)
|
|
|
|
return AnalysisResponse(**response_data)
|
|
|
|
|
|
@router.get("", response_model=List[AnalysisResponse])
|
|
async def get_analyses(
|
|
cluster_id: Optional[int] = None,
|
|
status: Optional[str] = None,
|
|
include_multi_cluster: bool = True,
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Get all analyses with optional filters
|
|
|
|
**Filters:**
|
|
- cluster_id: Filter by cluster (also matches multi-cluster analyses that include this cluster)
|
|
- status: Filter by status (draft, running, stopped, completed, failed)
|
|
- include_multi_cluster: Include analyses where cluster_id appears in cluster_ids array
|
|
"""
|
|
|
|
# Join with clusters to get cluster_name
|
|
query = """
|
|
SELECT a.*, c.name as cluster_name
|
|
FROM analyses a
|
|
LEFT JOIN clusters c ON a.cluster_id = c.id
|
|
WHERE 1=1
|
|
"""
|
|
params = {}
|
|
|
|
if cluster_id:
|
|
if include_multi_cluster:
|
|
# Match both primary cluster and multi-cluster analyses containing this cluster
|
|
query += """ AND (
|
|
a.cluster_id = :cluster_id
|
|
OR (a.is_multi_cluster = true AND a.cluster_ids::jsonb @> :cluster_id_json)
|
|
)"""
|
|
params["cluster_id"] = cluster_id
|
|
params["cluster_id_json"] = json.dumps([cluster_id])
|
|
else:
|
|
query += " AND a.cluster_id = :cluster_id"
|
|
params["cluster_id"] = cluster_id
|
|
|
|
if status:
|
|
query += " AND a.status = :status"
|
|
params["status"] = status
|
|
|
|
query += " ORDER BY a.created_at DESC"
|
|
|
|
results = await database.fetch_all(query, params)
|
|
|
|
# Parse cluster_ids and change_detection_types from JSON for each result
|
|
response_list = []
|
|
for row in results:
|
|
row_dict = dict(row)
|
|
if row_dict.get('cluster_ids'):
|
|
row_dict['cluster_ids'] = json.loads(row_dict['cluster_ids']) if isinstance(row_dict['cluster_ids'], str) else row_dict['cluster_ids']
|
|
if row_dict.get('change_detection_types'):
|
|
row_dict['change_detection_types'] = json.loads(row_dict['change_detection_types']) if isinstance(row_dict['change_detection_types'], str) else row_dict['change_detection_types']
|
|
if row_dict.get('l7_config') and isinstance(row_dict['l7_config'], str):
|
|
row_dict['l7_config'] = json.loads(row_dict['l7_config'])
|
|
response_list.append(AnalysisResponse(**row_dict))
|
|
|
|
return response_list
|
|
|
|
|
|
@router.get("/{analysis_id}", response_model=AnalysisResponse)
|
|
async def get_analysis(
|
|
analysis_id: int,
|
|
current_user: dict = Depends(get_current_user)
|
|
):
|
|
"""Get single analysis by ID with multi-cluster support"""
|
|
|
|
# Join with clusters to get cluster_name
|
|
query = """
|
|
SELECT a.*, c.name as cluster_name
|
|
FROM analyses a
|
|
LEFT JOIN clusters c ON a.cluster_id = c.id
|
|
WHERE a.id = :id
|
|
"""
|
|
result = await database.fetch_one(query, {"id": analysis_id})
|
|
|
|
if not result:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Analysis with id {analysis_id} not found"
|
|
)
|
|
|
|
# Parse cluster_ids and change_detection_types from JSON for response
|
|
row_dict = dict(result)
|
|
if row_dict.get('cluster_ids'):
|
|
row_dict['cluster_ids'] = json.loads(row_dict['cluster_ids']) if isinstance(row_dict['cluster_ids'], str) else row_dict['cluster_ids']
|
|
if row_dict.get('change_detection_types'):
|
|
row_dict['change_detection_types'] = json.loads(row_dict['change_detection_types']) if isinstance(row_dict['change_detection_types'], str) else row_dict['change_detection_types']
|
|
if row_dict.get('l7_config') and isinstance(row_dict['l7_config'], str):
|
|
row_dict['l7_config'] = json.loads(row_dict['l7_config'])
|
|
|
|
return AnalysisResponse(**row_dict)
|
|
|
|
|
|
@router.post("/{analysis_id}/start", response_model=AnalysisRunResponse)
|
|
async def start_analysis(
|
|
analysis_id: int,
|
|
request: Request,
|
|
current_user: dict = Depends(get_current_user)
|
|
):
|
|
"""
|
|
Start an analysis execution
|
|
|
|
**Sprint 5-6 Feature**
|
|
|
|
This will:
|
|
1. Create a new analysis_run record
|
|
2. Start eBPF data collection via Inspektor Gadget
|
|
3. Begin communication discovery
|
|
4. Update analysis status to 'running'
|
|
|
|
**Permissions Required:** analysis.start (Viewer role cannot start)
|
|
"""
|
|
# Check permission - Viewer role cannot start analyses
|
|
if not check_analysis_permission(current_user, "start"):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Permission denied. You do not have permission to start analyses. Required role: Admin, Operator, or Analyst."
|
|
)
|
|
|
|
# Get analysis
|
|
analysis_query = "SELECT * FROM analyses WHERE id = :id"
|
|
analysis = await database.fetch_one(analysis_query, {"id": analysis_id})
|
|
|
|
if not analysis:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Analysis with id {analysis_id} not found"
|
|
)
|
|
|
|
# Check if already running
|
|
if analysis["status"] == "running":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Analysis is already running"
|
|
)
|
|
|
|
# Get next run_number for this analysis
|
|
run_number_query = """
|
|
SELECT COALESCE(MAX(run_number), 0) + 1 as next_run_number
|
|
FROM analysis_runs WHERE analysis_id = :analysis_id
|
|
"""
|
|
run_number_result = await database.fetch_one(run_number_query, {"analysis_id": analysis_id})
|
|
next_run_number = run_number_result["next_run_number"] if run_number_result else 1
|
|
|
|
# Create analysis run
|
|
run_query = """
|
|
INSERT INTO analysis_runs (
|
|
analysis_id, run_number, status, start_time, events_collected,
|
|
workloads_discovered, communications_discovered, anomalies_detected, changes_detected
|
|
) VALUES (
|
|
:analysis_id, :run_number, 'running', NOW(), 0, 0, 0, 0, 0
|
|
) RETURNING *
|
|
"""
|
|
|
|
run_result = await database.fetch_one(run_query, {"analysis_id": analysis_id, "run_number": next_run_number})
|
|
|
|
# Update analysis status to running and reset started_at for auto-stop timing
|
|
# started_at must be updated on EVERY Start (not just first) for correct auto-stop behavior on restarts
|
|
update_query = "UPDATE analyses SET status = 'running', started_at = NOW(), updated_at = NOW() WHERE id = :id"
|
|
await database.execute(update_query, {"id": analysis_id})
|
|
|
|
analysis_level = (analysis.get("analysis_level") or "l4").strip().lower()
|
|
|
|
# Start analysis via Analysis Orchestrator (microservice)
|
|
try:
|
|
logger.info("Calling Analysis Orchestrator to start analysis",
|
|
analysis_id=analysis_id)
|
|
orchestrator_response = await analysis_orchestrator_client.start_analysis(
|
|
analysis_id, analysis_level=analysis_level
|
|
)
|
|
|
|
logger.info("Analysis Orchestrator started analysis successfully",
|
|
analysis_id=analysis_id,
|
|
tasks=len(orchestrator_response.get("task_assignments", [])))
|
|
|
|
# Log activity
|
|
await activity_service.log_activity(
|
|
user_id=current_user.get('user_id'),
|
|
username=current_user.get('username', 'system'),
|
|
action=ActivityService.ACTION_START,
|
|
resource_type=ActivityService.RESOURCE_ANALYSIS,
|
|
resource_id=str(analysis_id),
|
|
resource_name=analysis.get("name", f"Analysis {analysis_id}"),
|
|
ip_address=get_client_ip(request),
|
|
details={
|
|
"run_id": run_result['id'],
|
|
"run_number": next_run_number
|
|
}
|
|
)
|
|
|
|
return AnalysisRunResponse(**run_result)
|
|
|
|
except Exception as e:
|
|
# Rollback if orchestrator call failed
|
|
logger.error("Analysis Orchestrator failed to start analysis",
|
|
analysis_id=analysis_id,
|
|
error=str(e))
|
|
|
|
await database.execute(
|
|
"UPDATE analyses SET status = 'failed', updated_at = NOW() WHERE id = :id",
|
|
{"id": analysis_id}
|
|
)
|
|
await database.execute(
|
|
"UPDATE analysis_runs SET status = 'failed', error_message = :error WHERE id = :id",
|
|
{"id": run_result["id"], "error": f"Failed to start via orchestrator: {str(e)}"}
|
|
)
|
|
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Failed to start analysis via orchestrator: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.post("/{analysis_id}/stop")
|
|
async def stop_analysis(
|
|
analysis_id: int,
|
|
request: Request,
|
|
current_user: dict = Depends(get_current_user)
|
|
):
|
|
"""
|
|
Stop a running analysis
|
|
|
|
**Sprint 5-6 Feature**
|
|
|
|
Also fetches final statistics from ClickHouse and updates analysis_runs.
|
|
|
|
**Permissions Required:** analysis.stop (Viewer role cannot stop)
|
|
"""
|
|
# Check permission - Viewer role cannot stop analyses
|
|
if not check_analysis_permission(current_user, "stop"):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Permission denied. You do not have permission to stop analyses. Required role: Admin, Operator, or Analyst."
|
|
)
|
|
|
|
# Get analysis
|
|
analysis_query = "SELECT * FROM analyses WHERE id = :id"
|
|
analysis = await database.fetch_one(analysis_query, {"id": analysis_id})
|
|
|
|
if not analysis:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Analysis with id {analysis_id} not found"
|
|
)
|
|
|
|
if analysis["status"] != "running":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Analysis is not running"
|
|
)
|
|
|
|
cluster_id = analysis["cluster_id"]
|
|
|
|
# SAFETY FIRST: Stop the orchestrator BEFORE updating any status
|
|
# This ensures we don't show "stopped" while analysis is still running
|
|
|
|
# Step 1: Call orchestrator to ACTUALLY stop the analysis
|
|
# This is the critical operation - must succeed before status update
|
|
orchestrator_success = False
|
|
orchestrator_error = None
|
|
try:
|
|
logger.info("Calling Analysis Orchestrator to stop analysis", analysis_id=analysis_id)
|
|
orchestrator_success = await analysis_orchestrator_client.stop_analysis(analysis_id)
|
|
|
|
if not orchestrator_success:
|
|
logger.warning("Orchestrator returned False for stop_analysis, will verify status", analysis_id=analysis_id)
|
|
|
|
# Verify by checking actual status from orchestrator
|
|
await asyncio.sleep(0.5) # Brief wait for state to propagate
|
|
|
|
status_check = await analysis_orchestrator_client.get_analysis_status(analysis_id)
|
|
if status_check is None or status_check.get("status") in ["stopped", "completed", "error"]:
|
|
# Analysis is actually stopped (not found = not running = stopped)
|
|
logger.info("Status verification confirms analysis is stopped", analysis_id=analysis_id)
|
|
orchestrator_success = True
|
|
else:
|
|
logger.warning("Status verification shows analysis still running",
|
|
analysis_id=analysis_id, status=status_check.get("status"))
|
|
except Exception as e:
|
|
orchestrator_error = str(e)
|
|
logger.error("Failed to stop analysis via orchestrator", analysis_id=analysis_id, error=orchestrator_error)
|
|
|
|
# Even on exception, verify the actual status
|
|
try:
|
|
status_check = await analysis_orchestrator_client.get_analysis_status(analysis_id)
|
|
if status_check is None or status_check.get("status") in ["stopped", "completed", "error"]:
|
|
logger.info("Despite exception, status verification confirms analysis is stopped", analysis_id=analysis_id)
|
|
orchestrator_success = True
|
|
orchestrator_error = None
|
|
except Exception:
|
|
pass # Keep original error
|
|
|
|
# Step 2: Only update status to 'stopped' AFTER orchestrator confirms stop
|
|
# Even if orchestrator had issues, we proceed (but log warning) because:
|
|
# - The stop command was sent
|
|
# - Better to update status than leave in inconsistent state
|
|
# - User can check orchestrator logs if issues persist
|
|
|
|
stop_run_query = """
|
|
UPDATE analysis_runs
|
|
SET status = 'stopped', end_time = NOW()
|
|
WHERE analysis_id = :analysis_id AND status = 'running'
|
|
"""
|
|
await database.execute(stop_run_query, {"analysis_id": analysis_id})
|
|
|
|
update_query = "UPDATE analyses SET status = 'stopped', updated_at = NOW() WHERE id = :id"
|
|
await database.execute(update_query, {"id": analysis_id})
|
|
|
|
logger.info("Analysis status updated to stopped in database",
|
|
analysis_id=analysis_id,
|
|
orchestrator_success=orchestrator_success)
|
|
|
|
# Step 4: Get previous runs' total events (to calculate this run's events only)
|
|
previous_events_total = 0
|
|
previous_comms_total = 0
|
|
try:
|
|
prev_runs_query = """
|
|
SELECT COALESCE(SUM(events_collected), 0) as prev_events,
|
|
COALESCE(SUM(communications_discovered), 0) as prev_comms
|
|
FROM analysis_runs
|
|
WHERE analysis_id = :analysis_id
|
|
AND status IN ('stopped', 'completed', 'failed')
|
|
AND id != (SELECT id FROM analysis_runs WHERE analysis_id = :analysis_id ORDER BY start_time DESC LIMIT 1)
|
|
"""
|
|
prev_result = await database.fetch_one(prev_runs_query, {"analysis_id": analysis_id})
|
|
if prev_result:
|
|
previous_events_total = int(prev_result["prev_events"] or 0)
|
|
previous_comms_total = int(prev_result["prev_comms"] or 0)
|
|
logger.debug(f"Previous runs totals: events={previous_events_total}, comms={previous_comms_total}")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to get previous runs totals: {e}")
|
|
|
|
# Step 5: Fetch statistics from ClickHouse (can be slow, but analysis is already stopped)
|
|
events_collected = 0
|
|
communications_discovered = 0
|
|
total_events_clickhouse = 0
|
|
total_comms_clickhouse = 0
|
|
|
|
try:
|
|
from repositories.event_repository import get_event_repository
|
|
event_repo = get_event_repository(use_microservice=True)
|
|
|
|
# Get event counts - use simple count query for speed
|
|
event_counts = await event_repo.get_event_counts_by_type(
|
|
cluster_id=cluster_id,
|
|
analysis_id=analysis_id
|
|
)
|
|
total_events_clickhouse = int(sum(event_counts.values())) if event_counts else 0
|
|
|
|
# Calculate this run's events only (total - previous runs)
|
|
events_collected = max(0, total_events_clickhouse - previous_events_total)
|
|
|
|
# Get unique communications count from network_flows
|
|
try:
|
|
from config import get_clickhouse_config
|
|
from repositories.event_repository import ClickHouseEventRepository
|
|
|
|
ch_config = get_clickhouse_config()
|
|
direct_repo = ClickHouseEventRepository(
|
|
host=ch_config.get('host', 'clickhouse'),
|
|
port=ch_config.get('port', 8123),
|
|
database=ch_config.get('database', 'flowfish'),
|
|
user=ch_config.get('user', 'default'),
|
|
password=ch_config.get('password', ''),
|
|
)
|
|
|
|
# Count unique source->destination communication pairs
|
|
# Support multi-cluster format: analysis_id can be '{id}' or '{id}-{cluster_id}'
|
|
comm_query = f"""
|
|
SELECT count() as cnt FROM (
|
|
SELECT DISTINCT
|
|
source_namespace, source_pod,
|
|
dest_namespace, dest_pod, dest_port, protocol
|
|
FROM network_flows
|
|
WHERE analysis_id = '{analysis_id}' OR analysis_id LIKE '{analysis_id}-%'
|
|
)
|
|
"""
|
|
comm_result = await direct_repo._execute_query(comm_query)
|
|
if comm_result and len(comm_result) > 0:
|
|
total_comms_clickhouse = int(comm_result[0].get('cnt', 0))
|
|
# Calculate this run's communications only (total - previous runs)
|
|
communications_discovered = max(0, total_comms_clickhouse - previous_comms_total)
|
|
except Exception as comm_err:
|
|
logger.warning(f"Failed to get communications count: {comm_err}")
|
|
|
|
# L7 unique communications (dedup across HTTP/gRPC/DNS tables)
|
|
try:
|
|
aid_str = str(analysis_id)
|
|
l7_comm_query = f"""
|
|
SELECT count() as cnt FROM (
|
|
SELECT DISTINCT src_workload, src_namespace, dst_workload, dst_namespace FROM (
|
|
SELECT src_workload, src_namespace, dst_workload, dst_namespace
|
|
FROM l7_http_flows
|
|
WHERE analysis_id = '{aid_str}' OR analysis_id LIKE '{aid_str}-%'
|
|
UNION ALL
|
|
SELECT src_workload, src_namespace, dst_workload, dst_namespace
|
|
FROM l7_grpc_flows
|
|
WHERE analysis_id = '{aid_str}' OR analysis_id LIKE '{aid_str}-%'
|
|
UNION ALL
|
|
SELECT src_workload, src_namespace, dst_workload, dst_namespace
|
|
FROM l7_dns_flows
|
|
WHERE analysis_id = '{aid_str}' OR analysis_id LIKE '{aid_str}-%'
|
|
)
|
|
)
|
|
"""
|
|
l7_comm_result = await direct_repo._execute_query(l7_comm_query)
|
|
if l7_comm_result and len(l7_comm_result) > 0:
|
|
l7_comms = int(l7_comm_result[0].get('cnt', 0))
|
|
total_comms_clickhouse += l7_comms
|
|
communications_discovered = max(0, total_comms_clickhouse - previous_comms_total)
|
|
except Exception:
|
|
pass
|
|
|
|
logger.info("Fetched analysis statistics from ClickHouse",
|
|
analysis_id=analysis_id,
|
|
total_events_clickhouse=total_events_clickhouse,
|
|
total_comms_clickhouse=total_comms_clickhouse,
|
|
previous_events_total=previous_events_total,
|
|
previous_comms_total=previous_comms_total,
|
|
events_collected=events_collected,
|
|
communications_discovered=communications_discovered)
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Failed to fetch statistics from ClickHouse: {e}")
|
|
|
|
# Step 6: Update run with final statistics
|
|
update_stats_query = """
|
|
UPDATE analysis_runs
|
|
SET events_collected = :events_collected,
|
|
communications_discovered = :communications_discovered
|
|
WHERE analysis_id = :analysis_id AND status = 'stopped'
|
|
AND end_time = (SELECT MAX(end_time) FROM analysis_runs WHERE analysis_id = :analysis_id)
|
|
"""
|
|
await database.execute(update_stats_query, {
|
|
"analysis_id": analysis_id,
|
|
"events_collected": events_collected,
|
|
"communications_discovered": communications_discovered
|
|
})
|
|
|
|
# Log activity
|
|
await activity_service.log_activity(
|
|
user_id=current_user.get('user_id'),
|
|
username=current_user.get('username', 'system'),
|
|
action=ActivityService.ACTION_STOP,
|
|
resource_type=ActivityService.RESOURCE_ANALYSIS,
|
|
resource_id=str(analysis_id),
|
|
resource_name=analysis.get("name", f"Analysis {analysis_id}"),
|
|
ip_address=get_client_ip(request),
|
|
details={
|
|
"events_collected": events_collected,
|
|
"communications_discovered": communications_discovered,
|
|
"orchestrator_success": orchestrator_success
|
|
}
|
|
)
|
|
|
|
if orchestrator_success:
|
|
logger.info("Analysis stopped successfully via orchestrator",
|
|
analysis_id=analysis_id,
|
|
events_collected=events_collected,
|
|
communications_discovered=communications_discovered)
|
|
|
|
return {
|
|
"message": "Analysis stopped successfully",
|
|
"analysis_id": analysis_id,
|
|
"events_collected": events_collected,
|
|
"communications_discovered": communications_discovered,
|
|
"orchestrator_status": "success"
|
|
}
|
|
else:
|
|
logger.warning("Analysis stopped but orchestrator reported issues",
|
|
analysis_id=analysis_id,
|
|
orchestrator_error=orchestrator_error)
|
|
|
|
return {
|
|
"message": "Analysis stopped (orchestrator reported issues - please verify no data is being collected)",
|
|
"analysis_id": analysis_id,
|
|
"events_collected": events_collected,
|
|
"communications_discovered": communications_discovered,
|
|
"orchestrator_status": "warning",
|
|
"orchestrator_error": orchestrator_error
|
|
}
|
|
|
|
|
|
class DeleteAnalysisResponse(BaseModel):
|
|
"""Response model for analysis deletion"""
|
|
analysis_id: int
|
|
deleted: bool
|
|
postgresql: dict = {}
|
|
neo4j: dict = {}
|
|
clickhouse: dict = {}
|
|
redis: dict = {}
|
|
rabbitmq: dict = {}
|
|
duration_ms: int = 0
|
|
message: str = ""
|
|
|
|
|
|
@router.delete("/{analysis_id}", response_model=DeleteAnalysisResponse)
|
|
async def delete_analysis(
|
|
analysis_id: int,
|
|
request: Request,
|
|
current_user: dict = Depends(get_current_user)
|
|
):
|
|
"""
|
|
Delete an analysis and ALL associated data
|
|
|
|
This will delete:
|
|
- PostgreSQL: analyses, analysis_runs records
|
|
- Neo4j: All nodes and edges with this analysis_id (batch delete)
|
|
- ClickHouse: All events from all tables with this analysis_id (with polling)
|
|
- Redis: All cached data with this analysis_id
|
|
|
|
Returns detailed summary of what was deleted.
|
|
|
|
**Permissions Required:** analysis.delete (Viewer role cannot delete)
|
|
"""
|
|
# Check permission - Viewer role cannot delete analyses
|
|
if not check_analysis_permission(current_user, "delete"):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Permission denied. You do not have permission to delete analyses. Required role: Admin or Operator."
|
|
)
|
|
|
|
import time
|
|
start_time = time.time()
|
|
|
|
# Get analysis with cluster_id and analysis_level for comprehensive deletion
|
|
analysis_query = "SELECT status, name, cluster_id, is_scheduled, analysis_level FROM analyses WHERE id = :id"
|
|
analysis = await database.fetch_one(analysis_query, {"id": analysis_id})
|
|
|
|
if not analysis:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Analysis with id {analysis_id} not found"
|
|
)
|
|
|
|
if analysis["status"] == "running":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Cannot delete a running analysis. Stop it first."
|
|
)
|
|
|
|
# Remove schedule from orchestrator before deleting (best-effort)
|
|
if analysis.get("is_scheduled"):
|
|
try:
|
|
await analysis_orchestrator_client.unschedule_analysis(analysis_id)
|
|
except Exception as e:
|
|
logger.warning("Failed to unschedule before delete (scheduler will self-heal)",
|
|
analysis_id=analysis_id, error=str(e))
|
|
|
|
analysis_name = analysis["name"]
|
|
cluster_id = analysis["cluster_id"]
|
|
neo4j_result = {}
|
|
clickhouse_result = {}
|
|
redis_result = {}
|
|
|
|
# STEP -1: Best-effort L7 collection stop (safety net in case session lingers)
|
|
analysis_level = (analysis.get("analysis_level") or "l4").strip().lower()
|
|
if analysis_level in ("l7", "both"):
|
|
try:
|
|
await analysis_orchestrator_client.stop_analysis(analysis_id)
|
|
logger.info("Pre-delete L7 stop attempt completed", analysis_id=analysis_id)
|
|
except Exception as e:
|
|
logger.debug("Pre-delete stop attempt (expected if already stopped): %s", e)
|
|
|
|
# STEP 0: Mark analysis as deleted in Redis cache
|
|
# This prevents timeseries-writer and graph-writer from creating orphan data
|
|
# while we're deleting
|
|
try:
|
|
from database.redis import redis_client
|
|
|
|
deleted_key = f"flowfish:deleted_analysis:{analysis_id}"
|
|
await redis_client.setex(deleted_key, 86400, "1") # 24 hour TTL
|
|
logger.info("Marked analysis as deleted in Redis cache",
|
|
analysis_id=analysis_id,
|
|
key=deleted_key)
|
|
|
|
# Wait briefly for in-flight messages to be filtered out by writers
|
|
await asyncio.sleep(0.5) # 500ms should be enough for batch flush
|
|
|
|
except Exception as e:
|
|
logger.warning("Failed to mark analysis as deleted in Redis (continuing)",
|
|
analysis_id=analysis_id,
|
|
error=str(e))
|
|
|
|
# Delete from Neo4j (graph data) - batch processing with cluster_id fallback
|
|
try:
|
|
from database.neo4j import neo4j_service
|
|
neo4j_result = neo4j_service.delete_analysis_data(
|
|
analysis_id,
|
|
cluster_id=cluster_id,
|
|
batch_size=5000
|
|
)
|
|
# Remove duplicate keys from neo4j_result before logging
|
|
log_result = {k: v for k, v in neo4j_result.items() if k not in ('analysis_id', 'cluster_id')}
|
|
logger.info("Deleted Neo4j data",
|
|
analysis_id=analysis_id,
|
|
cluster_id=cluster_id,
|
|
analysis_name=analysis_name,
|
|
**log_result)
|
|
except Exception as e:
|
|
logger.warning("Failed to delete Neo4j data (continuing)",
|
|
analysis_id=analysis_id,
|
|
error=str(e))
|
|
neo4j_result = {"error": str(e)}
|
|
|
|
# Delete L7 data from Neo4j (L7Workload nodes) if applicable
|
|
try:
|
|
from database.neo4j import neo4j_service as _neo4j
|
|
l7_neo4j_result = _neo4j.delete_l7_analysis_data(analysis_id)
|
|
if l7_neo4j_result.get("deleted_nodes", 0) > 0:
|
|
logger.info("Deleted L7 Neo4j data",
|
|
analysis_id=analysis_id,
|
|
deleted_nodes=l7_neo4j_result.get("deleted_nodes", 0))
|
|
except Exception as e:
|
|
logger.warning("Failed to delete L7 Neo4j data (continuing)",
|
|
analysis_id=analysis_id, error=str(e))
|
|
|
|
# Delete from ClickHouse (time-series events) - with polling
|
|
# Try microservice first, then fallback to direct ClickHouse access
|
|
try:
|
|
from repositories.event_repository import get_event_repository, ClickHouseEventRepository
|
|
from config import get_clickhouse_config
|
|
|
|
# First try via microservice
|
|
event_repo = get_event_repository(use_microservice=True)
|
|
clickhouse_result = await event_repo.delete_analysis_data(
|
|
analysis_id,
|
|
wait_for_completion=True,
|
|
timeout_seconds=30
|
|
)
|
|
|
|
# Check if microservice deletion failed (returned error in result)
|
|
if clickhouse_result.get("error") or clickhouse_result.get("total_deleted", 0) == 0:
|
|
logger.warning("Microservice deletion failed or returned 0, trying direct ClickHouse access",
|
|
analysis_id=analysis_id,
|
|
microservice_result=clickhouse_result)
|
|
|
|
# Fallback to direct ClickHouse access
|
|
ch_config = get_clickhouse_config()
|
|
direct_repo = ClickHouseEventRepository(
|
|
host=ch_config.get('host', 'clickhouse'),
|
|
port=ch_config.get('port', 8123),
|
|
database=ch_config.get('database', 'flowfish'),
|
|
user=ch_config.get('user', 'default'),
|
|
password=ch_config.get('password', ''),
|
|
)
|
|
clickhouse_result = await direct_repo.delete_analysis_data(
|
|
analysis_id,
|
|
wait_for_completion=True,
|
|
timeout_seconds=60
|
|
)
|
|
|
|
logger.info("ClickHouse deletion completed",
|
|
analysis_id=analysis_id,
|
|
analysis_name=analysis_name,
|
|
**clickhouse_result)
|
|
except Exception as e:
|
|
logger.warning("Failed to delete ClickHouse data (continuing)",
|
|
analysis_id=analysis_id,
|
|
error=str(e))
|
|
clickhouse_result = {"error": str(e)}
|
|
|
|
# Delete L7 ClickHouse data (l7_http_flows, l7_grpc_flows, l7_dns_flows)
|
|
try:
|
|
from config import get_clickhouse_config
|
|
import httpx
|
|
ch_config = get_clickhouse_config()
|
|
ch_url = f"http://{ch_config.get('host', 'clickhouse')}:{ch_config.get('port', 8123)}"
|
|
ch_db = ch_config.get('database', 'flowfish')
|
|
l7_tables = ["l7_http_flows", "l7_grpc_flows", "l7_dns_flows", "l7_http_flows_5min_mv"]
|
|
aid_str = str(analysis_id)
|
|
l7_ch_errors = []
|
|
for tbl in l7_tables:
|
|
try:
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
sql = f"ALTER TABLE {ch_db}.{tbl} DELETE WHERE analysis_id = '{aid_str}' OR analysis_id LIKE '{aid_str}-%'"
|
|
resp = await client.post(ch_url, content=sql)
|
|
if resp.status_code >= 400:
|
|
l7_ch_errors.append(f"{tbl}: HTTP {resp.status_code}")
|
|
except Exception as tbl_err:
|
|
l7_ch_errors.append(f"{tbl}: {tbl_err}")
|
|
if l7_ch_errors:
|
|
logger.warning("L7 ClickHouse cleanup partial failures", analysis_id=analysis_id, errors=l7_ch_errors)
|
|
else:
|
|
logger.info("L7 ClickHouse cleanup done", analysis_id=analysis_id)
|
|
except Exception as e:
|
|
logger.warning("Failed to delete L7 ClickHouse data (continuing)",
|
|
analysis_id=analysis_id, error=str(e))
|
|
|
|
# Delete from Redis (cached data)
|
|
try:
|
|
from database.redis import redis_client
|
|
|
|
redis_deleted = 0
|
|
|
|
# Delete analysis-specific cache keys
|
|
# Pattern 1: analysis:status:{analysis_id}
|
|
status_key = f"analysis:status:{analysis_id}"
|
|
deleted = await redis_client.delete(status_key)
|
|
redis_deleted += deleted if deleted else 0
|
|
|
|
# Pattern 2: Any key containing the analysis_id
|
|
# Use scan to find keys matching pattern
|
|
analysis_patterns = [
|
|
f"*analysis*{analysis_id}*",
|
|
f"*:analysis:{analysis_id}*",
|
|
f"analysis:{analysis_id}:*",
|
|
f"flowfish:l7:stats:{analysis_id}",
|
|
f"flowfish:l7:graph:{analysis_id}:*",
|
|
f"flowfish:l7:cursor:{analysis_id}:*",
|
|
]
|
|
|
|
for pattern in analysis_patterns:
|
|
try:
|
|
matching_keys = await redis_client.keys(pattern)
|
|
if matching_keys:
|
|
for key in matching_keys:
|
|
await redis_client.delete(key)
|
|
redis_deleted += 1
|
|
except Exception as pattern_error:
|
|
logger.debug(f"Pattern {pattern} search failed: {pattern_error}")
|
|
|
|
redis_result = {"deleted_keys": redis_deleted}
|
|
logger.info("Redis cache cleared",
|
|
analysis_id=analysis_id,
|
|
deleted_keys=redis_deleted)
|
|
except Exception as e:
|
|
logger.warning("Failed to clear Redis cache (continuing)",
|
|
analysis_id=analysis_id,
|
|
error=str(e))
|
|
redis_result = {"error": str(e)}
|
|
|
|
# RabbitMQ note - messages are transient and consumed by workers
|
|
# We cannot selectively delete messages by analysis_id
|
|
# Any pending messages for this analysis will be processed but data
|
|
# will reference a non-existent analysis (which is fine)
|
|
rabbitmq_result = {
|
|
"note": "RabbitMQ messages are transient and consumed by workers within seconds. "
|
|
"Any pending messages for this analysis will be automatically discarded.",
|
|
"status": "not_applicable"
|
|
}
|
|
|
|
# Count analysis_runs before deletion (for PostgreSQL info)
|
|
runs_count_query = "SELECT COUNT(*) as cnt FROM analysis_runs WHERE analysis_id = :id"
|
|
runs_count = await database.fetch_one(runs_count_query, {"id": analysis_id})
|
|
runs_deleted = runs_count["cnt"] if runs_count else 0
|
|
|
|
# NOTE: change_workflow table removed from PostgreSQL
|
|
# Change events are now stored only in ClickHouse (deleted above)
|
|
|
|
# Clean up dependent tables before deleting the analysis record.
|
|
# blast_radius_assessments has a FK without ON DELETE CASCADE (fixed in migration,
|
|
# but explicit cleanup ensures safety even before migration runs).
|
|
# analysis_event_types has no FK at all, so orphan rows would accumulate.
|
|
blast_radius_deleted = 0
|
|
event_types_deleted = 0
|
|
|
|
try:
|
|
br_count_query = "SELECT COUNT(*) as cnt FROM blast_radius_assessments WHERE analysis_id = :id"
|
|
br_count = await database.fetch_one(br_count_query, {"id": analysis_id})
|
|
blast_radius_deleted = br_count["cnt"] if br_count else 0
|
|
if blast_radius_deleted > 0:
|
|
await database.execute(
|
|
"DELETE FROM blast_radius_assessments WHERE analysis_id = :id",
|
|
{"id": analysis_id}
|
|
)
|
|
logger.info("Deleted blast_radius_assessments",
|
|
analysis_id=analysis_id, count=blast_radius_deleted)
|
|
except Exception as e:
|
|
logger.warning("Failed to clean blast_radius_assessments",
|
|
analysis_id=analysis_id, error=str(e))
|
|
|
|
try:
|
|
aet_count_query = "SELECT COUNT(*) as cnt FROM analysis_event_types WHERE analysis_id = :id"
|
|
aet_count = await database.fetch_one(aet_count_query, {"id": analysis_id})
|
|
event_types_deleted = aet_count["cnt"] if aet_count else 0
|
|
if event_types_deleted > 0:
|
|
await database.execute(
|
|
"DELETE FROM analysis_event_types WHERE analysis_id = :id",
|
|
{"id": analysis_id}
|
|
)
|
|
except Exception as e:
|
|
logger.warning("Failed to clean analysis_event_types",
|
|
analysis_id=analysis_id, error=str(e))
|
|
|
|
# Delete from PostgreSQL (analysis record - cascade deletes analysis_runs)
|
|
try:
|
|
delete_query = "DELETE FROM analyses WHERE id = :id"
|
|
await database.execute(delete_query, {"id": analysis_id})
|
|
except Exception as e:
|
|
error_msg = str(e)
|
|
if "ForeignKeyViolationError" in error_msg or "IntegrityError" in error_msg:
|
|
logger.error("FK violation deleting analysis - dependent rows remain",
|
|
analysis_id=analysis_id, error=error_msg)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=(
|
|
f"Cannot delete analysis {analysis_id}: dependent records still exist in the database. "
|
|
"This may happen if a migration has not been applied. "
|
|
"Please contact your administrator."
|
|
)
|
|
)
|
|
raise
|
|
|
|
postgresql_result = {
|
|
"deleted_analyses": 1,
|
|
"deleted_runs": runs_deleted,
|
|
"deleted_blast_radius_assessments": blast_radius_deleted,
|
|
"deleted_event_types": event_types_deleted
|
|
}
|
|
|
|
duration_ms = int((time.time() - start_time) * 1000)
|
|
|
|
total_deleted = (
|
|
neo4j_result.get("deleted_edges", 0) +
|
|
neo4j_result.get("deleted_nodes", 0) +
|
|
clickhouse_result.get("total_deleted", 0) +
|
|
redis_result.get("deleted_keys", 0) +
|
|
postgresql_result.get("deleted_analyses", 0) +
|
|
postgresql_result.get("deleted_runs", 0)
|
|
)
|
|
|
|
# Check if Neo4j had 0 deletes but ClickHouse had data (analysis_id overwrite issue)
|
|
neo4j_deleted = neo4j_result.get("deleted_edges", 0) + neo4j_result.get("deleted_nodes", 0)
|
|
clickhouse_deleted = clickhouse_result.get("total_deleted", 0)
|
|
|
|
if neo4j_deleted == 0 and clickhouse_deleted > 0:
|
|
neo4j_result["warning"] = (
|
|
"Neo4j graph data may have been claimed by a later analysis. "
|
|
"This happens when multiple analyses run on the same cluster/pods. "
|
|
"Orphaned nodes will be cleaned up automatically."
|
|
)
|
|
logger.warning("Neo4j showed 0 deletes but ClickHouse had data - analysis_id may have been overwritten",
|
|
analysis_id=analysis_id,
|
|
clickhouse_deleted=clickhouse_deleted)
|
|
|
|
# STEP 6: Auto-cleanup orphaned data (data for analyses that no longer exist in PostgreSQL)
|
|
orphan_result = {"cleaned": 0}
|
|
try:
|
|
# Get all valid analysis IDs from PostgreSQL
|
|
valid_ids_query = "SELECT id FROM analyses"
|
|
valid_analyses = await database.fetch_all(valid_ids_query)
|
|
valid_analysis_ids = {str(row["id"]) for row in valid_analyses}
|
|
|
|
# Find and delete orphaned analysis_ids in ClickHouse
|
|
orphaned_ids = set()
|
|
tables = [
|
|
'network_flows', 'dns_queries', 'tcp_lifecycle', 'process_events',
|
|
'file_operations', 'capability_checks', 'oom_kills', 'bind_events',
|
|
'sni_events', 'mount_events', 'workload_metadata', 'communication_edges',
|
|
'l7_http_flows', 'l7_grpc_flows', 'l7_dns_flows',
|
|
]
|
|
|
|
ch_config = get_clickhouse_config()
|
|
direct_repo = ClickHouseEventRepository(
|
|
host=ch_config.get('host', 'clickhouse'),
|
|
port=ch_config.get('port', 8123),
|
|
database=ch_config.get('database', 'flowfish'),
|
|
user=ch_config.get('user', 'default'),
|
|
password=ch_config.get('password', ''),
|
|
)
|
|
|
|
for table in tables[:3]: # Check first 3 tables for speed
|
|
try:
|
|
distinct_query = f"SELECT DISTINCT analysis_id FROM {table} WHERE analysis_id != '' LIMIT 100"
|
|
result = await direct_repo._execute_query(distinct_query)
|
|
if result:
|
|
for row in result:
|
|
aid = str(row.get("analysis_id", ""))
|
|
if aid and aid not in valid_analysis_ids:
|
|
orphaned_ids.add(aid)
|
|
except:
|
|
pass
|
|
|
|
# Delete orphaned data — analysis_id can be numeric or composite (e.g. "42-3")
|
|
for orphan_id in orphaned_ids:
|
|
try:
|
|
aid = int(orphan_id) if orphan_id.isdigit() else orphan_id
|
|
result = await direct_repo.delete_analysis_data(
|
|
aid,
|
|
wait_for_completion=False,
|
|
timeout_seconds=10
|
|
)
|
|
orphan_result["cleaned"] += result.get("total_deleted", 0)
|
|
except Exception:
|
|
pass
|
|
|
|
if orphaned_ids:
|
|
logger.info("Auto-cleaned orphaned data",
|
|
orphaned_ids=list(orphaned_ids),
|
|
cleaned=orphan_result["cleaned"])
|
|
|
|
except Exception as e:
|
|
logger.debug(f"Orphan cleanup skipped: {e}")
|
|
|
|
total_deleted += orphan_result.get("cleaned", 0)
|
|
duration_ms = int((time.time() - start_time) * 1000)
|
|
|
|
logger.info("Analysis deleted completely",
|
|
analysis_id=analysis_id,
|
|
analysis_name=analysis_name,
|
|
total_deleted=total_deleted,
|
|
orphans_cleaned=orphan_result.get("cleaned", 0),
|
|
duration_ms=duration_ms)
|
|
|
|
# Log activity
|
|
await activity_service.log_activity(
|
|
user_id=current_user.get('user_id'),
|
|
username=current_user.get('username', 'system'),
|
|
action=ActivityService.ACTION_DELETE,
|
|
resource_type=ActivityService.RESOURCE_ANALYSIS,
|
|
resource_id=str(analysis_id),
|
|
resource_name=analysis_name,
|
|
ip_address=get_client_ip(request),
|
|
details={
|
|
"cluster_id": cluster_id,
|
|
"total_deleted": total_deleted,
|
|
"duration_ms": duration_ms
|
|
}
|
|
)
|
|
|
|
message_text = f"Analysis '{analysis_name}' deleted. Removed {total_deleted:,} records in {duration_ms}ms."
|
|
if orphan_result.get("cleaned", 0) > 0:
|
|
message_text += f" (including {orphan_result['cleaned']:,} orphaned records)"
|
|
|
|
return DeleteAnalysisResponse(
|
|
analysis_id=analysis_id,
|
|
deleted=True,
|
|
postgresql=postgresql_result,
|
|
neo4j=neo4j_result,
|
|
clickhouse=clickhouse_result,
|
|
redis=redis_result,
|
|
rabbitmq=rabbitmq_result,
|
|
duration_ms=duration_ms,
|
|
message=message_text
|
|
)
|
|
|
|
|
|
@router.get("/{analysis_id}/runs", response_model=List[AnalysisRunResponse])
|
|
async def get_analysis_runs(
|
|
analysis_id: int,
|
|
current_user: dict = Depends(get_current_user)
|
|
):
|
|
"""
|
|
Get all runs for an analysis
|
|
|
|
If analysis is running or events_collected is 0, fetches current counts from ClickHouse.
|
|
This ensures live updates while analysis is running.
|
|
"""
|
|
|
|
# Get analysis to get cluster_id and status
|
|
analysis_query = "SELECT cluster_id, status FROM analyses WHERE id = :id"
|
|
analysis = await database.fetch_one(analysis_query, {"id": analysis_id})
|
|
cluster_id = analysis["cluster_id"] if analysis else None
|
|
analysis_status = analysis["status"] if analysis else None
|
|
|
|
query = """
|
|
SELECT * FROM analysis_runs
|
|
WHERE analysis_id = :analysis_id
|
|
ORDER BY start_time DESC
|
|
"""
|
|
|
|
results = await database.fetch_all(query, {"analysis_id": analysis_id})
|
|
runs = [dict(row) for row in results]
|
|
|
|
# Fetch from ClickHouse if:
|
|
# 1. Analysis is currently running (for live updates)
|
|
# 2. Any run has 0 events (initial load or missing data)
|
|
is_running = analysis_status == 'running'
|
|
has_zero_events = any(run.get('events_collected', 0) == 0 for run in runs)
|
|
needs_update = is_running or has_zero_events
|
|
|
|
if needs_update and cluster_id:
|
|
try:
|
|
from repositories.event_repository import get_event_repository, ClickHouseEventRepository
|
|
from config import get_clickhouse_config
|
|
|
|
event_repo = get_event_repository(use_microservice=True)
|
|
|
|
# Get event counts from ClickHouse
|
|
event_counts = await event_repo.get_event_counts_by_type(
|
|
cluster_id=cluster_id,
|
|
analysis_id=analysis_id
|
|
)
|
|
total_events = int(sum(event_counts.values())) if event_counts else 0
|
|
|
|
# Get unique communications count from network_flows
|
|
communications_count = 0
|
|
try:
|
|
ch_config = get_clickhouse_config()
|
|
direct_repo = ClickHouseEventRepository(
|
|
host=ch_config.get('host', 'clickhouse'),
|
|
port=ch_config.get('port', 8123),
|
|
database=ch_config.get('database', 'flowfish'),
|
|
user=ch_config.get('user', 'default'),
|
|
password=ch_config.get('password', ''),
|
|
)
|
|
|
|
# Count unique source->destination communication pairs
|
|
# Support multi-cluster format: analysis_id can be '{id}' or '{id}-{cluster_id}'
|
|
comm_query = f"""
|
|
SELECT count() as cnt FROM (
|
|
SELECT DISTINCT
|
|
source_namespace, source_pod,
|
|
dest_namespace, dest_pod, dest_port, protocol
|
|
FROM network_flows
|
|
WHERE analysis_id = '{analysis_id}' OR analysis_id LIKE '{analysis_id}-%'
|
|
)
|
|
"""
|
|
comm_result = await direct_repo._execute_query(comm_query)
|
|
if comm_result and len(comm_result) > 0:
|
|
communications_count = int(comm_result[0].get('cnt', 0))
|
|
|
|
# L7 unique communications (dedup across L7 tables)
|
|
try:
|
|
aid = str(analysis_id)
|
|
l7_comm_query = f"""
|
|
SELECT count() as cnt FROM (
|
|
SELECT DISTINCT src_workload, src_namespace, dst_workload, dst_namespace FROM (
|
|
SELECT src_workload, src_namespace, dst_workload, dst_namespace
|
|
FROM l7_http_flows
|
|
WHERE analysis_id = '{aid}' OR analysis_id LIKE '{aid}-%'
|
|
UNION ALL
|
|
SELECT src_workload, src_namespace, dst_workload, dst_namespace
|
|
FROM l7_grpc_flows
|
|
WHERE analysis_id = '{aid}' OR analysis_id LIKE '{aid}-%'
|
|
UNION ALL
|
|
SELECT src_workload, src_namespace, dst_workload, dst_namespace
|
|
FROM l7_dns_flows
|
|
WHERE analysis_id = '{aid}' OR analysis_id LIKE '{aid}-%'
|
|
)
|
|
)
|
|
"""
|
|
l7_comm_result = await direct_repo._execute_query(l7_comm_query)
|
|
if l7_comm_result and len(l7_comm_result) > 0:
|
|
communications_count += int(l7_comm_result[0].get('cnt', 0))
|
|
except Exception:
|
|
pass
|
|
except Exception as e:
|
|
logger.debug(f"Failed to get communications count: {e}")
|
|
|
|
# Update runs with ClickHouse data
|
|
# Strategy: ClickHouse stores events by analysis_id (not per-run), so we can't
|
|
# know exactly which events belong to which run. Best approach:
|
|
# 1. Runs with DB values > 0: Keep their recorded values
|
|
# 2. Running run: Show (total - sum of recorded runs)
|
|
# 3. Completed/stopped runs with 0: Backfill with (total - sum of recorded runs)
|
|
# but only for the LATEST such run to avoid double-counting
|
|
|
|
# Sort runs by run_number to process in order
|
|
sorted_runs = sorted(runs, key=lambda r: r.get('run_number', 0))
|
|
|
|
# Calculate sum of runs that already have recorded values
|
|
recorded_events = sum(
|
|
(r.get('events_collected') or 0) for r in sorted_runs
|
|
if (r.get('events_collected') or 0) > 0
|
|
)
|
|
recorded_comms = sum(
|
|
(r.get('communications_discovered') or 0) for r in sorted_runs
|
|
if (r.get('communications_discovered') or 0) > 0
|
|
)
|
|
|
|
# Calculate unrecorded events (what's in ClickHouse but not in any run)
|
|
unrecorded_events = max(0, total_events - recorded_events)
|
|
unrecorded_comms = max(0, communications_count - recorded_comms)
|
|
|
|
# Find which run should get the unrecorded events:
|
|
# Priority: 1) Currently running run, 2) Latest run with 0 events
|
|
target_run_id = None
|
|
|
|
# First check for running run
|
|
for run in reversed(sorted_runs):
|
|
if run.get('status') == 'running':
|
|
target_run_id = run.get('id')
|
|
break
|
|
|
|
# If no running run, find the latest completed/stopped run with 0 events
|
|
if not target_run_id:
|
|
for run in reversed(sorted_runs):
|
|
if (run.get('events_collected') or 0) == 0:
|
|
target_run_id = run.get('id')
|
|
break
|
|
|
|
# Update the target run with unrecorded events
|
|
for run in sorted_runs:
|
|
run_id = run.get('id')
|
|
run_status = run.get('status', '')
|
|
current_events = run.get('events_collected') or 0
|
|
|
|
if run_id == target_run_id:
|
|
run['events_collected'] = unrecorded_events
|
|
run['communications_discovered'] = unrecorded_comms
|
|
|
|
# Persist to DB if this is a completed/stopped/failed run (backfill)
|
|
if run_status in ('stopped', 'completed', 'failed') and current_events == 0:
|
|
try:
|
|
update_query = """
|
|
UPDATE analysis_runs
|
|
SET events_collected = :events, communications_discovered = :comms
|
|
WHERE id = :run_id
|
|
"""
|
|
await database.execute(update_query, {
|
|
"run_id": run_id,
|
|
"events": unrecorded_events,
|
|
"comms": unrecorded_comms
|
|
})
|
|
logger.info(f"Backfilled run {run_id} with events={unrecorded_events}, comms={unrecorded_comms}")
|
|
except Exception as db_err:
|
|
logger.warning(f"Failed to backfill run {run_id}: {db_err}")
|
|
|
|
logger.info("Fetched live ClickHouse data for analysis runs",
|
|
analysis_id=analysis_id,
|
|
is_running=is_running,
|
|
total_events=total_events,
|
|
communications_count=communications_count,
|
|
recorded_events=recorded_events,
|
|
unrecorded_events=unrecorded_events,
|
|
target_run_id=target_run_id)
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Failed to fetch ClickHouse data for runs: {e}")
|
|
|
|
return [AnalysisRunResponse(**run) for run in runs]
|
|
|
|
|
|
class OrphanCleanupResponse(BaseModel):
|
|
"""Response model for orphan cleanup operation"""
|
|
orphaned_analysis_ids: List[str] = []
|
|
clickhouse_deleted: dict = {}
|
|
neo4j_deleted: dict = {}
|
|
total_deleted: int = 0
|
|
duration_ms: int = 0
|
|
message: str = ""
|
|
|
|
|
|
@router.delete("/cleanup/orphaned", response_model=OrphanCleanupResponse)
|
|
async def cleanup_orphaned_data(
|
|
current_user: dict = Depends(get_current_user)
|
|
):
|
|
"""
|
|
Clean up orphaned data - data in ClickHouse/Neo4j for analyses that no longer exist in PostgreSQL.
|
|
|
|
This is useful when:
|
|
- An analysis was partially deleted
|
|
- Data was written with an analysis_id that doesn't match any existing analysis
|
|
- Manual database cleanup left inconsistent state
|
|
|
|
WARNING: This will permanently delete all orphaned data!
|
|
"""
|
|
import time
|
|
start_time = time.time()
|
|
|
|
# Step 1: Get all valid analysis IDs from PostgreSQL
|
|
valid_ids_query = "SELECT id FROM analyses"
|
|
valid_analyses = await database.fetch_all(valid_ids_query)
|
|
valid_analysis_ids = {str(row["id"]) for row in valid_analyses}
|
|
|
|
logger.info("Valid analysis IDs in PostgreSQL", count=len(valid_analysis_ids), ids=list(valid_analysis_ids))
|
|
|
|
# Step 2: Find orphaned analysis IDs in ClickHouse
|
|
orphaned_ids = set()
|
|
clickhouse_result = {"tables": {}, "total_deleted": 0}
|
|
|
|
try:
|
|
from repositories.event_repository import get_event_repository, ClickHouseEventRepository
|
|
from config import get_clickhouse_config
|
|
|
|
ch_config = get_clickhouse_config()
|
|
direct_repo = ClickHouseEventRepository(
|
|
host=ch_config.get('host', 'clickhouse'),
|
|
port=ch_config.get('port', 8123),
|
|
database=ch_config.get('database', 'flowfish'),
|
|
user=ch_config.get('user', 'default'),
|
|
password=ch_config.get('password', ''),
|
|
)
|
|
|
|
# Query for distinct analysis_ids in ClickHouse
|
|
tables = [
|
|
'network_flows', 'dns_queries', 'tcp_lifecycle', 'process_events',
|
|
'file_operations', 'capability_checks', 'oom_kills', 'bind_events',
|
|
'sni_events', 'mount_events', 'workload_metadata', 'communication_edges',
|
|
'l7_http_flows', 'l7_grpc_flows', 'l7_dns_flows',
|
|
]
|
|
|
|
for table in tables:
|
|
try:
|
|
distinct_query = f"SELECT DISTINCT analysis_id FROM {table} WHERE analysis_id != ''"
|
|
result = await direct_repo._execute_query(distinct_query)
|
|
if result:
|
|
for row in result:
|
|
aid = str(row.get("analysis_id", ""))
|
|
if aid and aid not in valid_analysis_ids:
|
|
orphaned_ids.add(aid)
|
|
except Exception as e:
|
|
logger.debug(f"Query failed for {table}: {e}")
|
|
|
|
logger.info("Found orphaned analysis IDs in ClickHouse",
|
|
count=len(orphaned_ids),
|
|
ids=list(orphaned_ids))
|
|
|
|
# Step 3: Delete orphaned data from ClickHouse
|
|
for orphan_id in orphaned_ids:
|
|
try:
|
|
aid = int(orphan_id) if orphan_id.isdigit() else orphan_id
|
|
result = await direct_repo.delete_analysis_data(
|
|
aid,
|
|
wait_for_completion=True,
|
|
timeout_seconds=30
|
|
)
|
|
for table, count in result.get("tables", {}).items():
|
|
clickhouse_result["tables"][table] = clickhouse_result["tables"].get(table, 0) + count
|
|
clickhouse_result["total_deleted"] += result.get("total_deleted", 0)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to delete orphan {orphan_id} from ClickHouse: {e}")
|
|
|
|
except Exception as e:
|
|
logger.error("ClickHouse orphan cleanup failed", error=str(e))
|
|
clickhouse_result["error"] = str(e)
|
|
|
|
# Step 4: Clean up orphaned Neo4j data
|
|
neo4j_result = {"deleted_edges": 0, "deleted_nodes": 0}
|
|
|
|
try:
|
|
from database.neo4j import neo4j_service
|
|
|
|
for orphan_id in orphaned_ids:
|
|
try:
|
|
aid = int(orphan_id) if orphan_id.isdigit() else orphan_id
|
|
result = neo4j_service.delete_analysis_data(
|
|
aid,
|
|
batch_size=5000
|
|
)
|
|
neo4j_result["deleted_edges"] += result.get("deleted_edges", 0)
|
|
neo4j_result["deleted_nodes"] += result.get("deleted_nodes", 0)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to delete orphan {orphan_id} from Neo4j: {e}")
|
|
|
|
# Also cleanup nodes with no edges (truly orphaned)
|
|
cleanup_query = """
|
|
MATCH (n) WHERE NOT EXISTS { MATCH (n)-[]-() }
|
|
WITH n LIMIT 10000
|
|
DELETE n
|
|
RETURN count(n) as deleted
|
|
"""
|
|
try:
|
|
cleanup_result = neo4j_service._execute_query(cleanup_query, {})
|
|
if cleanup_result:
|
|
neo4j_result["orphaned_nodes"] = cleanup_result[0].get("deleted", 0)
|
|
neo4j_result["deleted_nodes"] += neo4j_result["orphaned_nodes"]
|
|
except:
|
|
pass
|
|
|
|
except Exception as e:
|
|
logger.error("Neo4j orphan cleanup failed", error=str(e))
|
|
neo4j_result["error"] = str(e)
|
|
|
|
duration_ms = int((time.time() - start_time) * 1000)
|
|
|
|
total_deleted = (
|
|
clickhouse_result.get("total_deleted", 0) +
|
|
neo4j_result.get("deleted_edges", 0) +
|
|
neo4j_result.get("deleted_nodes", 0)
|
|
)
|
|
|
|
logger.info("Orphan cleanup completed",
|
|
orphaned_ids=list(orphaned_ids),
|
|
clickhouse_deleted=clickhouse_result.get("total_deleted", 0),
|
|
neo4j_deleted=neo4j_result,
|
|
duration_ms=duration_ms)
|
|
|
|
return OrphanCleanupResponse(
|
|
orphaned_analysis_ids=list(orphaned_ids),
|
|
clickhouse_deleted=clickhouse_result,
|
|
neo4j_deleted=neo4j_result,
|
|
total_deleted=total_deleted,
|
|
duration_ms=duration_ms,
|
|
message=f"Cleaned up {len(orphaned_ids)} orphaned analyses. Deleted {total_deleted:,} records in {duration_ms}ms."
|
|
)
|
|
|
|
|
|
# ============================================
|
|
# Scheduled Analysis Endpoints
|
|
# ============================================
|
|
|
|
class ScheduleAnalysisRequest(BaseModel):
|
|
"""Request to schedule recurring analysis execution"""
|
|
cron_expression: str = Field(..., max_length=100, description="Cron expression (e.g. '0 2 * * *' for daily at 02:00)")
|
|
duration_seconds: int = Field(..., ge=60, le=86400, description="Per-run duration in seconds (1 min - 24 hours)")
|
|
max_runs: Optional[int] = Field(None, ge=0, description="Maximum scheduled runs (0 or null = unlimited)")
|
|
|
|
class ScheduleAnalysisResponse(BaseModel):
|
|
"""Response for schedule operation"""
|
|
analysis_id: int
|
|
is_scheduled: bool
|
|
schedule_expression: str
|
|
schedule_duration_seconds: int
|
|
next_run_at: Optional[str] = None
|
|
message: str
|
|
|
|
|
|
@router.post("/{analysis_id}/schedule", response_model=ScheduleAnalysisResponse)
|
|
async def schedule_analysis(
|
|
analysis_id: int,
|
|
schedule_req: ScheduleAnalysisRequest,
|
|
request: Request,
|
|
current_user: dict = Depends(get_current_user)
|
|
):
|
|
"""
|
|
Schedule an analysis for recurring execution.
|
|
|
|
Sets up a cron-based schedule that will automatically start the analysis
|
|
at the specified intervals. Each run uses the same configuration as the
|
|
original analysis and auto-stops after duration_seconds.
|
|
|
|
**Permissions Required:** analysis.start
|
|
"""
|
|
if not check_analysis_permission(current_user, "start"):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Permission denied. Required role: Admin, Operator, or Analyst."
|
|
)
|
|
|
|
analysis_query = "SELECT * FROM analyses WHERE id = :id"
|
|
analysis = await database.fetch_one(analysis_query, {"id": analysis_id})
|
|
|
|
if not analysis:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Analysis with id {analysis_id} not found"
|
|
)
|
|
|
|
max_runs = schedule_req.max_runs or 0
|
|
|
|
try:
|
|
orc_result = await analysis_orchestrator_client.schedule_analysis(
|
|
analysis_id=analysis_id,
|
|
cron_expression=schedule_req.cron_expression,
|
|
duration_seconds=schedule_req.duration_seconds,
|
|
max_runs=max_runs
|
|
)
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Failed to register schedule with orchestrator: {str(e)}"
|
|
)
|
|
|
|
if not orc_result.get("success"):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=orc_result.get("message", "Orchestrator rejected the schedule")
|
|
)
|
|
|
|
update_query = """
|
|
UPDATE analyses SET
|
|
is_scheduled = true,
|
|
schedule_expression = :cron,
|
|
schedule_duration_seconds = :duration,
|
|
max_scheduled_runs = :max_runs,
|
|
next_run_at = :next_run_at,
|
|
updated_at = NOW()
|
|
WHERE id = :id
|
|
"""
|
|
await database.execute(update_query, {
|
|
"id": analysis_id,
|
|
"cron": schedule_req.cron_expression,
|
|
"duration": schedule_req.duration_seconds,
|
|
"max_runs": max_runs if max_runs > 0 else None,
|
|
"next_run_at": orc_result.get("next_run_at")
|
|
})
|
|
|
|
await activity_service.log_activity(
|
|
user_id=current_user.get('user_id'),
|
|
username=current_user.get('username', 'system'),
|
|
action="schedule",
|
|
resource_type=ActivityService.RESOURCE_ANALYSIS,
|
|
resource_id=str(analysis_id),
|
|
resource_name=analysis.get("name", f"Analysis {analysis_id}"),
|
|
ip_address=get_client_ip(request),
|
|
details={
|
|
"cron_expression": schedule_req.cron_expression,
|
|
"duration_seconds": schedule_req.duration_seconds,
|
|
"max_runs": max_runs
|
|
}
|
|
)
|
|
|
|
return ScheduleAnalysisResponse(
|
|
analysis_id=analysis_id,
|
|
is_scheduled=True,
|
|
schedule_expression=schedule_req.cron_expression,
|
|
schedule_duration_seconds=schedule_req.duration_seconds,
|
|
next_run_at=orc_result.get("next_run_at"),
|
|
message="Analysis scheduled successfully"
|
|
)
|
|
|
|
|
|
@router.delete("/{analysis_id}/schedule")
|
|
async def unschedule_analysis(
|
|
analysis_id: int,
|
|
request: Request,
|
|
current_user: dict = Depends(get_current_user)
|
|
):
|
|
"""
|
|
Remove the schedule from an analysis.
|
|
|
|
Stops recurring execution. Does not affect a currently running analysis.
|
|
|
|
**Permissions Required:** analysis.stop
|
|
"""
|
|
if not check_analysis_permission(current_user, "stop"):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Permission denied. Required role: Admin, Operator, or Analyst."
|
|
)
|
|
|
|
analysis_query = "SELECT * FROM analyses WHERE id = :id"
|
|
analysis = await database.fetch_one(analysis_query, {"id": analysis_id})
|
|
|
|
if not analysis:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Analysis with id {analysis_id} not found"
|
|
)
|
|
|
|
try:
|
|
await analysis_orchestrator_client.unschedule_analysis(analysis_id)
|
|
except Exception as e:
|
|
logger.warning("Failed to unschedule via orchestrator (clearing DB anyway)",
|
|
analysis_id=analysis_id, error=str(e))
|
|
|
|
update_query = """
|
|
UPDATE analyses SET
|
|
is_scheduled = false,
|
|
schedule_expression = NULL,
|
|
schedule_duration_seconds = NULL,
|
|
next_run_at = NULL,
|
|
max_scheduled_runs = NULL,
|
|
schedule_run_count = 0,
|
|
updated_at = NOW()
|
|
WHERE id = :id
|
|
"""
|
|
await database.execute(update_query, {"id": analysis_id})
|
|
|
|
await activity_service.log_activity(
|
|
user_id=current_user.get('user_id'),
|
|
username=current_user.get('username', 'system'),
|
|
action="unschedule",
|
|
resource_type=ActivityService.RESOURCE_ANALYSIS,
|
|
resource_id=str(analysis_id),
|
|
resource_name=analysis.get("name", f"Analysis {analysis_id}"),
|
|
ip_address=get_client_ip(request),
|
|
details={}
|
|
)
|
|
|
|
return {"message": "Schedule removed", "analysis_id": analysis_id}
|