Configure OpenTelemetry on existing tests

kpunwatk · kpunwatk · commit 857b3dbd936c · 2025-09-01T11:34:32.000+01:00
modified:   tests/model_explainability/guardrails/conftest.py
	modified:   tests/model_explainability/guardrails/test_guardrails.py
diff --git a/tests/model_explainability/guardrails/conftest.py b/tests/model_explainability/guardrails/conftest.py
@@ -6,25 +6,31 @@
 import pytest
 from _pytest.fixtures import FixtureRequest
 from kubernetes.dynamic import DynamicClient
+from kubernetes.dynamic.exceptions import ResourceNotFoundError
+from ocp_resources.cluster_service_version import ClusterServiceVersion
 from ocp_resources.config_map import ConfigMap
 from ocp_resources.deployment import Deployment
 from ocp_resources.guardrails_orchestrator import GuardrailsOrchestrator
 from ocp_resources.inference_service import InferenceService
 from ocp_resources.namespace import Namespace
 from ocp_resources.pod import Pod
-from ocp_resources.resource import ResourceEditor
+from ocp_resources.resource import ResourceEditor, NamespacedResource
 from ocp_resources.route import Route
 from ocp_resources.secret import Secret
 from ocp_resources.serving_runtime import ServingRuntime
+from ocp_resources.subscription import Subscription
+from ocp_utilities.operators import install_operator, uninstall_operator
 from pytest_testconfig import py_config
+from timeout_sampler import TimeoutSampler
 
 from utilities.certificates_utils import create_ca_bundle_file
 from utilities.constants import (
     KServeDeploymentType,
     Labels,
+    Timeout,
 )
 from utilities.inference_utils import create_isvc
-
+from utilities.operator_utils import get_cluster_service_version
 
 GUARDRAILS_ORCHESTRATOR_NAME = "guardrails-orchestrator"
 
@@ -351,3 +357,197 @@ def hap_detector_route(
         service=hap_detector_isvc.name,
         wait_for_resource=True,
     )
