Real-time Monitoring and SSE Events

┌─────────────────────────────────────────────────────────────────┐
│                      FRONTEND CLIENTS                           │
│  ┌────────────┐  ┌────────────┐  ┌────────────┐                │
│  │  Browser 1 │  │  Browser 2 │  │  Browser N │                │
│  └──────┬─────┘  └──────┬─────┘  └──────┬─────┘                │
│         │                │                │                      │
│         │ SSE Connection │                │                      │
│         └────────────────┴────────────────┘                      │
│                          │                                       │
└──────────────────────────┼───────────────────────────────────────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────────────────┐
│                   BACKEND API SERVER                            │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │          SSE Endpoint (/api/v1/events/stream)            │  │
│  │  - Handles EventSource connections                       │  │
│  │  - Authenticates via JWT token in query parameter        │  │
│  │  - Subscribes to Redis pub/sub channel                   │  │
│  │  - Streams events to clients                             │  │
│  └─────────────────────┬────────────────────────────────────┘  │
│                        │                                         │
│                        ▼                                         │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │              Redis Pub/Sub (SSE Channel)                 │  │
│  │  - Channel: "sse:events"                                 │  │
│  │  - Cross-worker event distribution                       │  │
│  │  - Organization-scoped filtering                         │  │
│  └──────────────────────┬────────────────────────────────────┘ │
│                         │                                        │
│         ┌───────────────┴───────────────┐                       │
│         ▼                               ▼                       │
│  ┌─────────────┐               ┌──────────────┐                │
│  │  Services   │               │  ARQ Workers │                │
│  │  - Alerts   │               │  - Monitoring│                │
│  │  - Replica  │               │  - Tasks     │                │
│  │  - Tasks    │               │  - Backups   │                │
│  └─────────────┘               └──────────────┘                │
└─────────────────────────────────────────────────────────────────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────────────────┐
│                   NETDATA MONITORING                            │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │           Parent Netdata (on PaaSPortal server)          │  │
│  │  - Collects metrics from all child agents                │  │
│  │  - REST API for metrics queries                          │  │
│  │  - Webhook alerts to backend                             │  │
│  └─────────────────────┬────────────────────────────────────┘  │
│                        │                                         │
│                        ▼                                         │
│         ┌──────────────┴────────────────┐                       │
│         ▼                                ▼                       │
│  ┌─────────────┐                 ┌─────────────┐               │
│  │ Child Agent │                 │ Child Agent │               │
│  │ (Server 1)  │                 │ (Server N)  │               │
│  │ - CPU/RAM   │                 │ - CPU/RAM   │               │
│  │ - Disk      │                 │ - Disk      │               │
│  │ - Containers│                 │ - Containers│               │
│  └─────────────┘                 └─────────────┘               │
└─────────────────────────────────────────────────────────────────┘

# backend/api/v1/routes/events.py
 
import redis.asyncio as redis
from fastapi import APIRouter, Query, Request
from fastapi.responses import StreamingResponse
 
SSE_CHANNEL = "sse:events"
 
async def broadcast_to_organization(
    org_id: str | UUID,
    event_type: str,
    data: dict
) -> None:
    """Broadcast an event via Redis pub/sub to all connected SSE clients."""
    r = redis.from_url(get_redis_url())
    message = json.dumps({
        "org_id": str(org_id),
        "type": event_type,
        "data": data
    })
    await r.publish(SSE_CHANNEL, message)
    await r.aclose()

@router.get("/stream")
async def event_stream(
    request: Request,
    token: str = Query(..., description="JWT access token"),
):
    """SSE endpoint for real-time updates using Redis pub/sub."""
    user_id = await get_user_from_token(token)
 
    return StreamingResponse(
        event_generator(request, user_id),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",  # Disable nginx buffering
        },
    )

async def event_generator(
    request: Request,
    user_id: str
) -> AsyncGenerator[str, None]:
    """Generate SSE events for a client using Redis pub/sub."""
    r = redis.from_url(get_redis_url())
    pubsub = r.pubsub()
    await pubsub.subscribe(SSE_CHANNEL)
 
    # Send initial connection confirmation
    yield "event: connected\ndata: {\"status\": \"connected\"}\n\n"
 
    while True:
        # Check if client disconnected
        if await request.is_disconnected():
            break
 
        try:
            # Wait for Redis messages with timeout
            message = await asyncio.wait_for(
                pubsub.get_message(ignore_subscribe_messages=True, timeout=1.0),
                timeout=30.0
            )
 
            if message and message["type"] == "message":
                data = json.loads(message["data"])
                event_type = data.get("type", "message")
                event_data = json.dumps({
                    "type": event_type,
                    "data": data.get("data", {})
                })
 
                yield f"event: {event_type}\ndata: {event_data}\n\n"
 
        except TimeoutError:
            # Send keepalive ping every 30 seconds
            yield ": keepalive\n\n"
 
    await pubsub.unsubscribe(SSE_CHANNEL)
    await pubsub.aclose()
    await r.aclose()

