Add initial implementation of OpAMP

Just reports effective config at startup, but adds wiring for eventual dynamic config updates and reporting.

Author: pmcollins

.github/workflows/ci-main.yml

@@ -27,6 +27,10 @@ jobs:
     steps:
     - uses: actions/checkout@v6
 
+    - name: Enable long paths (Windows)
+      if: runner.os == 'Windows'
+      run: git config --system core.longpaths true
+
     - name: Set up Python ${{ matrix.python-version }}
       uses: actions/setup-python@v6
       with:

pyproject.toml

@@ -33,6 +33,8 @@ dependencies = [
   "opentelemetry-instrumentation-system-metrics==0.60b1",
   "opentelemetry-semantic-conventions==0.60b1",
   "protobuf>=6.31.1", # not our direct dep, prevents installing vulnerable proto versions (CVE‑2025‑4565)
+  "opentelemetry-opamp-client @ git+https://github.com/open-telemetry/opentelemetry-python-contrib.git@88e5bfc630baa9a0789dd28694319afd506eae09#subdirectory=opamp/opentelemetry-opamp-client", # unreleased, pinned to merge commit of PR #3635
+  "requests>=2.20.0", # OpAMP: HTTP transport (already transitive dep, making explicit)
 ]
 
 [project.urls]
@@ -46,6 +48,9 @@ configurator = "splunk_otel.configurator:SplunkConfigurator"
 [project.entry-points.opentelemetry_distro]
 splunk_distro = "splunk_otel.distro:SplunkDistro"
 
+[tool.hatch.metadata]
+allow-direct-references = true
+
 [tool.hatch.version]
 path = "src/splunk_otel/__about__.py"
 

src/splunk_otel/callgraphs/__init__.py

@@ -8,10 +8,30 @@
     SPLUNK_SNAPSHOT_SAMPLING_INTERVAL,
 )
 
+_DEFAULT_SNAPSHOT_SAMPLING_INTERVAL = 10
 
-def _configure_callgraphs_if_enabled(env=None):
+
+class CallgraphsState:
+    """Runtime state of the Callgraph (aka snapshot) profiler, for OpAMP reporting."""
+
+    def __init__(self, processor: "CallgraphsSpanProcessor | None", interval: int):
+        self._processor = processor
+        self._interval = interval
+
+    def is_enabled(self) -> bool:
+        return self._processor is not None
+
+    def interval(self) -> int:
+        if self._processor is not None:
+            return self._processor.interval_millis()
+        return self._interval
+
+
+def _configure_callgraphs_if_enabled(env=None) -> CallgraphsState:
     env = env or Env()
+    interval = env.getint(SPLUNK_SNAPSHOT_SAMPLING_INTERVAL, _DEFAULT_SNAPSHOT_SAMPLING_INTERVAL)
     if env.is_true(SPLUNK_SNAPSHOT_PROFILER_ENABLED):
-        trace.get_tracer_provider().add_span_processor(
-            CallgraphsSpanProcessor(env.getval(OTEL_SERVICE_NAME), env.getint(SPLUNK_SNAPSHOT_SAMPLING_INTERVAL, 10))
-        )
+        processor = CallgraphsSpanProcessor(env.getval(OTEL_SERVICE_NAME), interval)
+        trace.get_tracer_provider().add_span_processor(processor)
+        return CallgraphsState(processor, interval)
+    return CallgraphsState(None, interval)

src/splunk_otel/callgraphs/span_processor.py

@@ -74,6 +74,9 @@ def on_end(self, span: ReadableSpan) -> None:
             if len(self._span_id_to_trace_id) == 0:
                 self._profiler.pause_after(60.0)
 
+    def interval_millis(self) -> int:
+        return int(self._profiler.interval_seconds * 1000)
+
     def shutdown(self) -> None:
         self._profiler.stop()
 

src/splunk_otel/configurator.py

