Merge pull request #4 from MRoci/master

Add queue label in TASKS & TASKS_RUNTIME metrics

Author: SharpEdgeMarshall

.gitignore

@@ -10,5 +10,6 @@ __pycache__
 .cache/
 .python-version
 .pytest_cache
+htmlcov/
 
 .vscode

README.md

@@ -83,59 +83,59 @@ If you then look at the exposed metrics, you should see something like this:
 celery_workers{namespace="celery"} 1.0
 # HELP celery_tasks_total Number of tasks per state
 # TYPE celery_tasks_total gauge
-celery_tasks_total{name="my_app.tasks.calculate_something",namespace="celery",state="RECEIVED"} 0.0
-celery_tasks_total{name="my_app.tasks.calculate_something",namespace="celery",state="PENDING"} 0.0
-celery_tasks_total{name="my_app.tasks.calculate_something",namespace="celery",state="STARTED"} 0.0
-celery_tasks_total{name="my_app.tasks.calculate_something",namespace="celery",state="RETRY"} 0.0
-celery_tasks_total{name="my_app.tasks.calculate_something",namespace="celery",state="FAILURE"} 0.0
-celery_tasks_total{name="my_app.tasks.calculate_something",namespace="celery",state="REVOKED"} 0.0
-celery_tasks_total{name="my_app.tasks.calculate_something",namespace="celery",state="SUCCESS"} 1.0
-celery_tasks_total{name="my_app.tasks.fetch_some_data",namespace="celery",state="RECEIVED"} 3.0
-celery_tasks_total{name="my_app.tasks.fetch_some_data",namespace="celery",state="PENDING"} 0.0
-celery_tasks_total{name="my_app.tasks.fetch_some_data",namespace="celery",state="STARTED"} 1.0
-celery_tasks_total{name="my_app.tasks.fetch_some_data",namespace="celery",state="RETRY"} 2.0
-celery_tasks_total{name="my_app.tasks.fetch_some_data",namespace="celery",state="FAILURE"} 1.0
-celery_tasks_total{name="my_app.tasks.fetch_some_data",namespace="celery",state="REVOKED"} 0.0
-celery_tasks_total{name="my_app.tasks.fetch_some_data",namespace="celery",state="SUCCESS"} 7.0
+celery_tasks_total{name="my_app.tasks.calculate_something",namespace="celery",queue="celery",state="RECEIVED"} 0.0
+celery_tasks_total{name="my_app.tasks.calculate_something",namespace="celery",queue="celery",state="PENDING"} 0.0
+celery_tasks_total{name="my_app.tasks.calculate_something",namespace="celery",queue="celery",state="STARTED"} 0.0
+celery_tasks_total{name="my_app.tasks.calculate_something",namespace="celery",queue="celery",state="RETRY"} 0.0
+celery_tasks_total{name="my_app.tasks.calculate_something",namespace="celery",queue="celery",state="FAILURE"} 0.0
+celery_tasks_total{name="my_app.tasks.calculate_something",namespace="celery",queue="celery",state="REVOKED"} 0.0
+celery_tasks_total{name="my_app.tasks.calculate_something",namespace="celery",queue="celery",state="SUCCESS"} 1.0
+celery_tasks_total{name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery",state="RECEIVED"} 3.0
+celery_tasks_total{name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery",state="PENDING"} 0.0
+celery_tasks_total{name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery",state="STARTED"} 1.0
+celery_tasks_total{name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery",state="RETRY"} 2.0
+celery_tasks_total{name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery",state="FAILURE"} 1.0
+celery_tasks_total{name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery",state="REVOKED"} 0.0
+celery_tasks_total{name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery",state="SUCCESS"} 7.0
 # HELP celery_tasks_runtime_seconds Task runtime (seconds)
 # TYPE celery_tasks_runtime_seconds histogram
-celery_tasks_runtime_seconds_bucket{le="0.005",name="my_app.tasks.calculate_something",namespace="celery"} 29.0
-celery_tasks_runtime_seconds_bucket{le="0.01",name="my_app.tasks.calculate_something",namespace="celery"} 29.0
-celery_tasks_runtime_seconds_bucket{le="0.025",name="my_app.tasks.calculate_something",namespace="celery"} 29.0
-celery_tasks_runtime_seconds_bucket{le="0.05",name="my_app.tasks.calculate_something",namespace="celery"} 29.0
-celery_tasks_runtime_seconds_bucket{le="0.075",name="my_app.tasks.calculate_something",namespace="celery"} 29.0
-celery_tasks_runtime_seconds_bucket{le="0.1",name="my_app.tasks.calculate_something",namespace="celery"} 29.0
-celery_tasks_runtime_seconds_bucket{le="0.25",name="my_app.tasks.calculate_something",namespace="celery"} 29.0
-celery_tasks_runtime_seconds_bucket{le="0.5",name="my_app.tasks.calculate_something",namespace="celery"} 29.0
-celery_tasks_runtime_seconds_bucket{le="0.75",name="my_app.tasks.calculate_something",namespace="celery"} 29.0
-celery_tasks_runtime_seconds_bucket{le="1.0",name="my_app.tasks.calculate_something",namespace="celery"} 29.0
-celery_tasks_runtime_seconds_bucket{le="2.5",name="my_app.tasks.calculate_something",namespace="celery"} 29.0
-celery_tasks_runtime_seconds_bucket{le="5.0",name="my_app.tasks.calculate_something",namespace="celery"} 29.0
-celery_tasks_runtime_seconds_bucket{le="7.5",name="my_app.tasks.calculate_something",namespace="celery"} 29.0
-celery_tasks_runtime_seconds_bucket{le="10.0",name="my_app.tasks.calculate_something",namespace="celery"} 29.0
-celery_tasks_runtime_seconds_bucket{le="+Inf",name="my_app.tasks.calculate_something",namespace="celery"} 29.0
-celery_tasks_runtime_seconds_count{name="my_app.tasks.calculate_something",namespace="celery"} 29.0
-celery_tasks_runtime_seconds_sum{name="my_app.tasks.calculate_something",namespace="celery"} 0.04020289977779612
-celery_tasks_runtime_seconds_bucket{le="0.005",name="my_app.tasks.fetch_some_data",namespace="celery"} 2.0
-celery_tasks_runtime_seconds_bucket{le="0.01",name="my_app.tasks.fetch_some_data",namespace="celery"} 2.0
-celery_tasks_runtime_seconds_bucket{le="0.025",name="my_app.tasks.fetch_some_data",namespace="celery"} 2.0
-celery_tasks_runtime_seconds_bucket{le="0.05",name="my_app.tasks.fetch_some_data",namespace="celery"} 2.0
-celery_tasks_runtime_seconds_bucket{le="0.075",name="my_app.tasks.fetch_some_data",namespace="celery"} 2.0
-celery_tasks_runtime_seconds_bucket{le="0.1",name="my_app.tasks.fetch_some_data",namespace="celery"} 2.0
-celery_tasks_runtime_seconds_bucket{le="0.25",name="my_app.tasks.fetch_some_data",namespace="celery"} 2.0
-celery_tasks_runtime_seconds_bucket{le="0.5",name="my_app.tasks.fetch_some_data",namespace="celery"} 2.0
-celery_tasks_runtime_seconds_bucket{le="0.75",name="my_app.tasks.fetch_some_data",namespace="celery"} 2.0
-celery_tasks_runtime_seconds_bucket{le="1.0",name="my_app.tasks.fetch_some_data",namespace="celery"} 2.0
-celery_tasks_runtime_seconds_bucket{le="2.5",name="my_app.tasks.fetch_some_data",namespace="celery"} 2.0
-celery_tasks_runtime_seconds_bucket{le="5.0",name="my_app.tasks.fetch_some_data",namespace="celery"} 2.0
-celery_tasks_runtime_seconds_bucket{le="7.5",name="my_app.tasks.fetch_some_data",namespace="celery"} 2.0
-celery_tasks_runtime_seconds_bucket{le="10.0",name="my_app.tasks.fetch_some_data",namespace="celery"} 2.0
-celery_tasks_runtime_seconds_bucket{le="+Inf",name="my_app.tasks.fetch_some_data",namespace="celery"} 2.0
-celery_tasks_runtime_seconds_count{name="my_app.tasks.fetch_some_data",namespace="celery"} 2.0
-celery_tasks_runtime_seconds_sum{name="my_app.tasks.fetch_some_data",namespace="celery"} 0.00402028997777961
+celery_tasks_runtime_seconds_bucket{le="0.005",name="my_app.tasks.calculate_something",namespace="celery",queue="celery"} 29.0
+celery_tasks_runtime_seconds_bucket{le="0.01",name="my_app.tasks.calculate_something",namespace="celery",queue="celery"} 29.0
+celery_tasks_runtime_seconds_bucket{le="0.025",name="my_app.tasks.calculate_something",namespace="celery",queue="celery"} 29.0
+celery_tasks_runtime_seconds_bucket{le="0.05",name="my_app.tasks.calculate_something",namespace="celery",queue="celery"} 29.0
+celery_tasks_runtime_seconds_bucket{le="0.075",name="my_app.tasks.calculate_something",namespace="celery",queue="celery"} 29.0
+celery_tasks_runtime_seconds_bucket{le="0.1",name="my_app.tasks.calculate_something",namespace="celery",queue="celery"} 29.0
+celery_tasks_runtime_seconds_bucket{le="0.25",name="my_app.tasks.calculate_something",namespace="celery",queue="celery"} 29.0
+celery_tasks_runtime_seconds_bucket{le="0.5",name="my_app.tasks.calculate_something",namespace="celery",queue="celery"} 29.0
+celery_tasks_runtime_seconds_bucket{le="0.75",name="my_app.tasks.calculate_something",namespace="celery",queue="celery"} 29.0
+celery_tasks_runtime_seconds_bucket{le="1.0",name="my_app.tasks.calculate_something",namespace="celery",queue="celery"} 29.0
+celery_tasks_runtime_seconds_bucket{le="2.5",name="my_app.tasks.calculate_something",namespace="celery",queue="celery"} 29.0
+celery_tasks_runtime_seconds_bucket{le="5.0",name="my_app.tasks.calculate_something",namespace="celery",queue="celery"} 29.0
+celery_tasks_runtime_seconds_bucket{le="7.5",name="my_app.tasks.calculate_something",namespace="celery",queue="celery"} 29.0
+celery_tasks_runtime_seconds_bucket{le="10.0",name="my_app.tasks.calculate_something",namespace="celery",queue="celery"} 29.0
+celery_tasks_runtime_seconds_bucket{le="+Inf",name="my_app.tasks.calculate_something",namespace="celery",queue="celery"} 29.0
+celery_tasks_runtime_seconds_count{name="my_app.tasks.calculate_something",namespace="celery",queue="celery"} 29.0
+celery_tasks_runtime_seconds_sum{name="my_app.tasks.calculate_something",namespace="celery",queue="celery"} 0.04020289977779612
+celery_tasks_runtime_seconds_bucket{le="0.005",name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery"} 2.0
+celery_tasks_runtime_seconds_bucket{le="0.01",name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery"} 2.0
+celery_tasks_runtime_seconds_bucket{le="0.025",name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery"} 2.0
+celery_tasks_runtime_seconds_bucket{le="0.05",name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery"} 2.0
+celery_tasks_runtime_seconds_bucket{le="0.075",name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery"} 2.0
+celery_tasks_runtime_seconds_bucket{le="0.1",name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery"} 2.0
+celery_tasks_runtime_seconds_bucket{le="0.25",name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery"} 2.0
+celery_tasks_runtime_seconds_bucket{le="0.5",name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery"} 2.0
+celery_tasks_runtime_seconds_bucket{le="0.75",name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery"} 2.0
+celery_tasks_runtime_seconds_bucket{le="1.0",name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery"} 2.0
+celery_tasks_runtime_seconds_bucket{le="2.5",name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery"} 2.0
+celery_tasks_runtime_seconds_bucket{le="5.0",name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery"} 2.0
+celery_tasks_runtime_seconds_bucket{le="7.5",name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery"} 2.0
+celery_tasks_runtime_seconds_bucket{le="10.0",name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery"} 2.0
+celery_tasks_runtime_seconds_bucket{le="+Inf",name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery"} 2.0
+celery_tasks_runtime_seconds_count{name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery"} 2.0
+celery_tasks_runtime_seconds_sum{name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery"} 0.00402028997777961
 # TYPE celery_tasks_runtime_seconds_created gauge
-celery_tasks_runtime_seconds_created{name="my_app.tasks.calculate_something",namespace="celery"} 1.548944949810905e+09
-celery_tasks_runtime_seconds_created{name="my_app.tasks.fetch_some_data",namespace="celery"} 1.5489449550243628e+09
+celery_tasks_runtime_seconds_created{name="my_app.tasks.calculate_something",namespace="celery",queue="celery"} 1.548944949810905e+09
+celery_tasks_runtime_seconds_created{name="my_app.tasks.fetch_some_data",namespace="celery",queue="celery"} 1.5489449550243628e+09
 # HELP celery_tasks_latency_seconds Seconds between a task is received and started.
 # TYPE celery_tasks_latency_seconds histogram
 celery_tasks_latency_seconds_bucket{namespace="celery",le="0.005"} 2.0

celery_exporter/__main__.py

@@ -10,54 +10,107 @@
 
 __VERSION__ = (2, 0, 0)
 
-LOG_FORMAT = '[%(asctime)s] %(name)s:%(levelname)s: %(message)s'
+LOG_FORMAT = "[%(asctime)s] %(name)s:%(levelname)s: %(message)s"
 
-@click.command(context_settings={
-                    'auto_envvar_prefix':'CELERY_EXPORTER'
-                })
-@click.option('--broker-url', '-b', type=str, show_default=True, show_envvar=True,
-                default='redis://redis:6379/0',
-                help='URL to the Celery broker.')
-@click.option('--listen-address', '-l', type=str, show_default=True, show_envvar=True,
-                default='0.0.0.0:9540',
-                help='Address the HTTPD should listen on.')
-@click.option('--max-tasks', '-m', type=int, show_default=True, show_envvar=True,
-                default='10000',
-                help='Tasks cache size.')
-@click.option('--namespace', '-n', type=str, show_default=True, show_envvar=True,
-                default='celery',
-                help='Namespace for metrics.')
-@click.option('--transport-options', type=str, allow_from_autoenv=False,
-                help='JSON object with additional options passed to the underlying transport.')
-@click.option('--enable-events', is_flag=True, allow_from_autoenv=False,
-                help='Periodically enable Celery events.')
-@click.option('--tz', type=str, allow_from_autoenv=False,
-                help='Timezone used by the celery app.')
-@click.option('--verbose', is_flag=True, allow_from_autoenv=False,
-                help='Enable verbose logging.')
-@click.version_option(version='.'.join([str(x) for x in __VERSION__]))
-def main(broker_url, listen_address, max_tasks, namespace, transport_options, enable_events, tz, verbose):  # pragma: no cover
+
+@click.command(context_settings={"auto_envvar_prefix": "CELERY_EXPORTER"})
+@click.option(
+    "--broker-url",
+    "-b",
+    type=str,
+    show_default=True,
+    show_envvar=True,
+    default="redis://redis:6379/0",
+    help="URL to the Celery broker.",
+)
+@click.option(
+    "--listen-address",
+    "-l",
+    type=str,
+    show_default=True,
+    show_envvar=True,
+    default="0.0.0.0:9540",
+    help="Address the HTTPD should listen on.",
+)
+@click.option(
+    "--max-tasks",
+    "-m",
+    type=int,
+    show_default=True,
+    show_envvar=True,
+    default="10000",
+    help="Tasks cache size.",
+)
+@click.option(
+    "--namespace",
+    "-n",
+    type=str,
+    show_default=True,
+    show_envvar=True,
+    default="celery",
+    help="Namespace for metrics.",
+)
+@click.option(
+    "--transport-options",
+    type=str,
+    allow_from_autoenv=False,
+    help="JSON object with additional options passed to the underlying transport.",
+)
+@click.option(
+    "--enable-events",
+    is_flag=True,
+    allow_from_autoenv=False,
+    help="Periodically enable Celery events.",
+)
+@click.option(
+    "--tz", type=str, allow_from_autoenv=False, help="Timezone used by the celery app."
+)
+@click.option(
+    "--verbose", is_flag=True, allow_from_autoenv=False, help="Enable verbose logging."
+)
+@click.version_option(version=".".join([str(x) for x in __VERSION__]))
+def main(
+    broker_url,
+    listen_address,
+    max_tasks,
+    namespace,
+    transport_options,
+    enable_events,
+    tz,
+    verbose,
+):  # pragma: no cover
 
     if verbose:
         logging.basicConfig(level=logging.DEBUG, format=LOG_FORMAT)
     else:
         logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
 
     if tz:
-        os.environ['TZ'] = tz
+        os.environ["TZ"] = tz
         time.tzset()
 
     if transport_options:
         try:
             transport_options = json.loads(transport_options)
         except ValueError:
-            print("Error parsing broker transport options from JSON '{}'"
-                  .format(transport_options), file=sys.stderr)
-            sys.exit(1)      
+            print(
+                "Error parsing broker transport options from JSON '{}'".format(
+                    transport_options
+                ),
+                file=sys.stderr,
+            )
+            sys.exit(1)
 
-    celery_exporter = CeleryExporter(broker_url, listen_address, max_tasks, namespace, transport_options, enable_events)
+    celery_exporter = CeleryExporter(
+        broker_url,
+        listen_address,
+        max_tasks,
+        namespace,
+        transport_options,
+        enable_events,
+    )
     celery_exporter.start()
-    
+
     def shutdown(signum, frame):  # pragma: no cover
         """
         Shutdown is called if the process receives a TERM/INT signal.
@@ -71,5 +124,6 @@ def shutdown(signum, frame):  # pragma: no cover
     while True:
         signal.pause()
 
-if __name__ == '__main__':  # pragma: no cover
-    main() # pylint: disable=E1120
+
+if __name__ == "__main__":  # pragma: no cover
+    main()  # pylint: disable=E1120

celery_exporter/core.py

@@ -1,13 +1,26 @@
 import celery
 import logging
 import prometheus_client
-from .monitor import TaskThread, WorkerMonitoringThread, EnableEventsThread, setup_metrics
-
-__all__ = ('CeleryExporter',)
-
-class CeleryExporter():
-
-    def __init__(self, broker_url, listen_address, max_tasks=10000, namespace='celery', transport_options=None, enable_events=False):
+from .monitor import (
+    TaskThread,
+    WorkerMonitoringThread,
+    EnableEventsThread,
+    setup_metrics,
+)
+
+__all__ = ("CeleryExporter",)
+
+
+class CeleryExporter:
+    def __init__(
+        self,
+        broker_url,
+        listen_address,
+        max_tasks=10000,
+        namespace="celery",
+        transport_options=None,
+        enable_events=False,
+    ):
         self._listen_address = listen_address
         self._max_tasks = max_tasks
         self._namespace = namespace
@@ -22,11 +35,15 @@ def start(self):
 
         self._start_httpd()
 
-        t = TaskThread(app=self._app, namespace=self._namespace, max_tasks_in_memory=self._max_tasks)
+        t = TaskThread(
+            app=self._app,
+            namespace=self._namespace,
+            max_tasks_in_memory=self._max_tasks,
+        )
         t.daemon = True
         t.start()
 
-        w = WorkerMonitoringThread(app=self._app, namespace=self._namespace,)
+        w = WorkerMonitoringThread(app=self._app, namespace=self._namespace)
         w.daemon = True
         w.start()
 
@@ -40,8 +57,6 @@ def _start_httpd(self):  # pragma: no cover
         Starts the exposing HTTPD using the addr provided in a separate
         thread.
         """
-        host, port = self._listen_address.split(':')
-        logging.info('Starting HTTPD on {}:{}'.format(host, port))
+        host, port = self._listen_address.split(":")
+        logging.info("Starting HTTPD on {}:{}".format(host, port))
         prometheus_client.start_http_server(int(port), host)
-
-

celery_exporter/metrics.py

@@ -1,14 +1,18 @@
 import prometheus_client
 
 TASKS = prometheus_client.Counter(
-    'celery_tasks_total', 'Number of task events.',
-    ['namespace', 'name', 'state'])
+    "celery_tasks_total",
+    "Number of task events.",
+    ["namespace", "name", "state", "queue"],
+)
 TASKS_RUNTIME = prometheus_client.Histogram(
-    'celery_tasks_runtime_seconds', 'Task runtime.',
-    ['namespace', 'name'])
+    "celery_tasks_runtime_seconds", "Task runtime.", ["namespace", "name", "queue"]
+)
 LATENCY = prometheus_client.Histogram(
-    'celery_tasks_latency_seconds', 'Time between a task is received and started.',
-    ['namespace'])
+    "celery_tasks_latency_seconds",
+    "Time between a task is received and started.",
+    ["namespace"],
+)
 WORKERS = prometheus_client.Gauge(
-    'celery_workers', 'Number of alive workers',
-    ['namespace'])
+    "celery_workers", "Number of alive workers", ["namespace"]
+)