From 248eb313c038397294dd8cc4a8c213b820377bad Mon Sep 17 00:00:00 2001 From: Pablo Collins Date: Wed, 22 Jul 2026 14:28:25 -0400 Subject: [PATCH 01/10] Add provisional OpAMP impl --- CHANGELOG.md | 2 + pyproject.toml | 4 + src/splunk_otel/env.py | 3 + src/splunk_otel/opamp.py | 269 ++++++++++++++++++ tests/integration/opamp_effective_config.py | 139 +++++++++ tests/test_opamp.py | 300 ++++++++++++++++++++ 6 files changed, 717 insertions(+) create mode 100644 src/splunk_otel/opamp.py create mode 100644 tests/integration/opamp_effective_config.py create mode 100644 tests/test_opamp.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 452ab517..7bc1bd02 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- Add provisional OpAMP effective-config reporting through the OpenTelemetry SDK post-initialization hook + ## 2.12.0 - 2026-07-13 - Upgrade Otel dependencies to [1.44.0](https://github.com/open-telemetry/opentelemetry-python/releases/tag/v1.44.0) / [0.65b0](https://github.com/open-telemetry/opentelemetry-python-contrib/releases/tag/v0.65b0) diff --git a/pyproject.toml b/pyproject.toml index ce05edb8..86088351 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,6 +32,7 @@ dependencies = [ "opentelemetry-instrumentation-logging==0.65b0", "opentelemetry-instrumentation-system-metrics==0.65b0", "opentelemetry-semantic-conventions==0.65b0", + "opentelemetry-opamp-client==0.3b0", "protobuf>=6.33.5", # not our direct dep, prevents installing vulnerable proto versions (CVE‑2025‑4565, CVE-2026-0994) "wrapt>=2.2.2", ] @@ -52,6 +53,9 @@ configurator = "splunk_otel.configurator:SplunkConfigurator" [project.entry-points.opentelemetry_distro] splunk_distro = "splunk_otel.distro:SplunkDistro" +[project.entry-points._opentelemetry_opamp] +post_sdk_init_function = "splunk_otel.opamp:start_opamp" + [tool.hatch.version] path = "src/splunk_otel/__about__.py" diff --git a/src/splunk_otel/env.py b/src/splunk_otel/env.py index 774bdeea..f44940fe 100644 --- a/src/splunk_otel/env.py +++ b/src/splunk_otel/env.py @@ -45,6 +45,9 @@ SPLUNK_OTEL_SYSTEM_METRICS_ENABLED = "SPLUNK_OTEL_SYSTEM_METRICS_ENABLED" SPLUNK_ACCESS_TOKEN = "SPLUNK_ACCESS_TOKEN" # noqa: S105 SPLUNK_TRACE_RESPONSE_HEADER_ENABLED = "SPLUNK_TRACE_RESPONSE_HEADER_ENABLED" +SPLUNK_OPAMP_ENABLED = "SPLUNK_OPAMP_ENABLED" +SPLUNK_OPAMP_ENDPOINT = "SPLUNK_OPAMP_ENDPOINT" +SPLUNK_OPAMP_POLLING_INTERVAL = "SPLUNK_OPAMP_POLLING_INTERVAL" SPLUNK_PROFILER_ENABLED = "SPLUNK_PROFILER_ENABLED" SPLUNK_PROFILER_CALL_STACK_INTERVAL = "SPLUNK_PROFILER_CALL_STACK_INTERVAL" SPLUNK_PROFILER_LOGS_ENDPOINT = "SPLUNK_PROFILER_LOGS_ENDPOINT" diff --git a/src/splunk_otel/opamp.py b/src/splunk_otel/opamp.py new file mode 100644 index 00000000..b137a269 --- /dev/null +++ b/src/splunk_otel/opamp.py @@ -0,0 +1,269 @@ +# Copyright Splunk Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from opentelemetry._opamp.agent import OpAMPAgent +from opentelemetry._opamp.callbacks import MessageData, OpAMPCallbacks +from opentelemetry._opamp.client import OpAMPClient +from opentelemetry.environment_variables import ( + OTEL_LOGS_EXPORTER, + OTEL_METRICS_EXPORTER, + OTEL_TRACES_EXPORTER, +) +from opentelemetry.sdk.environment_variables import ( + OTEL_EXPORTER_OTLP_ENDPOINT, + OTEL_EXPORTER_OTLP_LOGS_ENDPOINT, + OTEL_EXPORTER_OTLP_LOGS_PROTOCOL, + OTEL_EXPORTER_OTLP_METRICS_ENDPOINT, + OTEL_EXPORTER_OTLP_METRICS_PROTOCOL, + OTEL_EXPORTER_OTLP_PROTOCOL, + OTEL_EXPORTER_OTLP_TRACES_ENDPOINT, + OTEL_EXPORTER_OTLP_TRACES_PROTOCOL, +) + +from splunk_otel.env import ( + Env, + SPLUNK_OPAMP_ENABLED, + SPLUNK_OPAMP_ENDPOINT, + SPLUNK_OPAMP_POLLING_INTERVAL, + SPLUNK_PROFILER_CALL_STACK_INTERVAL, + SPLUNK_PROFILER_ENABLED, + SPLUNK_SNAPSHOT_PROFILER_ENABLED, + SPLUNK_SNAPSHOT_SAMPLING_INTERVAL, +) + +if TYPE_CHECKING: + from collections.abc import Mapping + + from opentelemetry.sdk.resources import Resource + from opentelemetry.util.types import AnyValue + +logger = logging.getLogger(__name__) + +_CONFIG_FILENAME = "environment" +_CONFIG_CONTENT_TYPE = "text/plain; format=properties; vendor=splunk; v=1.0.0" +_DEFAULT_OPAMP_ENDPOINT = "http://localhost:4320/v1/opamp" +_DEFAULT_OPAMP_POLLING_INTERVAL_MS = 30000 +_DEFAULT_PROFILER_CALL_STACK_INTERVAL = 1000 +_DEFAULT_SNAPSHOT_SAMPLING_INTERVAL = 10 +_DEFAULT_GRPC_ENDPOINT = "http://localhost:4317" +_DEFAULT_HTTP_ENDPOINT = "http://localhost:4318/" +_OTLP_PROTOCOL_HTTP_PROTOBUF = "http/protobuf" +_OTLP_EXPORTER = "otlp" +_OTLP_PROTO_HTTP_EXPORTER = "otlp_proto_http" + +_SPLUNK_PROFILER_MEMORY_ENABLED = "SPLUNK_PROFILER_MEMORY_ENABLED" +_SPLUNK_SNAPSHOT_PROFILER_SAMPLING_INTERVAL = ( + "SPLUNK_SNAPSHOT_PROFILER_SAMPLING_INTERVAL" +) +_OTEL_CONFIG_FILE = "OTEL_CONFIG_FILE" +_OTEL_EXPERIMENTAL_CONFIG_FILE = "OTEL_EXPERIMENTAL_CONFIG_FILE" + +_SIGNAL_CONFIG = { + "traces": { + "endpoint": OTEL_EXPORTER_OTLP_TRACES_ENDPOINT, + "protocol": OTEL_EXPORTER_OTLP_TRACES_PROTOCOL, + "exporter": OTEL_TRACES_EXPORTER, + }, + "metrics": { + "endpoint": OTEL_EXPORTER_OTLP_METRICS_ENDPOINT, + "protocol": OTEL_EXPORTER_OTLP_METRICS_PROTOCOL, + "exporter": OTEL_METRICS_EXPORTER, + }, + "logs": { + "endpoint": OTEL_EXPORTER_OTLP_LOGS_ENDPOINT, + "protocol": OTEL_EXPORTER_OTLP_LOGS_PROTOCOL, + "exporter": OTEL_LOGS_EXPORTER, + }, +} + + +@dataclass(frozen=True) +class OpAMPConfig: + endpoint: str + polling_interval_ms: int + + @classmethod + def from_env(cls, env: Env) -> OpAMPConfig | None: + if not env.is_true(SPLUNK_OPAMP_ENABLED): + logger.debug("OpAMP disabled (%s is not true)", SPLUNK_OPAMP_ENABLED) + return None + + return cls( + endpoint=env.getval(SPLUNK_OPAMP_ENDPOINT, _DEFAULT_OPAMP_ENDPOINT), + polling_interval_ms=env.getint( + SPLUNK_OPAMP_POLLING_INTERVAL, + _DEFAULT_OPAMP_POLLING_INTERVAL_MS, + ), + ) + + +class _SplunkCallbacks(OpAMPCallbacks): + def on_connect_failed( + self, + _agent: OpAMPAgent, + _client: OpAMPClient, + error: Exception, + ) -> None: + logger.warning("Connection to OpAMP server failed", exc_info=error) + + def on_error( + self, + _agent: OpAMPAgent, + _client: OpAMPClient, + error_response, + ) -> None: + logger.warning("OpAMP server returned error: %s", error_response) + + def on_message( + self, + _agent: OpAMPAgent, + _client: OpAMPClient, + message: MessageData, + ) -> None: + logger.debug( + "ServerToAgent message received: remote_config=%s", + message.remote_config is not None, + ) + + +def start_opamp(resource: Resource) -> None: + """Start the Splunk OpAMP agent after the OpenTelemetry SDK starts.""" + env = Env() + config = OpAMPConfig.from_env(env) + if config is None: + return + + try: + client = _build_client(config, resource.attributes) + _start_agent(config, build_effective_config_report(env), client) + except Exception: + logger.exception("Failed to start OpAMP client") + + +def build_effective_config_report(env: Env) -> str: + values = ( + (OTEL_EXPORTER_OTLP_TRACES_ENDPOINT, _get_signal_endpoint(env, "traces")), + (OTEL_EXPORTER_OTLP_METRICS_ENDPOINT, _get_signal_endpoint(env, "metrics")), + (OTEL_EXPORTER_OTLP_LOGS_ENDPOINT, _get_signal_endpoint(env, "logs")), + ( + SPLUNK_PROFILER_ENABLED, + _bool_to_str(value=env.is_true(SPLUNK_PROFILER_ENABLED)), + ), + (_SPLUNK_PROFILER_MEMORY_ENABLED, "false"), + ( + SPLUNK_SNAPSHOT_PROFILER_ENABLED, + _bool_to_str(value=env.is_true(SPLUNK_SNAPSHOT_PROFILER_ENABLED)), + ), + ( + _SPLUNK_SNAPSHOT_PROFILER_SAMPLING_INTERVAL, + str( + env.getint( + SPLUNK_SNAPSHOT_SAMPLING_INTERVAL, + _DEFAULT_SNAPSHOT_SAMPLING_INTERVAL, + ) + ), + ), + ( + SPLUNK_PROFILER_CALL_STACK_INTERVAL, + str( + env.getint( + SPLUNK_PROFILER_CALL_STACK_INTERVAL, + _DEFAULT_PROFILER_CALL_STACK_INTERVAL, + ) + ), + ), + (_OTEL_CONFIG_FILE, "null"), + (_OTEL_EXPERIMENTAL_CONFIG_FILE, "null"), + ) + return "\n".join(f"{key}={value}" for key, value in values) + + +def _build_client( + config: OpAMPConfig, + resource_attributes: Mapping[str, AnyValue], + client_factory=OpAMPClient, +): + identifying_attributes = { + str(key): str(value) for key, value in resource_attributes.items() + } + return client_factory( + endpoint=config.endpoint, + headers={}, + agent_identifying_attributes=identifying_attributes, + agent_non_identifying_attributes={}, + ) + + +def _start_agent( + config: OpAMPConfig, + effective_config_report: str, + client, + agent_factory=OpAMPAgent, +): + client.update_effective_config( + {_CONFIG_FILENAME: effective_config_report}, + content_type=_CONFIG_CONTENT_TYPE, + ) + agent = agent_factory( + interval=config.polling_interval_ms / 1000, + callbacks=_SplunkCallbacks(), + client=client, + ) + agent.start() + logger.info("OpAMP client started: %s", config.endpoint) + return agent + + +def _get_signal_endpoint(env: Env, signal: str) -> str: + signal_config = _SIGNAL_CONFIG[signal] + endpoint = env.getval(signal_config["endpoint"]) + if endpoint: + return endpoint + + base_endpoint = env.getval(OTEL_EXPORTER_OTLP_ENDPOINT) + if _uses_http_protobuf(env, signal): + return _append_signal_path(base_endpoint or _DEFAULT_HTTP_ENDPOINT, signal) + + return base_endpoint or _DEFAULT_GRPC_ENDPOINT + + +def _uses_http_protobuf(env: Env, signal: str) -> bool: + signal_config = _SIGNAL_CONFIG[signal] + protocol = env.getval(signal_config["protocol"]) or env.getval( + OTEL_EXPORTER_OTLP_PROTOCOL + ) + if protocol: + return protocol.strip() == _OTLP_PROTOCOL_HTTP_PROTOBUF + + exporter = env.getval(signal_config["exporter"], _OTLP_EXPORTER).strip() + return exporter == _OTLP_PROTO_HTTP_EXPORTER + + +def _append_signal_path(endpoint: str, signal: str) -> str: + signal_path = f"v1/{signal}" + if endpoint.endswith(signal_path): + return endpoint + if not endpoint.endswith("/"): + endpoint += "/" + return endpoint + signal_path + + +def _bool_to_str(*, value: bool) -> str: + return "true" if value else "false" diff --git a/tests/integration/opamp_effective_config.py b/tests/integration/opamp_effective_config.py new file mode 100644 index 00000000..aeefa1b0 --- /dev/null +++ b/tests/integration/opamp_effective_config.py @@ -0,0 +1,139 @@ +import os +import tempfile +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +from lib import project_path + +_CONTENT_TYPE = "text/plain; format=properties; vendor=splunk; v=1.0.0" +_SERVICE_NAME = "opamp-effective-config-test" + + +def _assert_effective_config(request_path: Path) -> None: + from opentelemetry._opamp.proto import opamp_pb2 + + deadline = time.monotonic() + 10 + while not request_path.exists(): + assert time.monotonic() < deadline, "No OpAMP request received" + time.sleep(0.05) + + message = opamp_pb2.AgentToServer() + message.ParseFromString(request_path.read_bytes()) + + reports_effective_config = ( + opamp_pb2.AgentCapabilities.AgentCapabilities_ReportsEffectiveConfig + ) + assert message.capabilities & reports_effective_config + assert message.HasField("effective_config") + + identifying_attributes = { + attribute.key: attribute.value.string_value + for attribute in message.agent_description.identifying_attributes + } + assert identifying_attributes["service.name"] == _SERVICE_NAME + + config_file = message.effective_config.config_map.config_map["environment"] + assert config_file.content_type == _CONTENT_TYPE + + config = dict( + line.split("=", maxsplit=1) + for line in config_file.body.decode("utf-8").splitlines() + ) + assert config == { + "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT": "http://127.0.0.1:4318/v1/traces", + "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT": "http://127.0.0.1:4318/v1/metrics", + "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT": "http://127.0.0.1:4318/v1/logs", + "SPLUNK_PROFILER_ENABLED": "false", + "SPLUNK_PROFILER_MEMORY_ENABLED": "false", + "SPLUNK_SNAPSHOT_PROFILER_ENABLED": "false", + "SPLUNK_SNAPSHOT_PROFILER_SAMPLING_INTERVAL": "17", + "SPLUNK_PROFILER_CALL_STACK_INTERVAL": "1234", + "OTEL_CONFIG_FILE": "null", + "OTEL_EXPERIMENTAL_CONFIG_FILE": "null", + } + + +if __name__ == "__main__": + _assert_effective_config(Path(os.environ["OPAMP_TEST_REQUEST_FILE"])) + + +class _OpAMPRequestHandler(BaseHTTPRequestHandler): + request_path: Path + + def do_POST(self) -> None: + if ( + self.path != "/v1/opamp" + or self.headers.get_content_type() != "application/x-protobuf" + ): + self.send_error(400) + return + + content_length = int(self.headers["Content-Length"]) + body = self.rfile.read(content_length) + if not self.request_path.exists(): + pending_path = self.request_path.with_suffix(".pending") + pending_path.write_bytes(body) + pending_path.replace(self.request_path) + + self.send_response(200) + self.send_header("Content-Type", "application/x-protobuf") + self.send_header("Content-Length", "0") + self.end_headers() + + def log_message(self, _format: str, *args) -> None: + pass + + +class OpAMPEffectiveConfigOtelTest: + def __init__(self): + self._temp_dir = tempfile.TemporaryDirectory() + self._request_path = Path(self._temp_dir.name) / "request.pb" + _OpAMPRequestHandler.request_path = self._request_path + self._server = ThreadingHTTPServer(("127.0.0.1", 0), _OpAMPRequestHandler) + self._server_thread = threading.Thread( + target=self._server.serve_forever, + name="OpAMPTestServer", + daemon=True, + ) + self._server_thread.start() + + def requirements(self): + return (project_path(),) + + def environment_variables(self): + port = self._server.server_address[1] + return { + "NO_PROXY": "127.0.0.1,localhost", + "OTEL_EXPORTER_OTLP_ENDPOINT": "http://127.0.0.1:4318", + "OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf", + "OTEL_PYTHON_DISABLED_INSTRUMENTATIONS": "system_metrics", + "OTEL_SERVICE_NAME": _SERVICE_NAME, + "OPAMP_TEST_REQUEST_FILE": str(self._request_path), + "SPLUNK_OPAMP_ENABLED": "true", + "SPLUNK_OPAMP_ENDPOINT": f"http://127.0.0.1:{port}/v1/opamp", + "SPLUNK_OPAMP_POLLING_INTERVAL": "60000", + "SPLUNK_PROFILER_CALL_STACK_INTERVAL": "1234", + "SPLUNK_PROFILER_ENABLED": "false", + "SPLUNK_SNAPSHOT_PROFILER_ENABLED": "false", + "SPLUNK_SNAPSHOT_SAMPLING_INTERVAL": "17", + } + + def wrapper_command(self): + return "opentelemetry-instrument" + + def on_start(self): + return None + + def on_stop(self, _telemetry, stdout: str, stderr: str, returncode: int): + try: + assert returncode == 0, f"{stdout}\n{stderr}" + finally: + self._server.shutdown() + self._server.server_close() + self._server_thread.join(timeout=5) + self._temp_dir.cleanup() + + def is_http(self): + return True diff --git a/tests/test_opamp.py b/tests/test_opamp.py new file mode 100644 index 00000000..9eeea2da --- /dev/null +++ b/tests/test_opamp.py @@ -0,0 +1,300 @@ +# Copyright Splunk Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import logging + +from opentelemetry._opamp.client import OpAMPClient +from opentelemetry.environment_variables import OTEL_LOGS_EXPORTER +from opentelemetry.sdk.environment_variables import ( + OTEL_EXPORTER_OTLP_ENDPOINT, + OTEL_EXPORTER_OTLP_LOGS_ENDPOINT, + OTEL_EXPORTER_OTLP_METRICS_ENDPOINT, + OTEL_EXPORTER_OTLP_METRICS_PROTOCOL, + OTEL_EXPORTER_OTLP_PROTOCOL, + OTEL_EXPORTER_OTLP_TRACES_ENDPOINT, +) +from opentelemetry.sdk.resources import Resource +from opentelemetry.util._importlib_metadata import entry_points + +from splunk_otel.env import ( + Env, + SPLUNK_OPAMP_ENABLED, + SPLUNK_OPAMP_ENDPOINT, + SPLUNK_OPAMP_POLLING_INTERVAL, + SPLUNK_PROFILER_CALL_STACK_INTERVAL, + SPLUNK_PROFILER_ENABLED, + SPLUNK_SNAPSHOT_PROFILER_ENABLED, + SPLUNK_SNAPSHOT_SAMPLING_INTERVAL, +) +from splunk_otel.opamp import ( + OpAMPConfig, + _build_client, + _start_agent, + build_effective_config_report, + start_opamp, +) + + +class FakeClient: + def __init__(self, **kwargs): + self.init_kwargs = kwargs + self.effective_config_calls = [] + + def update_effective_config(self, config, content_type): + self.effective_config_calls.append((config, content_type)) + + +class FakeAgent: + def __init__(self, **kwargs): + self.init_kwargs = kwargs + self.started = False + + def start(self): + self.started = True + + +def parse_properties(content): + return dict(line.split("=", 1) for line in content.splitlines()) + + +def test_opamp_post_sdk_entry_point_is_registered(): + [entry_point] = entry_points( + group="_opentelemetry_opamp", name="post_sdk_init_function" + ) + + assert entry_point.value == "splunk_otel.opamp:start_opamp" + + +def test_opamp_config_returns_none_when_disabled(): + assert OpAMPConfig.from_env(Env({})) is None + + +def test_opamp_config_uses_defaults_when_enabled(): + config = OpAMPConfig.from_env(Env({SPLUNK_OPAMP_ENABLED: "true"})) + + assert config == OpAMPConfig( + endpoint="http://localhost:4320/v1/opamp", + polling_interval_ms=30000, + ) + + +def test_start_opamp_uses_resource_from_sdk_hook(monkeypatch): + resource = Resource.create({"service.name": "checkout"}) + captured = {} + + monkeypatch.setenv(SPLUNK_OPAMP_ENABLED, "true") + monkeypatch.setenv(SPLUNK_OPAMP_ENDPOINT, "http://host/opamp") + monkeypatch.setattr( + "splunk_otel.opamp._build_client", + lambda config, attributes: ( + captured.update(config=config, attributes=attributes) or FakeClient() + ), + ) + monkeypatch.setattr( + "splunk_otel.opamp._start_agent", + lambda _config, report, client: captured.update(report=report, client=client), + ) + + start_opamp(resource) + + assert captured["config"].endpoint == "http://host/opamp" + assert captured["attributes"]["service.name"] == "checkout" + assert captured["client"].effective_config_calls == [] + + +def test_start_opamp_logs_start_exception(monkeypatch, caplog): + monkeypatch.setenv(SPLUNK_OPAMP_ENABLED, "true") + monkeypatch.setattr( + "splunk_otel.opamp._build_client", + lambda *_args: (_ for _ in ()).throw(RuntimeError("boom")), + ) + + with caplog.at_level(logging.ERROR): + start_opamp(Resource.create({})) + + assert "Failed to start OpAMP client" in caplog.text + + +def test_start_agent_reports_effective_config_and_starts_agent(): + client = FakeClient() + + agent = _start_agent( + OpAMPConfig(endpoint="http://host/opamp", polling_interval_ms=5000), + f"{SPLUNK_PROFILER_ENABLED}=false", + client, + agent_factory=FakeAgent, + ) + + assert client.effective_config_calls == [ + ( + {"environment": f"{SPLUNK_PROFILER_ENABLED}=false"}, + "text/plain; format=properties; vendor=splunk; v=1.0.0", + ) + ] + assert agent.init_kwargs["interval"] == 5 + assert agent.started + + +def test_start_agent_builds_upstream_effective_config_message(): + class CapturingOpAMPClient(OpAMPClient): + effective_config = None + + def update_effective_config(self, config, content_type): + self.effective_config = super().update_effective_config( + config, content_type + ) + return self.effective_config + + client = CapturingOpAMPClient( + endpoint="http://host/opamp", + headers={}, + agent_identifying_attributes={}, + agent_non_identifying_attributes={}, + ) + + _start_agent( + OpAMPConfig(endpoint="http://host/opamp", polling_interval_ms=30000), + f"{SPLUNK_PROFILER_ENABLED}=false", + client, + agent_factory=FakeAgent, + ) + + config_file = client.effective_config.config_map.config_map["environment"] + assert ( + config_file.content_type + == "text/plain; format=properties; vendor=splunk; v=1.0.0" + ) + assert config_file.body == b"SPLUNK_PROFILER_ENABLED=false" + + +def test_client_stringifies_resource_attributes(): + client = _build_client( + OpAMPConfig(endpoint="http://host/opamp", polling_interval_ms=30000), + {"service.name": "checkout", "process.pid": 999}, + client_factory=FakeClient, + ) + + assert client.init_kwargs["agent_identifying_attributes"] == { + "service.name": "checkout", + "process.pid": "999", + } + assert client.init_kwargs["agent_non_identifying_attributes"] == {} + + +def test_effective_config_report_uses_defaults(): + assert parse_properties(build_effective_config_report(Env({}))) == { + "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT": "http://localhost:4317", + "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT": "http://localhost:4317", + "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT": "http://localhost:4317", + "SPLUNK_PROFILER_ENABLED": "false", + "SPLUNK_PROFILER_MEMORY_ENABLED": "false", + "SPLUNK_SNAPSHOT_PROFILER_ENABLED": "false", + "SPLUNK_SNAPSHOT_PROFILER_SAMPLING_INTERVAL": "10", + "SPLUNK_PROFILER_CALL_STACK_INTERVAL": "1000", + "OTEL_CONFIG_FILE": "null", + "OTEL_EXPERIMENTAL_CONFIG_FILE": "null", + } + + +def test_effective_config_report_uses_configured_splunk_values(): + report = parse_properties( + build_effective_config_report( + Env( + { + SPLUNK_PROFILER_ENABLED: "true", + SPLUNK_PROFILER_CALL_STACK_INTERVAL: "500", + SPLUNK_SNAPSHOT_PROFILER_ENABLED: "true", + SPLUNK_SNAPSHOT_SAMPLING_INTERVAL: "25", + } + ) + ) + ) + + assert report[SPLUNK_PROFILER_ENABLED] == "true" + assert report[SPLUNK_PROFILER_CALL_STACK_INTERVAL] == "500" + assert report[SPLUNK_SNAPSHOT_PROFILER_ENABLED] == "true" + assert report["SPLUNK_SNAPSHOT_PROFILER_SAMPLING_INTERVAL"] == "25" + + +def test_effective_config_report_uses_signal_specific_endpoints(): + report = parse_properties( + build_effective_config_report( + Env( + { + OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: "https://traces.example.com", + OTEL_EXPORTER_OTLP_METRICS_ENDPOINT: "https://metrics.example.com", + OTEL_EXPORTER_OTLP_LOGS_ENDPOINT: "https://logs.example.com", + } + ) + ) + ) + + assert report[OTEL_EXPORTER_OTLP_TRACES_ENDPOINT] == "https://traces.example.com" + assert report[OTEL_EXPORTER_OTLP_METRICS_ENDPOINT] == "https://metrics.example.com" + assert report[OTEL_EXPORTER_OTLP_LOGS_ENDPOINT] == "https://logs.example.com" + + +def test_effective_config_report_appends_http_signal_paths(): + report = parse_properties( + build_effective_config_report( + Env( + { + OTEL_EXPORTER_OTLP_ENDPOINT: "https://collector:4318", + OTEL_EXPORTER_OTLP_PROTOCOL: "http/protobuf", + } + ) + ) + ) + + assert ( + report[OTEL_EXPORTER_OTLP_TRACES_ENDPOINT] == "https://collector:4318/v1/traces" + ) + assert ( + report[OTEL_EXPORTER_OTLP_METRICS_ENDPOINT] + == "https://collector:4318/v1/metrics" + ) + assert report[OTEL_EXPORTER_OTLP_LOGS_ENDPOINT] == "https://collector:4318/v1/logs" + + +def test_effective_config_report_honors_signal_protocol_and_exporter(): + report = parse_properties( + build_effective_config_report( + Env( + { + OTEL_EXPORTER_OTLP_METRICS_PROTOCOL: "http/protobuf", + OTEL_LOGS_EXPORTER: "otlp_proto_http", + } + ) + ) + ) + + assert report[OTEL_EXPORTER_OTLP_TRACES_ENDPOINT] == "http://localhost:4317" + assert ( + report[OTEL_EXPORTER_OTLP_METRICS_ENDPOINT] + == "http://localhost:4318/v1/metrics" + ) + assert report[OTEL_EXPORTER_OTLP_LOGS_ENDPOINT] == "http://localhost:4318/v1/logs" + + +def test_opamp_config_reads_polling_interval(): + config = OpAMPConfig.from_env( + Env( + { + SPLUNK_OPAMP_ENABLED: "true", + SPLUNK_OPAMP_POLLING_INTERVAL: "2500", + } + ) + ) + + assert config.polling_interval_ms == 2500 From 4126c9eae180266886bfcbba274eb00ee628a1c0 Mon Sep 17 00:00:00 2001 From: Pablo Collins Date: Thu, 23 Jul 2026 15:49:36 -0400 Subject: [PATCH 02/10] Use oteltest OpAMP server in integration test --- tests/integration/opamp_effective_config.py | 122 ++++++-------------- 1 file changed, 33 insertions(+), 89 deletions(-) diff --git a/tests/integration/opamp_effective_config.py b/tests/integration/opamp_effective_config.py index aeefa1b0..8a45447a 100644 --- a/tests/integration/opamp_effective_config.py +++ b/tests/integration/opamp_effective_config.py @@ -1,118 +1,40 @@ import os import tempfile -import threading import time -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from lib import project_path -_CONTENT_TYPE = "text/plain; format=properties; vendor=splunk; v=1.0.0" _SERVICE_NAME = "opamp-effective-config-test" -def _assert_effective_config(request_path: Path) -> None: - from opentelemetry._opamp.proto import opamp_pb2 - +if __name__ == "__main__": + completion_path = Path(os.environ["OPAMP_TEST_COMPLETION_FILE"]) deadline = time.monotonic() + 10 - while not request_path.exists(): - assert time.monotonic() < deadline, "No OpAMP request received" + while not completion_path.exists(): + assert time.monotonic() < deadline, "No OpAMP callback received" time.sleep(0.05) - message = opamp_pb2.AgentToServer() - message.ParseFromString(request_path.read_bytes()) - - reports_effective_config = ( - opamp_pb2.AgentCapabilities.AgentCapabilities_ReportsEffectiveConfig - ) - assert message.capabilities & reports_effective_config - assert message.HasField("effective_config") - - identifying_attributes = { - attribute.key: attribute.value.string_value - for attribute in message.agent_description.identifying_attributes - } - assert identifying_attributes["service.name"] == _SERVICE_NAME - - config_file = message.effective_config.config_map.config_map["environment"] - assert config_file.content_type == _CONTENT_TYPE - - config = dict( - line.split("=", maxsplit=1) - for line in config_file.body.decode("utf-8").splitlines() - ) - assert config == { - "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT": "http://127.0.0.1:4318/v1/traces", - "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT": "http://127.0.0.1:4318/v1/metrics", - "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT": "http://127.0.0.1:4318/v1/logs", - "SPLUNK_PROFILER_ENABLED": "false", - "SPLUNK_PROFILER_MEMORY_ENABLED": "false", - "SPLUNK_SNAPSHOT_PROFILER_ENABLED": "false", - "SPLUNK_SNAPSHOT_PROFILER_SAMPLING_INTERVAL": "17", - "SPLUNK_PROFILER_CALL_STACK_INTERVAL": "1234", - "OTEL_CONFIG_FILE": "null", - "OTEL_EXPERIMENTAL_CONFIG_FILE": "null", - } - - -if __name__ == "__main__": - _assert_effective_config(Path(os.environ["OPAMP_TEST_REQUEST_FILE"])) - - -class _OpAMPRequestHandler(BaseHTTPRequestHandler): - request_path: Path - - def do_POST(self) -> None: - if ( - self.path != "/v1/opamp" - or self.headers.get_content_type() != "application/x-protobuf" - ): - self.send_error(400) - return - - content_length = int(self.headers["Content-Length"]) - body = self.rfile.read(content_length) - if not self.request_path.exists(): - pending_path = self.request_path.with_suffix(".pending") - pending_path.write_bytes(body) - pending_path.replace(self.request_path) - - self.send_response(200) - self.send_header("Content-Type", "application/x-protobuf") - self.send_header("Content-Length", "0") - self.end_headers() - - def log_message(self, _format: str, *args) -> None: - pass - class OpAMPEffectiveConfigOtelTest: def __init__(self): self._temp_dir = tempfile.TemporaryDirectory() - self._request_path = Path(self._temp_dir.name) / "request.pb" - _OpAMPRequestHandler.request_path = self._request_path - self._server = ThreadingHTTPServer(("127.0.0.1", 0), _OpAMPRequestHandler) - self._server_thread = threading.Thread( - target=self._server.serve_forever, - name="OpAMPTestServer", - daemon=True, - ) - self._server_thread.start() + self._completion_path = Path(self._temp_dir.name) / "complete" + self.effective_config_seen = False def requirements(self): return (project_path(),) def environment_variables(self): - port = self._server.server_address[1] return { "NO_PROXY": "127.0.0.1,localhost", "OTEL_EXPORTER_OTLP_ENDPOINT": "http://127.0.0.1:4318", "OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf", "OTEL_PYTHON_DISABLED_INSTRUMENTATIONS": "system_metrics", "OTEL_SERVICE_NAME": _SERVICE_NAME, - "OPAMP_TEST_REQUEST_FILE": str(self._request_path), + "OPAMP_TEST_COMPLETION_FILE": str(self._completion_path), "SPLUNK_OPAMP_ENABLED": "true", - "SPLUNK_OPAMP_ENDPOINT": f"http://127.0.0.1:{port}/v1/opamp", + "SPLUNK_OPAMP_ENDPOINT": "http://127.0.0.1:4320/v1/opamp", "SPLUNK_OPAMP_POLLING_INTERVAL": "60000", "SPLUNK_PROFILER_CALL_STACK_INTERVAL": "1234", "SPLUNK_PROFILER_ENABLED": "false", @@ -126,13 +48,35 @@ def wrapper_command(self): def on_start(self): return None + def on_opamp( + self, + effective_config, + remote_config_status, + remote_config_error, + ): + self.effective_config_seen = True + assert effective_config == { + "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT": "http://127.0.0.1:4318/v1/traces", + "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT": "http://127.0.0.1:4318/v1/metrics", + "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT": "http://127.0.0.1:4318/v1/logs", + "SPLUNK_PROFILER_ENABLED": "false", + "SPLUNK_PROFILER_MEMORY_ENABLED": "false", + "SPLUNK_SNAPSHOT_PROFILER_ENABLED": "false", + "SPLUNK_SNAPSHOT_PROFILER_SAMPLING_INTERVAL": "17", + "SPLUNK_PROFILER_CALL_STACK_INTERVAL": "1234", + "OTEL_CONFIG_FILE": "null", + "OTEL_EXPERIMENTAL_CONFIG_FILE": "null", + } + assert remote_config_status is None + assert remote_config_error is None + self._completion_path.touch() + return None + def on_stop(self, _telemetry, stdout: str, stderr: str, returncode: int): try: assert returncode == 0, f"{stdout}\n{stderr}" + assert self.effective_config_seen finally: - self._server.shutdown() - self._server.server_close() - self._server_thread.join(timeout=5) self._temp_dir.cleanup() def is_http(self): From e9a031bca2b41c0cd7285ef3fd2babfa8d4df27b Mon Sep 17 00:00:00 2001 From: Pablo Collins Date: Wed, 29 Jul 2026 09:21:37 -0400 Subject: [PATCH 03/10] Handle OpAMP resource attribute arrays --- src/splunk_otel/opamp.py | 111 ++++++++------- tests/integration/opamp_effective_config.py | 1 - tests/test_opamp.py | 150 +++++++++++++------- 3 files changed, 159 insertions(+), 103 deletions(-) diff --git a/src/splunk_otel/opamp.py b/src/splunk_otel/opamp.py index b137a269..11bd4ee0 100644 --- a/src/splunk_otel/opamp.py +++ b/src/splunk_otel/opamp.py @@ -14,8 +14,9 @@ from __future__ import annotations +import json import logging -from dataclasses import dataclass +from collections.abc import Mapping, Sequence from typing import TYPE_CHECKING from opentelemetry._opamp.agent import OpAMPAgent @@ -49,10 +50,7 @@ ) if TYPE_CHECKING: - from collections.abc import Mapping - from opentelemetry.sdk.resources import Resource - from opentelemetry.util.types import AnyValue logger = logging.getLogger(__name__) @@ -67,15 +65,14 @@ _OTLP_PROTOCOL_HTTP_PROTOBUF = "http/protobuf" _OTLP_EXPORTER = "otlp" _OTLP_PROTO_HTTP_EXPORTER = "otlp_proto_http" +_IDENTIFYING_RESOURCE_ATTRIBUTES = frozenset(("service.name", "service.namespace", "service.instance.id")) _SPLUNK_PROFILER_MEMORY_ENABLED = "SPLUNK_PROFILER_MEMORY_ENABLED" -_SPLUNK_SNAPSHOT_PROFILER_SAMPLING_INTERVAL = ( - "SPLUNK_SNAPSHOT_PROFILER_SAMPLING_INTERVAL" -) +_SPLUNK_SNAPSHOT_PROFILER_SAMPLING_INTERVAL = "SPLUNK_SNAPSHOT_PROFILER_SAMPLING_INTERVAL" _OTEL_CONFIG_FILE = "OTEL_CONFIG_FILE" _OTEL_EXPERIMENTAL_CONFIG_FILE = "OTEL_EXPERIMENTAL_CONFIG_FILE" -_SIGNAL_CONFIG = { +_SIGNAL_ENV_VARS = { "traces": { "endpoint": OTEL_EXPORTER_OTLP_TRACES_ENDPOINT, "protocol": OTEL_EXPORTER_OTLP_TRACES_PROTOCOL, @@ -94,26 +91,6 @@ } -@dataclass(frozen=True) -class OpAMPConfig: - endpoint: str - polling_interval_ms: int - - @classmethod - def from_env(cls, env: Env) -> OpAMPConfig | None: - if not env.is_true(SPLUNK_OPAMP_ENABLED): - logger.debug("OpAMP disabled (%s is not true)", SPLUNK_OPAMP_ENABLED) - return None - - return cls( - endpoint=env.getval(SPLUNK_OPAMP_ENDPOINT, _DEFAULT_OPAMP_ENDPOINT), - polling_interval_ms=env.getint( - SPLUNK_OPAMP_POLLING_INTERVAL, - _DEFAULT_OPAMP_POLLING_INTERVAL_MS, - ), - ) - - class _SplunkCallbacks(OpAMPCallbacks): def on_connect_failed( self, @@ -146,13 +123,23 @@ def on_message( def start_opamp(resource: Resource) -> None: """Start the Splunk OpAMP agent after the OpenTelemetry SDK starts.""" env = Env() - config = OpAMPConfig.from_env(env) - if config is None: + if not env.is_true(SPLUNK_OPAMP_ENABLED): + logger.debug("OpAMP disabled (%s is not true)", SPLUNK_OPAMP_ENABLED) return + endpoint = env.getval(SPLUNK_OPAMP_ENDPOINT, _DEFAULT_OPAMP_ENDPOINT) + polling_interval_ms = env.getint( + SPLUNK_OPAMP_POLLING_INTERVAL, + _DEFAULT_OPAMP_POLLING_INTERVAL_MS, + ) try: - client = _build_client(config, resource.attributes) - _start_agent(config, build_effective_config_report(env), client) + client = _build_client(endpoint, resource.attributes) + _start_agent( + polling_interval_ms, + build_effective_config_report(env), + client, + ) + logger.info("OpAMP client started: %s", endpoint) except Exception: logger.exception("Failed to start OpAMP client") @@ -196,23 +183,52 @@ def build_effective_config_report(env: Env) -> str: def _build_client( - config: OpAMPConfig, - resource_attributes: Mapping[str, AnyValue], + endpoint: str, + resource_attributes: Mapping[str, object], client_factory=OpAMPClient, ): - identifying_attributes = { - str(key): str(value) for key, value in resource_attributes.items() - } + identifying_attributes = {} + non_identifying_attributes = {} + for key, value in resource_attributes.items(): + encoded_value = _encode_resource_attribute(key, value) + if encoded_value is None: + continue + if key in _IDENTIFYING_RESOURCE_ATTRIBUTES: + identifying_attributes[key] = encoded_value + else: + non_identifying_attributes[key] = encoded_value + return client_factory( - endpoint=config.endpoint, + endpoint=endpoint, headers={}, agent_identifying_attributes=identifying_attributes, - agent_non_identifying_attributes={}, + agent_non_identifying_attributes=non_identifying_attributes, ) +def _encode_resource_attribute( + key: str, + value: object, +) -> str | bool | int | float | bytes | None: + if isinstance(value, (str, bool, int, float, bytes)): + return value + if isinstance(value, Sequence): + # opentelemetry-opamp-client 0.3b0 cannot encode sequence values. + try: + return json.dumps(value, separators=(",", ":")) + except (TypeError, ValueError): + pass + + logger.warning( + "Skipping OpAMP resource attribute %s with unsupported type %s", + key, + type(value).__name__, + ) + return None + + def _start_agent( - config: OpAMPConfig, + polling_interval_ms: int, effective_config_report: str, client, agent_factory=OpAMPAgent, @@ -222,18 +238,17 @@ def _start_agent( content_type=_CONFIG_CONTENT_TYPE, ) agent = agent_factory( - interval=config.polling_interval_ms / 1000, + interval=polling_interval_ms / 1000, callbacks=_SplunkCallbacks(), client=client, ) agent.start() - logger.info("OpAMP client started: %s", config.endpoint) return agent def _get_signal_endpoint(env: Env, signal: str) -> str: - signal_config = _SIGNAL_CONFIG[signal] - endpoint = env.getval(signal_config["endpoint"]) + signal_env_vars = _SIGNAL_ENV_VARS[signal] + endpoint = env.getval(signal_env_vars["endpoint"]) if endpoint: return endpoint @@ -245,14 +260,12 @@ def _get_signal_endpoint(env: Env, signal: str) -> str: def _uses_http_protobuf(env: Env, signal: str) -> bool: - signal_config = _SIGNAL_CONFIG[signal] - protocol = env.getval(signal_config["protocol"]) or env.getval( - OTEL_EXPORTER_OTLP_PROTOCOL - ) + signal_env_vars = _SIGNAL_ENV_VARS[signal] + protocol = env.getval(signal_env_vars["protocol"]) or env.getval(OTEL_EXPORTER_OTLP_PROTOCOL) if protocol: return protocol.strip() == _OTLP_PROTOCOL_HTTP_PROTOBUF - exporter = env.getval(signal_config["exporter"], _OTLP_EXPORTER).strip() + exporter = env.getval(signal_env_vars["exporter"], _OTLP_EXPORTER).strip() return exporter == _OTLP_PROTO_HTTP_EXPORTER diff --git a/tests/integration/opamp_effective_config.py b/tests/integration/opamp_effective_config.py index 8a45447a..666588bc 100644 --- a/tests/integration/opamp_effective_config.py +++ b/tests/integration/opamp_effective_config.py @@ -70,7 +70,6 @@ def on_opamp( assert remote_config_status is None assert remote_config_error is None self._completion_path.touch() - return None def on_stop(self, _telemetry, stdout: str, stderr: str, returncode: int): try: diff --git a/tests/test_opamp.py b/tests/test_opamp.py index 9eeea2da..e60a7866 100644 --- a/tests/test_opamp.py +++ b/tests/test_opamp.py @@ -15,6 +15,7 @@ import logging from opentelemetry._opamp.client import OpAMPClient +from opentelemetry._opamp.proto import opamp_pb2 from opentelemetry.environment_variables import OTEL_LOGS_EXPORTER from opentelemetry.sdk.environment_variables import ( OTEL_EXPORTER_OTLP_ENDPOINT, @@ -38,7 +39,6 @@ SPLUNK_SNAPSHOT_SAMPLING_INTERVAL, ) from splunk_otel.opamp import ( - OpAMPConfig, _build_client, _start_agent, build_effective_config_report, @@ -69,24 +69,39 @@ def parse_properties(content): def test_opamp_post_sdk_entry_point_is_registered(): - [entry_point] = entry_points( - group="_opentelemetry_opamp", name="post_sdk_init_function" - ) + [entry_point] = entry_points(group="_opentelemetry_opamp", name="post_sdk_init_function") assert entry_point.value == "splunk_otel.opamp:start_opamp" -def test_opamp_config_returns_none_when_disabled(): - assert OpAMPConfig.from_env(Env({})) is None +def test_start_opamp_returns_when_disabled(monkeypatch): + monkeypatch.setattr( + "splunk_otel.opamp._build_client", + lambda *_args: (_ for _ in ()).throw(AssertionError("unexpected call")), + ) + + start_opamp(Resource.create({})) -def test_opamp_config_uses_defaults_when_enabled(): - config = OpAMPConfig.from_env(Env({SPLUNK_OPAMP_ENABLED: "true"})) +def test_start_opamp_uses_defaults_when_enabled(monkeypatch): + captured = {} - assert config == OpAMPConfig( - endpoint="http://localhost:4320/v1/opamp", - polling_interval_ms=30000, + monkeypatch.setenv(SPLUNK_OPAMP_ENABLED, "true") + monkeypatch.setattr( + "splunk_otel.opamp._build_client", + lambda endpoint, _attributes: (captured.update(endpoint=endpoint) or FakeClient()), ) + monkeypatch.setattr( + "splunk_otel.opamp._start_agent", + lambda polling_interval_ms, _report, _client: captured.update(polling_interval_ms=polling_interval_ms), + ) + + start_opamp(Resource.create({})) + + assert captured == { + "endpoint": "http://localhost:4320/v1/opamp", + "polling_interval_ms": 30000, + } def test_start_opamp_uses_resource_from_sdk_hook(monkeypatch): @@ -95,20 +110,24 @@ def test_start_opamp_uses_resource_from_sdk_hook(monkeypatch): monkeypatch.setenv(SPLUNK_OPAMP_ENABLED, "true") monkeypatch.setenv(SPLUNK_OPAMP_ENDPOINT, "http://host/opamp") + monkeypatch.setenv(SPLUNK_OPAMP_POLLING_INTERVAL, "2500") monkeypatch.setattr( "splunk_otel.opamp._build_client", - lambda config, attributes: ( - captured.update(config=config, attributes=attributes) or FakeClient() - ), + lambda endpoint, attributes: (captured.update(endpoint=endpoint, attributes=attributes) or FakeClient()), ) monkeypatch.setattr( "splunk_otel.opamp._start_agent", - lambda _config, report, client: captured.update(report=report, client=client), + lambda polling_interval_ms, report, client: captured.update( + polling_interval_ms=polling_interval_ms, + report=report, + client=client, + ), ) start_opamp(resource) - assert captured["config"].endpoint == "http://host/opamp" + assert captured["endpoint"] == "http://host/opamp" + assert captured["polling_interval_ms"] == 2500 assert captured["attributes"]["service.name"] == "checkout" assert captured["client"].effective_config_calls == [] @@ -130,7 +149,7 @@ def test_start_agent_reports_effective_config_and_starts_agent(): client = FakeClient() agent = _start_agent( - OpAMPConfig(endpoint="http://host/opamp", polling_interval_ms=5000), + 5000, f"{SPLUNK_PROFILER_ENABLED}=false", client, agent_factory=FakeAgent, @@ -151,9 +170,7 @@ class CapturingOpAMPClient(OpAMPClient): effective_config = None def update_effective_config(self, config, content_type): - self.effective_config = super().update_effective_config( - config, content_type - ) + self.effective_config = super().update_effective_config(config, content_type) return self.effective_config client = CapturingOpAMPClient( @@ -164,32 +181,80 @@ def update_effective_config(self, config, content_type): ) _start_agent( - OpAMPConfig(endpoint="http://host/opamp", polling_interval_ms=30000), + 30000, f"{SPLUNK_PROFILER_ENABLED}=false", client, agent_factory=FakeAgent, ) config_file = client.effective_config.config_map.config_map["environment"] - assert ( - config_file.content_type - == "text/plain; format=properties; vendor=splunk; v=1.0.0" - ) + assert config_file.content_type == "text/plain; format=properties; vendor=splunk; v=1.0.0" assert config_file.body == b"SPLUNK_PROFILER_ENABLED=false" -def test_client_stringifies_resource_attributes(): +def test_client_partitions_resource_attributes_and_preserves_types(): client = _build_client( - OpAMPConfig(endpoint="http://host/opamp", polling_interval_ms=30000), - {"service.name": "checkout", "process.pid": 999}, + "http://host/opamp", + { + "service.name": "checkout", + "service.namespace": "store", + "service.instance.id": "checkout-1", + "process.pid": 999, + "host.name": "host-1", + }, client_factory=FakeClient, ) assert client.init_kwargs["agent_identifying_attributes"] == { "service.name": "checkout", - "process.pid": "999", + "service.namespace": "store", + "service.instance.id": "checkout-1", } + assert client.init_kwargs["agent_non_identifying_attributes"] == { + "process.pid": 999, + "host.name": "host-1", + } + + +def test_client_serializes_resource_attribute_arrays(): + resource = Resource( + { + "service.name": "checkout", + "host.ip": ["10.0.0.12", "127.0.0.1"], + "custom.flags": [True, False], + "custom.ports": [4317, 4318], + "custom.ratios": [0.5, 1.5], + } + ) + + client = _build_client( + "http://host/opamp", + resource.attributes, + ) + message = opamp_pb2.AgentToServer() + message.ParseFromString(client.build_full_state_message()) + attributes = {attribute.key: attribute.value for attribute in message.agent_description.non_identifying_attributes} + + assert attributes["host.ip"].string_value == '["10.0.0.12","127.0.0.1"]' + assert attributes["custom.flags"].string_value == "[true,false]" + assert attributes["custom.ports"].string_value == "[4317,4318]" + assert attributes["custom.ratios"].string_value == "[0.5,1.5]" + + +def test_client_skips_unsupported_resource_attribute(caplog): + with caplog.at_level(logging.WARNING): + client = _build_client( + "http://host/opamp", + { + "service.name": "checkout", + "custom.mapping": {"nested": True}, + }, + client_factory=FakeClient, + ) + + assert client.init_kwargs["agent_identifying_attributes"] == {"service.name": "checkout"} assert client.init_kwargs["agent_non_identifying_attributes"] == {} + assert "Skipping OpAMP resource attribute custom.mapping with unsupported type dict" in caplog.text def test_effective_config_report_uses_defaults(): @@ -257,13 +322,8 @@ def test_effective_config_report_appends_http_signal_paths(): ) ) - assert ( - report[OTEL_EXPORTER_OTLP_TRACES_ENDPOINT] == "https://collector:4318/v1/traces" - ) - assert ( - report[OTEL_EXPORTER_OTLP_METRICS_ENDPOINT] - == "https://collector:4318/v1/metrics" - ) + assert report[OTEL_EXPORTER_OTLP_TRACES_ENDPOINT] == "https://collector:4318/v1/traces" + assert report[OTEL_EXPORTER_OTLP_METRICS_ENDPOINT] == "https://collector:4318/v1/metrics" assert report[OTEL_EXPORTER_OTLP_LOGS_ENDPOINT] == "https://collector:4318/v1/logs" @@ -280,21 +340,5 @@ def test_effective_config_report_honors_signal_protocol_and_exporter(): ) assert report[OTEL_EXPORTER_OTLP_TRACES_ENDPOINT] == "http://localhost:4317" - assert ( - report[OTEL_EXPORTER_OTLP_METRICS_ENDPOINT] - == "http://localhost:4318/v1/metrics" - ) + assert report[OTEL_EXPORTER_OTLP_METRICS_ENDPOINT] == "http://localhost:4318/v1/metrics" assert report[OTEL_EXPORTER_OTLP_LOGS_ENDPOINT] == "http://localhost:4318/v1/logs" - - -def test_opamp_config_reads_polling_interval(): - config = OpAMPConfig.from_env( - Env( - { - SPLUNK_OPAMP_ENABLED: "true", - SPLUNK_OPAMP_POLLING_INTERVAL: "2500", - } - ) - ) - - assert config.polling_interval_ms == 2500 From 746b8ddc17023fe809809f826e5df388901588a0 Mon Sep 17 00:00:00 2001 From: Pablo Collins Date: Wed, 29 Jul 2026 11:19:15 -0400 Subject: [PATCH 04/10] Update docs for OpAMP support --- README.md | 10 ++++++++++ docs/opamp.md | 45 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 55 insertions(+) create mode 100644 docs/opamp.md diff --git a/README.md b/README.md index 551fee87..52997911 100644 --- a/README.md +++ b/README.md @@ -81,6 +81,16 @@ exporter. See [docs/secureapp.md](docs/secureapp.md) for setup details and [docs/examples/secureapp-collector-config.yaml](docs/examples/secureapp-collector-config.yaml) for a collector example. +## Fleet Management with OpAMP + +Splunk OTel Python provides OpAMP support. When enabled, the OpAMP client +connects to the configured endpoint and reports the Python agent's status and +settings. Agents connected to Splunk Observability Cloud appear on the Fleet +Management page. + +The default OpAMP endpoint is a local Splunk OpenTelemetry Collector. Set +`SPLUNK_OPAMP_ENDPOINT` to use another OpAMP endpoint. See +[OpAMP](docs/opamp.md) for setup and limits. # License diff --git a/docs/opamp.md b/docs/opamp.md new file mode 100644 index 00000000..a4c77c66 --- /dev/null +++ b/docs/opamp.md @@ -0,0 +1,45 @@ +# OpAMP + +Splunk OTel Python provides OpAMP support. When enabled, the OpAMP client +connects to the configured endpoint and reports the Python agent's status and +settings. Agents connected to Splunk Observability Cloud appear on the Fleet +Management page. + +OpAMP support is provisional. Remote configuration is not supported yet. + +## Default connection + +```text +Python process -> Splunk OpenTelemetry Collector -> Splunk Observability Cloud +``` + +The Python process sends OpAMP messages to +`http://localhost:4320/v1/opamp` by default. The collector must listen on that +address and forward the messages to Splunk Observability Cloud. + +## Enable OpAMP + +```sh +SPLUNK_OPAMP_ENABLED=true \ +opentelemetry-instrument python app.py +``` + +## Settings + +| Environment variable | Default | Description | +|------------------------------------|------------------------------------|--------------------------------| +| `SPLUNK_OPAMP_ENABLED` | `false` | Set to `true` to enable OpAMP. | +| `SPLUNK_OPAMP_ENDPOINT` | `http://localhost:4320/v1/opamp` | OpAMP endpoint. | +| `SPLUNK_OPAMP_POLLING_INTERVAL` | `30000` | Report interval, in milliseconds. | + + +## Troubleshooting + +Open **Data Management > Fleet Management > Instrumentation** in Splunk +Observability Cloud. If OpAMP is enabled, the Python agent should show `Connected`. + +If it does not: + +- When using the default endpoint, check that the collector listens on port `4320`. +- Check the collector endpoint and access token. +- Check the application logs for `Connection to OpAMP server failed`. From da0caf5f7cc2c03eccb20f3de6e700a003da8a45 Mon Sep 17 00:00:00 2001 From: Pablo Collins Date: Wed, 29 Jul 2026 11:28:31 -0400 Subject: [PATCH 05/10] Clarify OpAMP changelog entry --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7bc1bd02..d686eb75 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,7 @@ ## Unreleased -- Add provisional OpAMP effective-config reporting through the OpenTelemetry SDK post-initialization hook +- Add provisional OpAMP support for reporting Python agent status and settings ## 2.12.0 - 2026-07-13 - Upgrade Otel dependencies to [1.44.0](https://github.com/open-telemetry/opentelemetry-python/releases/tag/v1.44.0) / [0.65b0](https://github.com/open-telemetry/opentelemetry-python-contrib/releases/tag/v0.65b0) From f81d8e5265b7c2580b201ebd36693ec2ae30435a Mon Sep 17 00:00:00 2001 From: Pablo Collins Date: Wed, 29 Jul 2026 15:40:29 -0400 Subject: [PATCH 06/10] Sanitize OpAMP reported endpoints --- src/splunk_otel/opamp.py | 32 +++++++++++++++++++++++++++--- tests/test_opamp.py | 43 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 3 deletions(-) diff --git a/src/splunk_otel/opamp.py b/src/splunk_otel/opamp.py index 11bd4ee0..bd372250 100644 --- a/src/splunk_otel/opamp.py +++ b/src/splunk_otel/opamp.py @@ -146,9 +146,18 @@ def start_opamp(resource: Resource) -> None: def build_effective_config_report(env: Env) -> str: values = ( - (OTEL_EXPORTER_OTLP_TRACES_ENDPOINT, _get_signal_endpoint(env, "traces")), - (OTEL_EXPORTER_OTLP_METRICS_ENDPOINT, _get_signal_endpoint(env, "metrics")), - (OTEL_EXPORTER_OTLP_LOGS_ENDPOINT, _get_signal_endpoint(env, "logs")), + ( + OTEL_EXPORTER_OTLP_TRACES_ENDPOINT, + _sanitize_endpoint_for_reporting(_get_signal_endpoint(env, "traces")), + ), + ( + OTEL_EXPORTER_OTLP_METRICS_ENDPOINT, + _sanitize_endpoint_for_reporting(_get_signal_endpoint(env, "metrics")), + ), + ( + OTEL_EXPORTER_OTLP_LOGS_ENDPOINT, + _sanitize_endpoint_for_reporting(_get_signal_endpoint(env, "logs")), + ), ( SPLUNK_PROFILER_ENABLED, _bool_to_str(value=env.is_true(SPLUNK_PROFILER_ENABLED)), @@ -259,6 +268,23 @@ def _get_signal_endpoint(env: Env, signal: str) -> str: return base_endpoint or _DEFAULT_GRPC_ENDPOINT +def _sanitize_endpoint_for_reporting(endpoint: str) -> str: + sanitized_endpoint = endpoint.split("#", 1)[0].split("?", 1)[0] + if sanitized_endpoint.startswith("//"): + prefix = "//" + authority_and_path = sanitized_endpoint[2:] + else: + scheme, separator, authority_and_path = sanitized_endpoint.partition("://") + if not separator: + return sanitized_endpoint + prefix = f"{scheme}{separator}" + + authority, separator, path = authority_and_path.partition("/") + if "@" not in authority: + return sanitized_endpoint + return f"{prefix}{authority.rsplit('@', 1)[-1]}{separator}{path}" + + def _uses_http_protobuf(env: Env, signal: str) -> bool: signal_env_vars = _SIGNAL_ENV_VARS[signal] protocol = env.getval(signal_env_vars["protocol"]) or env.getval(OTEL_EXPORTER_OTLP_PROTOCOL) diff --git a/tests/test_opamp.py b/tests/test_opamp.py index e60a7866..925fdcea 100644 --- a/tests/test_opamp.py +++ b/tests/test_opamp.py @@ -40,6 +40,7 @@ ) from splunk_otel.opamp import ( _build_client, + _sanitize_endpoint_for_reporting, _start_agent, build_effective_config_report, start_opamp, @@ -257,6 +258,30 @@ def test_client_skips_unsupported_resource_attribute(caplog): assert "Skipping OpAMP resource attribute custom.mapping with unsupported type dict" in caplog.text +def test_sanitize_endpoint_for_reporting(): + assert ( + _sanitize_endpoint_for_reporting("https://collector.example.com:4318/v1/traces") + == "https://collector.example.com:4318/v1/traces" + ) + assert _sanitize_endpoint_for_reporting("localhost:4317") == "localhost:4317" + assert ( + _sanitize_endpoint_for_reporting("https://alice:secret@collector.example.com:4318/v1/traces") + == "https://collector.example.com:4318/v1/traces" + ) + assert ( + _sanitize_endpoint_for_reporting("https://collector.example.com:4318/v1/traces?token=abc#fragment") + == "https://collector.example.com:4318/v1/traces" + ) + assert ( + _sanitize_endpoint_for_reporting("//alice:secret@collector.example.com/v1/traces") + == "//collector.example.com/v1/traces" + ) + assert ( + _sanitize_endpoint_for_reporting("https://alice:secret@[invalid") + == "https://[invalid" + ) + + def test_effective_config_report_uses_defaults(): assert parse_properties(build_effective_config_report(Env({}))) == { "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT": "http://localhost:4317", @@ -310,6 +335,24 @@ def test_effective_config_report_uses_signal_specific_endpoints(): assert report[OTEL_EXPORTER_OTLP_LOGS_ENDPOINT] == "https://logs.example.com" +def test_effective_config_report_sanitizes_endpoint_credentials_and_parameters(): + report = parse_properties( + build_effective_config_report( + Env( + { + OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: "https://alice:secret@traces.example.com/v1/traces?token=abc#fragment", + OTEL_EXPORTER_OTLP_METRICS_ENDPOINT: "https://bob:password@metrics.example.com/v1/metrics?token=def", + OTEL_EXPORTER_OTLP_LOGS_ENDPOINT: "https://carol:pass@logs.example.com/v1/logs#fragment", + } + ) + ) + ) + + assert report[OTEL_EXPORTER_OTLP_TRACES_ENDPOINT] == "https://traces.example.com/v1/traces" + assert report[OTEL_EXPORTER_OTLP_METRICS_ENDPOINT] == "https://metrics.example.com/v1/metrics" + assert report[OTEL_EXPORTER_OTLP_LOGS_ENDPOINT] == "https://logs.example.com/v1/logs" + + def test_effective_config_report_appends_http_signal_paths(): report = parse_properties( build_effective_config_report( From 830723c5b370b7f5325a25d25a702ee592ad7a06 Mon Sep 17 00:00:00 2001 From: Pablo Collins Date: Wed, 29 Jul 2026 16:23:21 -0400 Subject: [PATCH 07/10] Validate OpAMP polling interval --- src/splunk_otel/opamp.py | 7 +++++++ tests/test_opamp.py | 24 ++++++++++++++++++++++++ 2 files changed, 31 insertions(+) diff --git a/src/splunk_otel/opamp.py b/src/splunk_otel/opamp.py index bd372250..fdb823f9 100644 --- a/src/splunk_otel/opamp.py +++ b/src/splunk_otel/opamp.py @@ -132,6 +132,13 @@ def start_opamp(resource: Resource) -> None: SPLUNK_OPAMP_POLLING_INTERVAL, _DEFAULT_OPAMP_POLLING_INTERVAL_MS, ) + if polling_interval_ms <= 0: + logger.warning( + "Invalid non-positive value for %s; using default %d ms", + SPLUNK_OPAMP_POLLING_INTERVAL, + _DEFAULT_OPAMP_POLLING_INTERVAL_MS, + ) + polling_interval_ms = _DEFAULT_OPAMP_POLLING_INTERVAL_MS try: client = _build_client(endpoint, resource.attributes) _start_agent( diff --git a/tests/test_opamp.py b/tests/test_opamp.py index 925fdcea..40803169 100644 --- a/tests/test_opamp.py +++ b/tests/test_opamp.py @@ -105,6 +105,30 @@ def test_start_opamp_uses_defaults_when_enabled(monkeypatch): } +def test_start_opamp_uses_default_for_non_positive_polling_interval(monkeypatch, caplog): + captured = [] + + monkeypatch.setenv(SPLUNK_OPAMP_ENABLED, "true") + monkeypatch.setattr( + "splunk_otel.opamp._build_client", + lambda *_args: FakeClient(), + ) + monkeypatch.setattr( + "splunk_otel.opamp._start_agent", + lambda polling_interval_ms, *_args: captured.append(polling_interval_ms), + ) + + for interval in ("0", "-1"): + monkeypatch.setenv(SPLUNK_OPAMP_POLLING_INTERVAL, interval) + start_opamp(Resource.create({})) + + assert captured == [30000, 30000] + assert caplog.messages == [ + "Invalid non-positive value for SPLUNK_OPAMP_POLLING_INTERVAL; using default 30000 ms", + "Invalid non-positive value for SPLUNK_OPAMP_POLLING_INTERVAL; using default 30000 ms", + ] + + def test_start_opamp_uses_resource_from_sdk_hook(monkeypatch): resource = Resource.create({"service.name": "checkout"}) captured = {} From 148649c33d87674c17de8e50876a90b6a2a053b7 Mon Sep 17 00:00:00 2001 From: Pablo Collins Date: Thu, 30 Jul 2026 10:35:09 -0400 Subject: [PATCH 08/10] Simplify OpAMP integration test synchronization --- tests/integration/opamp_effective_config.py | 20 +++----------------- 1 file changed, 3 insertions(+), 17 deletions(-) diff --git a/tests/integration/opamp_effective_config.py b/tests/integration/opamp_effective_config.py index 666588bc..50c6458e 100644 --- a/tests/integration/opamp_effective_config.py +++ b/tests/integration/opamp_effective_config.py @@ -1,7 +1,4 @@ -import os -import tempfile import time -from pathlib import Path from lib import project_path @@ -9,17 +6,11 @@ if __name__ == "__main__": - completion_path = Path(os.environ["OPAMP_TEST_COMPLETION_FILE"]) - deadline = time.monotonic() + 10 - while not completion_path.exists(): - assert time.monotonic() < deadline, "No OpAMP callback received" - time.sleep(0.05) + time.sleep(4) class OpAMPEffectiveConfigOtelTest: def __init__(self): - self._temp_dir = tempfile.TemporaryDirectory() - self._completion_path = Path(self._temp_dir.name) / "complete" self.effective_config_seen = False def requirements(self): @@ -32,7 +23,6 @@ def environment_variables(self): "OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf", "OTEL_PYTHON_DISABLED_INSTRUMENTATIONS": "system_metrics", "OTEL_SERVICE_NAME": _SERVICE_NAME, - "OPAMP_TEST_COMPLETION_FILE": str(self._completion_path), "SPLUNK_OPAMP_ENABLED": "true", "SPLUNK_OPAMP_ENDPOINT": "http://127.0.0.1:4320/v1/opamp", "SPLUNK_OPAMP_POLLING_INTERVAL": "60000", @@ -69,14 +59,10 @@ def on_opamp( } assert remote_config_status is None assert remote_config_error is None - self._completion_path.touch() def on_stop(self, _telemetry, stdout: str, stderr: str, returncode: int): - try: - assert returncode == 0, f"{stdout}\n{stderr}" - assert self.effective_config_seen - finally: - self._temp_dir.cleanup() + assert returncode == 0, f"{stdout}\n{stderr}" + assert self.effective_config_seen def is_http(self): return True From 376b68c11c011bd925813d30eb2793586184d89e Mon Sep 17 00:00:00 2001 From: Pablo Collins Date: Thu, 30 Jul 2026 10:37:37 -0400 Subject: [PATCH 09/10] Format OpAMP tests --- tests/test_opamp.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/tests/test_opamp.py b/tests/test_opamp.py index 40803169..83298ec8 100644 --- a/tests/test_opamp.py +++ b/tests/test_opamp.py @@ -300,10 +300,7 @@ def test_sanitize_endpoint_for_reporting(): _sanitize_endpoint_for_reporting("//alice:secret@collector.example.com/v1/traces") == "//collector.example.com/v1/traces" ) - assert ( - _sanitize_endpoint_for_reporting("https://alice:secret@[invalid") - == "https://[invalid" - ) + assert _sanitize_endpoint_for_reporting("https://alice:secret@[invalid") == "https://[invalid" def test_effective_config_report_uses_defaults(): From cde6cb8fe79b2184f568361b49fcf35e8b955838 Mon Sep 17 00:00:00 2001 From: Pablo Collins Date: Mon, 3 Aug 2026 14:49:20 -0400 Subject: [PATCH 10/10] Sanitize logged OpAMP endpoint --- src/splunk_otel/opamp.py | 2 +- tests/test_opamp.py | 21 +++++++++++++++++++++ 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/src/splunk_otel/opamp.py b/src/splunk_otel/opamp.py index fdb823f9..ab3497d7 100644 --- a/src/splunk_otel/opamp.py +++ b/src/splunk_otel/opamp.py @@ -146,7 +146,7 @@ def start_opamp(resource: Resource) -> None: build_effective_config_report(env), client, ) - logger.info("OpAMP client started: %s", endpoint) + logger.info("OpAMP client started: %s", _sanitize_endpoint_for_reporting(endpoint)) except Exception: logger.exception("Failed to start OpAMP client") diff --git a/tests/test_opamp.py b/tests/test_opamp.py index 83298ec8..6c2c25cc 100644 --- a/tests/test_opamp.py +++ b/tests/test_opamp.py @@ -105,6 +105,27 @@ def test_start_opamp_uses_defaults_when_enabled(monkeypatch): } +def test_start_opamp_logs_sanitized_endpoint(monkeypatch, caplog): + monkeypatch.setenv(SPLUNK_OPAMP_ENABLED, "true") + monkeypatch.setenv( + SPLUNK_OPAMP_ENDPOINT, + "https://alice:secret@opamp.example.com/v1/opamp?token=abc#fragment", + ) + monkeypatch.setattr( + "splunk_otel.opamp._build_client", + lambda *_args: FakeClient(), + ) + monkeypatch.setattr( + "splunk_otel.opamp._start_agent", + lambda *_args: None, + ) + + with caplog.at_level(logging.INFO): + start_opamp(Resource.create({})) + + assert caplog.messages == ["OpAMP client started: https://opamp.example.com/v1/opamp"] + + def test_start_opamp_uses_default_for_non_positive_polling_interval(monkeypatch, caplog): captured = []