@@ -12,14 +12,52 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
+from opentelemetry import trace
 from opentelemetry.sdk._configuration import _OTelSDKConfigurator
 
 from splunk_otel.profile import _start_profiling_if_enabled
 from splunk_otel.callgraphs import _configure_callgraphs_if_enabled
+from splunk_otel.opamp import _start_opamp_if_enabled
+from splunk_otel.opamp.config_registry import ConfigRegistry
+from splunk_otel.env import (
+    Env,
+    SPLUNK_PROFILER_ENABLED,
+    SPLUNK_PROFILER_CALL_STACK_INTERVAL,
+    SPLUNK_SNAPSHOT_PROFILER_ENABLED,
+    SPLUNK_SNAPSHOT_SAMPLING_INTERVAL,
+    OTEL_SERVICE_NAME,
+    OTEL_EXPORTER_OTLP_ENDPOINT,
+    OTEL_EXPORTER_OTLP_TRACES_ENDPOINT,
+    OTEL_EXPORTER_OTLP_METRICS_ENDPOINT,
+    OTEL_EXPORTER_OTLP_LOGS_ENDPOINT,
+)
 
 
 class SplunkConfigurator(_OTelSDKConfigurator):
     def _configure(self, **kwargs):
         super()._configure(**kwargs)
-        _start_profiling_if_enabled()
-        _configure_callgraphs_if_enabled()
+
+        env = Env()
+        profiling = _start_profiling_if_enabled(env)
+        callgraphs = _configure_callgraphs_if_enabled(env)
+
+        registry = self._build_registry(profiling, callgraphs, env)
+        resource = trace.get_tracer_provider().resource
+        _start_opamp_if_enabled(resource.attributes, registry, env)
+
+    def _build_registry(self, profiling_state, callgraphs_state, env) -> ConfigRegistry:
+        registry = ConfigRegistry()
+        registry.register(SPLUNK_PROFILER_ENABLED, getter=lambda: str(profiling_state.is_enabled()).lower())
+        registry.register(SPLUNK_PROFILER_CALL_STACK_INTERVAL, getter=lambda: str(profiling_state.interval_millis()))
+        registry.register(SPLUNK_SNAPSHOT_PROFILER_ENABLED, getter=lambda: str(callgraphs_state.is_enabled()).lower())
+        registry.register(SPLUNK_SNAPSHOT_SAMPLING_INTERVAL, getter=lambda: str(callgraphs_state.interval()))
+
+        for key in (
+            OTEL_SERVICE_NAME,
+            OTEL_EXPORTER_OTLP_ENDPOINT,
+            OTEL_EXPORTER_OTLP_TRACES_ENDPOINT,
+            OTEL_EXPORTER_OTLP_METRICS_ENDPOINT,
+            OTEL_EXPORTER_OTLP_LOGS_ENDPOINT,
+        ):
+            registry.register(key, getter=lambda k=key: env.getval(k) or None)
+        return registry

src/splunk_otel/env.py

@@ -55,6 +55,15 @@
 SPLUNK_SNAPSHOT_SAMPLING_INTERVAL = "SPLUNK_SNAPSHOT_SAMPLING_INTERVAL"
 SPLUNK_SNAPSHOT_SELECTION_PROBABILITY = "SPLUNK_SNAPSHOT_SELECTION_PROBABILITY"
 SPLUNK_REALM = "SPLUNK_REALM"
+SPLUNK_OPAMP_ENABLED = "SPLUNK_OPAMP_ENABLED"
+SPLUNK_OPAMP_ENDPOINT = "SPLUNK_OPAMP_ENDPOINT"
+SPLUNK_OPAMP_TOKEN = "SPLUNK_OPAMP_TOKEN"  # noqa: S105
+SPLUNK_OPAMP_POLLING_INTERVAL = "SPLUNK_OPAMP_POLLING_INTERVAL"
+OTEL_SERVICE_NAME = "OTEL_SERVICE_NAME"
+OTEL_EXPORTER_OTLP_ENDPOINT = "OTEL_EXPORTER_OTLP_ENDPOINT"
+OTEL_EXPORTER_OTLP_TRACES_ENDPOINT = "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"
+OTEL_EXPORTER_OTLP_METRICS_ENDPOINT = "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT"
+OTEL_EXPORTER_OTLP_LOGS_ENDPOINT = "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT"
 
 _pylogger = logging.getLogger(__name__)
 