+
+
+class OpenTelemetryCollector(NamespacedResource):
+    """
+    OpenTelemetryCollector is the Schema for the OpenTelemetry Collectors
+    """
+    api_group: str = "opentelemetry.io"
+
+    def __init__(self, **kwargs: Any) -> None:
+        super().__init__(**kwargs)
+
+
+class OpenTelemetryOperator(NamespacedResource):
+    """
+    OpenTelemetryOperator is the Schema for the opentelemetryoperators API
+    """
+
+    api_group: str = "opentelemetry.io"
+
+    def __init__(self, **kwargs: Any) -> None:
+        super().__init__(**kwargs)
+
+@pytest.fixture(scope="class")
+def installed_opentelemetry_operator(admin_client: DynamicClient, model_namespace: Namespace):
+    """
+    Install the Red Hat OpenTelemetry operator in the same namespace as the test
+    so CRs like OpenTelemetryCollector/Instrumentation are watched properly.
+    """
+    operator_ns = model_namespace
+    operator_name = "opentelemetry-operator"
+
+    otel_subscription = Subscription(
+        client=admin_client,
+        namespace=operator_ns.name,
+        name=operator_name,
+    )
+
+    if not otel_subscription.exists:
+        install_operator(
+            admin_client=admin_client,
+            target_namespaces=[operator_ns.name],
+            name=operator_name,
+            channel="stable",
+            source="redhat-operators",
+            operator_namespace=operator_ns.name,
+            timeout=Timeout.TIMEOUT_15MIN,
+            install_plan_approval="Automatic",
+            starting_csv="opentelemetry-operator.v0.127.0-2",
+        )
+
+        deployment = Deployment(
+            client=admin_client,
+            namespace=operator_ns.name,
+            name="opentelemetry-operator-controller-manager",
+            wait_for_resource=True,
+        )
+        deployment.wait_for_replicas()
+
+    yield
+
+    uninstall_operator(
+        admin_client=admin_client,
+        name=operator_name,
+        operator_namespace=operator_ns.name,
+        clean_up_namespace=True,
+    )
+
+@pytest.fixture(scope="class")
+def otel_operator_cr(
+    admin_client: DynamicClient,
+    installed_opentelemetry_operator: None,
+    model_namespace: Namespace,  # use the test namespace
+) -> Generator[OpenTelemetryCollector, Any, Any]:
+    """Create an OpenTelemetryCollector CR in the test namespace based on ALM examples from the CSV."""
+    otel_csv: ClusterServiceVersion = get_cluster_service_version(
+        client=admin_client,
+        prefix="opentelemetry",
+        namespace=model_namespace.name,  # fetch CSV from the same namespace as operator
+    )
+
+    alm_examples: list[dict[str, Any]] = otel_csv.get_alm_examples()
+    otel_cr_dict: dict[str, Any] = next(
+        example for example in alm_examples if example["kind"] == "OpenTelemetryCollector"
+    )
+
+    if not otel_cr_dict:
+        raise ResourceNotFoundError(f"No OpenTelemetryCollector dict found in alm_examples for CSV {otel_csv.name}")
+
+    otel_cr_dict["metadata"]["namespace"] = model_namespace.name  # create CR in test namespace
+
+    with OpenTelemetryCollector(kind_dict=otel_cr_dict) as otel_cr:
+        otel_cr.wait_for_condition(
+            condition="Available",
+            status=OpenTelemetryCollector.Condition.Status.TRUE,
+            timeout=Timeout.TIMEOUT_10MIN,
+        )
+        yield otel_cr
+
+
+class Jaeger(NamespacedResource):
+    """
+    Jaeger instance CR for Red Hat OpenShift distributed tracing platform.
+    """
+    api_group: str = "io.jaegertracing.openshift.v1"  # usually something like this
+
+    def __init__(self, **kwargs: Any) -> None:
+        super().__init__(**kwargs)
+
+@pytest.fixture(scope="session")
+def installed_jaeger_operator(admin_client: DynamicClient) -> Generator[None, Any, None]:
+    """Install Red Hat OpenShift distributed tracing platform (Jaeger operator)."""
+    operator_ns = Namespace(name="openshift-distributed-tracing", ensure_exists=True)
+    operator_name = "jaeger-product"
+
+    jaeger_subscription = Subscription(
+        client=admin_client,
+        namespace=operator_ns.name,
+        name=operator_name,
+        source="redhat-operators",
+        channel="stable",
+        starting_csv="jaeger-product.v1.65.0-4",
+        installPlanApproval="Automatic",
+    )
+
+    if not jaeger_subscription.exists:
+        jaeger_subscription.create()
+
+    yield
+
+@pytest.fixture(scope="class")
+def jaeger_instance(
+    admin_client: DynamicClient, model_namespace: Namespace
+) -> Generator[Jaeger, Any, None]:
+    """Create a Jaeger instance in the test namespace using default all-in-one strategy."""
+    jaeger_csv: ClusterServiceVersion = get_cluster_service_version(
+        client=admin_client,
+        prefix="jaeger",
+        namespace="openshift-distributed-tracing"
+    )
+    alm_examples: list[dict[str, Any]] = jaeger_csv.get_alm_examples()
+    jaeger_dict: dict[str, Any] = next(example for example in alm_examples if example["kind"] == "Jaeger")
+
+    if not jaeger_dict:
+        raise ResourceNotFoundError(f"No Jaeger dict found in alm_examples for CSV {jaeger_csv.name}")
+
+    jaeger_dict["metadata"]["namespace"] = model_namespace.name
+    jaeger_dict["metadata"]["name"] = "simplest"
+    jaeger_dict["spec"]["strategy"] = "allInOne"
+
+    with Jaeger(kind_dict=jaeger_dict) as jaeger:
+        wait_for_jaeger_pods(
+            client=admin_client,
+            jaeger_name=jaeger.name,
+            namespace=model_namespace.name,
+        )
+        yield jaeger
+
+def wait_for_jaeger_operator_deployments(namespace: str) -> None:
+    """
+    Wait for the Jaeger operator deployment to be ready.
+    """
+    operator_deployment_name = "jaeger-operator"
+
+    deployment = Deployment(name=operator_deployment_name, namespace=namespace)
+    deployment.wait_for_replicas()
+
+
+def wait_for_jaeger_pods(client: DynamicClient, jaeger_name: str, namespace: str, timeout: int = Timeout.TIMEOUT_15MIN) -> None:
+    """
+    Wait for pods created by a Jaeger instance to be ready.
+    """
+    def _get_jaeger_pods() -> list[Pod]:
+        return [
+            _pod
+            for _pod in Pod.get(
+                dyn_client=client,
+                namespace=namespace,
+                label_selector=f"app.kubernetes.io/instance={jaeger_name}",
+            )
+        ]
+
+    sampler = TimeoutSampler(wait_timeout=timeout, sleep=1, func=lambda: bool(_get_jaeger_pods()))
+
+    for sample in sampler:
+        if sample:
+            break
+
+    pods = _get_jaeger_pods()
+    for pod in pods:
+        pod.wait_for_condition(
+            condition=Pod.Condition.READY,
+            status="True",
+        )
+
diff --git a/tests/model_explainability/guardrails/test_guardrails.py b/tests/model_explainability/guardrails/test_guardrails.py
@@ -4,10 +4,12 @@
 import pytest
 import requests
 import yaml
