Add queues priority to metrics only works in redis

.gitignore

@@ -143,3 +143,4 @@ dmypy.json
 
 .virtualenv
 celery-exporter
+.idea

README.md

@@ -118,7 +118,10 @@ docker run -p 9808:9808 danihodovic/celery-exporter --broker-url=redis://redis.s
   --broker-transport-option global_keyprefix=danihodovic \
   --broker-transport-option visibility_timeout=7200
 ```
-
+In case you changed celery transport options to enable [priority](https://docs.celeryq.dev/en/stable/userguide/routing.html#redis-message-priorities) or changed separator just use same config here as well.
+```sh
+--broker-transport-option sep=':' --broker-transport-option priority_steps=[0,1,2,3,4,5,6,7,8,9]
+```
 In case of extended transport options, such as `sentinel_kwargs` you can pass JSON string:,
 for example:
 

conftest.py

@@ -41,6 +41,11 @@ def celery_config(broker):
     )
     if broker == "redis":
         config["broker_url"] = "redis://localhost:6379/"  # type: ignore
+        config["broker_transport_options"] = {
+            'priority_steps': list(range(10)),
+            'queue_order_strategy': 'priority',
+            'sep': ':'
+        }
     elif broker == "rabbitmq":
         config["broker_url"] = "amqp://guest:guest@localhost:5672"  # type: ignore
     elif broker == "memory":
@@ -70,7 +75,6 @@ def find_free_port():
     """
 
     def _find_free_port():
-
         s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
         s.bind(("0.0.0.0", 0))
@@ -88,7 +92,7 @@ def exporter_instance(find_free_port, celery_config, log_level):
         "host": "0.0.0.0",
         "port": find_free_port(),
         "broker_url": celery_config["broker_url"],
-        "broker_transport_option": ["visibility_timeout=7200"],
+        "broker_transport_option": ["visibility_timeout=7200", "sep=:", "priority_steps=[0,1,2,3,4,5,6,7,8,9]"],
         "broker_ssl_option": [],
         "retry_interval": 5,
         "log_level": log_level,

src/exporter.py

@@ -9,7 +9,12 @@
 from celery.utils import nodesplit  # type: ignore
 from kombu.exceptions import ChannelError  # type: ignore
 from loguru import logger
-from prometheus_client import CollectorRegistry, Counter, Gauge, Histogram
+from prometheus_client import (
+    CollectorRegistry,
+    Counter,
+    Gauge,
+    Histogram,
+)
 
 from .http_server import start_http_server
 
@@ -106,7 +111,7 @@ def __init__(self, buckets=None):
     def track_queue_metrics(self):
         with self.app.connection() as connection:  # type: ignore
             transport = connection.info()["transport"]
-            acceptable_transports = ["redis", "rediss", "amqp", "memory", "sentinel"]
+            acceptable_transports = ["redis", "rediss", "amqp", "amqps", "memory", "sentinel"]
             if transport not in acceptable_transports:
                 logger.debug(
                     f"Queue length tracking is only implemented for {acceptable_transports}"
@@ -117,9 +122,20 @@ def track_queue_metrics(self):
             # we need to cache queue info in exporter in case all workers are offline
             # so that no worker response to exporter will make active_queues return None
             queues = self.app.control.inspect().active_queues() or {}
+            queue_cache = set()
             for info_list in queues.values():
                 for queue_info in info_list:
                     self.queue_cache.add(queue_info["name"])
+                    queue_cache.add((queue_info['name']))
+
+            # Check celery queues based on worker separator & priority steps
+            separator = '\x06\x16'
+            if 'sep' in self.app.conf["broker_transport_options"]:
+                separator = self.app.conf["broker_transport_options"]['sep']
+            if 'priority_steps' in self.app.conf["broker_transport_options"]:
+                for queue in queue_cache:
+                    for step in self.app.conf['broker_transport_options']['priority_steps']:
+                        self.queue_cache.add(f'{queue}{separator}{str(step)}')
 
             track_length = lambda q, l: self.celery_queue_length.labels(
                 queue_name=q
@@ -128,7 +144,7 @@ def track_queue_metrics(self):
                 if transport in ["redis", "rediss", "sentinel"]:
                     queue_length = redis_queue_length(connection, queue)
                     track_length(queue, queue_length)
-                elif transport in ["amqp", "memory"]:
+                elif transport in ["amqp", "amqps", "memory"]:
                     queue_length = rabbitmq_queue_length(connection, queue)
                     track_length(queue, queue_length)