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 = (
"📊 گزارش وضعیت و متریکهای لحظهای سیستم (Grafana Metrics):\n"
f"🕒 زمان گزارش: {timestamp_str}\n\n"
"📥 ورودی از کانالهای مبدا (Ingest):\n"
f"• کل پستهای دریافت شده: {int(total_ingested):,}\n"
f"• نرخ ورودی لحظهای: {ingest_rate:.2f} پست در دقیقه\n\n"
"🚀 انتشار در کانالهای مقصد (Published):\n"
f"• کل پستهای منتشر شده: {int(total_published):,}\n"
f"• نرخ انتشار لحظهای: {publish_rate:.2f} پست در دقیقه\n\n"
"🧠 پردازش هوش مصنوعی (AI Engine):\n"
f"• کل درخواستها: {int(ai_total):,} (✅ {int(ai_success)} موفق | ❌ {int(ai_error)} خطا)\n"
f"• میانگین تاخیر پاسخ: {avg_latency:.2f}s\n"
f"• نرخ درخواست: {ai_rate:.2f} req/min\n\n"
"📬 وضعیت صف انتشار (Paced Queue):\n"
f"• کل پیامها در صف: {total_queue}\n"
)
if queue_breakdown:
details = " | ".join([f"{k}: {v}" for k, v in queue_breakdown.items()])
report += f" ({details})\n\n"
else:
report += " (صف خالی است)\n\n"
report += (
"🛡 پایش و سلامت سیستم:\n"
f"• پستهای تکراری شناساییشده: {int(duplicates):,}\n"
f"• خطاهای ثبتشده: {int(total_errors):,} (⚠️ {int(open_errors)} باز | ✅ {int(resolved_errors)} رفعشده)\n"
f"• اقدامات ادمین: {int(admin_actions):,}\n\n"
)
if mount_stats:
report += "💽 فضای ذخیرهسازی تفکیکی درایوها (Mount Points Storage):\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"• {mnt}: {free_mb:.0f} MB آزاد از {tot_mb:.0f} MB ({pct:.1f}% آزاد)\n"
else:
report += f"• {mnt}: {free_gb:.2f} GB آزاد از {tot_gb:.2f} GB ({pct:.1f}% آزاد)\n"
report += "\n"
report += "👇 برای دریافت نمودار تصویری متریکها روی بازه زمانی مورد نظر بزنید:"
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