feat(metrics): add copykar_redis_queue_size gauge and update grafana queue panels

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)

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",
+          "expr": "copykar_redis_queue_size",
-          "legendFormat": "Queue: {{status}}",
+          "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
 }

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)