{
  "type": "deployment.progress",
  "data": {
    "environment_id": "a1b2c3d4-...",
    "deployment_id": "e5f6g7h8-...",
    "progress_percent": 45,
    "current_step": "Building Docker image"
  }
}

{
  "type": "backup.completed",
  "data": {
    "environment_id": "a1b2c3d4-...",
    "backup_id": "b9c0d1e2-...",
    "backup_type": "manual",
    "size_bytes": 524288000,
    "duration_seconds": 45
  }
}

{
  "type": "replica.health_updated",
  "data": {
    "environment_id": "a1b2c3d4-...",
    "status": "online",
    "lag_bytes": 12582912,  // ~12MB
    "lag_seconds": 3.5,
    "last_check": "2024-12-11T10:30:00Z"
  }
}

{
  "type": "alert_triggered",
  "data": {
    "alert_id": "c3d4e5f6-...",
    "alert_type": "cpu_high",
    "severity": "warning",
    "title": "High CPU Usage",
    "message": "Server CPU usage at 85% for 10 minutes",
    "server_id": "d7e8f9g0-...",
    "server_name": "prod-server-1",
    "metric_value": 85.3,
    "threshold": 80.0,
    "triggered_at": "2024-12-11T10:35:00Z"
  }
}

{
  "type": "server.health_updated",
  "data": {
    "server_id": "f1g2h3i4-...",
    "server_name": "prod-server-2",
    "cpu_percent": 45.2,
    "memory_percent": 67.8,
    "disk_percent": 55.3,
    "status": "healthy"
  }
}

{
  "type": "task_status",
  "data": {
    "task_id": "g5h6i7j8-...",
    "task_type": "deploy",
    "status": "in_progress",
    "environment_id": "a1b2c3d4-...",
    "progress_percent": 60,
    "current_step": "Starting containers"
  }
}

{
  "type": "quota.warning",
  "data": {
    "organization_id": "h9i0j1k2-...",
    "resource_type": "ram_mb",
    "used": 15360,
    "limit": 16384,
    "percent_used": 93.75
  }
}

{
  "type": "permissions_changed",
  "data": {
    "user_id": "i3j4k5l6-...",
    "organization_id": "h9i0j1k2-...",
    "project_id": "m7n8o9p0-...",
    "changed_permissions": ["project.environments.create", "project.deployments.trigger"]
  }
}

{
  "type": "migration_progress",
  "data": {
    "migration_id": "j5k6l7m8-...",
    "status": "downloading",
    "progress_percent": 35,
    "current_step": "Downloading chunk 3 of 8",
    "error_message": null
  }
}

┌──────────────┐         ┌──────────────┐         ┌──────────────┐
│  Worker 1    │         │  Worker 2    │         │  Worker 3    │
│  (ARQ Task)  │         │  (API Server)│         │  (API Server)│
└──────┬───────┘         └──────┬───────┘         └──────┬───────┘
       │                        │                        │
       │ publish()              │ subscribe()            │ subscribe()
       └────────────────────────┼────────────────────────┘
                                │
                    ┌───────────▼──────────┐
                    │   Redis Pub/Sub      │
                    │  Channel: sse:events │
                    └───────────┬──────────┘
                                │
                  ┌─────────────┴─────────────┐
                  │                           │
          ┌───────▼────────┐         ┌───────▼────────┐
          │  Client A      │         │  Client B      │
          │  (Browser)     │         │  (Browser)     │
          └────────────────┘         └────────────────┘

{
  "org_id": "uuid-of-organization",
  "type": "event_type",
  "data": {
    "field1": "value1",
    "field2": "value2"
  }
}

import json
 
message = json.dumps({
    "org_id": str(org_id),
    "type": event_type,
    "data": data
})
 
await redis_client.publish(SSE_CHANNEL, message)

export function useEventStream() {
  const { isAuthenticated } = useAuthStore();
  const eventSourceRef = useRef<EventSource | null>(null);
  const reconnectAttempts = useRef(0);
  const maxReconnectAttempts = 10;
  const baseReconnectDelay = 1000; // 1 second
 
  const connect = useCallback(() => {
    const token = getAccessToken();
    if (!token || !isAuthenticated) return;
 
    const url = `${API_URL}/events/stream?token=${encodeURIComponent(token)}`;
    const eventSource = new EventSource(url);
    eventSourceRef.current = eventSource;
 
    eventSource.onopen = () => {
      console.log("[SSE] Connected");
      reconnectAttempts.current = 0;
    };
 
    eventSource.onerror = (error) => {
      console.error("[SSE] Error:", error);
      eventSource.close();
 
      // Exponential backoff reconnection
      if (reconnectAttempts.current < maxReconnectAttempts) {
        const delay = baseReconnectDelay * Math.pow(2, reconnectAttempts.current);
        setTimeout(() => {
          reconnectAttempts.current++;
          connect();
        }, delay);
      }
    };
 
    // Listen for specific event types
    eventSource.addEventListener("environment_status", (e) => {
      const parsed = JSON.parse(e.data);
      dispatchEvent(parsed);
    });
 
    eventSource.addEventListener("alert_triggered", (e) => {
      const parsed = JSON.parse(e.data);
      dispatchEvent({ type: "alert_triggered", data: parsed });
    });
 
    // ... more event listeners
  }, [isAuthenticated]);
 
  useEffect(() => {
    connect();
    return () => {
      if (eventSourceRef.current) {
        eventSourceRef.current.close();
      }
    };
  }, [connect]);
 
  return {
    isConnected: eventSourceRef.current?.readyState === EventSource.OPEN,
  };
}

