opamp: ack remote config status changes (#340)

* opamp: ack remote config status changes

Start reporting ReportsRemoteConfig agent capability and sending an
ack to the server when we update our config based on the remote one.
This rework the opamp handler signature to also take the agent as we
need to enqueue a remote config status message to ack the updated
configuration.

Author: xrmx

src/elasticotel/distro/config.py

@@ -17,6 +17,7 @@
 import logging
 
 from opentelemetry._opamp import messages
+from opentelemetry._opamp.agent import OpAMPAgent
 from opentelemetry._opamp.client import OpAMPClient
 from opentelemetry._opamp.proto import opamp_pb2 as opamp_pb2
 
@@ -34,10 +35,12 @@
 }
 
 
-def opamp_handler(client: OpAMPClient, message: opamp_pb2.ServerToAgent):
-    if not message.remote_config:
+def opamp_handler(agent: OpAMPAgent, client: OpAMPClient, message: opamp_pb2.ServerToAgent):
+    # we check config_hash because we need to track last received config and remote_config seems to be always truthy
+    if not message.remote_config or not message.remote_config.config_hash:
         return
 
+    error_message = ""
     for config_filename, config in messages._decode_remote_config(message.remote_config):
         # we don't have standardized config values so limit to configs coming from our backend
         if config_filename == "elastic":
@@ -51,7 +54,17 @@ def opamp_handler(client: OpAMPClient, message: opamp_pb2.ServerToAgent):
 
             if logging_level is None:
                 logger.warning("Logging level not handled: %s", config_logging_level)
+                error_message = f"Logging level not handled: {config_logging_level}"
             else:
                 # update upstream and distro logging levels
                 logging.getLogger("opentelemetry").setLevel(logging_level)
                 logging.getLogger("elasticotel").setLevel(logging_level)
+
+    status = opamp_pb2.RemoteConfigStatuses_FAILED if error_message else opamp_pb2.RemoteConfigStatuses_APPLIED
+    updated_remote_config = client._update_remote_config_status(
+        remote_config_hash=message.remote_config.config_hash, status=status, error_message=error_message
+    )
+    # if we changed the config send an ack to the server so we don't receive the same config at every heartbeat response
+    if updated_remote_config is not None:
+        payload = client._build_remote_config_status_response_message(updated_remote_config)
+        agent.send(payload=payload)

src/opentelemetry/_opamp/agent.py

@@ -71,7 +71,7 @@ def __init__(
         self,
         *,
         interval: float,
-        message_handler: Callable[[OpAMPClient, opamp_pb2.ServerToAgent], None],
+        message_handler: Callable[["OpAMPAgent", OpAMPClient, opamp_pb2.ServerToAgent], None],
         max_retries: int = 10,
         heartbeat_max_retries: int = 1,
         initial_backoff: float = 1.0,
@@ -178,10 +178,10 @@ def _run_worker(self) -> None:
                         logger.debug("Stop signaled, abandoning job %r", job.payload)
                         break
 
-            # we can't do much if the handler fails other than logging
             if message is not None:
+                # we can't do much if the handler fails other than logging
                 try:
-                    self._handler(self._client, message)
+                    self._handler(self, self._client, message)
                     logger.debug("Called Job message handler for: %r", message)
                 except Exception as exc:
                     logger.warning("Job %r handler failed with: %s", job.payload, exc)

src/opentelemetry/_opamp/client.py

@@ -41,6 +41,7 @@
     opamp_pb2.AgentCapabilities.AgentCapabilities_ReportsStatus
     | opamp_pb2.AgentCapabilities.AgentCapabilities_ReportsHeartbeat
     | opamp_pb2.AgentCapabilities.AgentCapabilities_AcceptsRemoteConfig
+    | opamp_pb2.AgentCapabilities.AgentCapabilities_ReportsRemoteConfig
 )
 
 
@@ -67,6 +68,7 @@ def __init__(
         )
         self._sequence_num: int = 0
         self._instance_uid: bytes = uuid7().bytes
