Add GCP structured log labels for task_id/run_id on backend logs

requirements/common.in

@@ -40,6 +40,9 @@ mozci[cache]==2.4.3
 # Dockerflow/CloudOps APIs
 dockerflow==2026.3.4
 
+# Structured logging to GCP Cloud Logging (task_id/run_id labels)
+google-cloud-logging==3.16.0
+
 # Measuring noise of perf data
 moz-measure-noise==2.70.0
 

tests/etl/taskcluster_pulse/test_handler.py

@@ -1,6 +1,8 @@
 import pytest
 
+from treeherder.etl.taskcluster_pulse import handler as tc_handler
 from treeherder.etl.taskcluster_pulse.handler import handle_message, handle_task_defined
+from treeherder.utils.logging_context import get_log_labels
 
 
 @pytest.mark.asyncio
@@ -43,6 +45,45 @@ async def test_handle_message_routes_task_defined():
     assert result[0]["result"] == "unknown"
 
 
+@pytest.mark.asyncio
+async def test_handle_message_sets_log_context(monkeypatch):
+    """handle_message wraps processing in a log_context carrying task_id/run_id."""
+    captured = {}
+
+    def capture_labels(*args, **kwargs):
+        captured.update(get_log_labels())
+        return {"state": "unscheduled", "result": "unknown"}
+
+    monkeypatch.setattr(tc_handler, "handle_task_defined", capture_labels)
+
+    task = {
+        "metadata": {"name": "t", "description": "d", "owner": "o@example.com"},
+        "created": "2025-01-01T00:00:00.000Z",
+        "workerType": "test-worker",
+        "tags": {},
+        "routes": ["tc-treeherder.v2.autoland.abc123"],
+        "extra": {"treeherder": {"symbol": "T", "tier": 1}},
+    }
+    message = {
+        "exchange": "exchange/taskcluster-queue/v1/task-defined",
+        "root_url": "https://firefox-ci-tc.services.mozilla.com",
+        "payload": {
+            "runId": 0,
+            "status": {"taskId": "AJBb7wqZT6K9kz4niYAatg", "state": "unscheduled", "runs": []},
+        },
+    }
+
+    await handle_message(message, task)
+
+    assert captured == {
+        "task_id": "AJBb7wqZT6K9kz4niYAatg",
+        "run_id": "0",
+        "component": "ingestion",
+    }
+    # context is cleaned up after the handler returns
+    assert get_log_labels() == {}
+
+
 def test_handle_task_defined():
     push_info = {
         "project": "autoland",

tests/utils/test_gcp_logging.py

@@ -0,0 +1,51 @@
+import io
+import json
+import logging
+
+import pytest
+
+pytest.importorskip("google.cloud.logging_v2")
+
+from treeherder.utils.gcp_logging import ContextLabelStructuredLogHandler
+from treeherder.utils.logging_context import log_context
+
+
+def emit_within_context(**context_labels):
+    """Emit one log line through the handler inside a log_context, return its GCP labels."""
+    stream = io.StringIO()
+    handler = ContextLabelStructuredLogHandler(stream=stream)
+    logger = logging.getLogger("test.gcp_logging")
+    logger.handlers = [handler]
+    logger.setLevel(logging.INFO)
+    logger.propagate = False
+    with log_context(**context_labels):
+        logger.info("hello world")
+    for line in stream.getvalue().splitlines():
+        entry = json.loads(line)
+        if entry.get("message") == "hello world":
+            return entry.get("logging.googleapis.com/labels", {})
+    raise AssertionError("log line not found in handler output")
+
+
+def test_handler_emits_context_labels_as_gcp_labels():
+    labels = emit_within_context(task_id="ABC", run_id="0", component="failureline")
+    assert labels["task_id"] == "ABC"
+    assert labels["run_id"] == "0"
+    assert labels["component"] == "failureline"
+
+
+def test_handler_emits_no_task_labels_without_context():
+    stream = io.StringIO()
+    handler = ContextLabelStructuredLogHandler(stream=stream)
+    logger = logging.getLogger("test.gcp_logging.nolabels")
+    logger.handlers = [handler]
+    logger.setLevel(logging.INFO)
+    logger.propagate = False
+    logger.info("plain message")
+    for line in stream.getvalue().splitlines():
+        entry = json.loads(line)
+        if entry.get("message") == "plain message":
+            labels = entry.get("logging.googleapis.com/labels", {})
+            assert "task_id" not in labels
+            return
+    raise AssertionError("log line not found in handler output")

tests/utils/test_logging_context.py

@@ -0,0 +1,108 @@
+import logging
+
+import pytest
+
+from treeherder.model import models as th_models
+from treeherder.utils.logging_context import (
+    apply_context_labels,
+    get_log_labels,
+    job_log_labels,
+    log_context,
+)
+
+
+def make_record():
+    return logging.LogRecord(
+        name="treeherder.test",
+        level=logging.INFO,
+        pathname=__file__,
+        lineno=1,
+        msg="hello",
+        args=(),
+        exc_info=None,
+    )
+
+
+def test_get_log_labels_empty_outside_context():
+    assert get_log_labels() == {}
+
+
+def test_log_context_sets_labels():
+    with log_context(task_id="ABC", run_id="0"):
+        assert get_log_labels() == {"task_id": "ABC", "run_id": "0"}
+    # restored after the block exits
+    assert get_log_labels() == {}
+
+
+def test_log_context_merges_and_restores_nested():
+    with log_context(task_id="ABC", component="ingestion"):
+        with log_context(run_id="2", component="failureline"):
+            assert get_log_labels() == {
+                "task_id": "ABC",
+                "run_id": "2",
+                "component": "failureline",
+            }
+        # inner block unwinds, outer labels remain
+        assert get_log_labels() == {"task_id": "ABC", "component": "ingestion"}
+    assert get_log_labels() == {}
+
+
+def test_log_context_ignores_none_values():
+    with log_context(task_id="ABC", run_id=None, job_id=None):
+        assert get_log_labels() == {"task_id": "ABC"}
+
+
+def test_get_log_labels_returns_a_copy():
+    with log_context(task_id="ABC"):
+        labels = get_log_labels()
+        labels["task_id"] = "MUTATED"
+        assert get_log_labels() == {"task_id": "ABC"}
+
+
+def test_log_context_usable_as_decorator():
+    @log_context(component="failureline")
+    def do_work():
+        return get_log_labels()
+
+    assert do_work() == {"component": "failureline"}
+    assert get_log_labels() == {}
+
+
+def test_apply_context_labels_stashes_current_labels_on_record():
+    record = make_record()
+    with log_context(task_id="ABC", run_id="0"):
+        apply_context_labels(record)
+    assert record.labels == {"task_id": "ABC", "run_id": "0"}
+
+
+def test_apply_context_labels_preserves_per_call_labels():
+    record = make_record()
+    # a label set explicitly on the call (via extra={"labels": ...}) wins over context
+    record.labels = {"task_id": "EXPLICIT"}
+    with log_context(task_id="CONTEXT", run_id="0"):
+        apply_context_labels(record)
+    assert record.labels == {"task_id": "EXPLICIT", "run_id": "0"}
+
+
+def test_apply_context_labels_no_context_is_noop():
+    record = make_record()
+    apply_context_labels(record)
+    assert getattr(record, "labels", {}) == {}
+
+
+@pytest.mark.django_db
+def test_job_log_labels_with_taskcluster_metadata(test_job):
+    meta = test_job.taskcluster_metadata
+    assert job_log_labels(test_job) == {
+        "task_id": meta.task_id,
+        "run_id": str(meta.retry_id),
+        "job_id": str(test_job.id),
+    }
+
+
+@pytest.mark.django_db
+def test_job_log_labels_without_metadata_omits_task_and_run(test_job):
+    th_models.TaskclusterMetadata.objects.filter(job=test_job).delete()
+    # re-fetch to drop the cached reverse relation
+    job = th_models.Job.objects.get(id=test_job.id)
+    assert job_log_labels(job) == {"job_id": str(job.id)}

treeherder/config/settings.py

@@ -25,6 +25,11 @@
 DEBUG = env.bool("TREEHERDER_DEBUG", default=False)
 LOGGING_LEVEL = env("LOGGING_LEVEL", default="INFO")
 
+# When enabled, the `treeherder` logger emits GCP-structured JSON to stdout with
+# task_id/run_id/component attached as queryable Cloud Logging labels (see
+# treeherder.utils.gcp_logging). Off by default so local/dev stays human-readable.
+GCP_STRUCTURED_LOGGING = env.bool("GCP_STRUCTURED_LOGGING", default=False)
+
 NEW_RELIC_INSIGHTS_API_KEY = env("NEW_RELIC_INSIGHTS_API_KEY", default=None)
 NEW_RELIC_INSIGHTS_API_URL = "https://insights-api.newrelic.com/v1/accounts/677903/query"
 
@@ -251,6 +256,16 @@
     },
 }
 