src/splunk_otel/opamp/__init__.py

@@ -0,0 +1,142 @@
+# 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.sdk.resources import (
+    TELEMETRY_SDK_LANGUAGE,
+    TELEMETRY_SDK_NAME,
+    TELEMETRY_SDK_VERSION,
+)
+from opentelemetry._opamp.agent import OpAMPAgent
+from opentelemetry._opamp.client import OpAMPClient
+from opentelemetry._opamp.proto import opamp_pb2
+
+from splunk_otel.env import (
+    Env,
+    SPLUNK_ACCESS_TOKEN,
+    SPLUNK_OPAMP_ENABLED,
+    SPLUNK_OPAMP_ENDPOINT,
+    SPLUNK_OPAMP_POLLING_INTERVAL,
+    SPLUNK_OPAMP_TOKEN,
+)
+from splunk_otel.distro import _DISTRO_NAME
+from splunk_otel.opamp.config_registry import ConfigRegistry
+
+logger = logging.getLogger(__name__)
+
+_IDENTIFYING_RESOURCE_KEYS = (
+    "service.name",
+    "service.namespace",
+    "service.instance.id",
+    "service.version",
+)
+
+_NON_IDENTIFYING_RESOURCE_KEYS = (
+    "os.type",
+    "os.name",
+    "os.version",
+    "host.name",
+    "host.arch",
+    "process.pid",
+    "process.runtime.name",
+    "process.runtime.version",
+    # Note: deployment.environment.name may need to move: splunk-otel-java puts this in identifying
+    "deployment.environment.name",
+)
+
+_DEFAULT_POLLING_INTERVAL_MS = 30000
+
+
+def _start_opamp_if_enabled(resource_attrs, registry: ConfigRegistry, env: Env) -> None:
+    if not env.is_true(SPLUNK_OPAMP_ENABLED):
+        logger.debug("OpAMP disabled (SPLUNK_OPAMP_ENABLED not set to true)")
+        return
+
+    endpoint = env.getval(SPLUNK_OPAMP_ENDPOINT)
+    if not endpoint:
+        logger.warning("SPLUNK_OPAMP_ENABLED=true but SPLUNK_OPAMP_ENDPOINT is not set; OpAMP disabled")
+        return
+
+    token = env.getval(SPLUNK_OPAMP_TOKEN) or env.getval(SPLUNK_ACCESS_TOKEN)
+    polling_interval_ms = env.getint(SPLUNK_OPAMP_POLLING_INTERVAL, _DEFAULT_POLLING_INTERVAL_MS)
+
+    logger.info("Starting OpAMP client: %s", endpoint)
+
+    try:
+        _start_opamp(endpoint, token, polling_interval_ms, registry, resource_attrs)
+
+    except Exception:
+        logger.exception("Failed to start OpAMP client")
+
+
+def _start_opamp(endpoint: str, token: str, polling_interval_ms: int, registry: ConfigRegistry, resource_attrs):
+    headers = {}
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+
+    identifying_attrs, non_identifying_attrs = _build_agent_attributes(resource_attrs)
+    client = OpAMPClient(
+        endpoint=endpoint,
+        headers=headers,
+        agent_identifying_attributes=identifying_attrs,
+        agent_non_identifying_attributes=non_identifying_attrs,
+    )
+    client.update_effective_config(
+        {"": registry.get_all()},
+        content_type="application/json",
+    )
+    agent = OpAMPAgent(
+        interval=polling_interval_ms / 1000,
+        message_handler=_handle_server_message,
+        client=client,
+    )
+    agent.start()
+    logger.info("OpAMP client started")
+
+
+def _build_agent_attributes(resource_attrs) -> tuple[dict, dict]:
+    from splunk_otel.__about__ import __version__ as distro_version
+
+    identifying_attrs = {}
+    for key in _IDENTIFYING_RESOURCE_KEYS:
+        val = resource_attrs.get(key)
+        if val is not None:
+            identifying_attrs[key] = str(val)
+
+    identifying_attrs.update(
+        {
+            TELEMETRY_SDK_LANGUAGE: "python",
+            TELEMETRY_SDK_NAME: "opentelemetry",
+            TELEMETRY_SDK_VERSION: str(resource_attrs.get(TELEMETRY_SDK_VERSION, "unknown")),
+            "telemetry.distro.name": _DISTRO_NAME,
+            "telemetry.distro.version": distro_version,
+        }
+    )
+
+    non_identifying_attrs = {}
+    for key in _NON_IDENTIFYING_RESOURCE_KEYS:
+        val = resource_attrs.get(key)
+        if val is not None:
+            non_identifying_attrs[key] = str(val)
+
+    return identifying_attrs, non_identifying_attrs
+
+
+def _handle_server_message(
+    _agent: OpAMPAgent,
+    _client: OpAMPClient,
+    message: opamp_pb2.ServerToAgent,
+) -> None:
+    logger.debug("ServerToAgent: flags=%s", message.flags)

