metrics: add host multi-mount storage metrics, Prometheus exporter, and Grafana panels
This commit is contained in:
mamad
@@ -0,0 +1,306 @@
|
||||
import os
|
||||
import io
|
||||
import time
|
||||
import httpx
|
||||
import logging
|
||||
from typing import Dict, Any, List, Optional, Tuple
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
PROMETHEUS_URL = os.getenv(
|
||||
"PROMETHEUS_URL",
|
||||
"http://prometheus:9090" if os.path.exists("/.dockerenv") else "http://localhost:9090"
|
||||
)
|
||||
|
||||
# Colors and style configuration for dark-mode graphs
|
||||
BG_COLOR = "#12171f"
|
||||
PANEL_BG = "#1a2230"
|
||||
GRID_COLOR = "#2c384d"
|
||||
TEXT_COLOR = "#e2e8f0"
|
||||
CYAN = "#38bdf8"
|
||||
GREEN = "#4ade80"
|
||||
PURPLE = "#a855f7"
|
||||
YELLOW = "#facc15"
|
||||
RED = "#f87171"
|
||||
ORANGE = "#fb923c"
|
||||
|
||||
|
||||
async def _query_instant(query: str) -> Optional[float]:
|
||||
url = f"{PROMETHEUS_URL.rstrip('/')}/api/v1/query"
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=4.0) as client:
|
||||
resp = await client.get(url, params={"query": query})
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
results = data.get("data", {}).get("result", [])
|
||||
if results:
|
||||
val = results[0].get("value", [0, "0"])[1]
|
||||
return float(val)
|
||||
except Exception as e:
|
||||
logger.debug(f"Prometheus instant query failed ({query}): {e}")
|
||||
return None
|
||||
|
||||
|
||||
async def _query_vector(query: str) -> List[Dict[str, Any]]:
|
||||
url = f"{PROMETHEUS_URL.rstrip('/')}/api/v1/query"
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=4.0) as client:
|
||||
resp = await client.get(url, params={"query": query})
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
return data.get("data", {}).get("result", [])
|
||||
except Exception as e:
|
||||
logger.debug(f"Prometheus vector query failed ({query}): {e}")
|
||||
return []
|
||||
|
||||
|
||||
async def _query_range(query: str, start: float, end: float, step: str) -> List[Tuple[float, float]]:
|
||||
url = f"{PROMETHEUS_URL.rstrip('/')}/api/v1/query_range"
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=8.0) as client:
|
||||
resp = await client.get(url, params={"query": query, "start": start, "end": end, "step": step})
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
results = data.get("data", {}).get("result", [])
|
||||
if results:
|
||||
values = results[0].get("values", [])
|
||||
return [(float(t), float(v)) for t, v in values]
|
||||
except Exception as e:
|
||||
logger.debug(f"Prometheus range query failed ({query}): {e}")
|
||||
return []
|
||||
|
||||
|
||||
async def get_instant_metrics_report() -> str:
|
||||
"""Fetch current real-time metrics from Prometheus / Grafana data source and format a comprehensive report."""
|
||||
# 1. Ingest metrics
|
||||
total_ingested = await _query_instant("sum(copykar_source_activity_total)") or 0
|
||||
ingest_rate = await _query_instant("sum(rate(copykar_source_activity_total[5m])) * 60") or 0.0
|
||||
|
||||
# 2. Publish metrics
|
||||
total_published = await _query_instant("sum(copykar_target_activity_total)") or 0
|
||||
publish_rate = await _query_instant("sum(rate(copykar_target_activity_total[5m])) * 60") or 0.0
|
||||
|
||||
# 3. AI metrics
|
||||
ai_success = await _query_instant('sum(copykar_ai_requests_total{status="success"})') or 0
|
||||
ai_error = await _query_instant('sum(copykar_ai_requests_total{status="error"})') or 0
|
||||
ai_total = ai_success + ai_error
|
||||
ai_latency_sum = await _query_instant("sum(copykar_ai_latency_seconds_sum)") or 0.0
|
||||
ai_latency_cnt = await _query_instant("sum(copykar_ai_latency_seconds_count)") or 0.0
|
||||
avg_latency = (ai_latency_sum / ai_latency_cnt) if ai_latency_cnt > 0 else 0.0
|
||||
ai_rate = await _query_instant("sum(rate(copykar_ai_requests_total[5m])) * 60") or 0.0
|
||||
|
||||
# 4. Queue breakdown
|
||||
queue_data = await _query_vector("copykar_posts_queue_gauge")
|
||||
queue_breakdown: Dict[str, int] = {}
|
||||
for item in queue_data:
|
||||
st = item.get("metric", {}).get("status", "unknown")
|
||||
val = int(float(item.get("value", [0, 0])[1]))
|
||||
queue_breakdown[st] = val
|
||||
total_queue = sum(queue_breakdown.values())
|
||||
|
||||
# 5. Duplicates & Errors
|
||||
duplicates = await _query_instant("sum(copykar_duplicates_detected_total)") or 0
|
||||
total_errors = await _query_instant("sum(copykar_errors_total)") or 0
|
||||
open_errors = await _query_instant("copykar_errors_open_total") or 0
|
||||
resolved_errors = await _query_instant("sum(copykar_errors_resolved_total)") or 0
|
||||
|
||||
# 6. Admin Actions
|
||||
admin_actions = await _query_instant("sum(copykar_admin_actions_total)") or 0
|
||||
|
||||
# 7. Disk Space per mount point
|
||||
disk_free_data = await _query_vector("copykar_disk_free_bytes")
|
||||
disk_total_data = await _query_vector("copykar_disk_total_bytes")
|
||||
disk_pct_data = await _query_vector("copykar_disk_free_percent")
|
||||
|
||||
mount_stats: Dict[str, Dict[str, Any]] = {}
|
||||
for item in disk_free_data:
|
||||
m = item.get("metric", {}).get("mountpoint", "/")
|
||||
val = float(item.get("value", [0, 0])[1])
|
||||
mount_stats.setdefault(m, {})["free_gb"] = val / (1024 ** 3)
|
||||
|
||||
for item in disk_total_data:
|
||||
m = item.get("metric", {}).get("mountpoint", "/")
|
||||
val = float(item.get("value", [0, 0])[1])
|
||||
mount_stats.setdefault(m, {})["total_gb"] = val / (1024 ** 3)
|
||||
|
||||
for item in disk_pct_data:
|
||||
m = item.get("metric", {}).get("mountpoint", "/")
|
||||
val = float(item.get("value", [0, 0])[1])
|
||||
mount_stats.setdefault(m, {})["pct"] = val
|
||||
|
||||
timestamp_str = time.strftime("%Y-%m-%d %H:%M:%S UTC", time.gmtime())
|
||||
|
||||
report = (
|
||||
"📊 <b>گزارش وضعیت و متریکهای لحظهای سیستم (Grafana Metrics):</b>\n"
|
||||
f"🕒 <i>زمان گزارش: {timestamp_str}</i>\n\n"
|
||||
"📥 <b>ورودی از کانالهای مبدا (Ingest):</b>\n"
|
||||
f"• کل پستهای دریافت شده: <b>{int(total_ingested):,}</b>\n"
|
||||
f"• نرخ ورودی لحظهای: <b>{ingest_rate:.2f}</b> پست در دقیقه\n\n"
|
||||
"🚀 <b>انتشار در کانالهای مقصد (Published):</b>\n"
|
||||
f"• کل پستهای منتشر شده: <b>{int(total_published):,}</b>\n"
|
||||
f"• نرخ انتشار لحظهای: <b>{publish_rate:.2f}</b> پست در دقیقه\n\n"
|
||||
"🧠 <b>پردازش هوش مصنوعی (AI Engine):</b>\n"
|
||||
f"• کل درخواستها: <b>{int(ai_total):,}</b> (✅ {int(ai_success)} موفق | ❌ {int(ai_error)} خطا)\n"
|
||||
f"• میانگین تاخیر پاسخ: <b>{avg_latency:.2f}s</b>\n"
|
||||
f"• نرخ درخواست: <b>{ai_rate:.2f}</b> req/min\n\n"
|
||||
"📬 <b>وضعیت صف انتشار (Paced Queue):</b>\n"
|
||||
f"• کل پیامها در صف: <b>{total_queue}</b>\n"
|
||||
)
|
||||
|
||||
if queue_breakdown:
|
||||
details = " | ".join([f"<code>{k}</code>: {v}" for k, v in queue_breakdown.items()])
|
||||
report += f" ({details})\n\n"
|
||||
else:
|
||||
report += " <i>(صف خالی است)</i>\n\n"
|
||||
|
||||
report += (
|
||||
"🛡 <b>پایش و سلامت سیستم:</b>\n"
|
||||
f"• پستهای تکراری شناساییشده: <b>{int(duplicates):,}</b>\n"
|
||||
f"• خطاهای ثبتشده: <b>{int(total_errors):,}</b> (⚠️ {int(open_errors)} باز | ✅ {int(resolved_errors)} رفعشده)\n"
|
||||
f"• اقدامات ادمین: <b>{int(admin_actions):,}</b>\n\n"
|
||||
)
|
||||
|
||||
if mount_stats:
|
||||
report += "💽 <b>فضای ذخیرهسازی تفکیکی درایوها (Mount Points Storage):</b>\n"
|
||||
for mnt, data in sorted(mount_stats.items()):
|
||||
free_gb = data.get("free_gb", 0.0)
|
||||
tot_gb = data.get("total_gb", 0.0)
|
||||
pct = data.get("pct", 0.0)
|
||||
if tot_gb > 0 and tot_gb < 1.0:
|
||||
free_mb = free_gb * 1024
|
||||
tot_mb = tot_gb * 1024
|
||||
report += f"• <code>{mnt}</code>: <b>{free_mb:.0f} MB</b> آزاد از <b>{tot_mb:.0f} MB</b> (<b>{pct:.1f}%</b> آزاد)\n"
|
||||
else:
|
||||
report += f"• <code>{mnt}</code>: <b>{free_gb:.2f} GB</b> آزاد از <b>{tot_gb:.2f} GB</b> (<b>{pct:.1f}%</b> آزاد)\n"
|
||||
report += "\n"
|
||||
|
||||
report += "<i>👇 برای دریافت نمودار تصویری متریکها روی بازه زمانی مورد نظر بزنید:</i>"
|
||||
return report
|
||||
|
||||
|
||||
def _generate_chart_image(
|
||||
time_range_label: str,
|
||||
ingest_pts: List[Tuple[float, float]],
|
||||
publish_pts: List[Tuple[float, float]],
|
||||
ai_pts: List[Tuple[float, float]],
|
||||
queue_pts: List[Tuple[float, float]],
|
||||
error_pts: List[Tuple[float, float]],
|
||||
) -> bytes:
|
||||
"""Generate dark-mode multi-panel metrics graph using matplotlib in memory."""
|
||||
import matplotlib
|
||||
matplotlib.use("Agg")
|
||||
import matplotlib.pyplot as plt
|
||||
import matplotlib.dates as mdates
|
||||
from datetime import datetime
|
||||
|
||||
plt.style.use("dark_background")
|
||||
fig, axes = plt.subplots(2, 2, figsize=(12, 7.5), dpi=140)
|
||||
fig.patch.set_facecolor(BG_COLOR)
|
||||
fig.suptitle(f"Copykar System Metrics Dashboard ({time_range_label})", fontsize=15, color=TEXT_COLOR, fontweight="bold", y=0.98)
|
||||
|
||||
for ax in axes.flat:
|
||||
ax.set_facecolor(PANEL_BG)
|
||||
ax.tick_params(colors=TEXT_COLOR, labelsize=8)
|
||||
ax.grid(True, linestyle="--", alpha=0.3, color=GRID_COLOR)
|
||||
for spine in ax.spines.values():
|
||||
spine.set_color(GRID_COLOR)
|
||||
|
||||
# 1. Ingest & Publish Rates
|
||||
ax1 = axes[0, 0]
|
||||
ax1.set_title("Ingest vs Publish Rate (posts/min)", fontsize=10, color=CYAN, fontweight="bold")
|
||||
if ingest_pts:
|
||||
t1 = [datetime.fromtimestamp(p[0]) for p in ingest_pts]
|
||||
v1 = [p[1] for p in ingest_pts]
|
||||
ax1.plot(t1, v1, label="Ingested (posts/m)", color=CYAN, linewidth=1.8)
|
||||
if publish_pts:
|
||||
t2 = [datetime.fromtimestamp(p[0]) for p in publish_pts]
|
||||
v2 = [p[1] for p in publish_pts]
|
||||
ax1.plot(t2, v2, label="Published (posts/m)", color=GREEN, linewidth=1.8)
|
||||
if ingest_pts or publish_pts:
|
||||
ax1.legend(loc="upper left", fontsize=8, facecolor=PANEL_BG, edgecolor=GRID_COLOR)
|
||||
|
||||
# 2. AI Request Rate
|
||||
ax2 = axes[0, 1]
|
||||
ax2.set_title("AI Request Rate (req/min)", fontsize=10, color=PURPLE, fontweight="bold")
|
||||
if ai_pts:
|
||||
t_ai = [datetime.fromtimestamp(p[0]) for p in ai_pts]
|
||||
v_ai = [p[1] for p in ai_pts]
|
||||
ax2.plot(t_ai, v_ai, label="AI Calls / min", color=PURPLE, linewidth=1.8)
|
||||
ax2.fill_between(t_ai, v_ai, color=PURPLE, alpha=0.2)
|
||||
ax2.legend(loc="upper left", fontsize=8, facecolor=PANEL_BG, edgecolor=GRID_COLOR)
|
||||
|
||||
# 3. Queue Depth
|
||||
ax3 = axes[1, 0]
|
||||
ax3.set_title("Queue Depth (Active Posts)", fontsize=10, color=YELLOW, fontweight="bold")
|
||||
if queue_pts:
|
||||
t_q = [datetime.fromtimestamp(p[0]) for p in queue_pts]
|
||||
v_q = [p[1] for p in queue_pts]
|
||||
ax3.plot(t_q, v_q, label="Queue Size", color=YELLOW, linewidth=1.8)
|
||||
ax3.fill_between(t_q, v_q, color=YELLOW, alpha=0.2)
|
||||
ax3.legend(loc="upper left", fontsize=8, facecolor=PANEL_BG, edgecolor=GRID_COLOR)
|
||||
|
||||
# 4. Error Rate
|
||||
ax4 = axes[1, 1]
|
||||
ax4.set_title("Error Rate (errors/min)", fontsize=10, color=RED, fontweight="bold")
|
||||
if error_pts:
|
||||
t_err = [datetime.fromtimestamp(p[0]) for p in error_pts]
|
||||
v_err = [p[1] for p in error_pts]
|
||||
ax4.plot(t_err, v_err, label="Errors / min", color=RED, linewidth=1.8)
|
||||
ax4.fill_between(t_err, v_err, color=RED, alpha=0.2)
|
||||
ax4.legend(loc="upper left", fontsize=8, facecolor=PANEL_BG, edgecolor=GRID_COLOR)
|
||||
|
||||
|
||||
# Formatting date axes
|
||||
for ax in axes.flat:
|
||||
ax.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M"))
|
||||
fig.autofmt_xdate(rotation=25)
|
||||
|
||||
plt.tight_layout(rect=[0, 0.03, 1, 0.95])
|
||||
buf = io.BytesIO()
|
||||
plt.savefig(buf, format="jpg", facecolor=BG_COLOR, edgecolor="none", bbox_inches="tight", pil_kwargs={"quality": 95})
|
||||
plt.close(fig)
|
||||
buf.seek(0)
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
async def generate_metrics_graph(time_range: str = "15m") -> Tuple[bytes, str]:
|
||||
"""Fetch time-series range metrics and return a JPG chart image along with its exact timestamped filename."""
|
||||
now = time.time()
|
||||
now_dt = time.strftime("%Y-%m-%d_%H-%M-%S", time.localtime(now))
|
||||
filename = f"copykar_metrics_{time_range}_{now_dt}.jpg"
|
||||
|
||||
if time_range == "15m":
|
||||
start = now - 15 * 60
|
||||
step = "15s"
|
||||
label = "Last 15 Minutes"
|
||||
elif time_range == "3h":
|
||||
start = now - 3 * 3600
|
||||
step = "1m"
|
||||
label = "Last 3 Hours"
|
||||
elif time_range == "24h":
|
||||
start = now - 24 * 3600
|
||||
step = "5m"
|
||||
label = "Last 24 Hours"
|
||||
else:
|
||||
start = now - 15 * 60
|
||||
step = "15s"
|
||||
label = "Last 15 Minutes"
|
||||
|
||||
rate_window = "1m" if time_range in ("15m", "3h") else "5m"
|
||||
|
||||
ingest_pts = await _query_range(f"sum(rate(copykar_source_activity_total[{rate_window}])) * 60", start, now, step)
|
||||
publish_pts = await _query_range(f"sum(rate(copykar_target_activity_total[{rate_window}])) * 60", start, now, step)
|
||||
ai_pts = await _query_range(f"sum(rate(copykar_ai_requests_total[{rate_window}])) * 60", start, now, step)
|
||||
queue_pts = await _query_range("sum(copykar_posts_queue_gauge)", start, now, step)
|
||||
error_pts = await _query_range(f"sum(rate(copykar_errors_total[{rate_window}])) * 60", start, now, step)
|
||||
|
||||
chart_bytes = _generate_chart_image(
|
||||
time_range_label=label,
|
||||
ingest_pts=ingest_pts,
|
||||
publish_pts=publish_pts,
|
||||
ai_pts=ai_pts,
|
||||
queue_pts=queue_pts,
|
||||
error_pts=error_pts
|
||||
)
|
||||
return chart_bytes, filename
|
||||
|
||||
Reference in New Issue
Block a user