Add timeouts to all api calls

Signed-off-by: Mathias L. Baumann <[email protected]>

Author: Marenz

RELEASE_NOTES.md

@@ -11,6 +11,9 @@
 ## New Features
 
 * `dispatch-cli` supports now the parameter `--type` and `--running` to filter the list of running services by type and status, respectively.
+* Every call now has a default timeout of 60 seconds, streams terminate after five minutes. This can be influenced by the two new parameters for`DispatchApiClient.__init__()`:
+    * `default_timeout: timedelta` (default: 60 seconds)
+    * `stream_timeout: timedelta` (default: 5 minutes)
 
 ## Bug Fixes
 

src/frequenz/client/dispatch/_client.py

@@ -49,6 +49,12 @@
 # pylint: enable=no-name-in-module
 DEFAULT_DISPATCH_PORT = 443
 
+DEFAULT_GRPC_CALL_TIMEOUT = timedelta(seconds=60)
+"""Default gRPC call timeout."""
+
+DEFAULT_GRPC_STREAM_TIMEOUT = timedelta(seconds=5)
+"""Default gRPC stream timeout."""
+
 
 class DispatchApiClient(BaseApiClient[dispatch_pb2_grpc.MicrogridDispatchServiceStub]):
     """Dispatch API client."""
@@ -59,13 +65,17 @@ def __init__(
         server_url: str,
         key: str,
         connect: bool = True,
+        call_timeout: timedelta | None = None,
+        stream_timeout: timedelta | None = None,
     ) -> None:
         """Initialize the client.
 
         Args:
             server_url: The URL of the server to connect to.
             key: API key to use for authentication.
             connect: Whether to connect to the service immediately.
+            call_timeout: Timeout for gRPC calls, default is 60 seconds.
+            stream_timeout: Timeout for gRPC streams, default is 5 minutes.
         """
         super().__init__(
             server_url,
@@ -82,6 +92,11 @@ def __init__(
         ] = {}
         """A dictionary of streamers, keyed by microgrid_id."""
 
+        self._call_timeout = (call_timeout or DEFAULT_GRPC_CALL_TIMEOUT).total_seconds()
+        self._stream_timeout = (
+            stream_timeout or DEFAULT_GRPC_STREAM_TIMEOUT
+        ).total_seconds()
+
     @property
     def stub(self) -> dispatch_pb2_grpc.MicrogridDispatchServiceAsyncStub:
         """The stub for the service."""
@@ -177,7 +192,9 @@ def to_interval(
         while True:
             response = await cast(
                 Awaitable[ListMicrogridDispatchesResponse],
-                self.stub.ListMicrogridDispatches(request, metadata=self._metadata),
+                self.stub.ListMicrogridDispatches(
+                    request, metadata=self._metadata, timeout=self._call_timeout
+                ),
             )
 
             yield (Dispatch.from_protobuf(dispatch) for dispatch in response.dispatches)
@@ -234,7 +251,9 @@ def _get_stream(
                 stream_method=lambda: cast(
                     AsyncIterator[StreamMicrogridDispatchesResponse],
                     self.stub.StreamMicrogridDispatches(
-                        request, metadata=self._metadata
+                        request,
+                        metadata=self._metadata,
+                        timeout=self._stream_timeout,
                     ),
                 ),
                 transform=DispatchEvent.from_protobuf,
@@ -303,7 +322,9 @@ async def create(  # pylint: disable=too-many-positional-arguments
         response = await cast(
             Awaitable[CreateMicrogridDispatchResponse],
             self.stub.CreateMicrogridDispatch(
-                request.to_protobuf(), metadata=self._metadata
+                request.to_protobuf(),
+                metadata=self._metadata,
+                timeout=self._call_timeout,
             ),
         )
 
@@ -394,7 +415,9 @@ async def update(
 
         response = await cast(
             Awaitable[UpdateMicrogridDispatchResponse],
-            self.stub.UpdateMicrogridDispatch(msg, metadata=self._metadata),
+            self.stub.UpdateMicrogridDispatch(
+                msg, metadata=self._metadata, timeout=self._call_timeout
+            ),
         )
 
         return Dispatch.from_protobuf(response.dispatch)
@@ -414,7 +437,9 @@ async def get(self, *, microgrid_id: int, dispatch_id: int) -> Dispatch:
         )
         response = await cast(
             Awaitable[GetMicrogridDispatchResponse],
-            self.stub.GetMicrogridDispatch(request, metadata=self._metadata),
+            self.stub.GetMicrogridDispatch(
+                request, metadata=self._metadata, timeout=self._call_timeout
+            ),
         )
         return Dispatch.from_protobuf(response.dispatch)
 
@@ -430,5 +455,7 @@ async def delete(self, *, microgrid_id: int, dispatch_id: int) -> None:
         )
         await cast(
             Awaitable[None],
-            self.stub.DeleteMicrogridDispatch(request, metadata=self._metadata),
+            self.stub.DeleteMicrogridDispatch(
+                request, metadata=self._metadata, timeout=self._call_timeout
+            ),
         )