export function SSEProvider({ children }: { children: ReactNode }) {
  const { isConnected } = useEventStream();
 
  return (
    <SSEContext.Provider value={{ isConnected }}>
      {children}
    </SSEContext.Provider>
  );
}

import { useSSEEvent } from "@/hooks/useEventStream";
 
export function DeploymentStatus({ environmentId }: Props) {
  const [status, setStatus] = useState<string>("idle");
  const [progress, setProgress] = useState<number>(0);
 
  // Subscribe to deployment progress events
  useSSEEvent("deployment.progress", (event) => {
    if (event.data.environment_id === environmentId) {
      setProgress(event.data.progress_percent);
      setStatus(event.data.current_step);
    }
  });
 
  // Subscribe to deployment completion
  useSSEEvent("deployment.completed", (event) => {
    if (event.data.environment_id === environmentId) {
      setStatus("completed");
      setProgress(100);
      toast.success("Deployment completed successfully!");
    }
  });
 
  return (
    <div>
      <ProgressBar percent={progress} />
      <span>{status}</span>
    </div>
  );
}

useSSEEvent("*", (event) => {
  console.log("Received event:", event.type, event.data);
 
  // Log all events for debugging
  if (isDevelopment) {
    console.table({
      Type: event.type,
      Data: JSON.stringify(event.data, null, 2),
    });
  }
});

// Subscribe to specific event in one component
useSSEEvent("replica.health_updated", handleReplicaUpdate);

// Subscribe to multiple related events
useSSEEvent("deployment.started", handleDeploymentStart);
useSSEEvent("deployment.progress", handleDeploymentProgress);
useSSEEvent("deployment.completed", handleDeploymentComplete);
useSSEEvent("deployment.failed", handleDeploymentFailed);

// Subscribe and filter by resource ID
useSSEEvent("alert_triggered", (event) => {
  if (event.data.server_id === currentServerId) {
    showAlert(event.data);
  }
});

// Show toast notifications for critical events
useSSEEvent("alert_triggered", (event) => {
  if (event.data.severity === "critical") {
    toast.error(event.data.title, {
      description: event.data.message,
    });
  }
});

┌─────────────────────────────────────────────────────────────────┐
│              PAASPORTAL SERVER (Parent Netdata)                 │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │          Netdata Parent Container                        │  │
│  │  - Port: 19999                                           │  │
│  │  - Collects metrics from all child agents                │  │
│  │  - REST API: /api/v1/...                                 │  │
│  │  - Web UI: https://monitoring.paasportal.io              │  │
│  └──────────────────────────────────────────────────────────┘  │
└────────────────────────────────┬────────────────────────────────┘
                                 │ Streaming (port 19999)
                 ┌───────────────┴───────────────┐
                 │                               │
┌────────────────▼─────────┐    ┌───────────────▼────────────┐
│  MANAGED SERVER 1        │    │  MANAGED SERVER N          │
│  ┌────────────────────┐  │    │  ┌────────────────────┐    │
│  │ Netdata Child Agent│  │    │  │ Netdata Child Agent│    │
│  │ - Machine GUID     │  │    │  │ - Machine GUID     │    │
│  │ - Streams to parent│  │    │  │ - Streams to parent│    │
│  │ - Collects metrics │  │    │  │ - Collects metrics │    │
│  └────────────────────┘  │    │  └────────────────────┘    │
│                          │    │                            │
│  Docker Containers:      │    │  Docker Containers:        │
│  - Odoo                  │    │  - Odoo                    │
│  - PostgreSQL            │    │  - PostgreSQL              │
│  - Redis                 │    │  - Redis                   │
└──────────────────────────┘    └────────────────────────────┘

[STREAM_API_KEY]
    # Enable accepting streamed metrics
    enabled = yes
 
    # Allow connections from child agents
    allow from = *
 
    # API key for authentication
    api key = YOUR_STREAM_API_KEY
 
    # Retention for streamed metrics
    history = 3600
    memory mode = dbengine

# In bootstrap_server.sh
 
