From 4f5570850e71cb5b8ae7c7333ed4d22f69c5336d Mon Sep 17 00:00:00 2001 From: mamad Date: Thu, 27 Aug 2026 21:18:28 +0330 Subject: [PATCH] feat(metrics): add copykar_redis_queue_size gauge and update grafana queue panels --- core/metrics.py | 5 ++ monitoring/grafana/dashboards/copykar.json | 83 +++++++++++++++++----- services/queue_consumer.py | 4 +- 3 files changed, 74 insertions(+), 18 deletions(-) diff --git a/core/metrics.py b/core/metrics.py index 5552067..4fd95b4 100644 --- a/core/metrics.py +++ b/core/metrics.py @@ -48,6 +48,11 @@ QUEUE_POSTS_GAUGE = Gauge( ["status"] ) +REDIS_QUEUE_SIZE_GAUGE = Gauge( + "copykar_redis_queue_size", + "Current number of posts waiting in Redis incoming queue" +) + def start_metrics_server(port: int = 8000): try: start_http_server(port) diff --git a/monitoring/grafana/dashboards/copykar.json b/monitoring/grafana/dashboards/copykar.json index f152241..d63e307 100644 --- a/monitoring/grafana/dashboards/copykar.json +++ b/monitoring/grafana/dashboards/copykar.json @@ -13,14 +13,14 @@ "collapsed": false, "gridPos": { "h": 1, "w": 24, "x": 0, "y": 0 }, "id": 100, - "title": "📌 Key Performance Indicators", + "title": "📌 Key Performance Indicators & Real-Time Queues", "type": "row" }, { "collapsed": false, - "gridPos": { "h": 4, "w": 4, "x": 0, "y": 1 }, + "gridPos": { "h": 4, "w": 3, "x": 0, "y": 1 }, "id": 1, - "title": "Total Posts Collected", + "title": "Total Collected", "type": "stat", "targets": [ { @@ -32,16 +32,42 @@ "fieldConfig": { "defaults": { "color": { "mode": "palette-classic" }, - "mappings": [], "thresholds": { "mode": "absolute", "steps": [{ "color": "blue", "value": null }] } } } }, { "collapsed": false, - "gridPos": { "h": 4, "w": 4, "x": 4, "y": 1 }, + "gridPos": { "h": 4, "w": 3, "x": 3, "y": 1 }, + "id": 15, + "title": "📥 Redis Queue Depth", + "type": "stat", + "targets": [ + { + "expr": "copykar_redis_queue_size or vector(0)", + "legendFormat": "Waiting Posts", + "refId": "A" + } + ], + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null }, + { "color": "orange", "value": 10 }, + { "color": "red", "value": 50 } + ] + } + } + } + }, + { + "collapsed": false, + "gridPos": { "h": 4, "w": 3, "x": 6, "y": 1 }, "id": 2, - "title": "Duplicates Intercepted", + "title": "Duplicates Blocked", "type": "stat", "targets": [ { @@ -52,14 +78,13 @@ ], "fieldConfig": { "defaults": { - "color": { "mode": "palette-classic" }, "thresholds": { "mode": "absolute", "steps": [{ "color": "orange", "value": null }] } } } }, { "collapsed": false, - "gridPos": { "h": 4, "w": 4, "x": 8, "y": 1 }, + "gridPos": { "h": 4, "w": 3, "x": 9, "y": 1 }, "id": 3, "title": "Admin Approvals", "type": "stat", @@ -78,7 +103,7 @@ }, { "collapsed": false, - "gridPos": { "h": 4, "w": 4, "x": 12, "y": 1 }, + "gridPos": { "h": 4, "w": 3, "x": 12, "y": 1 }, "id": 4, "title": "Admin Rejections", "type": "stat", @@ -97,7 +122,7 @@ }, { "collapsed": false, - "gridPos": { "h": 4, "w": 4, "x": 16, "y": 1 }, + "gridPos": { "h": 4, "w": 3, "x": 15, "y": 1 }, "id": 5, "title": "Posts Published", "type": "stat", @@ -116,7 +141,7 @@ }, { "collapsed": false, - "gridPos": { "h": 4, "w": 4, "x": 20, "y": 1 }, + "gridPos": { "h": 4, "w": 3, "x": 18, "y": 1 }, "id": 6, "title": "AI Requests Total", "type": "stat", @@ -133,11 +158,30 @@ } } }, + { + "collapsed": false, + "gridPos": { "h": 4, "w": 3, "x": 21, "y": 1 }, + "id": 16, + "title": "Publish Queue Depth", + "type": "stat", + "targets": [ + { + "expr": "copykar_posts_queue_gauge{status=\"approved\"} or vector(0)", + "legendFormat": "Approved Queued", + "refId": "A" + } + ], + "fieldConfig": { + "defaults": { + "thresholds": { "mode": "absolute", "steps": [{ "color": "yellow", "value": null }] } + } + } + }, { "collapsed": false, "gridPos": { "h": 1, "w": 24, "x": 0, "y": 5 }, "id": 101, - "title": "📈 Pipeline Ingestion, Queues & Publishing", + "title": "📈 Pipeline Rates & Queue Depths", "type": "row" }, { @@ -172,13 +216,18 @@ "collapsed": false, "gridPos": { "h": 8, "w": 8, "x": 16, "y": 6 }, "id": 9, - "title": "Active Queue Depth", + "title": "Queue Depth Over Time", "type": "timeseries", "targets": [ { - "expr": "copykar_posts_queue_gauge", - "legendFormat": "Queue: {{status}}", + "expr": "copykar_redis_queue_size", + "legendFormat": "Redis Incoming Queue (2m Pacing)", "refId": "A" + }, + { + "expr": "copykar_posts_queue_gauge", + "legendFormat": "State: {{status}}", + "refId": "B" } ] }, @@ -280,7 +329,7 @@ "refresh": "5s", "schemaVersion": 38, "style": "dark", - "tags": ["copykar", "telegram", "ai", "telemetry"], + "tags": ["copykar", "telegram", "ai", "telemetry", "redis"], "time": { "from": "now-1h", "to": "now" @@ -289,5 +338,5 @@ "timezone": "browser", "title": "Copykar Telegram Fleet Executive Dashboard", "uid": "copykar-executive-dashboard", - "version": 2 + "version": 3 } diff --git a/services/queue_consumer.py b/services/queue_consumer.py index 3df2138..07cb7a0 100644 --- a/services/queue_consumer.py +++ b/services/queue_consumer.py @@ -4,7 +4,7 @@ import logging from typing import Optional from core.queue import RedisQueue from services.ai_processor import AIProcessor -from core.metrics import QUEUE_POSTS_GAUGE +from core.metrics import QUEUE_POSTS_GAUGE, REDIS_QUEUE_SIZE_GAUGE logger = logging.getLogger(__name__) @@ -29,6 +29,7 @@ class QueueConsumerService: try: qsize = await self.queue.qsize() QUEUE_POSTS_GAUGE.labels(status="redis_incoming").set(qsize) + REDIS_QUEUE_SIZE_GAUGE.set(qsize) if qsize > 0: post_id = await self.queue.pop() @@ -37,6 +38,7 @@ class QueueConsumerService: await self.ai_processor.process_post(post_id) new_qsize = await self.queue.qsize() QUEUE_POSTS_GAUGE.labels(status="redis_incoming").set(new_qsize) + REDIS_QUEUE_SIZE_GAUGE.set(new_qsize) except Exception as e: logger.error(f"Error in queue consumer loop: {e}", exc_info=True)