feat: add --generic-hostname-worker-task-metric flag

Mirror --generic-hostname-task-sent-metric for the worker-executed metrics:
label the celery_task_* counters and celery_task_runtime with a generic
hostname instead of the executing worker's, to bound label cardinality when
workers are autoscaled (for example Kubernetes/KEDA pods). track_task_event
builds a single labels dict that feeds every counter and the runtime
observation, so one gated assignment collapses them together. celery_task_sent
keeps its own flag, and celery_worker_up / celery_worker_tasks_active are
unaffected since they do not pass through track_task_event.

Refs #387

Co-Authored-By: Claude <noreply@anthropic.com>

Author: bafulton

src/cli.py

@@ -111,6 +111,16 @@ def _eq_sign_separated_argument_to_dict(_ctx, _param, value):
     "Knowing which client sent a task might not be useful for many use cases as for example in "
     "Kubernetes environments where the client's hostname is a random string.",
 )
+@click.option(
+    "--generic-hostname-worker-task-metric",
+    default=False,
+    is_flag=True,
+    help="The celery_task_* counters and celery_task_runtime will be labeled with a generic "
+    "hostname instead of the executing worker's hostname. This option helps with label "
+    "cardinality when using a dynamic number of workers, as for example in Kubernetes "
+    "environments where the worker's hostname is a random string. celery_task_sent is not "
+    "affected; use --generic-hostname-task-sent-metric for the client side.",
+)
 @click.option(
     "-Q",
     "--queues",
@@ -155,6 +165,7 @@ def cli(  # pylint: disable=too-many-arguments,too-many-positional-arguments,too
     worker_timeout,
     purge_offline_worker_metrics,
     generic_hostname_task_sent_metric,
+    generic_hostname_worker_task_metric,
     queues,
     metric_prefix,
     default_queue_name,
@@ -167,6 +178,7 @@ def cli(  # pylint: disable=too-many-arguments,too-many-positional-arguments,too
         worker_timeout,
         purge_offline_worker_metrics,
         generic_hostname_task_sent_metric,
+        generic_hostname_worker_task_metric,
         queues,
         metric_prefix,
         default_queue_name,

src/exporter.py

@@ -27,6 +27,7 @@ def __init__(
         worker_timeout_seconds=5 * 60,
         purge_offline_worker_metrics_seconds=10 * 60,
         generic_hostname_task_sent_metric=False,
+        generic_hostname_worker_task_metric=False,
         initial_queues=None,
         metric_prefix="celery_",
         default_queue_name="celery",
@@ -40,6 +41,7 @@ def __init__(
             purge_offline_worker_metrics_seconds
         )
         self.generic_hostname_task_sent_metric = generic_hostname_task_sent_metric
+        self.generic_hostname_worker_task_metric = generic_hostname_worker_task_metric
         self.default_queue_name = default_queue_name
 
         # Static labels
@@ -284,7 +286,10 @@ def track_task_event(self, event):
             "queue_name": getattr(task, "queue", self.default_queue_name),
             **self.static_label,
         }
-        if event["type"] == "task-sent" and self.generic_hostname_task_sent_metric:
+        if event["type"] == "task-sent":
+            if self.generic_hostname_task_sent_metric:
+                labels["hostname"] = "generic"
+        elif self.generic_hostname_worker_task_metric:
             labels["hostname"] = "generic"
 
         for counter_name, counter in self.state_counters.items():

src/test_metrics.py

@@ -258,3 +258,84 @@ def succeed():
             )
             is None
         )
+
+
+def test_worker_generic_task_hostname(threaded_exporter, celery_app, hostname):
+    threaded_exporter.generic_hostname_worker_task_metric = True
+    time.sleep(5)
+
+    @celery_app.task
+    def succeed():
+        pass
+
+    succeed.apply_async()
+
+    with start_worker(celery_app, without_heartbeat=False):
+        time.sleep(5)
+
+        # The worker-executed counter and the runtime histogram carry the generic
+        # hostname, not the executing worker's.
+        assert (
+            threaded_exporter.registry.get_sample_value(
+                "celery_task_succeeded_total",
+                labels={
+                    "hostname": "generic",
+                    "name": "src.test_metrics.succeed",
+                    "queue_name": "celery",
+                },
+            )
+            == 1.0
+        )
+        assert (
+            threaded_exporter.registry.get_sample_value(
+                "celery_task_runtime_count",
+                labels={
+                    "hostname": "generic",
+                    "name": "src.test_metrics.succeed",
+                    "queue_name": "celery",
+                },
+            )
+            == 1.0
+        )
+        # The runtime histogram is only ever touched by observe() on the collapsed
+        # worker event, so no series exists under the real worker hostname.
+        assert (
+            threaded_exporter.registry.get_sample_value(
+                "celery_task_runtime_count",
+                labels={
+                    "hostname": hostname,
+                    "name": "src.test_metrics.succeed",
+                    "queue_name": "celery",
+                },
+            )
+            is None
+        )
+
+        # celery_task_sent is client-side and governed by its own flag, so this flag
+        # leaves it labeled with the real hostname.
+        assert (
+            threaded_exporter.registry.get_sample_value(
+                "celery_task_sent_total",
+                labels={
+                    "hostname": hostname,
+                    "name": "src.test_metrics.succeed",
+                    "queue_name": "celery",
+                },
+            )
+            == 1.0
+        )
+
+        # celery_worker_up does not pass through track_task_event, so it keeps the real
+        # per-worker hostname that KEDA's scaler depends on.
+        assert (
+            threaded_exporter.registry.get_sample_value(
+                "celery_worker_up", labels={"hostname": hostname}
+            )
+            == 1.0
+        )
+        assert (
+            threaded_exporter.registry.get_sample_value(
+                "celery_worker_up", labels={"hostname": "generic"}
+            )
+            is None
+        )