Link span of queue item to trigger event span

This change links the span of the queue item to the span of the
forwarded trigger event.

This adds a currently empty span context field to trigger events which
will be later populated by the drivers.

We also need to add add a span context field to
`ChangeManagementEvent`s, as some drivers add dequeue events to the
pipeline event queues directly.

Change-Id: Icd240712b86cc22e55fb67f6787a0974d5308043

3 files changed, 31 insertions, 2 deletions

diff --git a/zuul/manager/__init__.py b/zuul/manager/__init__.py
index ab83afc3d..d98b871be 100644
--- a/zuul/manager/__init__.py
+++ b/zuul/manager/__init__.py
@@ -584,8 +584,14 @@ class PipelineManager(metaclass=ABCMeta):
                      (change, change_queue, self.pipeline))
             if enqueue_time is None:
                 enqueue_time = time.time()
+
+            event_span = tracing.restoreSpanContext(event.span_context)
+            link_attributes = {"rel": type(event).__name__}
+            link = trace.Link(event_span.get_span_context(),
+                              attributes=link_attributes)
             span_info = tracing.startSavedSpan(
-                'QueueItem', start_time=enqueue_time)
+                'QueueItem', start_time=enqueue_time, links=[link])
+
             item = change_queue.enqueueChange(change, event,
                                               span_info=span_info,
                                               enqueue_time=enqueue_time)
diff --git a/zuul/model.py b/zuul/model.py
index 9b4496ece..23a4502a5 100644
--- a/zuul/model.py
+++ b/zuul/model.py
@@ -6239,6 +6239,8 @@ class ChangeManagementEvent(ManagementEvent):
         self.oldrev = oldrev or '0000000000000000000000000000000000000000'
         self.newrev = newrev or '0000000000000000000000000000000000000000'
         self.timestamp = time.time()
+        span = trace.get_current_span()
+        self.span_context = tracing.getSpanContext(span)
 
     def toDict(self):
         d = super().toDict()
@@ -6252,12 +6254,14 @@ class ChangeManagementEvent(ManagementEvent):
         d["oldrev"] = self.oldrev
         d["newrev"] = self.newrev
         d["timestamp"] = self.timestamp
+        d["span_context"] = self.span_context
         return d
 
     def updateFromDict(self, d):
         super().updateFromDict(d)
         self.type = d.get("type")
         self.timestamp = d.get("timestamp")
+        self.span_context = d.get("span_context")
 
     @classmethod
     def fromDict(cls, data):
@@ -6544,6 +6548,8 @@ class TriggerEvent(AbstractEvent):
         self.arrived_at_scheduler_timestamp = None
         self.driver_name = None
         self.branch_cache_ltime = -1
+        span = trace.get_current_span()
+        self.span_context = tracing.getSpanContext(span)
 
     def toDict(self):
         return {
@@ -6578,6 +6584,7 @@ class TriggerEvent(AbstractEvent):
             ),
             "driver_name": self.driver_name,
             "branch_cache_ltime": self.branch_cache_ltime,
+            "span_context": self.span_context,
         }
 
     def updateFromDict(self, d):
@@ -6612,6 +6619,7 @@ class TriggerEvent(AbstractEvent):
         )
         self.driver_name = d["driver_name"]
         self.branch_cache_ltime = d.get("branch_cache_ltime", -1)
+        self.span_context = d.get("span_context")
 
     @property
     def canonical_project_name(self):
diff --git a/zuul/scheduler.py b/zuul/scheduler.py
index 9548cb6d2..d37ef9539 100644
--- a/zuul/scheduler.py
+++ b/zuul/scheduler.py
@@ -31,6 +31,7 @@ from collections import defaultdict, OrderedDict
 from apscheduler.schedulers.background import BackgroundScheduler
 from apscheduler.triggers.interval import IntervalTrigger
 from kazoo.exceptions import NotEmptyError
+from opentelemetry import trace
 
 from zuul import configloader, exceptions
 from zuul.lib import commandsocket
@@ -176,6 +177,8 @@ class Scheduler(threading.Thread):
     """
 
     log = logging.getLogger("zuul.Scheduler")
+    tracer = trace.get_tracer("zuul")
+
     _stats_interval = 30
     _semaphore_cleanup_interval = IntervalTrigger(minutes=60, jitter=60)
     _general_cleanup_interval = IntervalTrigger(minutes=60, jitter=60)
@@ -2217,7 +2220,14 @@ class Scheduler(threading.Thread):
                     log = get_annotated_logger(self.log, event.zuul_event_id)
                     log.debug("Forwarding trigger event %s", event)
                     try:
-                        self._forward_trigger_event(event, tenant)
+                        trigger_span = tracing.restoreSpanContext(
+                            event.span_context)
+                        with self.tracer.start_as_current_span(
+                                "TenantTriggerEventProcessing",
+                                links=[
+                                    trace.Link(trigger_span.get_span_context())
+                                ]):
+                            self._forward_trigger_event(event, tenant)
                     except Exception:
                         log.exception("Unable to forward event %s "
                                       "to tenant %s", event, tenant.name)
@@ -2310,6 +2320,11 @@ class Scheduler(threading.Thread):
 
         event.min_reconfigure_ltime = self.trigger_events[
             tenant.name].last_reconfigure_event_ltime
+
+        span = trace.get_current_span()
+        span.set_attribute("reconfigure_tenant", reconfigure_tenant)
+        event.span_context = tracing.getSpanContext(span)
+
         for pipeline in tenant.layout.pipelines.values():
             if (
                 pipeline.manager.eventMatches(event, change)