mirror of
https://github.com/taylanbakircioglu/flowfish.git
synced 2026-09-22 02:23:33 +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
464 lines
14 KiB
Python
464 lines
14 KiB
Python
"""
|
|
Base classes for cluster connections
|
|
"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass, field
|
|
from typing import Dict, List, Optional, Any
|
|
from datetime import datetime
|
|
import structlog
|
|
|
|
logger = structlog.get_logger()
|
|
|
|
|
|
@dataclass
|
|
class ConnectionConfig:
|
|
"""Configuration for cluster connection"""
|
|
cluster_id: int
|
|
name: str
|
|
connection_type: str # 'in-cluster', 'token', 'kubeconfig'
|
|
api_server_url: Optional[str] = None
|
|
token: Optional[str] = None
|
|
ca_cert: Optional[str] = None
|
|
kubeconfig: Optional[str] = None
|
|
skip_tls_verify: bool = False
|
|
gadget_namespace: Optional[str] = None # Namespace where gadget is deployed (from UI)
|
|
gadget_endpoint: Optional[str] = None # Deprecated - not used anymore
|
|
|
|
@property
|
|
def is_remote(self) -> bool:
|
|
"""Check if this is a remote cluster connection"""
|
|
return self.connection_type.lower().replace('_', '-') in ['token', 'kubeconfig']
|
|
|
|
|
|
@dataclass
|
|
class ClusterInfo:
|
|
"""Cluster information"""
|
|
k8s_version: Optional[str] = None
|
|
platform: Optional[str] = None
|
|
total_nodes: int = 0
|
|
total_pods: int = 0
|
|
total_namespaces: int = 0
|
|
error: Optional[str] = None
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"k8s_version": self.k8s_version,
|
|
"platform": self.platform,
|
|
"total_nodes": self.total_nodes,
|
|
"total_pods": self.total_pods,
|
|
"total_namespaces": self.total_namespaces,
|
|
"error": self.error
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class GadgetHealth:
|
|
"""Inspector Gadget health status"""
|
|
health_status: str = "not_installed" # healthy, degraded, unhealthy, unknown, not_installed
|
|
version: Optional[str] = None
|
|
pods_ready: int = 0
|
|
pods_total: int = 0
|
|
error: Optional[str] = None
|
|
details: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"health_status": self.health_status,
|
|
"version": self.version,
|
|
"pods_ready": self.pods_ready,
|
|
"pods_total": self.pods_total,
|
|
"error": self.error,
|
|
"details": self.details
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class BeylaHealth:
|
|
"""Grafana Beyla L7 agent health status"""
|
|
health_status: str = "not_installed" # healthy, degraded, unhealthy, unknown, not_installed
|
|
version: str = ""
|
|
daemonset_ready: int = 0
|
|
daemonset_total: int = 0
|
|
collector_ready: bool = False
|
|
issues: List[str] = field(default_factory=list)
|
|
error: str = ""
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"health_status": self.health_status,
|
|
"version": self.version,
|
|
"daemonset_ready": self.daemonset_ready,
|
|
"daemonset_total": self.daemonset_total,
|
|
"collector_ready": self.collector_ready,
|
|
"issues": self.issues,
|
|
"error": self.error,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class Namespace:
|
|
"""Namespace information"""
|
|
name: str
|
|
uid: Optional[str] = None
|
|
status: str = "Active"
|
|
labels: Dict[str, str] = field(default_factory=dict)
|
|
created_at: Optional[str] = None
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"name": self.name,
|
|
"uid": self.uid,
|
|
"status": self.status,
|
|
"labels": self.labels,
|
|
"created_at": self.created_at
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class Deployment:
|
|
"""Deployment information"""
|
|
name: str
|
|
namespace: str
|
|
uid: Optional[str] = None
|
|
replicas: int = 0
|
|
available_replicas: int = 0
|
|
labels: Dict[str, str] = field(default_factory=dict)
|
|
annotations: Dict[str, str] = field(default_factory=dict)
|
|
image: Optional[str] = None
|
|
created_at: Optional[str] = None
|
|
spec_hash: str = ""
|
|
containers: List[Dict[str, Any]] = field(default_factory=list)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"name": self.name,
|
|
"namespace": self.namespace,
|
|
"uid": self.uid,
|
|
"replicas": self.replicas,
|
|
"available_replicas": self.available_replicas,
|
|
"labels": self.labels,
|
|
"annotations": self.annotations,
|
|
"image": self.image,
|
|
"created_at": self.created_at,
|
|
"workload_type": "deployment",
|
|
"spec_hash": self.spec_hash,
|
|
"containers": self.containers,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class Pod:
|
|
"""Pod information"""
|
|
name: str
|
|
namespace: str
|
|
uid: Optional[str] = None
|
|
status: str = "Unknown"
|
|
node_name: Optional[str] = None
|
|
labels: Dict[str, str] = field(default_factory=dict)
|
|
ip: Optional[str] = None
|
|
image: Optional[str] = None
|
|
created_at: Optional[str] = None
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"name": self.name,
|
|
"namespace": self.namespace,
|
|
"uid": self.uid,
|
|
"status": self.status,
|
|
"node_name": self.node_name,
|
|
"labels": self.labels,
|
|
"ip": self.ip,
|
|
"created_at": self.created_at
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class Service:
|
|
"""Service information"""
|
|
name: str
|
|
namespace: str
|
|
uid: Optional[str] = None
|
|
type: str = "ClusterIP"
|
|
cluster_ip: Optional[str] = None
|
|
ports: List[Dict[str, Any]] = field(default_factory=list)
|
|
labels: Dict[str, str] = field(default_factory=dict)
|
|
selector: Dict[str, str] = field(default_factory=dict)
|
|
created_at: Optional[str] = None
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"name": self.name,
|
|
"namespace": self.namespace,
|
|
"uid": self.uid,
|
|
"type": self.type,
|
|
"cluster_ip": self.cluster_ip,
|
|
"ports": self.ports,
|
|
"labels": self.labels,
|
|
"selector": self.selector,
|
|
"created_at": self.created_at
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class ConfigMap:
|
|
"""ConfigMap information with data hash (not raw data)"""
|
|
name: str
|
|
namespace: str
|
|
uid: Optional[str] = None
|
|
data_hash: str = "empty"
|
|
created_at: Optional[str] = None
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"name": self.name,
|
|
"namespace": self.namespace,
|
|
"uid": self.uid,
|
|
"data_hash": self.data_hash,
|
|
"workload_type": "configmap",
|
|
"created_at": self.created_at
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class Secret:
|
|
"""Secret information with data hash (not raw data)"""
|
|
name: str
|
|
namespace: str
|
|
uid: Optional[str] = None
|
|
data_hash: str = "empty"
|
|
type: str = "Opaque"
|
|
created_at: Optional[str] = None
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"name": self.name,
|
|
"namespace": self.namespace,
|
|
"uid": self.uid,
|
|
"data_hash": self.data_hash,
|
|
"type": self.type,
|
|
"workload_type": "secret",
|
|
"created_at": self.created_at
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class StatefulSet:
|
|
"""StatefulSet information"""
|
|
name: str
|
|
namespace: str
|
|
uid: Optional[str] = None
|
|
replicas: int = 0
|
|
ready_replicas: int = 0
|
|
labels: Dict[str, str] = field(default_factory=dict)
|
|
annotations: Dict[str, str] = field(default_factory=dict)
|
|
image: Optional[str] = None
|
|
created_at: Optional[str] = None
|
|
spec_hash: str = ""
|
|
containers: List[Dict[str, Any]] = field(default_factory=list)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"name": self.name,
|
|
"namespace": self.namespace,
|
|
"uid": self.uid,
|
|
"replicas": self.replicas,
|
|
"ready_replicas": self.ready_replicas,
|
|
"labels": self.labels,
|
|
"annotations": self.annotations,
|
|
"workload_type": "statefulset",
|
|
"image": self.image,
|
|
"created_at": self.created_at,
|
|
"spec_hash": self.spec_hash,
|
|
"containers": self.containers,
|
|
}
|
|
|
|
|
|
class ClusterConnection(ABC):
|
|
"""
|
|
Abstract base class for cluster connections.
|
|
|
|
Each connection type (in-cluster, token, kubeconfig) implements this interface.
|
|
The ClusterConnectionManager uses these connections to interact with clusters.
|
|
"""
|
|
|
|
def __init__(self, config: ConnectionConfig):
|
|
self.config = config
|
|
self.cluster_id = config.cluster_id
|
|
self._connected = False
|
|
self._last_used: Optional[datetime] = None
|
|
self._error_count = 0
|
|
|
|
@property
|
|
def is_connected(self) -> bool:
|
|
return self._connected
|
|
|
|
@property
|
|
def connection_type(self) -> str:
|
|
return self.config.connection_type
|
|
|
|
async def connect(self) -> bool:
|
|
"""
|
|
Establish connection to the cluster.
|
|
Returns True if successful, False otherwise.
|
|
"""
|
|
try:
|
|
await self._do_connect()
|
|
self._connected = True
|
|
self._error_count = 0
|
|
logger.info("Connected to cluster", cluster_id=self.cluster_id, type=self.connection_type)
|
|
return True
|
|
except Exception as e:
|
|
self._connected = False
|
|
self._error_count += 1
|
|
logger.error("Failed to connect to cluster", cluster_id=self.cluster_id, error=str(e))
|
|
return False
|
|
|
|
async def disconnect(self) -> None:
|
|
"""Close connection and cleanup resources"""
|
|
try:
|
|
await self._do_disconnect()
|
|
self._connected = False
|
|
logger.info("Disconnected from cluster", cluster_id=self.cluster_id)
|
|
except Exception as e:
|
|
logger.warning("Error during disconnect", cluster_id=self.cluster_id, error=str(e))
|
|
|
|
def mark_used(self) -> None:
|
|
"""Mark this connection as recently used"""
|
|
self._last_used = datetime.utcnow()
|
|
|
|
# Abstract methods to be implemented by subclasses
|
|
|
|
@abstractmethod
|
|
async def _do_connect(self) -> None:
|
|
"""Internal connect implementation"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def _do_disconnect(self) -> None:
|
|
"""Internal disconnect implementation"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def get_cluster_info(self) -> ClusterInfo:
|
|
"""Get cluster information (nodes, pods, namespaces count)"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def check_gadget_health(self) -> GadgetHealth:
|
|
"""Check Inspector Gadget health status"""
|
|
pass
|
|
|
|
async def check_beyla_health(self, beyla_namespace: str = "") -> BeylaHealth:
|
|
"""Check Grafana Beyla DaemonSet + L7 Collector health.
|
|
|
|
Default implementation queries pods via get_pods(); subclasses may
|
|
override with a dedicated gRPC call when available.
|
|
"""
|
|
if not beyla_namespace:
|
|
return BeylaHealth(health_status="not_installed", error="beyla_namespace not configured")
|
|
try:
|
|
pods = await self.get_pods(namespace=beyla_namespace, label_selector="app=beyla")
|
|
ds_total = len(pods)
|
|
ds_ready = sum(1 for p in pods if p.status == "Running")
|
|
|
|
collector_pods = await self.get_pods(namespace=beyla_namespace, label_selector="app=flowfish-l7-collector")
|
|
if not collector_pods:
|
|
collector_pods = await self.get_pods(namespace=beyla_namespace, label_selector="app=l7-collector")
|
|
collector_ready = any(p.status == "Running" for p in collector_pods)
|
|
|
|
issues: List[str] = []
|
|
if ds_total == 0:
|
|
issues.append("No Beyla pods found")
|
|
if ds_ready < ds_total:
|
|
issues.append(f"Only {ds_ready}/{ds_total} Beyla pods ready")
|
|
if not collector_ready:
|
|
issues.append("flowfish-l7-collector not running")
|
|
|
|
if ds_total == 0 and not collector_ready:
|
|
h_status = "not_installed"
|
|
elif ds_total == 0 and collector_ready:
|
|
h_status = "degraded"
|
|
issues.append("Collector running but no Beyla DaemonSet pods found")
|
|
elif ds_ready == ds_total and collector_ready:
|
|
h_status = "healthy"
|
|
elif ds_ready > 0 or collector_ready:
|
|
h_status = "degraded"
|
|
else:
|
|
h_status = "unhealthy"
|
|
|
|
version = ""
|
|
for p in pods:
|
|
img = p.image or ""
|
|
if "beyla" in img and ":" in img:
|
|
tag = img.rsplit(":", 1)[-1]
|
|
version = tag if tag.startswith("v") else f"v{tag}"
|
|
break
|
|
|
|
return BeylaHealth(
|
|
health_status=h_status,
|
|
version=version,
|
|
daemonset_ready=ds_ready,
|
|
daemonset_total=ds_total,
|
|
collector_ready=collector_ready,
|
|
issues=issues,
|
|
)
|
|
except Exception as e:
|
|
return BeylaHealth(health_status="unknown", error=str(e))
|
|
|
|
@abstractmethod
|
|
async def get_namespaces(self) -> List[Namespace]:
|
|
"""Get list of namespaces"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def get_deployments(self, namespace: Optional[str] = None) -> List[Deployment]:
|
|
"""Get list of deployments"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def get_pods(self, namespace: Optional[str] = None, label_selector: Optional[str] = None) -> List[Pod]:
|
|
"""Get list of pods"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def get_services(self, namespace: Optional[str] = None) -> List[Service]:
|
|
"""Get list of services"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def get_statefulsets(self, namespace: Optional[str] = None) -> List[StatefulSet]:
|
|
"""Get list of statefulsets"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def get_configmaps(self, namespace: Optional[str] = None) -> List[ConfigMap]:
|
|
"""Get list of configmaps with data hash"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def get_secrets(self, namespace: Optional[str] = None) -> List[Secret]:
|
|
"""Get list of secrets with data hash"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def get_network_policies(self, namespace: Optional[str] = None) -> List[Dict[str, Any]]:
|
|
"""Get list of network policies with spec hash"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def get_ingresses(self, namespace: Optional[str] = None) -> List[Dict[str, Any]]:
|
|
"""Get list of ingresses with spec hash"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def get_routes(self, namespace: Optional[str] = None) -> List[Dict[str, Any]]:
|
|
"""Get list of OpenShift routes with spec hash"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def get_labels(self, namespace: Optional[str] = None) -> List[str]:
|
|
"""Get unique labels from resources"""
|
|
pass
|
|
|