Perf - Add otel supporting functionality to framework context objects (open-telemetry#552)

The PR adds support for tracing / logging / metrics in the various
framework context objects.

E.g. automatic creation / end of spans with status, ability to fetch
otel tracers, meters, and logging handler inside plugins, decorating
spans with ctx metadata, and generating span events.

---------

Co-authored-by: Laurent Quérel <l.querel@f5.com>

Author: clhain

tools/pipeline_perf_test/orchestrator/lib/core/context/base.py

@@ -5,17 +5,25 @@
 different implementations of Contexts throughout the orchestrator.
 """
 
+import logging
 import datetime
 import json
-import time
 import traceback
 from enum import Enum
 
+from contextlib import AbstractContextManager
 from dataclasses import dataclass, field
 from typing import Dict, List, Optional, TYPE_CHECKING
 
+from opentelemetry.trace import Span, Status, StatusCode
+
+from ..telemetry.telemetry_runtime import TelemetryRuntime
+from ..telemetry.test_event import TestEvent
+from ..telemetry.telemetry_client import TelemetryClient
 
 if TYPE_CHECKING:
+    from opentelemetry.sdk.metrics import Meter
+    from opentelemetry.sdk.trace import Tracer
     from ..test_framework.test_suite import TestSuite
     from ..component.component import Component
 
@@ -43,18 +51,126 @@ class BaseContext:
     name: Optional[str] = None
     status: Optional[ExecutionStatus] = ExecutionStatus.PENDING
     error: Optional[Exception] = None
-    start_time: Optional[float] = None
-    end_time: Optional[float] = None
+    start_time: Optional[datetime.datetime] = None
+    end_time: Optional[datetime.datetime] = None
     metadata: dict = field(default_factory=dict)
 
+    start_event_type: TestEvent = field(init=False, default=TestEvent.SUITE_START)
+    end_event_type: TestEvent = field(init=False, default=TestEvent.SUITE_END)
+
+    span: Optional[Span] = field(default=None, init=False)
+    span_cm: Optional[AbstractContextManager] = field(default=None, init=False)
+    span_name: Optional[str] = None
+
     parent_ctx: Optional["BaseContext"] = None
     child_contexts: List["BaseContext"] = field(default_factory=list)
 
+    def __post_init__(self):
+        """
+        Initializes default metadata after object creation.
+
+        Sets the 'test.ctx.type' to the class name and 'test.ctx.name' to the object's name
+        in the metadata dictionary, if they are not already defined.
+        """
+        self.metadata.setdefault("test.ctx.type", self.__class__.__name__)
+        self.metadata.setdefault("test.ctx.name", self.name)
+
+    def __enter__(self):
+        """
+        Enters the context for use in a 'with' statement.
+
+        Starts the process or operation associated with this object
+        and returns the object itself for use within the context block.
+        """
+        self.start()
+        return self
+
+    def __exit__(self, exc_type, exc_value, _traceback):
+        """
+        Exits the context, handling completion or exceptions.
+
+        Updates the context status to ERROR if an exception occurred,
+        or SUCCESS if the status was still RUNNING. Finalizes the context
+        by calling the end() method.
+
+        Args:
+            exc_type: The exception type, if any occurred.
+            exc_value: The exception instance.
+            _traceback: The traceback object.
+        """
+        if exc_type:
+            self.status = ExecutionStatus.ERROR
+            self.error = exc_value
+        elif self.status == ExecutionStatus.RUNNING:
+            self.status = ExecutionStatus.SUCCESS
+        self.end()
+
+    def start(self):
+        """
+        Marks the beginning of the context or operation.
+
+        - Sets the execution status to RUNNING.
+        - Records the current UTC time as the start time.
+        - Initializes a tracing span (if a tracer is available) using the object's class name
+        and optional `name` attribute for the span name.
+        - Attaches contextual attributes to the tracing span.
+        - Records a start event with a precise timestamp in nanoseconds.
+
+        This method is typically called at the beginning of a timed or monitored execution block,
+        such as within a context manager (`__enter__`).
+        """
+        self.status = ExecutionStatus.RUNNING
+        self.start_time = datetime.datetime.now(tz=datetime.timezone.utc)
+
+        tracer = self.get_tracer("test-framework")
+        if tracer:
+            span_name = getattr(
+                self,
+                "span_name",
+                f"{self.__class__.__name__} - {getattr(self, 'name', 'unnamed')}",
+            )
+            self.span_cm = tracer.start_as_current_span(span_name)
+            self.span = self.span_cm.__enter__()
+            attrs = self.merge_ctx_metadata()
+            for k, v in attrs.items():
+                self.span.set_attribute(k, v)
+
+        timestamp_unix_nanos = int(self.start_time.timestamp() * 1_000_000_000)
+        self._record_start_event(timestamp_unix_nanos)
+
+    def end(self):
+        """
+        Marks the end of the context or operation and logs its duration.
+
+        - Records the current UTC time as the end time.
+        - Logs an end event with a precise timestamp in nanoseconds.
+        - If a tracing span is active:
+            - Sets the span's status based on the current execution status (SUCCESS, ERROR, etc.).
+            - Closes the span context manager to finalize the trace.
+
+        This method is typically called at the end of a monitored or timed execution block,
+        such as within a context manager (`__exit__`).
+        """
+        self.end_time = datetime.datetime.now(tz=datetime.timezone.utc)
+        timestamp_unix_nanos = int(self.end_time.timestamp() * 1_000_000_000)
+        self._record_end_event(timestamp_unix_nanos)
+        if self.span_cm and self.span:
+            if self.status == ExecutionStatus.SUCCESS:
+                self.span.set_status(StatusCode.OK)
+            elif self.status == ExecutionStatus.ERROR:
+                self.span.set_status(
+                    Status(StatusCode.ERROR, str(self.error) or "Context failed")
+                )
+            else:
+                # Optional: Handle PENDING or other states explicitly
+                self.span.set_status(Status(StatusCode.UNSET))
+            self.span_cm.__exit__(None, None, None)
+
     @property
     def duration(self) -> Optional[float]:
         """Duration of the context between start/stop calls or none if not run/stopped."""
         if self.start_time is not None and self.end_time is not None:
-            return self.end_time - self.start_time
+            return (self.end_time - self.start_time).total_seconds()
         return None
 
     def add_child_ctx(self, ctx: "BaseContext"):
@@ -84,7 +200,7 @@ def get_component_by_name(self, name: str) -> Optional["Component"]:
         Returns: The named component or none if not found.
         """
         if hasattr(self, "parent_ctx"):
-            return self.parent_ctx.get_component(name)
+            return self.parent_ctx.get_component_by_name(name)
         raise NotImplementedError("This context does not support get_component_by_name")
 
     def get_test_suite(self) -> Optional["TestSuite"]:
@@ -93,40 +209,145 @@ def get_test_suite(self) -> Optional["TestSuite"]:
             return self.parent_ctx.get_test_suite()
         raise NotImplementedError("This context does not support get_test_suite")
 
-    def log(self, message: str):
+    def record_event(
+        self, event_name: str, timestamp_unix_nanos: Optional[int] = None, **kwargs
+    ):
         """
-        Logs a message both to the logger and stores it in the context log list.
-
-        Args:
-            message: the message to log
+        Record an event, enriching it with context-specific metadata.
         """
-        msg_dict = {"ctx_type": self.__class__.__name__, "message": message}
-        if self.name:
-            msg_dict["name"] = self.name
+        if not self.span or not self.span.is_recording():
+            return
         if self.status:
-            msg_dict["status"] = self.status
+            kwargs.setdefault("test.ctx.status", self.status.value)
+        if self.error:
+            kwargs.setdefault("test.ctx.error", str(self.error))
         if self.duration:
-            msg_dict["duration"] = f"{self.duration:.4f}"
+            kwargs.setdefault("test.ctx.duration", self.duration)
+        kwargs = self.merge_ctx_metadata(**kwargs)
+        self.span.add_event(event_name, kwargs, timestamp=timestamp_unix_nanos)
+
+    def merge_ctx_metadata(self, **kwargs: dict):
+        """Merge context metadata and status with the supplied arguments."""
+        if self.error:
+            kwargs.setdefault("test.ctx.error", str(self.error))
+        for key, value in self.metadata.items():
+            if value:
+                kwargs.setdefault(key, value)
+        return kwargs
+
+    def get_logger(self, logger_name: str = __name__) -> logging.LoggerAdapter:
+        """
+        Returns a context-aware logger with enriched metadata.
 
-        log_entry = f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {json.dumps(msg_dict)}"
-        print(log_entry)
+        - Retrieves a base logger using the specified logger name.
+        - Merges contextual metadata from the object.
+        - Adds non-empty metadata fields as 'extra' context to the logger.
+        - Returns a `LoggerAdapter` that injects this context into all log records.
 
-    def start(self):
-        """Mark start of context (e.g., for timing)."""
-        self.log("Context Starting...")
-        self.status = ExecutionStatus.RUNNING
-        self.start_time = time.time()
+        Args:
+            logger_name (str): The name of the logger to retrieve. Defaults to the current module's name.
 
-    def end(self):
-        """Mark end of context and log duration."""
-        self.end_time = time.time()
-        self.log("Context Ended")
+        Returns:
+            logging.LoggerAdapter: A logger adapter that includes contextual metadata for structured logging.
+        """
+        base_logger = logging.getLogger(logger_name)
+        extra = {}
+        attrs = self.merge_ctx_metadata()
+        for k, v in attrs.items():
+            if v:
+                extra[k] = v
+        return logging.LoggerAdapter(base_logger, extra)
+
+    def get_tracer(
+        self, name="default", runtime_name: str = TelemetryRuntime.type
+    ) -> Optional["Tracer"]:
+        """
+        Retrieves a tracer instance from the telemetry runtime.
+
+        - Accesses the current test suite and retrieves the specified telemetry runtime.
+        - Returns a tracer identified by the given name from that runtime.
+        - If the telemetry runtime is not found, returns None.
+
+        Args:
+            name (str): The name of the tracer to retrieve. Defaults to "default".
+            runtime_name (str): The name/type of the telemetry runtime to use. Defaults to the class-level `TelemetryRuntime.type`.
+
+        Returns:
+            Optional[Tracer]: A tracer instance for telemetry, or None if unavailable.
+        """
+        ts = self.get_test_suite()
+        telemetry_runtime: TelemetryRuntime = ts.get_runtime(runtime_name)
+        if not telemetry_runtime:
+            return
+        return telemetry_runtime.get_tracer(name)
+
+    def get_meter(
+        self, name="default", runtime_name: str = TelemetryRuntime.type
+    ) -> Optional["Meter"]:
+        """
+        Retrieves a meter instance from the telemetry runtime.
+
+        - Accesses the current test suite and obtains the specified telemetry runtime.
+        - Returns a meter identified by the given name from that runtime.
+        - If the telemetry runtime is not available, returns None.
+
+        Args:
+            name (str): The name of the meter to retrieve. Defaults to "default".
+            runtime_name (str): The name/type of the telemetry runtime to use. Defaults to the class-level `TelemetryRuntime.type`.
+
+        Returns:
+            Optional[Meter]: A meter instance for recording metrics, or None if unavailable.
+        """
+        ts = self.get_test_suite()
+        telemetry_runtime: TelemetryRuntime = ts.get_runtime(runtime_name)
+        if not telemetry_runtime:
+            return
+        return telemetry_runtime.get_meter(name)
+
+    def get_telemetry_client(
+        self, runtime_name: str = TelemetryRuntime.type
+    ) -> Optional[TelemetryClient]:
+        """
+        Retrieves the telemetry client from the specified telemetry runtime.
+
+        - Accesses the current test suite to retrieve the telemetry runtime by name.
+        - Returns the telemetry client associated with that runtime.
+        - If the runtime is not found, returns None.
+
+        Args:
+            runtime_name (str): The name/type of the telemetry runtime to use.
+                                Defaults to the class-level `TelemetryRuntime.type`.
+
+        Returns:
+            Optional[TelemetryClient]: The telemetry client instance, or None if unavailable.
+        """
+        ts = self.get_test_suite()
+        telemetry_runtime: TelemetryRuntime = ts.get_runtime(runtime_name)
+        if not telemetry_runtime:
+            return
+        return telemetry_runtime.get_client()
+
+    def get_metadata(self) -> dict:
+        """Return metadata specific to this context level (e.g. test_name, test_step)."""
+        return self.metadata
+
+    def _record_start_event(self, timestamp_unix_nanos: Optional[int]):
+        """Hook to record a context-specific start event."""
+        self.record_event(
+            self.start_event_type.namespaced(), timestamp=timestamp_unix_nanos
+        )
+
+    def _record_end_event(self, timestamp_unix_nanos: Optional[int]):
+        """Hook to record a context-specific end event."""
+        self.record_event(
+            self.end_event_type.namespaced(), timestamp=timestamp_unix_nanos
+        )
 
     @staticmethod
-    def _format_time(timestamp: Optional[float]) -> str:
+    def _format_time(timestamp: Optional[datetime.datetime]) -> str:
         if timestamp is None:
             return "None"
-        return datetime.datetime.fromtimestamp(timestamp).isoformat()
+        return timestamp.isoformat()
 
     def summary_string(self, indent: int = 2) -> str:
         """
@@ -168,8 +389,8 @@ def to_dict(self) -> dict:
             "name": self.name,
             "status": self.status.value if self.status else None,
             "error": str(self.error) if self.error else None,
-            "start_time": self.start_time,
-            "end_time": self.end_time,
+            "start_time": self._format_time(self.start_time),
+            "end_time": self._format_time(self.end_time),
             "duration": self.duration,
             "metadata": self.metadata,
             "child_contexts": [child.to_dict() for child in self.child_contexts],

tools/pipeline_perf_test/orchestrator/lib/core/context/component_hook_context.py

@@ -31,10 +31,11 @@
 from enum import Enum
 from typing import Optional, TYPE_CHECKING
 
+from ..telemetry.test_event import TestEvent
 from .base import BaseContext
-from .test_contexts import TestStepContext
 
 if TYPE_CHECKING:
+    from .test_contexts import TestStepContext
     from ..component.component import Component
 
 
@@ -81,13 +82,27 @@ class ComponentHookContext(BaseContext):
     parent_ctx: Optional["TestStepContext"] = None
     phase: Optional["HookableComponentPhase"] = None
 
+    def __post_init__(self):
+        """
+        Performs additional initialization after object creation.
+        """
+        super().__post_init__()
+        self.start_event_type = TestEvent.HOOK_START
+        self.end_event_type = TestEvent.HOOK_END
+        if self.parent_ctx:
+            merged_metadata = {**self.parent_ctx.metadata, **self.metadata}
+            merged_metadata["test.ctx.component"] = self.parent_ctx.step.component.name
+            self.metadata = merged_metadata
+        self.metadata["test.ctx.phase"] = self.phase.value
+        self.span_name = f"Run Component Hook: {self.name}"
+
     def get_step_component(self) -> Optional["Component"]:
         """Fetches the component instance on which this hook is firing.
 
         Returns: the component instance or none.
         """
         if self.parent_ctx is None:
             raise RuntimeError(
-                "LifecycleHookContext.parent_ctx must be set to access the step component."
+                "ComponentHookContext.parent_ctx must be set to access the step component."
             )
         return self.parent_ctx.step.component