bugsnag: propagate cloud_provider into contextual logging for queued executions

Adds CLOUD_PROVIDER_ANNOTATION_KEY to common_annotations so the annotation
key has a single definition in the OSS layer.

In the queued-execution processing span, reads the cloud_provider value from
task_spec annotations (already in memory) and adds it to the contextual
logging context when present. The existing _before_notify hook then includes
it automatically in the tangle_context tab on every Bugsnag event raised
during that execution's processing.

Author: morgan-wowk

cloud_pipelines_backend/instrumentation/contextual_logging.py

@@ -22,9 +22,14 @@
 """
 
 import contextvars
+import typing
 from contextlib import contextmanager
 from typing import Any, Optional
 
+from .. import backend_types_sql as bts
+
+_CLOUD_PROVIDER_ANNOTATION_KEY = "cloud-pipelines.net/orchestration/cloud_provider"
+
 # Single context variable to store all metadata as a dictionary
 _context_metadata: contextvars.ContextVar[dict[str, Any]] = contextvars.ContextVar(
     "context_metadata", default={}
@@ -125,3 +130,21 @@ def logging_context(**metadata: Any):
     finally:
         # Restore previous metadata
         _context_metadata.set(prev_metadata)
+
+
+def execution_logging_context(execution: bts.ExecutionNode):
+    """Return a logging context populated with metadata for *execution*.
+
+    Always sets ``execution_id``.  Also sets ``cloud_provider`` when the
+    ``cloud-pipelines.net/orchestration/cloud_provider`` annotation is present
+    on the task spec.
+    """
+    ctx: dict[str, typing.Any] = {"execution_id": execution.id}
+    cloud_provider = (
+        (execution.task_spec or {})
+        .get("annotations", {})
+        .get(_CLOUD_PROVIDER_ANNOTATION_KEY)
+    )
+    if cloud_provider is not None:
+        ctx["cloud_provider"] = cloud_provider
+    return logging_context(**ctx)

cloud_pipelines_backend/orchestrator_sql.py

@@ -131,7 +131,7 @@ def internal_process_queued_executions_queue(self, session: orm.Session):
             self._queued_executions_queue_idle = False
             start_timestamp = time.monotonic_ns()
 
-            with contextual_logging.logging_context(execution_id=queued_execution.id):
+            with contextual_logging.execution_logging_context(queued_execution):
                 _logger.info("Before processing queued execution")
                 try:
                     self.internal_process_one_queued_execution(