Start opentelemetry spans on host start instead of task start

v2_playbook_on_task_start does not have the host information, so spans
would always start at the same time for every host in that task, even if
they started at different times, like when hosts > forks with strategy
host_pinned. This also hides the duration of the task for that host.

This change uses the newer v2_runner_on_start callback and adds the acutal
host start time to the span. The change is backwards compatible with ansible
versions that do not have v2_runner_on_start and makes no assumptions around
the ordering of v2_runner_on_start and v2_playbook_on_task_start.

plugins/callback/opentelemetry.py

@@ -185,6 +185,7 @@ class TaskData:
                 # concatenate task include output from multiple items
                 host.result = f"{self.host_data[host.uuid].result}\n{host.result}"
             else:
+                self.host_data[host.uuid].update(host)
                 return
 
         self.host_data[host.uuid] = host
@@ -195,12 +196,20 @@ class HostData:
     Data about an individual host.
     """
 
-    def __init__(self, uuid, name, status, result):
+    def __init__(self, uuid, name, status, result, start=None):
         self.uuid = uuid
         self.name = name
         self.status = status
         self.result = result
         self.finish = time_ns()
+        self.start = start
+
+    def update(self, host):
+        self.status = host.status
+        self.result = host.result
+        self.finish = host.finish
+        if host.start is not None:
+            self.start = host.start
 
 
 class OpenTelemetrySource:
@@ -221,13 +230,17 @@ class OpenTelemetrySource:
         carrier["traceparent"] = traceparent
         return TraceContextTextMapPropagator().extract(carrier=carrier)
 
-    def start_task(self, tasks_data, hide_task_arguments, play_name, task):
+    def start_task(self, tasks_data, hide_task_arguments, play_name, task, host=None):
         """record the start of a task for one or more hosts"""
 
         uuid = task._uuid
 
         if uuid in tasks_data:
-            return
+            if host:
+                tasks_data[uuid].add_host(HostData(host._uuid, host.name, "started", None, time_ns()))
+                return
+            else:
+                return
 
         name = task.get_name().strip()
         path = task.get_path()
@@ -238,6 +251,8 @@ class OpenTelemetrySource:
             args = task.args
 
         tasks_data[uuid] = TaskData(uuid, name, path, play_name, action, args)
+        if host:
+            tasks_data[uuid].add_host(HostData(host._uuid, host.name, "started", None, time_ns()))
 
     def finish_task(self, tasks_data, status, result, dump):
         """record the results of a task for a single host"""
@@ -310,7 +325,8 @@ class OpenTelemetrySource:
             parent.set_attribute("ansible.host.user", self.user)
             for task in tasks:
                 for host_data in task.host_data.values():
-                    with tracer.start_as_current_span(task.name, start_time=task.start, end_on_exit=False) as span:
+                    start = host_data.start or task.start
+                    with tracer.start_as_current_span(task.name, start_time=start, end_on_exit=False) as span:
                         self.update_span_data(task, host_data, span, disable_logs, disable_attributes_in_logs)
 
         return otel_exporter
@@ -565,6 +581,9 @@ class CallbackModule(CallbackBase):
     def v2_playbook_on_task_start(self, task, is_conditional):
         self.opentelemetry.start_task(self.tasks_data, self.hide_task_arguments, self.play_name, task)
 
+    def v2_runner_on_start(self, host, task):
+        self.opentelemetry.start_task(self.tasks_data, self.hide_task_arguments, self.play_name, task, host)
+
     def v2_playbook_on_cleanup_task_start(self, task):
         self.opentelemetry.start_task(self.tasks_data, self.hide_task_arguments, self.play_name, task)
 

tests/unit/plugins/callback/test_opentelemetry.py

@@ -50,6 +50,27 @@ class TestOpentelemetry(unittest.TestCase):
         self.assertEqual(task_data.action, "myaction")
         self.assertEqual(task_data.args, {})
 
+    def test_run_task_with_host(self):
+        tasks_data = OrderedDict()
+
+        self.opentelemetry.start_task(tasks_data, False, "myplay", self.mock_task, self.mock_host)
+
+        task_data = tasks_data["myuuid"]
+        self.assertEqual(task_data.uuid, "myuuid")
+        self.assertEqual(task_data.name, "mytask")
+        self.assertEqual(task_data.path, "/mypath")
+        self.assertEqual(task_data.play, "myplay")
+        self.assertEqual(task_data.action, "myaction")
+        self.assertEqual(task_data.args, {})
+
+        host_data = task_data.host_data["myhost_uuid"]
+        self.assertEqual(host_data.uuid, "myhost_uuid")
+        self.assertEqual(host_data.name, "myhost")
+        self.assertIsNotNone(host_data.start)
+
+        self.opentelemetry.finish_task(tasks_data, "ok", self.my_task_result, "")
+        self.assertEqual(host_data.status, "ok")
+
     def test_finish_task_with_a_host_match(self):
         tasks_data = OrderedDict()
         tasks_data["myuuid"] = self.my_task