diff --git a/CHANGES.md b/CHANGES.md index a2cd1b1..855de20 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -2,7 +2,7 @@ -## 1.6.0 (unreleased) +## 2.0.0 (unreleased) - Nothing changed yet. diff --git a/README.md b/README.md index 46b8b51..e127127 100644 --- a/README.md +++ b/README.md @@ -33,10 +33,35 @@ We export the following metrics: * `celery_task_queuetime_seconds{queue}`, histogram (only if `task_send_sent_event` is enabled in Celery) * `celery_task_runtime_seconds{queue}`, histogram -If you pass `--queuelength-interval=x` then every x seconds the queue lengths will be checked (NOTE: this only works with redis as the broker), resulting in this additional metric: +If you pass `--queue-length`, the queue lengths are read from the broker on each scrape (NOTE: this only works with redis as the broker), resulting in this additional metric: * `celery_queue_length{queue="..."}`, gauge +While the broker cannot be reached, `celery_queue_length` is left out. +So an alert like `celery_queue_length > 100` silently stops firing; +add one that fires when the metric is missing. +With a single exporter per alert selector, `absent()` does that: + +```yaml +- alert: CeleryQueueLengthMissing + expr: absent(celery_queue_length{namespace="myapp"}) + for: 5m +``` + +`absent()` only fires if *no* series matches the selector at all. +If the selector covers several exporters (e.g. one per broker), +one of them failing goes unnoticed as long as another still reports. +Compare against the `up` series Prometheus records for each scrape target instead, +which yields one alert per exporter that is down or cannot read its broker: + +```yaml +- alert: CeleryQueueLengthMissing + expr: up{job="myapp/celery-metrics"} unless on(namespace, pod) celery_queue_length + for: 5m +``` + +The `up` selector must match only exporters started with `--queue-length`, and `on(...)` +must name labels that identify the exporter on both sides. ## Run tests diff --git a/changelog/+queue-length-flag.change b/changelog/+queue-length-flag.change new file mode 100644 index 0000000..500fae7 --- /dev/null +++ b/changelog/+queue-length-flag.change @@ -0,0 +1 @@ +Replace `--queuelength-interval=x` with the flag `--queue-length`. The old option is rejected, so update the exporter arguments when upgrading diff --git a/changelog/+scrape.change b/changelog/+scrape.change new file mode 100644 index 0000000..9a35900 --- /dev/null +++ b/changelog/+scrape.change @@ -0,0 +1 @@ +Read queue lengths from the broker on each scrape instead of polling them in a background thread. While the broker cannot be reached, `celery_queue_length` is left out instead of keeping stale values; see the README on how to alert on that diff --git a/changelog/+timeout.change b/changelog/+timeout.change new file mode 100644 index 0000000..a563252 --- /dev/null +++ b/changelog/+timeout.change @@ -0,0 +1 @@ +Give up reading queue lengths on a hanging broker after 3 seconds per connect, handshake and query (so about 9 seconds at worst), to still answer within the default 10 second scrape timeout. Configurable via `socket_timeout`/`socket_connect_timeout` in `broker_transport_options` diff --git a/pyproject.toml b/pyproject.toml index 76ea375..a5192b3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,12 +1,12 @@ [project] name = "celery_redis_prometheus" description = "Exports task execution metrics in Prometheus format" -version = "1.6.0.dev0" +version = "2.0.0.dev0" dependencies = [ "celery >= 5.0.0.dev0", "click", - "prometheus_client", + "prometheus_client >= 0.14.0", ] authors = [{name="DIE ZEIT", email="engineering@zeit.de"}] diff --git a/src/celery_redis_prometheus/exporter.py b/src/celery_redis_prometheus/exporter.py index 3120aa3..39b5e24 100644 --- a/src/celery_redis_prometheus/exporter.py +++ b/src/celery_redis_prometheus/exporter.py @@ -1,35 +1,38 @@ from functools import wraps -import _thread import collections import json import logging -import threading import time import celery.bin.base import click import prometheus_client +import prometheus_client.core +import prometheus_client.registry log = logging.getLogger(__name__) -# Remove any `process_` and `python_` metrics, since we're proxying for the -# whole celery machinery, but those would only be about this process. -prometheus_client.REGISTRY.unregister(prometheus_client.GC_COLLECTOR) -prometheus_client.REGISTRY.unregister(prometheus_client.PLATFORM_COLLECTOR) -prometheus_client.REGISTRY.unregister(prometheus_client.PROCESS_COLLECTOR) +# Our own registry instead of prometheus_client's default one, so we don't +# export its `process_` and `python_` metrics: we're proxying for the whole +# celery machinery, but those would only be about this process. +REGISTRY = prometheus_client.CollectorRegistry() +# These are rendered by the HTTP server started in main() on each request. +# They will be updated later by processing Celery events and the next scrape +# will pick up the new values. STATS = { - 'tasks': prometheus_client.Counter('celery_tasks_total', 'Number of tasks', ['queue', 'state']), + 'tasks': prometheus_client.Counter( + 'celery_tasks_total', 'Number of tasks', ['queue', 'state'], registry=REGISTRY + ), 'queuetime': prometheus_client.Histogram( - 'celery_task_queuetime_seconds', 'Task queue wait time', ['queue'] + 'celery_task_queuetime_seconds', 'Task queue wait time', ['queue'], registry=REGISTRY ), 'runtime': prometheus_client.Histogram( - 'celery_task_runtime_seconds', 'Task runtime', ['queue'] + 'celery_task_runtime_seconds', 'Task runtime', ['queue'], registry=REGISTRY ), - 'queues': prometheus_client.Gauge('celery_queue_length', 'Queue length', ['queue']), } @@ -37,46 +40,30 @@ @click.option('--host', default='0.0.0.0', help='Listen host') @click.option('--port', default=9691, help='Listen port') @click.option( - '--queuelength-interval', default=0, help='Check queue lengths every x seconds (0=disabled)' + '--queue-length', + is_flag=True, + help='Export queue lengths, read from the broker on each scrape (redis only)', ) @click.option('--verbose', is_flag=True, help='Enable debug logging') @click.pass_context -def main(ctx, host, port, queuelength_interval, verbose): +def main(ctx, host, port, queue_length, verbose): app = ctx.obj.app app.log.setup(logging.DEBUG if verbose else logging.INFO) - queuelength_thread = None - if queuelength_interval: - queuelength_thread = QueueLengthMonitor(app, queuelength_interval) - queuelength_thread.start() + if queue_length: + REGISTRY.register(QueueLengthCollector(app)) - receiver = CeleryEventReceiver(app) - prometheus_client.start_http_server(port, host) + # Serves REGISTRY from a daemon thread. + prometheus_client.start_http_server(port, host, registry=REGISTRY) log.info('Listening on %s:%s', host, port) - try_interval = 1 - while True: - try: - try_interval *= 2 - receiver() - try_interval = 1 - except (KeyboardInterrupt, SystemExit): - log.info('Exiting') - if queuelength_thread: - queuelength_thread.stop() - _thread.interrupt_main() - break - except Exception as e: - log.error( - 'Failed to capture events: "%s", trying again in %s seconds.', - e, - try_interval, - exc_info=True, - ) - time.sleep(try_interval) + CeleryEventReceiver(app).run_forever() def task_handler(fn): + """Wrapper to first apply the event, get the relevant task and forward + to the relevant handler.""" + @wraps(fn) def wrapper(self, event): self.state.event(event) @@ -87,6 +74,19 @@ def wrapper(self, event): class CeleryEventReceiver: + """Turns the Celery event stream into the metrics in STATS. + + Workers publish an event to the broker for each task state change, if + they are started with `-E` (or `worker_send_task_events`). We subscribe + to these and record them in a `celery.events.State`, which collects the + events of each task. The handlers only update the metrics, Prometheus + picks up their current values on its next scrape. + + This goes against the recommended Prometheus pattern of "observe on scrape", + but we can't really collect the Celery task metrics without keeping state. + Redis metrics can be collected on scrape, QueueLengthCollector implements that. + """ + def __init__(self, app): self.app = app @@ -123,7 +123,26 @@ def on_task_retried(self, event, task): STATS['tasks'].labels(task.routing_key, 'retried').inc() self.record_runtime(task) - def __call__(self, *args, **kw): + def run_forever(self): + try_interval = 1 + while True: + try: + try_interval *= 2 + self.capture() + try_interval = 1 + except (KeyboardInterrupt, SystemExit): + log.info('Exiting') + break + except Exception as e: + log.error( + 'Failed to capture events: "%s", trying again in %s seconds.', + e, + try_interval, + exc_info=True, + ) + time.sleep(try_interval) + + def capture(self, *args, **kw): self.state = self.app.events.State() kw.setdefault('wakeup', False) @@ -141,43 +160,65 @@ def __call__(self, *args, **kw): recv.capture(*args, **kw) -class QueueLengthMonitor(threading.Thread): - def __init__(self, app, interval): - super().__init__() - self.app = app - self.interval = interval - self.running = True +class QueueLengthCollector(prometheus_client.registry.Collector): + """ + Reads the queue lengths from Redis and exposes them on each scrape. + This is a separate collector because we don't need to keep Celery state + and we want to be able to tell if the broker is down. + """ - def run(self): - while self.running: - try: - lengths = collections.Counter() - - with self.app.connection() as connection: - pipe = connection.channel().client.pipeline(transaction=False) - for queue in self.app.conf['task_queues']: - # Not claimed by any worker yet - pipe.llen(queue.name) - # Claimed by worker but not acked/processed yet - pipe.hvals('unacked') - - result = pipe.execute() - - unacked = result.pop() - for task in unacked: - data = json.loads(task.decode('utf-8')) - queue = data[-1] - lengths[queue] += 1 + # By default, the broker connection waits forever for a Redis that stops + # answering, but we have to respond within the Prometheus scrape timeout + # (10s by default). Connecting, the handshake and the query can each take + # one full timeout, so 3 x 3s at worst. + TRANSPORT_OPTIONS = {'socket_timeout': 3, 'socket_connect_timeout': 3} - for llen, queue in zip(result, self.app.conf['task_queues']): - lengths[queue.name] += llen - - for queue, length in lengths.items(): - STATS['queues'].labels(queue).set(length) + def __init__(self, app): + self.app = app - time.sleep(self.interval) - except Exception: - log.error('Uncaught exception, preventing thread from crashing.', exc_info=True) + # Without describe(), register() would call collect() and thus hit redis. + def describe(self): + yield self._lengths_metric() - def stop(self): - self.running = False + def collect(self): + try: + lengths = self.queue_lengths() + except Exception: + log.error('Failed to read queue lengths', exc_info=True) + return + metric = self._lengths_metric() + for queue, length in lengths.items(): + metric.add_metric([queue], length) + yield metric + + def queue_lengths(self): + # Celery puts the options passed here over `broker_transport_options`, + # so merge ours underneath to let the configured ones win. + options = dict( + self.TRANSPORT_OPTIONS, **self.app.conf.get('broker_transport_options') or {} + ) + queues = self.app.conf['task_queues'] + with self.app.connection(transport_options=options) as connection: + # channel() would otherwise connect with one retry after sleeping 2s, + # which doubles the time we hang on an unresponsive broker. + connection.ensure_connection(max_retries=0) + pipe = connection.channel().client.pipeline(transaction=False) + for queue in queues: + # Not claimed by any worker yet + pipe.llen(queue.name) + # Claimed by worker but not acked/processed yet + pipe.hvals('unacked') + result = pipe.execute() + + lengths = collections.Counter() + for task in result.pop(): + data = json.loads(task.decode('utf-8')) + lengths[data[-1]] += 1 + for llen, queue in zip(result, queues): + lengths[queue.name] += llen + return lengths + + def _lengths_metric(self): + return prometheus_client.core.GaugeMetricFamily( + 'celery_queue_length', 'Queue length', labels=['queue'] + ) diff --git a/src/celery_redis_prometheus/tests/test_exporter.py b/src/celery_redis_prometheus/tests/test_exporter.py index cb05d58..8f5bb68 100644 --- a/src/celery_redis_prometheus/tests/test_exporter.py +++ b/src/celery_redis_prometheus/tests/test_exporter.py @@ -1,5 +1,9 @@ +import json import threading +import unittest.mock +import kombu +import prometheus_client import pytest import celery_redis_prometheus.exporter @@ -10,7 +14,7 @@ def test_collects_task_events(celery_worker): receiver = celery_redis_prometheus.exporter.CeleryEventReceiver(conftest.CELERY) # 3 = recived + started + succeeded - thread = threading.Thread(target=lambda: receiver(limit=3)) + thread = threading.Thread(target=lambda: receiver.capture(limit=3)) thread.start() conftest.celery_ping.delay().get() thread.join() @@ -22,7 +26,7 @@ def test_collects_task_events(celery_worker): def test_sets_separate_state_for_retry_failed(celery_worker): receiver = celery_redis_prometheus.exporter.CeleryEventReceiver(conftest.CELERY) # 6 = received + started + retry + received + started + failed - thread = threading.Thread(target=lambda: receiver(limit=6)) + thread = threading.Thread(target=lambda: receiver.capture(limit=6)) thread.start() with pytest.raises(Exception): conftest.provoke_retry.delay().get() @@ -30,3 +34,122 @@ def test_sets_separate_state_for_retry_failed(celery_worker): data = celery_redis_prometheus.exporter.STATS['tasks'].collect() item = [x for x in data[0].samples if x.labels.get('state', '') == 'retries-exceeded'] assert item[0].value == 1 + + +def redis_app(*results): + """Returns a mock app whose redis pipeline answers each scrape with the + next of `results`: `[llen per queue..., hvals('unacked')]`. + + The collector reads them via `with app.connection() as connection:` and + `connection.channel().client.pipeline(...)`. + """ + pipe = unittest.mock.Mock() + pipe.execute.side_effect = [list(x) for x in results] + connection = unittest.mock.MagicMock() + connection.__enter__.return_value = connection + connection.channel.return_value.client.pipeline.return_value = pipe + app = unittest.mock.Mock() + app.connection.return_value = connection + app.conf = {'task_queues': [kombu.Queue('default'), kombu.Queue('other')]} + return app + + +def broken_app(**conf): + app = unittest.mock.MagicMock() + app.conf = {'task_queues': [kombu.Queue('default')], **conf} + app.connection.side_effect = ConnectionError('broker down') + return app + + +def registry_with_queue_metrics(app): + registry = prometheus_client.CollectorRegistry(auto_describe=True) + registry.register(celery_redis_prometheus.exporter.QueueLengthCollector(app)) + return registry + + +def unacked(queue): + return json.dumps([{}, '', queue]).encode('utf-8') + + +def scrape(registry): + return { + (sample.name, tuple(sample.labels.values())): sample.value + for metric in registry.collect() + for sample in metric.samples + } + + +def test_adds_unacked_tasks_to_queue_length(): + app = redis_app([3, 0, [unacked('default'), unacked('default')]]) + assert scrape(registry_with_queue_metrics(app)) == { + ('celery_queue_length', ('default',)): 5, + ('celery_queue_length', ('other',)): 0, + } + + +def test_reads_queue_lengths_on_each_scrape_only(): + app = redis_app([1, 0, []], [2, 0, []]) + registry = registry_with_queue_metrics(app) + app.connection.assert_not_called() + + assert registry.get_sample_value('celery_queue_length', {'queue': 'default'}) == 1 + assert registry.get_sample_value('celery_queue_length', {'queue': 'default'}) == 2 + + +def test_failed_read_leaves_out_queue_lengths(): + assert scrape(registry_with_queue_metrics(broken_app())) == {} + + +def test_recovers_after_failed_read(): + app = redis_app([1, 0, []]) + registry = registry_with_queue_metrics(app) + app.connection.side_effect = [ConnectionError('broker down'), app.connection.return_value] + + assert registry.get_sample_value('celery_queue_length', {'queue': 'default'}) is None + assert registry.get_sample_value('celery_queue_length', {'queue': 'default'}) == 1 + + +def test_socket_configuration_set_by_default(): + app = broken_app() + scrape(registry_with_queue_metrics(app)) + + options = app.connection.call_args.kwargs['transport_options'] + assert options == {'socket_timeout': 3, 'socket_connect_timeout': 3} + + +def test_does_not_retry_connecting(): + app = redis_app([1, 0, []]) + scrape(registry_with_queue_metrics(app)) + + connection = app.connection.return_value + connection.ensure_connection.assert_called_once_with(max_retries=0) + + +def test_configured_transport_options_override_default_timeout(): + app = broken_app(broker_transport_options={'socket_timeout': 2}) + scrape(registry_with_queue_metrics(app)) + + options = app.connection.call_args.kwargs['transport_options'] + assert options == {'socket_timeout': 2, 'socket_connect_timeout': 3} + + +def test_does_not_export_metrics_about_exporter_process(): + names = {metric.name for metric in celery_redis_prometheus.exporter.REGISTRY.collect()} + assert not [x for x in names if x.startswith(('process_', 'python_'))] + assert 'celery_tasks' in names + + +def test_broker_outage_does_not_impact_task_metrics(): + collector = celery_redis_prometheus.exporter.QueueLengthCollector(broken_app()) + # main() registers on the exporter's registry, next to the event metrics. + registry = celery_redis_prometheus.exporter.REGISTRY + registry.register(collector) + try: + STATS = celery_redis_prometheus.exporter.STATS + STATS['tasks'].labels('coexist', 'succeeded').inc() + + labels = {'queue': 'coexist', 'state': 'succeeded'} + assert registry.get_sample_value('celery_tasks_total', labels) == 1 + assert registry.get_sample_value('celery_queue_length', {'queue': 'default'}) is None + finally: + registry.unregister(collector) diff --git a/uv.lock b/uv.lock index d96ac94..3e77aaf 100644 --- a/uv.lock +++ b/uv.lock @@ -185,7 +185,7 @@ wheels = [ [[package]] name = "celery-redis-prometheus" -version = "1.6.0.dev0" +version = "2.0.0.dev0" source = { editable = "." } dependencies = [ { name = "celery", version = "5.2.7", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.8'" }, @@ -219,7 +219,7 @@ release = [ requires-dist = [ { name = "celery", specifier = ">=5.0.0.dev0" }, { name = "click" }, - { name = "prometheus-client" }, + { name = "prometheus-client", specifier = ">=0.14.0" }, ] [package.metadata.requires-dev]