From 8eb5ca0a3f38f9484ded58a20b14973c9e42aa07 Mon Sep 17 00:00:00 2001 From: mamad Date: Thu, 27 Aug 2026 22:42:19 +0330 Subject: [PATCH] feat(observability): implement global database error logging, Prometheus error metrics, and max 3-panel Grafana dashboard --- core/error_logger.py | 38 ++++ core/metrics.py | 6 + db/database.py | 13 ++ monitoring/grafana/dashboards/copykar.json | 244 +++++++++++++-------- services/admin_bot.py | 1 + services/ai_processor.py | 3 +- services/collector.py | 83 +++++-- services/publisher.py | 3 +- 8 files changed, 268 insertions(+), 123 deletions(-) create mode 100644 core/error_logger.py diff --git a/core/error_logger.py b/core/error_logger.py new file mode 100644 index 0000000..9712ef7 --- /dev/null +++ b/core/error_logger.py @@ -0,0 +1,38 @@ +import json +import logging +import traceback +from typing import Optional, Dict, Any +from db.database import get_db_pool +from core.metrics import ERRORS_TOTAL + +logger = logging.getLogger("copykar.errors") + +async def log_exception(service_name: str, error: Exception, context: Optional[Dict[str, Any]] = None): + """Log an exception to Database, Prometheus metrics, and Python logs.""" + error_type = type(error).__name__ + error_msg = str(error) + tb_str = traceback.format_exc() + ctx_json = json.dumps(context or {}, default=str) + + # 1. Prometheus Metric + try: + ERRORS_TOTAL.labels(service=service_name, error_type=error_type).inc() + except Exception as e: + logger.debug(f"Failed to increment error metric: {e}") + + # 2. Python standard logger + logger.error(f"[{service_name}] {error_type}: {error_msg}\nContext: {ctx_json}\n{tb_str}") + + # 3. PostgreSQL Database + try: + pool = await get_db_pool() + async with pool.acquire() as conn: + await conn.execute( + """ + INSERT INTO error_logs (service_name, error_type, error_message, traceback, context) + VALUES ($1, $2, $3, $4, $5::jsonb); + """, + service_name, error_type, error_msg, tb_str, ctx_json + ) + except Exception as db_err: + logger.critical(f"Failed to write error to database: {db_err}") diff --git a/core/metrics.py b/core/metrics.py index c146da3..df767d1 100644 --- a/core/metrics.py +++ b/core/metrics.py @@ -46,6 +46,12 @@ POSTS_PUBLISHED_TOTAL = Counter( ["target_channel_id"] ) +ERRORS_TOTAL = Counter( + "copykar_errors_total", + "Total exceptions and errors caught across services", + ["service", "error_type"] +) + # Histograms AI_LATENCY_SECONDS = Histogram( "copykar_ai_latency_seconds", diff --git a/db/database.py b/db/database.py index d63bb6a..42b9892 100644 --- a/db/database.py +++ b/db/database.py @@ -71,6 +71,19 @@ CREATE TABLE IF NOT EXISTS settings ( description TEXT ); +CREATE TABLE IF NOT EXISTS error_logs ( + id BIGSERIAL PRIMARY KEY, + service_name VARCHAR(64) NOT NULL, + error_type VARCHAR(128) NOT NULL, + error_message TEXT NOT NULL, + traceback TEXT, + context JSONB DEFAULT '{}'::jsonb, + created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP +); + +CREATE INDEX IF NOT EXISTS idx_error_logs_created_at ON error_logs(created_at DESC); +CREATE INDEX IF NOT EXISTS idx_error_logs_service ON error_logs(service_name); + -- Migration safety for existing tables ALTER TABLE targets ADD COLUMN IF NOT EXISTS personality TEXT DEFAULT ''; ALTER TABLE targets ADD COLUMN IF NOT EXISTS custom_footer TEXT DEFAULT ''; diff --git a/monitoring/grafana/dashboards/copykar.json b/monitoring/grafana/dashboards/copykar.json index 07d3e40..8bda1f4 100644 --- a/monitoring/grafana/dashboards/copykar.json +++ b/monitoring/grafana/dashboards/copykar.json @@ -13,15 +13,15 @@ "collapsed": false, "gridPos": { "h": 1, "w": 24, "x": 0, "y": 0 }, "id": 100, - "title": "📌 شاخص‌های کلیدی عملکرد و وضعیت صف‌ها (KPIs & Queues)", + "title": "📌 شاخص‌های کلیدی عملکرد و صف‌های سیستم (KPIs & Queues)", "type": "row" }, { "collapsed": false, - "gridPos": { "h": 4, "w": 6, "x": 0, "y": 1 }, + "gridPos": { "h": 4, "w": 8, "x": 0, "y": 1 }, "id": 1, "title": "📥 کل پست‌های دریافتی از مبدا", - "description": "تعداد کل پست‌های خامی که توسط ربات جمع‌آوری‌کننده از کانال‌های مبدا دریافت شده است.", + "description": "تعداد کل پیام‌های دریافت شده توسط ربات از کانال‌های مبدا تحت مانیتور.", "type": "stat", "targets": [ { @@ -39,10 +39,10 @@ }, { "collapsed": false, - "gridPos": { "h": 4, "w": 6, "x": 6, "y": 1 }, + "gridPos": { "h": 4, "w": 8, "x": 8, "y": 1 }, "id": 2, "title": "⏳ مجموع پست‌های در صف ارسال مقصد", - "description": "تعداد پست‌های تایید شده توسط ادمین که در صف ردیس کانال‌های مقصد منتظر رسیدن نوبت ارسال (فاصله زمانی یا پایان ساعت خواب) هستند.", + "description": "تعداد پست‌های تایید شده که در صف ردیس کانال‌های مقصد منتظر رسیدن نوبت ارسال یا پایان ساعت خواب هستند.", "type": "stat", "targets": [ { @@ -67,30 +67,10 @@ }, { "collapsed": false, - "gridPos": { "h": 4, "w": 6, "x": 12, "y": 1 }, + "gridPos": { "h": 4, "w": 8, "x": 16, "y": 1 }, "id": 3, - "title": "✅ پست‌های تایید شده ادمین", - "description": "تعداد پست‌هایی که ادمین بازنویسی آن‌ها را تایید کرده و به صف انتشار فرستاده است.", - "type": "stat", - "targets": [ - { - "expr": "sum(copykar_admin_actions_total{action=\"approved\"}) or vector(0)", - "legendFormat": "تایید شده", - "refId": "A" - } - ], - "fieldConfig": { - "defaults": { - "thresholds": { "mode": "absolute", "steps": [{ "color": "green", "value": null }] } - } - } - }, - { - "collapsed": false, - "gridPos": { "h": 4, "w": 6, "x": 18, "y": 1 }, - "id": 4, "title": "🚀 کل پست‌های منتشر شده نهایی", - "description": "تعداد پست‌هایی که با موفقیت در کانال‌های مقصد ارسال و منتشر شده‌اند.", + "description": "تعداد پست‌هایی که با موفقیت در کانال‌های مقصد نهایی ارسال و منتشر شده‌اند.", "type": "stat", "targets": [ { @@ -109,13 +89,88 @@ "collapsed": false, "gridPos": { "h": 1, "w": 24, "x": 0, "y": 5 }, "id": 101, + "title": "📋 تصمیمات ادمین و وضعیت سلامت خطاها (Admin Actions & Error Telemetry)", + "type": "row" + }, + { + "collapsed": false, + "gridPos": { "h": 4, "w": 8, "x": 0, "y": 6 }, + "id": 4, + "title": "✅ پست‌های تایید شده ادمین", + "description": "تعداد پست‌هایی که ادمین بازنویسی آن‌ها را تایید کرده و به صف ارسال فرستاده است.", + "type": "stat", + "targets": [ + { + "expr": "sum(copykar_admin_actions_total{action=\"approved\"}) or vector(0)", + "legendFormat": "تایید شده", + "refId": "A" + } + ], + "fieldConfig": { + "defaults": { + "thresholds": { "mode": "absolute", "steps": [{ "color": "green", "value": null }] } + } + } + }, + { + "collapsed": false, + "gridPos": { "h": 4, "w": 8, "x": 8, "y": 6 }, + "id": 5, + "title": "❌ پست‌های رد و بایگانی شده", + "description": "تعداد پست‌هایی که توسط ادمین در کانال بررسی رد شده‌اند.", + "type": "stat", + "targets": [ + { + "expr": "sum(copykar_admin_actions_total{action=\"rejected\"}) or vector(0)", + "legendFormat": "رد شده", + "refId": "A" + } + ], + "fieldConfig": { + "defaults": { + "thresholds": { "mode": "absolute", "steps": [{ "color": "red", "value": null }] } + } + } + }, + { + "collapsed": false, + "gridPos": { "h": 4, "w": 8, "x": 16, "y": 6 }, + "id": 6, + "title": "⚠️ مجموع خطاهای ثبت شده در سیستم", + "description": "تعداد کل استثناها و خطاهایی که در دیتابیس PostgreSQL و لاگ‌ها ثبت شده است.", + "type": "stat", + "targets": [ + { + "expr": "sum(copykar_errors_total) or vector(0)", + "legendFormat": "خطاها", + "refId": "A" + } + ], + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null }, + { "color": "yellow", "value": 1 }, + { "color": "red", "value": 5 } + ] + } + } + } + }, + { + "collapsed": false, + "gridPos": { "h": 1, "w": 24, "x": 0, "y": 10 }, + "id": 102, "title": "📡 فعالیت به تفکیک کانال‌های مبدا و مقصد (Channels Activity Breakdown)", "type": "row" }, { "collapsed": false, - "gridPos": { "h": 8, "w": 12, "x": 0, "y": 6 }, - "id": 5, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 11 }, + "id": 7, "title": "📥 پست‌های دریافتی به تفکیک کانال مبدا", "description": "نمودار زمانی و حجم پست‌های جمع‌آوری شده از هر یک از کانال‌های مبدا تحت مانیتور.", "type": "timeseries", @@ -129,8 +184,8 @@ }, { "collapsed": false, - "gridPos": { "h": 8, "w": 12, "x": 12, "y": 6 }, - "id": 6, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 11 }, + "id": 8, "title": "🚀 پست‌های منتشر شده به تفکیک کانال مقصد", "description": "نمودار زمانی تعداد پست‌های ارسال شده به هر یک از کانال‌های مقصد نهایی.", "type": "timeseries", @@ -144,15 +199,15 @@ }, { "collapsed": false, - "gridPos": { "h": 1, "w": 24, "x": 0, "y": 14 }, - "id": 102, - "title": "🤖 عملیات هوش مصنوعی و عملکرد سیستم (AI Operations & Latency)", + "gridPos": { "h": 1, "w": 24, "x": 0, "y": 19 }, + "id": 103, + "title": "🤖 عملیات هوش مصنوعی و تحلیل خطاها (AI Operations & Error Breakdown)", "type": "row" }, { "collapsed": false, - "gridPos": { "h": 8, "w": 12, "x": 0, "y": 15 }, - "id": 7, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 20 }, + "id": 9, "title": "🤖 تعداد و وضعیت درخواست‌های هوش مصنوعی", "description": "نرخ بازنویسی‌های ارسالی به هوش مصنوعی به تفکیک وضعیت پاسخ و نوع عملیات.", "type": "timeseries", @@ -166,8 +221,60 @@ }, { "collapsed": false, - "gridPos": { "h": 8, "w": 12, "x": 12, "y": 15 }, - "id": 8, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 20 }, + "id": 10, + "title": "⚠️ خطاهای سیستم به تفکیک سرویس و نوع استثنا", + "description": "تحلیل آماری و نموداری خطاهای رخ داده در سیستم و ثبت شده در پایگاه داده.", + "type": "timeseries", + "targets": [ + { + "expr": "sum by (service, error_type) (copykar_errors_total)", + "legendFormat": "{{service}}: {{error_type}}", + "refId": "A" + } + ] + }, + { + "collapsed": false, + "gridPos": { "h": 1, "w": 24, "x": 0, "y": 28 }, + "id": 104, + "title": "⚡ مانیتورینگ منابع سرور و کانتینر (Container Telemetry)", + "type": "row" + }, + { + "collapsed": false, + "gridPos": { "h": 7, "w": 8, "x": 0, "y": 29 }, + "id": 11, + "title": "⚡ درصد مصرف پردازنده (CPU %)", + "description": "میزان مصرف پردازنده توسط سرویس کپی‌کار.", + "type": "timeseries", + "targets": [ + { + "expr": "rate(process_cpu_seconds_total{job=\"copykar\"}[1m]) * 100", + "legendFormat": "مصرف CPU %", + "refId": "A" + } + ] + }, + { + "collapsed": false, + "gridPos": { "h": 7, "w": 8, "x": 8, "y": 29 }, + "id": 12, + "title": "💾 میزان مصرف حافظه رم (RAM MB)", + "description": "حافظه رم اشغال شده توسط پردازش برنامه به مگابایت.", + "type": "timeseries", + "targets": [ + { + "expr": "process_resident_memory_bytes{job=\"copykar\"} / 1024 / 1024", + "legendFormat": "حافظه رم (MB)", + "refId": "A" + } + ] + }, + { + "collapsed": false, + "gridPos": { "h": 7, "w": 8, "x": 16, "y": 29 }, + "id": 13, "title": "⏱ مدت زمان پاسخ‌دهی هوش مصنوعی (ثانیه)", "description": "میانگین زمان و صدک ۹۵ ام برای بازنویسی پست‌ها متناسب با شخصیت هر کانال.", "type": "timeseries", @@ -183,69 +290,12 @@ "refId": "B" } ] - }, - { - "collapsed": false, - "gridPos": { "h": 1, "w": 24, "x": 0, "y": 23 }, - "id": 103, - "title": "⚡ مانیتورینگ منابع سرور و کانتینر (Container Telemetry)", - "type": "row" - }, - { - "collapsed": false, - "gridPos": { "h": 7, "w": 8, "x": 0, "y": 24 }, - "id": 9, - "title": "⚡ درصد مصرف پردازنده (CPU %)", - "description": "میزان مصرف پردازنده توسط سرویس کپی‌کار.", - "type": "timeseries", - "targets": [ - { - "expr": "rate(process_cpu_seconds_total{job=\"copykar\"}[1m]) * 100", - "legendFormat": "مصرف CPU %", - "refId": "A" - } - ] - }, - { - "collapsed": false, - "gridPos": { "h": 7, "w": 8, "x": 8, "y": 24 }, - "id": 10, - "title": "💾 میزان مصرف حافظه رم (RAM MB)", - "description": "حافظه رم اشغال شده توسط پردازش برنامه به مگابایت.", - "type": "timeseries", - "targets": [ - { - "expr": "process_resident_memory_bytes{job=\"copykar\"} / 1024 / 1024", - "legendFormat": "حافظه رم (MB)", - "refId": "A" - } - ] - }, - { - "collapsed": false, - "gridPos": { "h": 7, "w": 8, "x": 16, "y": 24 }, - "id": 11, - "title": "🧵 تعداد تردهای فعال و فایل‌های باز", - "description": "تعداد Thread های فعال و File Descriptor های باز پردازش.", - "type": "timeseries", - "targets": [ - { - "expr": "process_threads_total{job=\"copykar\"}", - "legendFormat": "تعداد تردها", - "refId": "A" - }, - { - "expr": "process_open_fds{job=\"copykar\"}", - "legendFormat": "فایل دیسکریپتورها", - "refId": "B" - } - ] } ], "refresh": "5s", "schemaVersion": 38, "style": "dark", - "tags": ["copykar", "telegram", "ai", "telemetry", "persian"], + "tags": ["copykar", "telegram", "ai", "telemetry", "persian", "errors"], "time": { "from": "now-1h", "to": "now" @@ -254,5 +304,5 @@ "timezone": "browser", "title": "داشبورد مدیریت ناوگان کپی‌کار (Copykar Executive Dashboard)", "uid": "copykar-executive-dashboard", - "version": 5 + "version": 6 } diff --git a/services/admin_bot.py b/services/admin_bot.py index 636757c..cef4ee2 100644 --- a/services/admin_bot.py +++ b/services/admin_bot.py @@ -10,6 +10,7 @@ from db.repository import Repository from core.queue import RedisQueue from core.metrics import ADMIN_ACTIONS_TOTAL, TARGET_ACTIVITY_TOTAL from core.proxy import get_telegram_proxy +from core.error_logger import log_exception logger = logging.getLogger(__name__) diff --git a/services/ai_processor.py b/services/ai_processor.py index 4f8d185..7ff1403 100644 --- a/services/ai_processor.py +++ b/services/ai_processor.py @@ -5,6 +5,7 @@ from db.models import Post, TargetChannel from db.repository import Repository from core.llm import LLMClient from core.metrics import DUPLICATES_DETECTED_TOTAL +from core.error_logger import log_exception logger = logging.getLogger(__name__) @@ -57,7 +58,7 @@ class AIProcessor: if rewritten: return rewritten.strip() except Exception as e: - logger.error(f"Failed to rewrite post for target {target.id} ({target.title}): {e}") + await log_exception("ai_processor.rewrite", e, {"target_id": target.id, "target_title": target.title}) # Fallback if AI fails: clean basic @mentions and append footer fallback = raw_text diff --git a/services/collector.py b/services/collector.py index 1d155e1..7b6e141 100644 --- a/services/collector.py +++ b/services/collector.py @@ -9,6 +9,7 @@ from core.dedup import compute_content_hash, compute_file_hash from core.queue import RedisQueue from core.metrics import COLLECTED_POSTS_TOTAL, SOURCE_ACTIVITY_TOTAL from core.proxy import get_telegram_proxy +from core.error_logger import log_exception logger = logging.getLogger(__name__) @@ -41,29 +42,33 @@ class CollectorService: async def start(self, notify_fn: Optional[Callable[[str], Awaitable[None]]] = None): logger.info("Initializing Collector Userbot client...") - await self.client.connect() + try: + await self.client.connect() - if await self.client.is_user_authorized(): - me = await self.client.get_me() - logger.info(f"Collector Userbot is authorized as: {me.first_name} (@{me.username})") - self._register_handlers() - return True + if await self.client.is_user_authorized(): + me = await self.client.get_me() + logger.info(f"Collector Userbot is authorized as: {me.first_name} (@{me.username})") + self._register_handlers() + return True - logger.warning("Collector Userbot is not authorized. Requesting login code...") - if self.phone and notify_fn: - try: - sent = await self.client.send_code_request(self.phone) - self.phone_code_hash = sent.phone_code_hash - await notify_fn( - f"🔐 نیاز به ورود ربات جمع‌آوری‌کننده\n\n" - f"کد تایید تلگرام به شماره {self.phone} ارسال شد.\n\n" - f"لطفا با دستور زیر پاسخ دهید:\n" - f"/code 12345" - ) - except Exception as e: - logger.error(f"Failed to send login code request: {e}") - await notify_fn(f"❌ خطا در ارسال کد ورود: {e}") - return False + logger.warning("Collector Userbot is not authorized. Requesting login code...") + if self.phone and notify_fn: + try: + sent = await self.client.send_code_request(self.phone) + self.phone_code_hash = sent.phone_code_hash + await notify_fn( + f"🔐 نیاز به ورود ربات جمع‌آوری‌کننده\n\n" + f"کد تایید تلگرام به شماره {self.phone} ارسال شد.\n\n" + f"لطفا با دستور زیر پاسخ دهید:\n" + f"/code 12345" + ) + except Exception as e: + await log_exception("collector.login", e, {"phone": self.phone}) + await notify_fn(f"❌ خطا در ارسال کد ورود: {e}") + return False + except Exception as e: + await log_exception("collector.start", e) + return False async def submit_code(self, code: str) -> str: if not self.phone or not self.phone_code_hash: @@ -78,6 +83,7 @@ class CollectorService: except SessionPasswordNeededError: return "🔐 رمز دو مرحله‌ای فعال است. لطفا با این دستور رمز را وارد کنید: /password رمز_عبور" except Exception as e: + await log_exception("collector.submit_code", e) return f"❌ خطا در ورود: {e}" async def submit_password(self, password: str) -> str: @@ -87,6 +93,7 @@ class CollectorService: self._register_handlers() return f"✅ تایید دو مرحله‌ای موفق بود! حساب فعال: {me.first_name}." except Exception as e: + await log_exception("collector.submit_password", e) return f"❌ خطا در تایید رمز دو مرحله‌ای: {e}" def _register_handlers(self): @@ -100,6 +107,33 @@ class CollectorService: self._handlers_registered = True logger.info("Collector real-time event handlers registered.") + async def _resolve_channel_entity(self, channel_id: int, username: Optional[str] = None): + """Robustly resolve channel entity even if not yet cached in local Telethon session.""" + if username: + try: + clean_user = username.replace("@", "").strip() + return await self.client.get_entity(clean_user) + except Exception: + pass + + try: + return await self.client.get_entity(channel_id) + except Exception: + pass + + try: + # Refresh dialogs cache + await self.client.get_dialogs(limit=50) + return await self.client.get_entity(channel_id) + except Exception: + pass + + try: + raw_str = str(channel_id).replace("-100", "").replace("-", "") + return await self.client.get_entity(int(raw_str)) + except Exception as e: + raise ValueError(f"Could not resolve entity for channel {channel_id} (@{username}): {e}") + async def _handle_message(self, event: events.NewMessage.Event): try: chat_id = event.chat_id @@ -146,7 +180,7 @@ class CollectorService: if self.on_post_received: await self.on_post_received(post_id) except Exception as e: - logger.error(f"Error handling message from {event.chat_id}: {e}", exc_info=True) + await log_exception("collector.handle_message", e, {"chat_id": event.chat_id, "msg_id": getattr(event.message, "id", None)}) async def scrape_channel_history( self, @@ -162,11 +196,12 @@ class CollectorService: source = await self.repo.get_source_by_channel_id(channel_id) source_title = source.title if source else str(channel_id) + username = source.username if source else None collected_count = 0 skipped_count = 0 try: - entity = await self.client.get_input_entity(channel_id) + entity = await self._resolve_channel_entity(channel_id, username) messages = [] async for msg in self.client.iter_messages(entity, limit=limit): messages.append(msg) @@ -225,7 +260,7 @@ class CollectorService: ) return collected_count except Exception as e: - logger.error(f"Error scraping history from {channel_id}: {e}", exc_info=True) + await log_exception("collector.scrape_history", e, {"channel_id": channel_id, "limit": limit}) if progress_callback: await progress_callback(f"❌ خطا در دریافت پست‌های کانال {channel_id}: {e}") return collected_count diff --git a/services/publisher.py b/services/publisher.py index 3382444..502d66e 100644 --- a/services/publisher.py +++ b/services/publisher.py @@ -9,6 +9,7 @@ from db.repository import Repository from core.queue import RedisQueue from core.metrics import TARGET_ACTIVITY_TOTAL, QUEUE_POSTS_GAUGE, REDIS_QUEUE_SIZE_GAUGE from core.proxy import get_telegram_proxy +from core.error_logger import log_exception logger = logging.getLogger(__name__) @@ -129,7 +130,7 @@ class PublisherService: TARGET_ACTIVITY_TOTAL.labels(channel_id=str(target.channel_id), title=target.title or '').inc() logger.info(f"Published post ID {post_id} to Target {target.title} ({target.channel_id})") except Exception as e: - logger.error(f"Failed to publish queued post {post_id} to target {target.channel_id}: {e}", exc_info=True) + await log_exception("publisher.publish", e, {"post_id": post_id, "target_id": target.id, "channel_id": target.channel_id}) REDIS_QUEUE_SIZE_GAUGE.set(total_queued)