diff --git a/CHANGELOG.md b/CHANGELOG.md index 452ab517..d686eb75 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- 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) 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`. 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..ab3497d7 --- /dev/null +++ b/src/splunk_otel/opamp.py @@ -0,0 +1,315 @@ +# 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 json +import logging +from collections.abc import Mapping, Sequence +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 opentelemetry.sdk.resources import Resource + +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" +_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" +_OTEL_CONFIG_FILE = "OTEL_CONFIG_FILE" +_OTEL_EXPERIMENTAL_CONFIG_FILE = "OTEL_EXPERIMENTAL_CONFIG_FILE" + +_SIGNAL_ENV_VARS = { + "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, + }, +} + + +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() + 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, + ) + 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( + polling_interval_ms, + build_effective_config_report(env), + client, + ) + logger.info("OpAMP client started: %s", _sanitize_endpoint_for_reporting(endpoint)) + except Exception: + logger.exception("Failed to start OpAMP client") + + +def build_effective_config_report(env: Env) -> str: + values = ( + ( + 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)), + ), + (_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( + endpoint: str, + resource_attributes: Mapping[str, object], + client_factory=OpAMPClient, +): + 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=endpoint, + headers={}, + agent_identifying_attributes=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( + polling_interval_ms: int, + 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=polling_interval_ms / 1000, + callbacks=_SplunkCallbacks(), + client=client, + ) + agent.start() + return agent + + +def _get_signal_endpoint(env: Env, signal: str) -> str: + signal_env_vars = _SIGNAL_ENV_VARS[signal] + endpoint = env.getval(signal_env_vars["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 _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) + if protocol: + return protocol.strip() == _OTLP_PROTOCOL_HTTP_PROTOBUF + + exporter = env.getval(signal_env_vars["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..50c6458e --- /dev/null +++ b/tests/integration/opamp_effective_config.py @@ -0,0 +1,68 @@ +import time + +from lib import project_path + +_SERVICE_NAME = "opamp-effective-config-test" + + +if __name__ == "__main__": + time.sleep(4) + + +class OpAMPEffectiveConfigOtelTest: + def __init__(self): + self.effective_config_seen = False + + def requirements(self): + return (project_path(),) + + def environment_variables(self): + 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, + "SPLUNK_OPAMP_ENABLED": "true", + "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", + "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_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 + + def on_stop(self, _telemetry, stdout: str, stderr: str, returncode: int): + assert returncode == 0, f"{stdout}\n{stderr}" + assert self.effective_config_seen + + def is_http(self): + return True diff --git a/tests/test_opamp.py b/tests/test_opamp.py new file mode 100644 index 00000000..6c2c25cc --- /dev/null +++ b/tests/test_opamp.py @@ -0,0 +1,429 @@ +# 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._opamp.proto import opamp_pb2 +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 ( + _build_client, + _sanitize_endpoint_for_reporting, + _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_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_start_opamp_uses_defaults_when_enabled(monkeypatch): + captured = {} + + 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_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 = [] + + 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 = {} + + 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 endpoint, attributes: (captured.update(endpoint=endpoint, attributes=attributes) or FakeClient()), + ) + monkeypatch.setattr( + "splunk_otel.opamp._start_agent", + lambda polling_interval_ms, report, client: captured.update( + polling_interval_ms=polling_interval_ms, + report=report, + client=client, + ), + ) + + start_opamp(resource) + + assert captured["endpoint"] == "http://host/opamp" + assert captured["polling_interval_ms"] == 2500 + 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( + 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( + 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_partitions_resource_attributes_and_preserves_types(): + client = _build_client( + "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", + "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_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", + "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_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( + 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"