# Install Netdata
bash <(curl -Ss https://my-netdata.io/kickstart.sh) \
  --stable-channel \
  --disable-telemetry \
  --claim-token ${NETDATA_CLAIM_TOKEN} \
  --claim-url https://app.netdata.cloud
 
# Configure streaming to parent
cat > /etc/netdata/stream.conf <<EOF
[stream]
    enabled = yes
    destination = ${NETDATA_PARENT_HOST}:19999
    api key = ${NETDATA_STREAM_API_KEY}
    timeout seconds = 60
    buffer size bytes = 1048576
    reconnect delay seconds = 5
EOF
 
# Restart Netdata
systemctl restart netdata

class VM(Base):
    # ... other fields ...
 
    # Netdata streaming configuration
    netdata_streaming_enabled = Column(Boolean, default=False)
    netdata_streaming_status = Column(String(50), nullable=True)
    netdata_machine_guid = Column(String(36), nullable=True)
    netdata_version = Column(String(20), nullable=True)
    netdata_installed_at = Column(DateTime(timezone=True), nullable=True)

# backend/services/monitoring_service.py
 
CPU_WARNING_THRESHOLD = 80
CPU_CRITICAL_THRESHOLD = 95
MEMORY_WARNING_THRESHOLD = 80
MEMORY_CRITICAL_THRESHOLD = 95
DISK_WARNING_THRESHOLD = 80
DISK_CRITICAL_THRESHOLD = 90
 
def _calculate_status(cpu: float, memory: float, disk: float) -> str:
    """Calculate overall health status from metrics."""
    if (cpu >= CPU_CRITICAL_THRESHOLD or
        memory >= MEMORY_CRITICAL_THRESHOLD or
        disk >= DISK_CRITICAL_THRESHOLD):
        return "critical"
 
    if (cpu >= CPU_WARNING_THRESHOLD or
        memory >= MEMORY_WARNING_THRESHOLD or
        disk >= DISK_WARNING_THRESHOLD):
        return "warning"
 
    return "healthy"

class MonitoringService:
    """Service for querying metrics on-demand from Netdata."""
 
    async def get_server_summary(self, server_id: UUID) -> MetricsSummary:
        """Level 1: Quick summary for cards."""
        server = await self._get_server(server_id)
 
        async with get_netdata_client(server) as client:
            system_stats = await client.get_system_stats()
            container_stats = await client.get_container_stats()
 
        return self._to_summary(system_stats, container_stats)
 
    async def get_server_metrics(
        self,
        server_id: UUID,
        include_timeseries: bool = False,
        time_range: str = "24h"
    ) -> ServerMetricsResponse:
        """Level 2: Full metrics with optional charts."""
        server = await self._get_server(server_id)
 
        async with get_netdata_client(server) as client:
            system_stats = await client.get_system_stats()
            container_stats = await client.get_container_stats()
 
            timeseries = None
            if include_timeseries:
                history = await client.get_historical_stats(minutes=parse_time_range(time_range))
                timeseries = self._format_timeseries(history)
 
        return ServerMetricsResponse(
            server_id=server_id,
            server_name=server.name,
            summary=self._to_summary(system_stats, container_stats),
            containers=[self._to_container_metrics(c) for c in container_stats],
            timeseries=timeseries,
            queried_at=datetime.now(UTC),
        )

class NetdataClient:
    """Client for querying Netdata API with dual-mode support."""
 
    def __init__(self, vm: VM, use_parent: bool | None = None):
        """
        Initialize client.
 
        Args:
            vm: VM to query metrics for
            use_parent: Override mode. If None, auto-detect from vm.netdata_streaming_enabled
        """
        self.vm = vm
        self.use_parent = use_parent if use_parent is not None else vm.netdata_streaming_enabled
 
    async def _query_parent_api(self, endpoint: str) -> dict:
        """Query parent Netdata for this VM's metrics."""
        if not self.vm.netdata_machine_guid:
            raise NetdataClientError("VM has no machine_guid")
 
        # Parent exposes child nodes at /host/{machine_guid}/...
        url = f"{PARENT_URL}/host/{self.vm.netdata_machine_guid}{endpoint}"
 
        response = await self._http_client.get(url)
        return response.json()
 
    async def get_system_stats(self) -> dict:
        """Get current system statistics."""
        # Query CPU, memory, disk, network from Netdata charts
        cpu_data = await self.get_chart_data("system.cpu", after=-1, points=1)
        mem_data = await self.get_chart_data("system.ram", after=-1, points=1)
        disk_data = await self.get_chart_data("disk_space._", after=-1, points=1)
        net_data = await self.get_chart_data("system.net", after=-1, points=1)
 
        return {
            "cpu": {"percent": parse_cpu_percent(cpu_data)},
            "memory": {"used": parse_mem_used(mem_data), "total": parse_mem_total(mem_data), "percent": parse_mem_percent(mem_data)},
            "disk": {"used": parse_disk_used(disk_data), "total": parse_disk_total(disk_data), "percent": parse_disk_percent(disk_data)},
            "network": {"rx_bytes": parse_net_rx(net_data), "tx_bytes": parse_net_tx(net_data)},
            "timestamp": datetime.now(UTC).isoformat(),
        }

async def get_container_stats(self, container_name: str | None = None) -> list[dict]:
    """Get container statistics from Netdata or docker ps."""
 
    if self.use_parent:
        # Query cgroup charts from Netdata
        charts = await self._query_api("/api/v1/charts")
        containers = []
 
        for chart_name in charts["charts"].keys():
            if chart_name.startswith("cgroup_") and ".cpu" in chart_name:
                cname = chart_name.replace("cgroup_", "").rsplit(".cpu", 1)[0]
 
                # Get CPU, memory, network for this container
                cpu_percent = await self._get_container_cpu(cname)
                memory_used, memory_limit = await self._get_container_memory(cname)
                network_rx, network_tx = await self._get_container_network(cname)
 
                containers.append({
                    "container_name": cname,
                    "cpu_percent": cpu_percent,
                    "memory_used": memory_used,
                    "memory_limit": memory_limit,
                    "network_rx": network_rx,
                    "network_tx": network_tx,
                    "status": "running",
                })
 
        return containers
 
    else:
        # SSH mode: Use docker ps
        cmd = "docker ps -a --format '{{.Names}}|{{.Status}}|{{.RunningFor}}'"
        exit_code, stdout, stderr = self.ssh.execute_command(cmd)
 
        containers = []
        for line in stdout.strip().split('\n'):
            parts = line.split('|')
            containers.append({
                "container_name": parts[0],
                "status": parse_docker_status(parts[1]),
                "uptime": parse_uptime(parts[2]),
            })
 
        return containers

# backend/services/replication_monitor.py
 
LAG_WARNING_BYTES = 50 * 1024 * 1024    # 50 MB
LAG_CRITICAL_BYTES = 100 * 1024 * 1024  # 100 MB
LAG_WARNING_SECONDS = 15
LAG_CRITICAL_SECONDS = 30
 
if lag_bytes > LAG_CRITICAL_BYTES or lag_seconds > LAG_CRITICAL_SECONDS:
    status = "lagging"  # Critical lag
elif lag_bytes > LAG_WARNING_BYTES or lag_seconds > LAG_WARNING_SECONDS:
    status = "lagging"  # Warning lag
else:
    status = "online"   # Healthy

async def _query_replication_stats(self, environment: ProjectEnvironment) -> dict:
    """Query pg_stat_replication view on primary PostgreSQL container."""
 
    primary_container = f"postgres-primary-{environment.id}"
 
    sql_query = """
        SELECT
            state,
            pg_wal_lsn_diff(sent_lsn, flush_lsn) AS lag_bytes,
            EXTRACT(EPOCH FROM write_lag) AS lag_seconds
        FROM pg_stat_replication
        WHERE application_name = 'replica'
        LIMIT 1;
    """
 
    query_cmd = (
        f"docker exec {primary_container} "
        f"psql -U odoo -d postgres -t -A -F '|' "
        f"-c \"{sql_query}\""
    )
 
    ssh_manager = get_ssh_manager_for_vm(environment.vm)
    ssh_manager.connect()
 
    exit_code, stdout, stderr = ssh_manager.execute_command(query_cmd, timeout=30)
 
    # Parse output: state|lag_bytes|lag_seconds
    parts = stdout.strip().split("|")
    state = parts[0].strip()
    lag_bytes = int(float(parts[1])) if parts[1] else 0
    lag_seconds = float(parts[2]) if parts[2] else 0.0
 
    return {
        "state": state,
        "lag_bytes": lag_bytes,
        "lag_seconds": lag_seconds,
    }

async def _emit_health_updated_event(
    self,
    environment: ProjectEnvironment,
    status: str,
    lag_bytes: int | None,
    lag_seconds: float | None
) -> None:
    """Emit SSE event for replica health update."""
 
    org_id = environment.project.organization_id
 
    event_data = {
        "environment_id": str(environment.id),
        "status": status,
        "lag_bytes": lag_bytes,
        "lag_seconds": lag_seconds,
        "last_check": environment.replica_last_check.isoformat(),
    }
 
    await broadcast_to_organization(
        org_id=org_id,
        event_type="replica.health_updated",
        data=event_data
    )

// Subscribe to replica health updates
useSSEEvent("replica.health_updated", (event) => {
  if (event.data.environment_id === environmentId) {
    setReplicaStatus(event.data.status);
    setLagBytes(event.data.lag_bytes);
    setLagSeconds(event.data.lag_seconds);
 
    // Show alert if lagging
    if (event.data.status === "lagging") {
      toast.warning("Database replica is lagging", {
        description: `Lag: ${formatBytes(event.data.lag_bytes)} / ${event.data.lag_seconds}s`,
      });
    }
  }
});

# backend/models/deployment.py
 
class Deployment(Base):
    id = Column(UUID, primary_key=True)
    environment_id = Column(UUID, ForeignKey("project_environments.id"))
 
    status = Column(SQLEnum(DeploymentStatus))
    created_at = Column(DateTime(timezone=True), default=datetime.now(UTC))
    started_at = Column(DateTime(timezone=True), nullable=True)
    completed_at = Column(DateTime(timezone=True), nullable=True)
 
    @property
    def duration_seconds(self) -> int | None:
        """Calculate deployment duration in seconds."""
        if not self.started_at or not self.completed_at:
            return None
        return int((self.completed_at - self.started_at).total_seconds())

{
  "type": "deployment.completed",
  "data": {
    "deployment_id": "...",
    "duration_seconds": 127
  }
}

# backend/models/backup.py
 
class Backup(Base):
    id = Column(UUID, primary_key=True)
    environment_id = Column(UUID, ForeignKey("project_environments.id"))
 
    started_at = Column(DateTime(timezone=True))
    completed_at = Column(DateTime(timezone=True), nullable=True)
    size_bytes = Column(BigInteger, nullable=True)
 
    @property
    def duration_seconds(self) -> int | None:
        """Calculate backup duration."""
        if not self.started_at or not self.completed_at:
            return None
        return int((self.completed_at - self.started_at).total_seconds())

# backend/core/middleware.py
 
@app.middleware("http")
async def add_timing_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    duration_ms = (time.time() - start_time) * 1000
    response.headers["X-Response-Time"] = f"{duration_ms:.2f}ms"
    return response

async def get_historical_stats(self, minutes: int = 60) -> dict:
    """Get historical statistics for charts."""
    after = -minutes * 60  # Convert to seconds
 
    cpu_data = await self.get_chart_data("system.cpu", after=after, points=minutes)
    mem_data = await self.get_chart_data("system.ram", after=after, points=minutes)
 
    return {
        "cpu": [{"timestamp": ts, "value": val} for ts, val in cpu_data],
        "memory": [{"timestamp": ts, "value": val} for ts, val in mem_data],
    }

const { data } = useQuery({
  queryKey: ["server-metrics-history", serverId, timeRange],
  queryFn: () => api.get(`/servers/${serverId}/metrics`, {
    params: { include_timeseries: true, time_range: timeRange }
  }),
});
 
// Render with uPlot or Recharts
<LineChart data={data.timeseries.cpu} />

# backend/models/alert.py
 
class AlertRule(Base):
    id = Column(UUID, primary_key=True)
    organization_id = Column(UUID, ForeignKey("organizations.id"), nullable=True)
 
    name = Column(String(255), nullable=False)
    description = Column(Text)
 
    alert_type = Column(String(50), nullable=False)
    severity = Column(SQLEnum(AlertSeverity), nullable=False)
 
    # Threshold configuration
    metric_name = Column(String(100), nullable=False)
    operator = Column(String(10), default=">")  # >, <, >=, <=, ==
    threshold = Column(Float, nullable=False)
    duration_seconds = Column(Integer, default=300)  # Must breach for this long
 
    # Notification
    notify_channels = Column(JSON, default=list)  # ["email", "slack"]
    cooldown_seconds = Column(Integer, default=3600)  # 1 hour between alerts
 
    is_active = Column(Boolean, default=True)

DEFAULT_ALERT_RULES = [
    {
        "name": "High CPU",
        "alert_type": "cpu_high",
        "severity": "warning",
        "metric_name": "cpu_percent",
        "operator": ">",
        "threshold": 80,
        "duration_seconds": 600,  # 10 minutes
        "notify_channels": ["email"],
        "cooldown_seconds": 3600,  # 1 hour
    },
    {
        "name": "Disk Space Critical",
        "alert_type": "disk_high",
        "severity": "critical",
        "metric_name": "disk_percent",
        "operator": ">",
        "threshold": 95,
        "duration_seconds": 60,
        "notify_channels": ["email", "slack"],
        "cooldown_seconds": 1800,  # 30 minutes
    },
]

# backend/services/alert_service.py
 
class AlertService:
    async def create_alert(
        self,
        server_id: UUID,
        alert_type: AlertType,
        severity: AlertSeverity,
        title: str,
        message: str | None = None,
        metric_value: float | None = None,
        threshold: float | None = None,
    ) -> AlertEvent:
        """Create alert with deduplication."""
 
        # Generate fingerprint for deduplication
        fingerprint = self._generate_fingerprint(server_id, alert_type, metric_name)
 
        # Check for existing unresolved alert within cooldown
        existing = await self._find_existing_alert(fingerprint, cooldown_seconds=3600)
 
        if existing:
            # Update occurrence count instead of creating new
            existing.occurrence_count += 1
            existing.last_occurrence = datetime.now(UTC)
            existing.metric_value = metric_value
            await self.db.commit()
            return existing
 
        # Create new alert
        alert = AlertEvent(
            organization_id=server.organization_id,
            server_id=server_id,
            alert_type=alert_type,
            severity=severity,
            title=title,
            message=message,
            metric_value=metric_value,
            threshold=threshold,
            status=AlertStatus.TRIGGERED,
            fingerprint=fingerprint,
            occurrence_count=1,
        )
 
        self.db.add(alert)
        await self.db.commit()
 
        # Broadcast via SSE
        await broadcast_to_organization(
            org_id=alert.organization_id,
            event_type="alert_triggered",
            data={
                "alert_id": str(alert.id),
                "alert_type": alert_type.value,
                "severity": severity.value,
                "title": title,
                "metric_value": metric_value,
                "threshold": threshold,
            },
        )
 
        return alert

class AlertStatus(str, Enum):
    TRIGGERED = "triggered"      # New alert, not yet seen
    ACKNOWLEDGED = "acknowledged"  # User acknowledged, investigating
    RESOLVED = "resolved"          # Issue fixed
 
async def acknowledge_alert(self, alert_id: UUID, user_id: UUID) -> AlertEvent:
    """Mark alert as acknowledged."""
    alert = await self.db.get(AlertEvent, alert_id)
 
    alert.status = AlertStatus.ACKNOWLEDGED
    alert.acknowledged_at = datetime.now(UTC)
    alert.acknowledged_by = user_id
 
    await self.db.commit()
 
    # Broadcast status change
    await broadcast_to_organization(
        org_id=alert.organization_id,
        event_type="alert_acknowledged",
        data={
            "alert_id": str(alert.id),
            "acknowledged_by": str(user_id),
        },
    )
 
    return alert
 
async def resolve_alert(self, alert_id: UUID) -> AlertEvent:
    """Mark alert as resolved."""
    alert = await self.db.get(AlertEvent, alert_id)
 
    alert.status = AlertStatus.RESOLVED
    alert.resolved_at = datetime.now(UTC)
 
    await self.db.commit()
 
    # Broadcast resolution
    await broadcast_to_organization(
        org_id=alert.organization_id,
        event_type="alert_resolved",
        data={"alert_id": str(alert.id)},
    )
 
    return alert

await email_service.send_alert_notification(
    to=user.email,
    alert=alert_event,
    template="alert_triggered"
)

await broadcast_to_organization(
    org_id=alert.organization_id,
    event_type="alert_triggered",
    data=alert_data
)

async def send_slack_alert(webhook_url: str, alert: AlertEvent):
    payload = {
        "attachments": [{
            "color": "#FF0000" if alert.severity == "critical" else "#FFCC00",
            "title": f"{alert.severity.upper()}: {alert.title}",
            "text": alert.message,
            "fields": [
                {"title": "Server", "value": alert.server.name, "short": True},
                {"title": "Value", "value": str(alert.metric_value), "short": True},
            ],
        }]
    }
 
    async with httpx.AsyncClient() as client:
        await client.post(webhook_url, json=payload)

const { data: rules } = useQuery({
  queryKey: ["alert-rules"],
  queryFn: () => api.get("/monitoring/alerts/rules"),
});
 
// Create new rule
await api.post("/monitoring/alerts/rules", {
  name: "High Memory (Custom)",
  alert_type: "memory_high",
  severity: "warning",
  metric_name: "memory_percent",
  operator: ">",
  threshold: 85,
  duration_seconds: 300,
  notify_channels: ["email", "slack"],
  cooldown_seconds: 1800,
});

# .env.production
 
# Netdata parent URL (public-facing)
NETDATA_PARENT_PUBLIC_URL=https://monitoring.paasportal.io
 
# Netdata parent API URL (internal)
NETDATA_PARENT_URL=http://netdata-parent:19999
 
# API key for authentication
NETDATA_PARENT_API_KEY=your-api-key-here

services:
  netdata-parent:
    image: netdata/netdata:stable
    container_name: netdata-parent
    hostname: netdata-parent
    restart: unless-stopped
 
    ports:
      - "19999:19999"
 
    volumes:
      - netdata-config:/etc/netdata
      - netdata-lib:/var/lib/netdata
      - netdata-cache:/var/cache/netdata
 
    environment:
      - NETDATA_CLAIM_TOKEN=${NETDATA_CLAIM_TOKEN}
      - NETDATA_CLAIM_URL=https://app.netdata.cloud
 
    cap_add:
      - SYS_PTRACE
 
    security_opt:
      - apparmor:unconfined
 
    labels:
      - "traefik.enable=true"
      - "traefik.http.routers.netdata.rule=Host(`monitoring.paasportal.io`)"
      - "traefik.http.routers.netdata.tls.certresolver=letsencrypt"
      - "traefik.http.services.netdata.loadbalancer.server.port=19999"
 
volumes:
  netdata-config:
  netdata-lib:
  netdata-cache:

[STREAM_API_KEY]
    # Enable receiving streamed metrics
    enabled = yes
 
    # Allow from any IP (child agents)
    allow from = *
 
    # Stream API key (must match child configuration)
    api key = YOUR_STREAM_API_KEY
 
    # Retention for streamed metrics (1 hour = 3600 seconds)
    history = 3600
 
    # Storage mode
    memory mode = dbengine
 
    # Health monitoring
    health enabled = yes

[stream]
    # Enable streaming to parent
    enabled = yes
 
    # Parent Netdata address
    destination = monitoring.paasportal.io:19999
 
    # Stream API key (must match parent)
    api key = YOUR_STREAM_API_KEY
 
    # Connection settings
    timeout seconds = 60
    buffer size bytes = 1048576
    reconnect delay seconds = 5
 
    # Send all charts
    send charts matching = *

# On parent server
curl http://localhost:19999/api/v1/info | jq .
 
# Response includes streaming hosts
{
  "streaming": {
    "status": "enabled",
    "nodes": [
      {
        "guid": "abc123...",
        "hostname": "prod-server-1",
        "status": "connected"
      }
    ]
  }
}

{
  "cpu_percent": 45.2,
  "memory_percent": 67.8,
  "memory_used_bytes": 5637144576,
  "memory_total_bytes": 8318521344,
  "disk_percent": 55.3,
  "disk_used_bytes": 55301234688,
  "disk_total_bytes": 100000000000,
  "network_rx_rate": 1248576,
  "network_tx_rate": 895632,
  "container_count": 8,
  "containers_healthy": 7,
  "containers_unhealthy": 1,
  "status": "healthy",
  "queried_at": "2024-12-11T10:30:00Z"
}

{
  "server_id": "a1b2c3d4-...",
  "server_name": "prod-server-1",
  "summary": {
    "cpu_percent": 45.2,
    "memory_percent": 67.8,
    "disk_percent": 55.3,
    "status": "healthy",
    "queried_at": "2024-12-11T10:30:00Z"
  },
  "containers": [
    {
      "container_id": "abc123",
      "container_name": "prod-env-123_odoo",
      "cpu_percent": 15.3,
      "memory_percent": 45.0,
      "memory_used_bytes": 536870912,
      "memory_limit_bytes": 1073741824,
      "network_rx_bytes": 1248576,
      "network_tx_bytes": 895632,
      "status": "running",
      "restart_count": 0,
      "uptime_seconds": 86400
    }
  ],
  "timeseries": {
    "cpu": {
      "metric_name": "cpu",
      "points": [
        {"timestamp": "2024-12-11T10:00:00Z", "value": 42.5},
        {"timestamp": "2024-12-11T10:05:00Z", "value": 45.2}
      ],
      "unit": "%"
    },
    "memory": {
      "metric_name": "memory",
      "points": [
        {"timestamp": "2024-12-11T10:00:00Z", "value": 65.3},
        {"timestamp": "2024-12-11T10:05:00Z", "value": 67.8}
      ],
      "unit": "%"
    }
  },
  "netdata_dashboard_url": "https://monitoring.prod-server-1.paasportal.io",
  "queried_at": "2024-12-11T10:30:00Z"
}

[
  {
    "container_id": "abc123",
    "container_name": "env-prod_odoo",
    "cpu_percent": 15.3,
    "memory_percent": 45.0,
    "memory_used_bytes": 536870912,
    "memory_limit_bytes": 1073741824,
    "network_rx_bytes": 1248576,
    "network_tx_bytes": 895632,
    "status": "running",
    "restart_count": 0,
    "uptime_seconds": 86400
  },
  {
    "container_id": "def456",
    "container_name": "env-prod_db",
    "cpu_percent": 8.7,
    "memory_percent": 52.3,
    "memory_used_bytes": 1073741824,
    "memory_limit_bytes": 2147483648,
    "network_rx_bytes": 524288,
    "network_tx_bytes": 315392,
    "status": "running",
    "restart_count": 0,
    "uptime_seconds": 86400
  }
]

{
  "status": "online",
  "lag_bytes": 12582912,
  "lag_seconds": 3.5,
  "error_message": null,
  "last_check": "2024-12-11T10:30:00Z"
}

# SSE endpoint - disable buffering
location /api/v1/events/stream {
    proxy_pass http://backend:8000;
 
    # SSE requires these headers
    proxy_set_header Connection '';
    proxy_http_version 1.1;
    chunked_transfer_encoding off;
 
    # Disable buffering for SSE
    proxy_buffering off;
    proxy_cache off;
 
    # Long timeout for SSE (24 hours)
    proxy_read_timeout 86400s;
    proxy_send_timeout 86400s;
 
    # Forward client info
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Forwarded-Proto $scheme;
}

return StreamingResponse(
    event_generator(request, user_id),
    media_type="text/event-stream",
    headers={
        "Cache-Control": "no-cache",
        "Connection": "keep-alive",
        "X-Accel-Buffering": "no",  # Instructs Nginx to disable buffering
    },
)