app.py

import gradio as gr
import matplotlib.pyplot as plt
import random
import math
from collections import deque

# -----------------------------
# Simulated time-series storage
# -----------------------------
MAX_POINTS = 60

timestamps = deque(maxlen=MAX_POINTS)
cpu_data = deque(maxlen=MAX_POINTS)
memory_data = deque(maxlen=MAX_POINTS)
request_rate_data = deque(maxlen=MAX_POINTS)
latency_data = deque(maxlen=MAX_POINTS)
error_rate_data = deque(maxlen=MAX_POINTS)
disk_data = deque(maxlen=MAX_POINTS)

tick = 0


def clamp(value, low, high):
    return max(low, min(high, value))


def seed_data():
    global tick
    for _ in range(MAX_POINTS):
        tick += 1
        add_datapoint()


def add_datapoint():
    global tick

    t = tick
    timestamps.append(t)

    cpu = 45 + 20 * math.sin(t / 5) + random.uniform(-6, 6)
    memory = 58 + 10 * math.sin(t / 9 + 1.5) + random.uniform(-3, 3)
    request_rate = 220 + 90 * math.sin(t / 6 + 0.5) + random.uniform(-20, 20)
    latency = 120 + 35 * math.sin(t / 7 + 2.2) + random.uniform(-10, 10)
    error_rate = 1.5 + 1.2 * abs(math.sin(t / 8)) + random.uniform(-0.2, 0.3)
    disk = 72 + 4 * math.sin(t / 18) + random.uniform(-0.8, 0.8)

    if random.random() < 0.08:
        cpu += random.uniform(12, 25)
        latency += random.uniform(20, 60)
        error_rate += random.uniform(1.0, 3.5)

    if random.random() < 0.05:
        request_rate += random.uniform(80, 150)

    cpu_data.append(clamp(cpu, 0, 100))
    memory_data.append(clamp(memory, 0, 100))
    request_rate_data.append(clamp(request_rate, 50, 500))
    latency_data.append(clamp(latency, 50, 350))
    error_rate_data.append(clamp(error_rate, 0, 10))
    disk_data.append(clamp(disk, 0, 100))


def get_status(cpu, memory, latency, error_rate):
    if error_rate > 4 or latency > 240 or cpu > 90:
        return "Critical"
    if error_rate > 2.5 or latency > 180 or cpu > 75 or memory > 80:
        return "Warning"
    return "Healthy"


def make_timeseries_plot(title, y_label, y_data, threshold=None):
    fig, ax = plt.subplots(figsize=(10, 3.2))
    ax.plot(list(timestamps), list(y_data), linewidth=2)
    ax.set_title(title)
    ax.set_xlabel("Time")
    ax.set_ylabel(y_label)
    ax.grid(True, alpha=0.3)

    if threshold is not None:
        ax.axhline(threshold, linestyle="--", linewidth=1)

    plt.tight_layout()
    return fig


def make_dashboard():
    global tick
    tick += 1
    add_datapoint()

    cpu = cpu_data[-1]
    memory = memory_data[-1]
    request_rate = request_rate_data[-1]
    latency = latency_data[-1]
    error_rate = error_rate_data[-1]
    disk = disk_data[-1]

    status = get_status(cpu, memory, latency, error_rate)

    summary_md = f"""
# Prometheus + Grafana Simulation

### Cluster Status: **{status}**
- **CPU Usage:** {cpu:.1f}%
- **Memory Usage:** {memory:.1f}%
- **Request Rate:** {request_rate:.0f} req/s
- **Latency:** {latency:.0f} ms
- **Error Rate:** {error_rate:.2f}%
- **Disk Usage:** {disk:.1f}%
"""

    alerts = []
    if cpu > 80:
        alerts.append(f"High CPU detected: {cpu:.1f}%")
    if memory > 80:
        alerts.append(f"High memory usage: {memory:.1f}%")
    if latency > 200:
        alerts.append(f"Latency spike detected: {latency:.0f} ms")
    if error_rate > 3:
        alerts.append(f"Elevated error rate: {error_rate:.2f}%")
    if disk > 85:
        alerts.append(f"Disk usage warning: {disk:.1f}%")

    alert_text = "\n".join(f"- {a}" for a in alerts) if alerts else "- No active alerts"

    cpu_plot = make_timeseries_plot("CPU Usage", "%", cpu_data, threshold=80)
    memory_plot = make_timeseries_plot("Memory Usage", "%", memory_data, threshold=80)
    req_plot = make_timeseries_plot("Request Rate", "req/s", request_rate_data)
    latency_plot = make_timeseries_plot("Latency", "ms", latency_data, threshold=200)
    error_plot = make_timeseries_plot("Error Rate", "%", error_rate_data, threshold=3)
    disk_plot = make_timeseries_plot("Disk Usage", "%", disk_data, threshold=85)

    return (
        summary_md,
        alert_text,
        cpu_plot,
        memory_plot,
        req_plot,
        latency_plot,
        error_plot,
        disk_plot,
    )


seed_data()

with gr.Blocks(title="Grafana + Prometheus Simulator") as demo:
    gr.Markdown("# Live Observability Dashboard")
    gr.Markdown(
        "This Gradio app simulates a Prometheus-monitored service with a Grafana-style dashboard that updates automatically."
    )

    with gr.Row():
        summary = gr.Markdown()
        alerts = gr.Markdown()

    with gr.Row():
        cpu_chart = gr.Plot()
        memory_chart = gr.Plot()

    with gr.Row():
        req_chart = gr.Plot()
        latency_chart = gr.Plot()

    with gr.Row():
        error_chart = gr.Plot()
        disk_chart = gr.Plot()

    # initial page load
    demo.load(
        fn=make_dashboard,
        inputs=None,
        outputs=[
            summary,
            alerts,
            cpu_chart,
            memory_chart,
            req_chart,
            latency_chart,
            error_chart,
            disk_chart,
        ],
    )

    # repeating refresh every 2 seconds
    timer = gr.Timer(value=2.0, active=True)
    timer.tick(
        fn=make_dashboard,
        inputs=None,
        outputs=[
            summary,
            alerts,
            cpu_chart,
            memory_chart,
            req_chart,
            latency_chart,
            error_chart,
            disk_chart,
        ],
    )

demo.launch()