+        self._remote_config_status: opamp_pb2.RemoteConfigStatus | None = None
 
     def _build_connection_message(self) -> bytes:
         message = messages._build_presentation_message(
@@ -94,6 +96,37 @@ def _build_heartbeat_message(self) -> bytes:
         data = messages._encode_message(message)
         return data
 
+    def _update_remote_config_status(
+        self, remote_config_hash: bytes, status: opamp_pb2.RemoteConfigStatuses.ValueType, error_message: str = ""
+    ) -> opamp_pb2.RemoteConfigStatus | None:
+        status_changed = (
+            not self._remote_config_status
+            or self._remote_config_status.last_remote_config_hash != remote_config_hash
+            or self._remote_config_status.status != status
+            or self._remote_config_status.error_message != error_message
+        )
+        # if the status changed update we return the RemoteConfigStatus message so that we can send it to the server
+        if status_changed:
+            _logger.debug("Update remote config status changed for %s", remote_config_hash)
+            self._remote_config_status = messages._build_remote_config_status_message(
+                last_remote_config_hash=remote_config_hash,
+                status=status,
+                error_message=error_message,
+            )
+            return self._remote_config_status
+        else:
+            return None
+
+    def _build_remote_config_status_response_message(self, remote_config_status: opamp_pb2.RemoteConfigStatus) -> bytes:
+        message = messages._build_remote_config_status_response_message(
+            instance_uid=self._instance_uid,
+            sequence_num=self._sequence_num,
+            capabilities=_HANDLED_CAPABILITIES,
+            remote_config_status=remote_config_status,
+        )
+        data = messages._encode_message(message)
+        return data
+
     def _send(self, data: bytes):
         try:
             response = self._transport.send(

src/opentelemetry/_opamp/messages.py

@@ -93,6 +93,28 @@ def _build_agent_disconnect_message(
     return command
 
 
+def _build_remote_config_status_message(
+    last_remote_config_hash: bytes, status: opamp_pb2.RemoteConfigStatuses.ValueType, error_message: str = ""
+) -> opamp_pb2.RemoteConfigStatus:
+    return opamp_pb2.RemoteConfigStatus(
+        last_remote_config_hash=last_remote_config_hash,
+        status=status,
+        error_message=error_message,
+    )
+
+
+def _build_remote_config_status_response_message(
+    instance_uid: bytes, sequence_num: int, capabilities: int, remote_config_status: opamp_pb2.RemoteConfigStatus
+) -> opamp_pb2.AgentToServer:
+    command = opamp_pb2.AgentToServer(
+        instance_uid=instance_uid,
+        sequence_num=sequence_num,
+        remote_config_status=remote_config_status,
+        capabilities=capabilities,
+    )
+    return command
+
+
 def _encode_message(data: opamp_pb2.AgentToServer) -> bytes:
     return data.SerializeToString()
 

tests/distro/test_distro.py

@@ -217,48 +217,68 @@ def test_configurator_ignores_opamp_without_endpoint(self, client_mock, agent_mo
 
 class TestOpAMPHandler(TestCase):
     @mock.patch.object(logging, "getLogger")
-    def test_does_not_nothing_without_remote_config(self, get_logger_mock):
+    def test_does_nothing_without_remote_config(self, get_logger_mock):
         message = opamp_pb2.ServerToAgent()
+        agent = mock.Mock()
         client = mock.Mock()
-        opamp_handler(client, message)
+        opamp_handler(agent, client, message)
 
         get_logger_mock.assert_not_called()
 
     @mock.patch.object(logging, "getLogger")
     def test_ignores_non_elastic_filename(self, get_logger_mock):
+        agent = mock.Mock()
         client = mock.Mock()
         config = opamp_pb2.AgentConfigMap()
         config.config_map["non-elastic"].body = json.dumps({"logging_level": "trace"}).encode()
         config.config_map["non-elastic"].content_type = "application/json"
-        remote_config = opamp_pb2.AgentRemoteConfig(config=config)
+        remote_config = opamp_pb2.AgentRemoteConfig(config=config, config_hash=b"1234")
         message = opamp_pb2.ServerToAgent(remote_config=remote_config)
-        opamp_handler(client, message)
+        opamp_handler(agent, client, message)
 
         get_logger_mock.assert_not_called()
 
+        client._update_remote_config_status.assert_called_once_with(
+            remote_config_hash=b"1234", status=opamp_pb2.RemoteConfigStatuses_APPLIED, error_message=""
+        )
+        client._build_remote_config_status_response_message.assert_called_once_with(
+            client._update_remote_config_status()
+        )
+        agent.send.assert_called_once_with(payload=mock.ANY)
+
     @mock.patch.object(logging, "getLogger")
     def test_sets_matching_logging_level(self, get_logger_mock):
+        agent = mock.Mock()
         client = mock.Mock()
         config = opamp_pb2.AgentConfigMap()
         config.config_map["elastic"].body = json.dumps({"logging_level": "trace"}).encode()
         config.config_map["elastic"].content_type = "application/json"
-        remote_config = opamp_pb2.AgentRemoteConfig(config=config)
+        remote_config = opamp_pb2.AgentRemoteConfig(config=config, config_hash=b"1234")
         message = opamp_pb2.ServerToAgent(remote_config=remote_config)
-        opamp_handler(client, message)
+        opamp_handler(agent, client, message)
 
         get_logger_mock.assert_has_calls(
             [mock.call("opentelemetry"), mock.call().setLevel(5), mock.call("elasticotel"), mock.call().setLevel(5)]
         )
 
+        client._update_remote_config_status.assert_called_once_with(
+            remote_config_hash=b"1234", status=opamp_pb2.RemoteConfigStatuses_APPLIED, error_message=""
+        )
+        client._build_remote_config_status_response_message.assert_called_once_with(
+            client._update_remote_config_status()
+        )
+        agent.send.assert_called_once_with(payload=mock.ANY)
+
     @mock.patch.object(logging, "getLogger")
     def test_sets_logging_to_default_info_without_logging_level_entry_in_config(self, get_logger_mock):
+        agent = mock.Mock()
         client = mock.Mock()
         config = opamp_pb2.AgentConfigMap()
         config.config_map["elastic"].body = json.dumps({}).encode()
         config.config_map["elastic"].content_type = "application/json"
-        remote_config = opamp_pb2.AgentRemoteConfig(config=config)
+        remote_config = opamp_pb2.AgentRemoteConfig(config=config, config_hash=b"1234")
         message = opamp_pb2.ServerToAgent(remote_config=remote_config)
-        opamp_handler(client, message)
+        opamp_handler(agent, client, message)
 
         get_logger_mock.assert_has_calls(
             [
@@ -269,14 +289,28 @@ def test_sets_logging_to_default_info_without_logging_level_entry_in_config(self
             ]
         )
 
+        client._update_remote_config_status.assert_called_once_with(
+            remote_config_hash=b"1234", status=opamp_pb2.RemoteConfigStatuses_APPLIED, error_message=""
+        )
+        client._build_remote_config_status_response_message.assert_called_once_with(
+            client._update_remote_config_status()
+        )
+        agent.send.assert_called_once_with(payload=mock.ANY)
+
     @mock.patch.object(logging, "getLogger")
     def test_warns_if_logging_level_does_not_match_our_map(self, get_logger_mock):
+        agent = mock.Mock()
         client = mock.Mock()
         config = opamp_pb2.AgentConfigMap()
         config.config_map["elastic"].body = json.dumps({"logging_level": "unexpected"}).encode()
         config.config_map["elastic"].content_type = "application/json"
-        remote_config = opamp_pb2.AgentRemoteConfig(config=config)
+        remote_config = opamp_pb2.AgentRemoteConfig(config=config, config_hash=b"1234")
         message = opamp_pb2.ServerToAgent(remote_config=remote_config)
 
         with self.assertLogs(config_logger, logging.WARNING):
-            opamp_handler(client, message)
+            opamp_handler(agent, client, message)
+
+        client._build_remote_config_status_response_message.assert_called_once_with(
+            client._update_remote_config_status()
+        )
+        agent.send.assert_called_once_with(payload=mock.ANY)

tests/opamp/cassettes/test_connection_heartbeat_disconnection.yaml renamed to tests/opamp/cassettes/test_connection_remote_config_status_heartbeat_disconnection.yaml

@@ -1,8 +1,8 @@
 interactions:
 - request:
     body: !!binary |
-      ChABl44aq8F1QqCWeJJQNmiKGj0KFQoMc2VydmljZS5uYW1lEgUKA2ZvbwokChtkZXBsb3ltZW50
-      LmVudmlyb25tZW50Lm5hbWUSBQoDZm9vIINA
+      ChABl89ktxVxE7d60nbzJJzzGj0KFQoMc2VydmljZS5uYW1lEgUKA2ZvbwokChtkZXBsb3ltZW50
+      LmVudmlyb25tZW50Lm5hbWUSBQoDZm9vIINg
     headers:
       Accept:
       - '*/*'
@@ -21,22 +21,23 @@ interactions:
   response:
     body:
       string: !!binary |
-        ChABl44aq8F1QqCWeJJQNmiKGmYKOgo4CgdlbGFzdGljEi0KGXsibG9nZ2luZ19sZXZlbCI6ImRl
-        YnVnIn0SEGFwcGxpY2F0aW9uL2pzb24SKGJmOThhY2M2NDNhMzJjNTcyZWFhYmVhNGIwZWU5ZTEw
-        NmFkYWRiZDY4AlIA
+        ChABl89ktxVxE7d60nbzJJzzGmYKOgo4CgdlbGFzdGljEi0KGXsibG9nZ2luZ19sZXZlbCI6ImRl
+        YnVnIn0SEGFwcGxpY2F0aW9uL2pzb24SKGY3MzA5M2VjZDEyNjkzZGMxNDUxYWQ2MjdlZDA2MWJl
+        ZWM5ZjU1OWM4AlIA
     headers:
       Content-Length:
       - '126'
       Content-Type:
       - application/x-protobuf
       Date:
-      - Fri, 20 Jun 2025 16:10:02 GMT
+      - Thu, 03 Jul 2025 08:26:13 GMT
     status:
       code: 200
       message: OK
 - request:
     body: !!binary |
-      ChABl44aq8F1QqCWeJJQNmiKEAE=
+      ChABl89ktxVxE7d60nbzJJzzEAEgg2A6LAooZjczMDkzZWNkMTI2OTNkYzE0NTFhZDYyN2VkMDYx
+      YmVlYzlmNTU5YxAB
     headers:
       Accept:
       - '*/*'
@@ -45,7 +46,7 @@ interactions:
       Connection:
       - keep-alive
       Content-Length:
-      - '20'
+      - '69'
       Content-Type:
       - application/x-protobuf
       User-Agent:
@@ -55,48 +56,78 @@ interactions:
   response:
     body:
       string: !!binary |
-        ChABl44aq8F1QqCWeJJQNmiKGmYKOgo4CgdlbGFzdGljEi0KGXsibG9nZ2luZ19sZXZlbCI6ImRl
-        YnVnIn0SEGFwcGxpY2F0aW9uL2pzb24SKGJmOThhY2M2NDNhMzJjNTcyZWFhYmVhNGIwZWU5ZTEw
-        NmFkYWRiZDY4AlIA
+        ChABl89ktxVxE7d60nbzJJzzOAJSAA==
     headers:
       Content-Length:
-      - '126'
+      - '22'
       Content-Type:
       - application/x-protobuf
       Date:
-      - Fri, 20 Jun 2025 16:10:03 GMT
+      - Thu, 03 Jul 2025 08:26:13 GMT
     status:
       code: 200
       message: OK
 - request:
     body: !!binary |
-      ChABl44aq8F1QqCWeJJQNmiKEAJKAA==
+      ChABl89ktxVxE7d60nbzJJzzEAIgg2A=
     headers:
       Accept:
       - '*/*'
       Accept-Encoding:
       - gzip, deflate
       Connection:
       - keep-alive
+      Content-Length:
+      - '23'
+      Content-Type:
+      - application/x-protobuf
+      User-Agent:
+      - OTel-OpAMP-Python/0.0.1
+    method: POST
+    uri: http://localhost:4320/v1/opamp
+  response:
+    body:
+      string: !!binary |
+        ChABl89ktxVxE7d60nbzJJzzOAJSAA==
+    headers:
       Content-Length:
       - '22'
       Content-Type:
       - application/x-protobuf
+      Date:
+      - Thu, 03 Jul 2025 08:26:14 GMT
+    status:
+      code: 200
+      message: OK
+- request:
+    body: !!binary |
+      ChABl89ktxVxE7d60nbzJJzzEAMgg2BKAA==
+    headers:
+      Accept:
+      - '*/*'
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      Content-Length:
+      - '25'
+      Content-Type:
+      - application/x-protobuf
       User-Agent:
       - OTel-OpAMP-Python/0.0.1
     method: POST
     uri: http://localhost:4320/v1/opamp
   response:
     body:
       string: !!binary |
-        ChABl44aq8F1QqCWeJJQNmiKOAJSAA==
+        ChABl89ktxVxE7d60nbzJJzzOAJSAA==
     headers:
       Content-Length:
       - '22'
       Content-Type:
       - application/x-protobuf
       Date:
-      - Fri, 20 Jun 2025 16:10:03 GMT
+      - Thu, 03 Jul 2025 08:26:15 GMT
     status:
       code: 200
       message: OK