Update OpAMP client to have additional callback methods (open-telemetry#4322)

* Update callbacks to use format similar to existing java and go clients

* Add changelog entry

* Update Callbacks to use ABC, add MessageData class and ReportFullState flag handling

* Fix docs build

* Check error_response before processing other ServerToAgent fields per OpAMP spec

---------

Co-authored-by: Riccardo Magliocchetti <riccardo.magliocchetti@gmail.com>

Author: kelseyma

docs/conf.py

@@ -160,6 +160,10 @@
         "py:class",
         "opamp_pb2.EffectiveConfig",
     ),
+    (
+        "py:class",
+        "opamp_pb2.AgentRemoteConfig",
+    ),
 ]
 
 cfg = ConfigParser()

opamp/opentelemetry-opamp-client/CHANGELOG.md

@@ -9,3 +9,5 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Initial implementation
   ([#3635](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3635))
+- Update client to have additional callback methods
+  ([#4322](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4322))

opamp/opentelemetry-opamp-client/src/opentelemetry/_opamp/__init__.py

@@ -35,7 +35,7 @@
 
 Since OpAMP APIs, config options or environment variables are not standardizes the distros are required
 to provide code doing so.
-OTel Python distros would need to provide their own message handler callback that implements the actual
+OTel Python distros would need to provide their own Callbacks subclass that implements the actual
 change of whatever configuration their backends sends.
 
 Please note that the API is not finalized yet and so the name is called ``_opamp`` with the underscore.
@@ -48,15 +48,18 @@
     import os
 
     from opentelemetry._opamp.agent import OpAMPAgent
+    from opentelemetry._opamp.callbacks import Callbacks
     from opentelemetry._opamp.client import OpAMPClient
-    from opentelemetry._opamp.proto import opamp_pb2 as opamp_pb2
     from opentelemetry.sdk._configuration import _OTelSDKConfigurator
     from opentelemetry.sdk.resources import OTELResourceDetector
 
 
-    def opamp_handler(agent: OpAMPAgent, client: OpAMPClient, message: opamp_pb2.ServerToAgent):
-        for config_filename, config in message.remote_config.config.config_map.items():
-            print("do something")
+    class MyCallbacks(Callbacks):
+        def on_message(self, agent, client, message):
+            if message.remote_config is None:
+                return
+            for config_filename, config in message.remote_config.config.config_map.items():
+                print("do something")
 
 
     class MyOpenTelemetryConfigurator(_OTelSDKConfigurator):
@@ -79,7 +82,7 @@ def _configure(self, **kwargs):
                 )
                 opamp_agent = OpAMPAgent(
                     interval=30,
-                    message_handler=opamp_handler,
+                    callbacks=MyCallbacks(),
                     client=opamp_client,
                 )
                 opamp_agent.start()
@@ -90,6 +93,7 @@ def _configure(self, **kwargs):
 """
 
 from opentelemetry._opamp.agent import OpAMPAgent
+from opentelemetry._opamp.callbacks import Callbacks, MessageData
 from opentelemetry._opamp.client import OpAMPClient
 
-__all__ = ["OpAMPAgent", "OpAMPClient"]
+__all__ = ["Callbacks", "MessageData", "OpAMPAgent", "OpAMPClient"]

opamp/opentelemetry-opamp-client/src/opentelemetry/_opamp/agent.py

@@ -21,12 +21,24 @@
 import threading
 from typing import Any, Callable
 
+from opentelemetry._opamp.callbacks import Callbacks, MessageData
 from opentelemetry._opamp.client import OpAMPClient
 from opentelemetry._opamp.proto import opamp_pb2
 
 logger = logging.getLogger(__name__)
 
 
+def _safe_invoke(function: Callable[..., Any], *args: Any) -> None:
+    function_name = "<unknown>"
+    try:
+        function_name = function.__name__
+        function(*args)
+    except Exception as exc:  # pylint: disable=broad-exception-caught
+        logger.error(
+            "Error when invoking function '%s'", function_name, exc_info=exc
+        )
+
+
 class _Job:
     """
     Represents a single request job, with retry/backoff metadata.
@@ -73,24 +85,22 @@ def __init__(
         self,
         *,
         interval: float = 30,
-        message_handler: Callable[
-            ["OpAMPAgent", OpAMPClient, opamp_pb2.ServerToAgent], None
-        ],
+        callbacks: Callbacks,
         max_retries: int = 10,
         heartbeat_max_retries: int = 1,
         initial_backoff: float = 1.0,
         client: OpAMPClient,
     ):
         """
         :param interval: seconds between heartbeat calls
-        :param message_handler: user provided function that takes the received ServerToAgent message
+        :param callbacks: Callbacks instance for receiving client events
         :param max_retries: how many times to retry a failed job for ad-hoc messages
         :param heartbeat_max_retries: how many times to retry an heartbeat failed job
         :param initial_backoff: base seconds for exponential backoff
         :param client: an OpAMPClient instance
         """
         self._interval = interval
-        self._handler = message_handler
+        self._callbacks = callbacks
         self._max_retries = max_retries
         self._heartbeat_max_retries = heartbeat_max_retries
         self._initial_backoff = initial_backoff
@@ -186,23 +196,31 @@ def _run_worker(self) -> None:
             while job.should_retry() and not self._stop.is_set():
                 try:
                     message = self._client.send(job.payload)
+                    _safe_invoke(
+                        self._callbacks.on_connect, self, self._client
+                    )
                     logger.debug("Job succeeded: %r", job.payload)
                     break
                 except Exception as exc:
                     job.attempt += 1
+                    _safe_invoke(
+                        self._callbacks.on_connect_failed,
+                        self,
+                        self._client,
+                        exc,
+                    )
                     logger.warning(
                         "Job %r failed attempt %d/%d: %s",
                         job.payload,
                         job.attempt,
-                        job.max_retries,
+                        job.max_retries + 1,
                         exc,
                     )
 
                     if not job.should_retry():
                         logger.error(
                             "Job %r dropped after max retries", job.payload
                         )
-                        logger.exception(exc)
                         break
 
                     # exponential backoff with +/- 20% jitter, interruptible by stop event
@@ -216,14 +234,7 @@ def _run_worker(self) -> None:
                         break
 
             if message is not None:
-                # we can't do much if the handler fails other than logging
-                try:
-                    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
-                    )
+                self._process_message(message)
 
             try:
                 if job.callback is not None:
@@ -233,6 +244,29 @@ def _run_worker(self) -> None:
             finally:
                 self._queue.task_done()
 
+    def _process_message(self, message: opamp_pb2.ServerToAgent) -> None:
+        if message.HasField("error_response"):
+            _safe_invoke(
+                self._callbacks.on_error,
+                self,
+                self._client,
+                message.error_response,
+            )
+            return
+
+        if message.flags & opamp_pb2.ServerToAgentFlags_ReportFullState:
+            logger.debug("Server requested full state report")
+            payload = self._client.build_full_state_message()
+            self.send(payload)
+
+        msg_data = MessageData.from_server_message(message)
+        _safe_invoke(
+            self._callbacks.on_message,
+            self,
+            self._client,
+            msg_data,
+        )
+
     def stop(self, timeout: float | None = None) -> None:
         """
         Signal server we are disconnecting and then threads to exit

opamp/opentelemetry-opamp-client/src/opentelemetry/_opamp/callbacks.py

@@ -0,0 +1,94 @@
+# Copyright The OpenTelemetry Authors
+#
+# 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
+
+from abc import ABC
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from opentelemetry._opamp.proto import opamp_pb2
+
+if TYPE_CHECKING:
+    from opentelemetry._opamp.agent import OpAMPAgent
+    from opentelemetry._opamp.client import OpAMPClient
+
+
+@dataclass
+class MessageData:
+    """Structured view of a ServerToAgent message for callback consumption.
+
+    Only fields the agent is expected to act on are exposed. Flags and
+    error_response are handled internally by the client before this
+    object reaches the callback.
+    """
+
+    remote_config: opamp_pb2.AgentRemoteConfig | None = None
+
+    @classmethod
+    def from_server_message(
+        cls, message: opamp_pb2.ServerToAgent
+    ) -> MessageData:
+        return cls(
+            remote_config=message.remote_config
+            if message.HasField("remote_config")
+            else None,
+        )
+
+
+class Callbacks(ABC):
+    """OpAMP client callbacks with no-op defaults.
+
+    All methods have no-op defaults so that subclasses only need to
+    override the callbacks they care about. New callbacks can be added
+    in the future without breaking existing subclasses.
+    """
+
+    def on_connect(self, agent: OpAMPAgent, client: OpAMPClient) -> None:
+        """Called when the connection is successfully established to the
+        Server. For HTTP clients this is called for any request if the
+        response status is OK.
+        """
+
+    def on_connect_failed(
+        self,
+        agent: OpAMPAgent,
+        client: OpAMPClient,
+        error: Exception,
+    ) -> None:
+        """Called when the connection to the Server cannot be established.
+        May also be called if the connection is lost and reconnection
+        attempt fails.
+        """
+
+    def on_error(
+        self,
+        agent: OpAMPAgent,
+        client: OpAMPClient,
+        error_response: opamp_pb2.ServerErrorResponse,
+    ) -> None:
+        """Called when the Server reports an error in response to a
+        previously sent request. Useful for logging purposes. The Agent
+        should not attempt to process the error by reconnecting or
+        retrying previous operations. The client handles the UNAVAILABLE
+        case internally by performing retries as necessary.
+        """
+
+    def on_message(
+        self,
+        agent: OpAMPAgent,
+        client: OpAMPClient,
+        message: MessageData,
+    ) -> None:
+        """Called when the Agent receives a message that needs processing."""

opamp/opentelemetry-opamp-client/src/opentelemetry/_opamp/transport/requests.py

@@ -59,8 +59,7 @@ def send(
             )
             response.raise_for_status()
         except Exception as exc:
-            logger.error(str(exc))
-            raise OpAMPException
+            raise OpAMPException(str(exc)) from exc
 
         message = messages.decode_message(response.content)
 

opamp/opentelemetry-opamp-client/tests/opamp/cassettes/test_connection_remote_config_status_heartbeat_disconnection.yaml

@@ -56,10 +56,10 @@ interactions:
     body:
       string: !!binary |
         ChABnL5b5k5046/FazbAJ7r4GioKBgoECgASABIg47DEQpj8HBSa+/TImW+5JCeuQeRkm5NMpJWZ
-        G3hSuFUwAVIA
+        G3hSuFVSAA==
     headers:
       Content-Length:
-      - '66'
+      - '64'
       Content-Type:
       - application/x-protobuf
       Date:
@@ -89,10 +89,10 @@ interactions:
     body:
       string: !!binary |
         ChABnL5b5k5046/FazbAJ7r4GioKBgoECgASABIg47DEQpj8HBSa+/TImW+5JCeuQeRkm5NMpJWZ
-        G3hSuFUwAVIA
+        G3hSuFVSAA==
     headers:
       Content-Length:
-      - '66'
+      - '64'
       Content-Type:
       - application/x-protobuf
       Date:
@@ -122,10 +122,10 @@ interactions:
     body:
       string: !!binary |
         ChABnL5b5k5046/FazbAJ7r4GioKBgoECgASABIg47DEQpj8HBSa+/TImW+5JCeuQeRkm5NMpJWZ
-        G3hSuFUwAVIA
+        G3hSuFVSAA==
     headers:
       Content-Length:
-      - '66'
+      - '64'
       Content-Type:
       - application/x-protobuf
       Date: