Merge pull request #6 from MRoci/master

Support wildcard task_routes & add sample grafana dashboard

README.md

@@ -8,9 +8,9 @@ Celery Exporter is a Prometheus metrics exporter for Celery 4, written in python
 Here the list of exposed metrics:
 
 * `celery_tasks_total` exposes the number of tasks currently known to the queue
-  labeled by `name`, `state` and `namespace`.
+  labeled by `name`, `state`, `queue` and `namespace`.
 * `celery_tasks_runtime_seconds` tracks the number of seconds tasks take
-  until completed as histogram labeled by `name` and `namespace`
+  until completed as histogram labeled by `name`, `queue` and `namespace`
 * `celery_tasks_latency_seconds` exposes a histogram of task latency, i.e. the time until
   tasks are picked up by a worker
 * `celery_workers` exposes the number of currently probably alive workers

celery_exporter/state.py

@@ -19,6 +19,15 @@ def __init__(self, max_tasks_in_memory=10000):
         self._queue_by_task = {}
         self._mutex = threading.Lock()
 
+    @classmethod
+    def _gen_wildcards(self, name):
+        chunked = name.split(".")
+        res = [name]
+        for elem in reversed(chunked):
+            chunked.pop()
+            res.append(".".join(chunked + ["*"]))
+        return res
+
     @classmethod
     def get_config(self, app):
         res = dict()
@@ -32,12 +41,17 @@ def get_config(self, app):
             for conf in confs.values():
                 default = conf.get("task_default_queue", CELERY_DEFAULT_QUEUE)
                 if task_name in res and res[task_name] != default:
-                    continue
+                    break
 
-                try:
+                task_wildcard_names = self._gen_wildcards(task_name)
+                if "task_routes" in conf:
                     routes = conf["task_routes"]
-                    res[task_name] = routes[task_name]["queue"]
-                except KeyError:
+                    res[task_name] = default
+                    for i in task_wildcard_names:
+                        if i in routes and "queue" in routes[i]:
+                            res[task_name] = routes[i]["queue"]
+                            break
+                else:
                     res[task_name] = default
         return res