+if GCP_STRUCTURED_LOGGING:
+    # Route the `treeherder` logger through the GCP-structured handler so its
+    # records carry task_id/run_id/component labels. Added conditionally so the
+    # google-cloud-logging import only happens when the feature is enabled.
+    LOGGING["handlers"]["gcp_structured"] = {
+        "class": "treeherder.utils.gcp_logging.ContextLabelStructuredLogHandler",
+        "level": "DEBUG",
+    }
+    LOGGING["loggers"]["treeherder"]["handlers"] = ["gcp_structured"]
+
 # SECURITY
 
 CSRF_TRUSTED_ORIGINS = env.list(

treeherder/etl/taskcluster_pulse/handler.py

@@ -12,6 +12,7 @@
 
 from treeherder.etl.schema import get_json_schema
 from treeherder.etl.taskcluster_pulse.parse_route import parse_route
+from treeherder.utils.logging_context import log_context
 
 env = environ.Env()
 logger = logging.getLogger(__name__)
@@ -176,45 +177,51 @@ async def handle_message(message, task_definition=None):
         task_id = message["payload"]["status"]["taskId"]
         # task-defined messages don't have runId since the task has no runs yet
         run_id = message["payload"].get("runId", 0)
-        if task_definition:
-            task = task_definition
-        else:
-            async_queue = taskcluster.aio.Queue({"rootUrl": message["root_url"]}, session=session)
-            with settings.STATSD_CLIENT.timer("taskcluster_fetch_definition"):
-                task = await async_queue.task(task_id)
 
-        try:
-            parsed_route = parse_route_info("tc-treeherder", task_id, task["routes"], task)
-        except PulseHandlerError as e:
-            logger.debug("%s", str(e))
-            return jobs
+        # Attach task_id/run_id as GCP log labels to every line emitted while
+        # ingesting this task (route parsing, task filtering, validation, etc.).
+        with log_context(task_id=task_id, run_id=str(run_id), component="ingestion"):
+            if task_definition:
+                task = task_definition
+            else:
+                async_queue = taskcluster.aio.Queue(
+                    {"rootUrl": message["root_url"]}, session=session
+                )
+                with settings.STATSD_CLIENT.timer("taskcluster_fetch_definition"):
+                    task = await async_queue.task(task_id)
 
-        if ignore_task(task, task_id, message["root_url"], parsed_route["project"]):
-            return jobs
-
-        logger.debug("Message received for task %s", task_id)
+            try:
+                parsed_route = parse_route_info("tc-treeherder", task_id, task["routes"], task)
+            except PulseHandlerError as e:
+                logger.debug("%s", str(e))
+                return jobs
+
+            if ignore_task(task, task_id, message["root_url"], parsed_route["project"]):
+                return jobs
+
+            logger.debug("Message received for task %s", task_id)
+
+            # Validation failures are common and logged, so do nothing more.
+            if not validate_task(task, task_id, run_id):
+                return jobs
+
+            task_type = EXCHANGE_EVENT_MAP.get(message["exchange"])
+
+            if not task_type:
+                raise Exception("Unknown exchange: {exchange}".format(exchange=message["exchange"]))
+            elif task_type == "unscheduled":
+                jobs.append(handle_task_defined(parsed_route, task, message))
+            elif task_type == "pending":
+                jobs.append(handle_task_pending(parsed_route, task, message))
+            elif task_type == "running":
+                jobs.append(handle_task_running(parsed_route, task, message))
+            elif task_type in ("completed", "failed"):
+                jobs.append(await handle_task_completed(parsed_route, task, message, session))
+            elif task_type == "exception":
+                jobs.append(await handle_task_exception(parsed_route, task, message, session))
 
-        # Validation failures are common and logged, so do nothing more.
-        if not validate_task(task, task_id, run_id):
             return jobs
 
-        task_type = EXCHANGE_EVENT_MAP.get(message["exchange"])
-
-        if not task_type:
-            raise Exception("Unknown exchange: {exchange}".format(exchange=message["exchange"]))
-        elif task_type == "unscheduled":
-            jobs.append(handle_task_defined(parsed_route, task, message))
-        elif task_type == "pending":
-            jobs.append(handle_task_pending(parsed_route, task, message))
-        elif task_type == "running":
-            jobs.append(handle_task_running(parsed_route, task, message))
-        elif task_type in ("completed", "failed"):
-            jobs.append(await handle_task_completed(parsed_route, task, message, session))
-        elif task_type == "exception":
-            jobs.append(await handle_task_exception(parsed_route, task, message, session))
-
-        return jobs
-
 
 # Builds the basic Treeherder job message that's universal for all
 # messsage types.