+from ocp_resources.pod import Pod
 from simple_logger.logger import get_logger
 from timeout_sampler import retry
 
 from tests.model_explainability.constants import MNT_MODELS
+from tests.model_explainability.guardrails.conftest import wait_for_jaeger_pods
 from tests.model_explainability.guardrails.constants import (
     QWEN_ISVC_NAME,
     CHAT_GENERATION_CONFIG,
@@ -481,3 +483,58 @@ def test_guardrails_several_detector_negative_detection(
         )
 
         verify_negative_detection_response(response=response)
+
+@pytest.mark.parametrize(
+    "model_namespace, orchestrator_config, guardrails_orchestrator",
+    [
+        pytest.param(
+            {"name": "test-guardrails-opentelemetry"},
+            {
+                "orchestrator_config_data": {
+                    "config.yaml": yaml.dump({
+                        "chat_generation": CHAT_GENERATION_CONFIG,
+                        "detectors": BUILTIN_DETECTOR_CONFIG,
+                    })
+                },
+            },
+            {"enable_built_in_detectors": False, "enable_guardrails_gateway": False},
+        )
+    ],
+    indirect=True,
+)
+@pytest.mark.rawdeployment
+class TestGuardrailsOrchestratorWithOpenTelemetry:
+    """
+    Tests that Guardrails Orchestrator can be instrumented with OpenTelemetry.
+    """
+
+    def test_guardrails_with_opentelemetry(
+            self,
+            guardrails_orchestrator,
+            otel_operator_cr,
+            jaeger_instance,
+    ):
+        orchestrator = guardrails_orchestrator
+        assert orchestrator.exists
+
+        # Wait for all orchestrator pods to be ready
+        pods = Pod.get(
+            dyn_client=orchestrator.client,
+            namespace=orchestrator.namespace,
+            label_selector=f"app={orchestrator.name}",
+        )
+
+        for pod in pods:
+            pod.wait_for_condition(
+                condition=Pod.Condition.READY,
+                status="True",
+                timeout=Timeout.TIMEOUT_10MIN
+            )
+
+        # Wait for Jaeger instance pods to be ready
+        wait_for_jaeger_pods(
+            client=jaeger_instance.client,
+            jaeger_name=jaeger_instance.name,
+            namespace=jaeger_instance.namespace,
+            timeout=Timeout.TIMEOUT_10MIN
+        )