src/splunk_otel/opamp/config_registry.py

@@ -0,0 +1,99 @@
+# 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.
+
+"""
+Config registry for OpAMP EffectiveConfig reporting (and eventually remote updates).
+
+Feature modules register the keys they own at startup. OpAMP reads all registered
+values when building EffectiveConfig, and will call write callbacks for remote
+config updates in Phase 2.
+
+Usage:
+    registry = ConfigRegistry()
+
+    # in profile.py
+    registry.register("SPLUNK_PROFILER_ENABLED", getter=lambda: str(ctx.running))
+
+    # in opamp/__init__.py
+    registry.get_all()  # -> {"SPLUNK_PROFILER_ENABLED": "true", ...}
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import Callable
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class _ConfigEntry:
+    getter: Callable[[], str | None]
+    setter: Callable[[str], None] | None = None
+
+
+class ConfigRegistry:
+    def __init__(self):
+        self._entries: dict[str, _ConfigEntry] = {}
+
+    def register(
+        self,
+        key: str,
+        *,
+        getter: Callable[[], str],
+        setter: Callable[[str], None] | None = None,
+    ) -> None:
+        """
+        Register a config key with read (and optionally write) callbacks.
+
+        Args:
+            key: The config key name (e.g. "SPLUNK_PROFILER_ENABLED")
+            getter: Returns the current string value of this key.
+            setter: Applies an updated string value. None means read-only (restart required).
+        """
+        self._entries[key] = _ConfigEntry(getter=getter, setter=setter)
+
+    def get_all(self) -> dict[str, str]:
+        """Return current values for all registered keys, omitting keys whose getter returns None."""
+        out = {}
+        for key, entry in self._entries.items():
+            try:
+                val = entry.getter()
+                if val is not None:
+                    out[key] = val
+            except Exception:  # noqa: BLE001
+                logger.warning("Failed to read config key %s", key, exc_info=True)
+        return out
+
+    def update(self, updates: dict[str, str]) -> list[str]:
+        """
+        Update registry with a dict of keys to values.
+        """
+        updated_keys = []
+        for key, value in updates.items():
+            entry = self._entries.get(key)
+            if entry is None:
+                logger.warning("Ignoring unknown config key: %s", key)
+                continue
+            if entry.setter is None:
+                logger.warning("Ignoring read-only config key: %s", key)
+                continue
+            try:
+                entry.setter(value)
+                updated_keys.append(key)
+                logger.info("Applied config update: %s=%s", key, value)
+            except Exception:
+                logger.exception("Failed to apply config key %s=%s", key, value)
+        return updated_keys