From 120d33392b1d3457e3bf8350875da2419b00fa1e Mon Sep 17 00:00:00 2001 From: Tomuta Gabriel Date: Fri, 26 Sep 2025 16:16:16 +0300 Subject: [PATCH 1/8] Events initial implementation --- ankaios_sdk/_components/__init__.py | 1 + ankaios_sdk/_components/control_interface.py | 12 +- ankaios_sdk/_components/event_campaign.py | 100 ++++++++++++++++ ankaios_sdk/_components/request.py | 51 +++++++++ ankaios_sdk/_components/response.py | 72 +++++++++++- ankaios_sdk/ankaios.py | 113 +++++++++++++++++++ examples/apps/state_event.py | 85 ++++++++++++++ tests/response/test_response.py | 14 ++- tests/test_control_interface.py | 34 +++++- 9 files changed, 463 insertions(+), 19 deletions(-) create mode 100644 ankaios_sdk/_components/event_campaign.py create mode 100644 examples/apps/state_event.py diff --git a/ankaios_sdk/_components/__init__.py b/ankaios_sdk/_components/__init__.py index 7d5ddfee..dfc71b7e 100644 --- a/ankaios_sdk/_components/__init__.py +++ b/ankaios_sdk/_components/__init__.py @@ -43,6 +43,7 @@ from .response import * from .manifest import * from .log_campaign import * +from .event_campaign import * from .control_interface import * from .file import * diff --git a/ankaios_sdk/_components/control_interface.py b/ankaios_sdk/_components/control_interface.py index bb1ac8b7..353b8691 100644 --- a/ankaios_sdk/_components/control_interface.py +++ b/ankaios_sdk/_components/control_interface.py @@ -101,7 +101,7 @@ class ControlInterface: "(str): The base path for the Ankaios control interface." def __init__( - self, add_response_callback: Callable, add_log_callback: Callable + self, add_response_callback: Callable, add_log_callback: Callable, add_event_callback: Callable ) -> None: """ Initialize the ControlInterface object. This is used @@ -124,6 +124,7 @@ def __init__( self._add_response_callback = add_response_callback self._add_log_callback = add_log_callback + self._add_event_callback = add_event_callback self._logger = get_logger() @@ -397,6 +398,15 @@ def _handle_response(self, response: Response) -> None: ) return + # Filter out the events + if response.content_type in [ + ResponseType.EVENT_RESPONSE, + ]: + self._add_event_callback( + response.get_request_id(), response.content + ) + return + # Send out the response to the Ankaios class self._add_response_callback(response) diff --git a/ankaios_sdk/_components/event_campaign.py b/ankaios_sdk/_components/event_campaign.py new file mode 100644 index 00000000..d1af38ee --- /dev/null +++ b/ankaios_sdk/_components/event_campaign.py @@ -0,0 +1,100 @@ +# Copyright (c) 2025 Elektrobit Automotive GmbH +# +# This program and the accompanying materials are made available under the +# terms of the Apache License, Version 2.0 which is available at +# https://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. +# +# SPDX-License-Identifier: Apache-2.0 + + +""" +This module defines the EventQueue class for handling events. + +Classes +------- + +- EventQueue: + Represents a queue of events received from the Ankaios system. + Inherits from the standard Queue class. + +Usage +----- + +- Check current state of the followed fields: + .. code-block:: python + + event_queue: EventQueue + current_state = event_queue.complete_state + +- Get the events out of the queue: + .. code-block:: python + + event_queue: EventQueue + event: EventEntry = event_queue.get() # blocks until an event is available +""" + +__all__ = ["EventQueue"] + + +from queue import Queue +from .request import EventsRequest, EventsCancelRequest +from .response import EventEntry +from .complete_state import CompleteState + + +class EventQueue(Queue): + """ + Represents a queue of updates through the log campaign. + Inherits from the standard Queue class. + All objects in this queue are of type :py:type:`EventEntry`. + """ + + # pylint: disable=too-many-arguments + def __init__( + self, + request: EventsRequest, + ) -> None: + """ + Initializes the EventQueue with the given parameters. + + Args: + request (EventRequest): The request object containing the log + campaign parameters. + """ + super().__init__() + self.complete_state: CompleteState = None + self._request = request + + def add_event(self, event: "EventEntry") -> None: + """ + Adds an event to the queue and updates the known complete state. + + Args: + event (EventEntry): The event to be added to the queue. + """ + self.complete_state = event.complete_state + self.put(event) + + def _get_request(self) -> EventsRequest: + """ + Returns the EventRequest object. + + Returns: + EventRequest: The EventRequest object. + """ + return self._request + + def _get_cancel_request(self) -> EventsCancelRequest: + """ + Returns the EventCancelRequest object. + + Returns: + EventCancelRequest: The EventCancelRequest object. + """ + return EventsCancelRequest(request_id=self._request.get_id()) diff --git a/ankaios_sdk/_components/request.py b/ankaios_sdk/_components/request.py index 05ccfbd2..4f546cc7 100644 --- a/ankaios_sdk/_components/request.py +++ b/ankaios_sdk/_components/request.py @@ -77,6 +77,7 @@ class that should be subclassed for specific request types. "UpdateStateRequest", "LogsRequest", "LogsCancelRequest", + "EventsRequest", ] import uuid @@ -163,6 +164,7 @@ def __init__(self, masks: list = []) -> None: """ super().__init__() self._request.completeStateRequest.fieldMask[:] = masks + self._request.completeStateRequest.subscribeForEvents = False self.logger.debug( "Created request of type GetState with id %s", @@ -282,3 +284,52 @@ def __init__(self, request_id: str) -> None: "Created request of type LogsCancelRequest with id %s", self._request.requestId, ) + + +# pylint: disable=too-few-public-methods, dangerous-default-value +class EventsRequest(Request): + """ + Represents a request for registering on events configured by the + given masks. + """ + + def __init__(self, masks: list = []) -> None: + """ + Initializes a EventsRequest instance. + + Args: + masks (list): The masks to subscribe to for the event. + """ + super().__init__() + self._request.completeStateRequest.fieldMask[:] = masks + self._request.completeStateRequest.subscribeForEvents = True + + self.logger.debug( + "Created request of type EventsRequest with id %s", + self._request.requestId, + ) + + +# pylint: disable=too-few-public-methods, dangerous-default-value +class EventsCancelRequest(Request): + """ + Represents a request for unregistering from the event stream + of a specific events campaign in the Ankaios system. + """ + + def __init__(self, request_id: str) -> None: + """ + Initializes an EventsCancelRequest instance. + + Args: + id (str): The request ID. + """ + super().__init__(_id=request_id) + self._request.eventsCancelRequest.CopyFrom( + _ank_base.EventsCancelRequest() + ) + + self.logger.debug( + "Created request of type EventsCancelRequest with id %s", + self._request.requestId, + ) diff --git a/ankaios_sdk/_components/response.py b/ankaios_sdk/_components/response.py index 1e26fc29..99449d3b 100644 --- a/ankaios_sdk/_components/response.py +++ b/ankaios_sdk/_components/response.py @@ -29,13 +29,15 @@ - LogsStopResponse: Represents a response for marking the end of the log stream from a workload instance. +- EventEntry: + Represents an event notification. Enums ----- - ResponseType: - Enumeration for the different types of response. It includes - ERROR, COMPLETE_STATE, and UPDATE_STATE_SUCCESS and CONNECTION_CLOSED. + Enumeration for the different types of response. It includes responses + like ERROR, CONNECTION_CLOSED, COMPLETE_STATE and so on. Union Types ----------- @@ -73,6 +75,7 @@ "LogEntry", "LogsStopResponse", "LogResponse", + "EventEntry", ] from dataclasses import dataclass @@ -158,9 +161,17 @@ def _from_proto(self) -> None: if self._response.HasField("error"): self.content_type = ResponseType.ERROR self.content = self._response.error.message - elif self._response.HasField("completeState"): - self.content_type = ResponseType.COMPLETE_STATE - self.content = CompleteState(_proto=self._response.completeState) + elif self._response.HasField("completeStateResponse"): + if self._response.completeStateResponse.HasField("alteredFields"): + self.content_type = ResponseType.EVENT_RESPONSE + self.content = EventEntry._from_response( + self._response.completeStateResponse + ) + else: + self.content_type = ResponseType.COMPLETE_STATE + self.content = CompleteState( + _proto=self._response.completeStateResponse.completeState + ) elif self._response.HasField("UpdateStateSuccess"): update_state_msg = self._response.UpdateStateSuccess self.content_type = ResponseType.UPDATE_STATE_SUCCESS @@ -203,6 +214,9 @@ def _from_proto(self) -> None: elif self._response.HasField("logsCancelAccepted"): self.content_type = ResponseType.LOGS_CANCEL_ACCEPTED self.content = None + elif self._response.HasField("eventsCancelAccepted"): + self.content_type = ResponseType.EVENT_CANCEL_ACCEPTED + self.content = None else: raise ResponseException("Invalid response type.") @@ -254,7 +268,11 @@ class ResponseType(Enum): "(int): Got logs stop response." LOGS_CANCEL_ACCEPTED = 8 "(int): Logs cancel request accepted." - CONNECTION_CLOSED = 9 + EVENT_RESPONSE = 9 + "(int): Got an event response." + EVENT_CANCEL_ACCEPTED = 10 + "(int): Event cancel request accepted." + CONNECTION_CLOSED = 20 "(int): Connection closed by the server." def __str__(self) -> str: @@ -396,3 +414,45 @@ def _from_stop_response( LogResponse = Union[LogEntry, LogsStopResponse] + + +@dataclass +class EventEntry: + """ + Represents an event notification. + """ + + complete_state: CompleteState + """The complete state of the event.""" + added_fields: list[str] + """The list of added fields of the state.""" + updated_fields: list[str] + """The list of updated fields of the state.""" + removed_fields: list[str] + """The list of removed fields of the state.""" + + def __str__(self) -> str: + """ + Converts the EventEntry to a string. + + Returns: + str: The string representation of the EventEntry. + """ + ret = "Event notification:\n" + if self.added_fields: + ret += f" Added fields: {self.added_fields}\n" + if self.updated_fields: + ret += f" Updated fields: {self.updated_fields}\n" + if self.removed_fields: + ret += f" Deleted fields: {self.removed_fields}\n" + + @staticmethod + def _from_response( + response: _ank_base.CompleteStateResponse, + ) -> "EventEntry": + return EventEntry( + CompleteState(_proto=response.completeState), + response.alteredFields.addedFields, + response.alteredFields.updatedFields, + response.alteredFields.removedFields, + ) diff --git a/ankaios_sdk/ankaios.py b/ankaios_sdk/ankaios.py index a03a5ff2..184fade2 100644 --- a/ankaios_sdk/ankaios.py +++ b/ankaios_sdk/ankaios.py @@ -110,6 +110,17 @@ print(f"State not reached in time.") else: print(f"State reached.") + +Field masks +----------- + +Some of the methods support field masks to filter the state. +Some examples of field masks include: +- "desiredState.workloads" # All workloads in the desired state +- "desiredState.workloads." # Workload with the specific name +- "desiredState.configs" # All configs in the desired state +- "workloadStates" # All workload states +- "workloadStates.." # State of a specific workload """ __all__ = ["Ankaios"] @@ -143,6 +154,9 @@ LogQueue, LogResponse, LogsRequest, + EventsRequest, + EventQueue, + EventEntry, ) from .utils import ( AnkaiosLogLevel, @@ -184,6 +198,7 @@ def __init__( # Thread safe queue for responses and logs self._responses: Queue = Queue() self._logs_callbacks: dict[str, Callable] = {} + self._events_callbacks: dict[str, Callable] = {} self.logger = get_logger() self.set_logger_level(log_level) @@ -192,6 +207,7 @@ def __init__( self._control_interface = ControlInterface( add_response_callback=self._add_response, add_log_callback=self._add_logs, + add_event_callback=self._add_events, ) self._control_interface.connect() @@ -257,6 +273,18 @@ def _add_logs(self, request_id: str, logs: list[LogResponse]) -> None: "Received logs for unknown request id %s", request_id ) + def _add_events(self, request_id: str, event: EventEntry) -> None: + """ + Method will be called automatically from the Control Interface + when an event is received. + """ + if request_id in self._events_callbacks: + self._events_callbacks[request_id](event) + else: + self.logger.warning( + "Received event with unknown request id %s", request_id + ) + def _get_response_by_id( self, request_id: str, timeout: float = DEFAULT_TIMEOUT ) -> Response: @@ -1092,3 +1120,88 @@ def stop_receiving_logs( self._logs_callbacks.pop(request.get_id(), None) return None raise AnkaiosProtocolException("Received unexpected content type.") + + def register_event( + self, field_masks: list[str], timeout: float = DEFAULT_TIMEOUT + ) -> "EventQueue": + """ + Register an event. + + Args: + field_masks (list[str]): The masks to filter the state for events. + timeout (float): The maximum time to wait for the response, + in seconds. + + Raises: + ControlInterfaceException: If not connected. + TimeoutError: If the request timed out. + AnkaiosResponseError: If the response is an error. + AnkaiosProtocolException: If the response has unexpected + content type. + ConnectionClosedException: If the connection is closed. + """ + event_request = EventsRequest(masks=field_masks) + + # Create the event queue and get the request + event_queue = EventQueue(event_request) + request = event_queue._get_request() + + try: + response = self._send_request(request, timeout) + except TimeoutError as e: + self.logger.error("%s", e) + raise e + + # Interpret response + (content_type, content) = response.get_content() + if content_type == ResponseType.ERROR: + self.logger.error( + "Error while trying to register event: %s", content + ) + raise AnkaiosResponseError(f"Received error: {content}") + if content_type == ResponseType.COMPLETE_STATE: + self.logger.info("Event registered successfully, state received.") + self._events_callbacks[request.get_id()] = event_queue.add_event + return event_queue + raise AnkaiosProtocolException("Received unexpected content type.") + + def unregister_event( + self, event_queue: "EventQueue", timeout: float = DEFAULT_TIMEOUT + ) -> None: + """ + Unregister an event. + + Args: + event_queue (EventQueue): The event queue to be unregistered. + timeout (float): The maximum time to wait for the response, + in seconds. + + Raises: + ControlInterfaceException: If not connected. + TimeoutError: If the request timed out. + AnkaiosResponseError: If the response is an error. + AnkaiosProtocolException: If the response has unexpected + content type. + ConnectionClosedException: If the connection is closed. + """ + # Get the cancel request + request = event_queue._get_cancel_request() + + try: + response = self._send_request(request, timeout) + except TimeoutError as e: + self.logger.error("%s", e) + raise e + + # Interpret response + (content_type, content) = response.get_content() + if content_type == ResponseType.ERROR: + self.logger.error( + "Error while trying to unregister event: %s", content + ) + raise AnkaiosResponseError(f"Received error: {content}") + if content_type == ResponseType.EVENT_CANCEL_ACCEPTED: + self.logger.info("Event unregister request accepted.") + self._events_callbacks.pop(request.get_id(), None) + return None + raise AnkaiosProtocolException("Received unexpected content type.") diff --git a/examples/apps/state_event.py b/examples/apps/state_event.py new file mode 100644 index 00000000..f1988171 --- /dev/null +++ b/examples/apps/state_event.py @@ -0,0 +1,85 @@ +# Copyright (c) 2025 Elektrobit Automotive GmbH +# +# This program and the accompanying materials are made available under the +# terms of the Apache License, Version 2.0 which is available at +# https://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. +# +# SPDX-License-Identifier: Apache-2.0 + +from ankaios_sdk import ( + Ankaios, + AnkaiosLogLevel, + AnkaiosException, + Workload, + EventEntry, +) +import sys, signal + +# Create a new Ankaios object. +# The connection to the control interface is automatically done at this step. +# The Ankaios class supports context manager syntax: +with Ankaios(log_level=AnkaiosLogLevel.DEBUG) as ankaios: + + def signal_handler(sig, frame): + global ankaios + del ankaios + sys.exit(0) + + # Add a SIGTERM handler to allow a clean shutdown + signal.signal(signal.SIGTERM, signal_handler) + + # Create a new workload + workload = ( + Workload.builder() + .workload_name("count_to_five") + .agent_name("agent_Py_SDK") + .runtime("podman") + .restart_policy("NEVER") + .runtime_config( + 'image: ghcr.io/eclipse-ankaios/tests/alpine:latest\ncommandOptions: [ "--entrypoint", "/bin/sh" ]\ncommandArgs: [ "-c", "echo -e \'1\\n2\\n3\\n4\\n5\';" ]' + ) + .build() + ) + + try: + # Run the workload + update_response = ankaios.apply_workload(workload) + + # Get the WorkloadInstanceName to check later if the workload is running + workload_instance_name = update_response.added_workloads[0] + + # Register to events to the workload + try: + event_queue = ankaios.register_event( + field_masks=["desiredState.workloads.count_to_five"], + ) + except AnkaiosException as e: + print("Ankaios Exception occurred during event registration: ", e) + sys.exit(1) + + while True: + # Get the events out of the queue + event: EventEntry = event_queue.get() + + # Interpret the event + print("Received event:") + if event.added_fields: + print(f" - Added fields: {event.added_fields}") + if event.updated_fields: + print(f" - Updated fields: {event.updated_fields}") + if event.removed_fields: + print(f" - Removed fields: {event.removed_fields}") + print(f"Current complete state: {event.complete_state}\n") + + # Unregister the events + ankaios.unregister_event(event_queue) + + # Catch the AnkaiosException in case something went wrong + except AnkaiosException as e: + print("Ankaios Exception occurred: ", e) diff --git a/tests/response/test_response.py b/tests/response/test_response.py index 53f9ffd2..260810b0 100644 --- a/tests/response/test_response.py +++ b/tests/response/test_response.py @@ -42,13 +42,15 @@ MESSAGE_BUFFER_COMPLETE_STATE = _control_api.FromAnkaios( response=_ank_base.Response( requestId="2233", - completeState=_ank_base.CompleteState( - desiredState=_ank_base.State( - apiVersion="v0.1", - workloads=_ank_base.WorkloadMap( - workloads={}, + completeStateResponse=_ank_base.CompleteStateResponse( + completeState=_ank_base.CompleteState( + desiredState=_ank_base.State( + apiVersion="v0.1", + workloads=_ank_base.WorkloadMap( + workloads={}, + ), ), - ) + ), ), ) ).SerializeToString() diff --git a/tests/test_control_interface.py b/tests/test_control_interface.py index f803ff7e..4917ccab 100644 --- a/tests/test_control_interface.py +++ b/tests/test_control_interface.py @@ -46,7 +46,9 @@ def test_state(): Test the state enum and the changing of the state. """ ci = ControlInterface( - add_response_callback=lambda _: None, add_log_callback=lambda _: None + add_response_callback=lambda _: None, + add_log_callback=lambda _: None, + add_event_callback=lambda _: None, ) ci._logger = MagicMock() assert ci._state == ControlInterfaceState.TERMINATED @@ -69,7 +71,9 @@ def test_connection(): Test the connect / disconnect functionality. """ ci = ControlInterface( - add_response_callback=lambda _: None, add_log_callback=lambda _: None + add_response_callback=lambda _: None, + add_log_callback=lambda _: None, + add_event_callback=lambda _: None, ) ci._state = ControlInterfaceState.CONNECTED assert ci.connected @@ -169,6 +173,7 @@ def test_read_thread_general(): ci = ControlInterface( add_response_callback=lambda _: None, add_log_callback=lambda _: None, + add_event_callback=lambda _: None, ) with pytest.raises( ControlInterfaceException, match="Error while opening input fifo" @@ -191,6 +196,7 @@ def test_read_thread_general(): ci = ControlInterface( add_response_callback=lambda _: None, add_log_callback=lambda _: None, + add_event_callback=lambda _: None, ) # Start thread (similar to _connect) @@ -231,6 +237,7 @@ def test_read_thread_agent_disconnected(): ci = ControlInterface( add_response_callback=lambda _: None, add_log_callback=lambda _: None, + add_event_callback=lambda _: None, ) # Start thread (similar to _connect) @@ -277,6 +284,7 @@ def test_read_thread_connection_closed(): ci = ControlInterface( add_response_callback=lambda _: None, add_log_callback=lambda _: None, + add_event_callback=lambda _: None, ) # Start thread (similar to _connect) @@ -306,6 +314,7 @@ def test_handle_response(): ci = ControlInterface( add_response_callback=response_callback, add_log_callback=lambda _: None, + add_event_callback=lambda _: None, ) ci._logger = MagicMock() @@ -351,6 +360,7 @@ def test_handle_response_control_interface_accepted(): ci = ControlInterface( add_response_callback=response_callback, add_log_callback=lambda _: None, + add_event_callback=lambda _: None, ) ci._state = ControlInterfaceState.INITIALIZED @@ -363,6 +373,7 @@ def test_handle_response_control_interface_accepted(): ci = ControlInterface( add_response_callback=response_callback, add_log_callback=lambda _: None, + add_event_callback=lambda _: None, ) ci._logger = MagicMock() ci._state = ControlInterfaceState.CONNECTED @@ -389,6 +400,7 @@ def test_handle_response_connection_closed(): ci = ControlInterface( add_response_callback=response_callback, add_log_callback=lambda _: None, + add_event_callback=lambda _: None, ) ci._state = ControlInterfaceState.INITIALIZED @@ -402,6 +414,7 @@ def test_handle_response_connection_closed(): ci = ControlInterface( add_response_callback=response_callback, add_log_callback=lambda _: None, + add_event_callback=lambda _: None, ) ci._state = ControlInterfaceState.CONNECTED @@ -420,6 +433,7 @@ def test_handle_response_logs(): ci = ControlInterface( add_response_callback=response_callback, add_log_callback=logs_callback, + add_event_callback=lambda _: None, ) ci._state = ControlInterfaceState.CONNECTED @@ -434,7 +448,9 @@ def test_agent_gone_routine(): Test the _agent_gone_routine method of the ControlInterface class. """ ci = ControlInterface( - add_response_callback=lambda _: None, add_log_callback=lambda _: None + add_response_callback=lambda _: None, + add_log_callback=lambda _: None, + add_event_callback=lambda _: None, ) ci._state = ControlInterfaceState.CONNECTED with patch( @@ -472,7 +488,9 @@ def test_write_to_pipe(): Test the _write_to_pipe method of the ControlInterface class. """ ci = ControlInterface( - add_response_callback=lambda _: None, add_log_callback=lambda _: None + add_response_callback=lambda _: None, + add_log_callback=lambda _: None, + add_event_callback=lambda _: None, ) ci._output_file = None @@ -495,7 +513,9 @@ def test_write_request(): Test the write_request method of the ControlInterface class. """ ci = ControlInterface( - add_response_callback=lambda _: None, add_log_callback=lambda _: None + add_response_callback=lambda _: None, + add_log_callback=lambda _: None, + add_event_callback=lambda _: None, ) ci._state = ControlInterfaceState.TERMINATED @@ -522,7 +542,9 @@ def test_send_initial_hello(): Test the _send_initial_hello method of the Ankaios class. """ ci = ControlInterface( - add_response_callback=lambda _: None, add_log_callback=lambda _: None + add_response_callback=lambda _: None, + add_log_callback=lambda _: None, + add_event_callback=lambda _: None, ) with patch("ankaios_sdk.ControlInterface._write_to_pipe") as mock_write: initial_hello = _control_api.ToAnkaios( From d7e12d4d14253aa6fe388cb4233a60697ace6f36 Mon Sep 17 00:00:00 2001 From: Tomuta Gabriel Date: Thu, 2 Oct 2025 17:06:11 +0300 Subject: [PATCH 2/8] Add utest and doc, fix lint --- ankaios_sdk/_components/control_interface.py | 5 +- ankaios_sdk/_components/event_campaign.py | 6 +- ankaios_sdk/_components/request.py | 6 + ankaios_sdk/_components/response.py | 2 + docs/source/event_campaign.rst | 14 ++ docs/source/index.rst | 1 + docs/source/request.rst | 18 +++ tests/response/test_response.py | 53 ++++++- tests/test_ankaios.py | 144 +++++++++++++++++- tests/test_complete_state.py | 8 +- tests/test_control_interface.py | 23 +++ tests/test_event_campaign.py | 81 ++++++++++ ...test_log_queue.py => test_log_campaign.py} | 0 13 files changed, 348 insertions(+), 13 deletions(-) create mode 100644 docs/source/event_campaign.rst create mode 100644 tests/test_event_campaign.py rename tests/{test_log_queue.py => test_log_campaign.py} (100%) diff --git a/ankaios_sdk/_components/control_interface.py b/ankaios_sdk/_components/control_interface.py index 353b8691..932115c0 100644 --- a/ankaios_sdk/_components/control_interface.py +++ b/ankaios_sdk/_components/control_interface.py @@ -101,7 +101,10 @@ class ControlInterface: "(str): The base path for the Ankaios control interface." def __init__( - self, add_response_callback: Callable, add_log_callback: Callable, add_event_callback: Callable + self, + add_response_callback: Callable, + add_log_callback: Callable, + add_event_callback: Callable, ) -> None: """ Initialize the ControlInterface object. This is used diff --git a/ankaios_sdk/_components/event_campaign.py b/ankaios_sdk/_components/event_campaign.py index d1af38ee..44ea1a10 100644 --- a/ankaios_sdk/_components/event_campaign.py +++ b/ankaios_sdk/_components/event_campaign.py @@ -36,7 +36,7 @@ .. code-block:: python event_queue: EventQueue - event: EventEntry = event_queue.get() # blocks until an event is available + event: EventEntry = event_queue.get() # blocking call """ __all__ = ["EventQueue"] @@ -50,7 +50,7 @@ class EventQueue(Queue): """ - Represents a queue of updates through the log campaign. + Represents a queue of updates through the event campaign. Inherits from the standard Queue class. All objects in this queue are of type :py:type:`EventEntry`. """ @@ -64,7 +64,7 @@ def __init__( Initializes the EventQueue with the given parameters. Args: - request (EventRequest): The request object containing the log + request (EventRequest): The request object containing the event campaign parameters. """ super().__init__() diff --git a/ankaios_sdk/_components/request.py b/ankaios_sdk/_components/request.py index 4f546cc7..0a8a1409 100644 --- a/ankaios_sdk/_components/request.py +++ b/ankaios_sdk/_components/request.py @@ -31,6 +31,11 @@ class that should be subclassed for specific request types. - LogsCancelRequest: Represents a request to stop the real-time log stream from the Ankaios system. +- EventsRequest: + Represents a request to register for events from the Ankaios system. +- EventsCancelRequest: + Represents a request to unregister from the event stream of a specific + events campaign in the Ankaios system. Usage ----- @@ -78,6 +83,7 @@ class that should be subclassed for specific request types. "LogsRequest", "LogsCancelRequest", "EventsRequest", + "EventsCancelRequest", ] import uuid diff --git a/ankaios_sdk/_components/response.py b/ankaios_sdk/_components/response.py index 99449d3b..6d4d6533 100644 --- a/ankaios_sdk/_components/response.py +++ b/ankaios_sdk/_components/response.py @@ -149,6 +149,7 @@ def _parse_response(self) -> None: self.get_request_id(), ) + # pylint: disable=too-many-branches def _from_proto(self) -> None: """ Converts the parsed protobuf message to a Response object. @@ -445,6 +446,7 @@ def __str__(self) -> str: ret += f" Updated fields: {self.updated_fields}\n" if self.removed_fields: ret += f" Deleted fields: {self.removed_fields}\n" + return ret @staticmethod def _from_response( diff --git a/docs/source/event_campaign.rst b/docs/source/event_campaign.rst new file mode 100644 index 00000000..919b34c3 --- /dev/null +++ b/docs/source/event_campaign.rst @@ -0,0 +1,14 @@ +Event Campaign +============== + +.. automodule:: ankaios_sdk._components.event_campaign + +EventQueue Class +---------------- + +.. autoclass:: ankaios_sdk._components.event_campaign.EventQueue + :special-members: __init__ + :members: + :inherited-members: + :undoc-members: + :show-inheritance: diff --git a/docs/source/index.rst b/docs/source/index.rst index d4ca8053..12f51778 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -22,6 +22,7 @@ response file log_campaign + event_campaign control_interface utils exceptions diff --git a/docs/source/request.rst b/docs/source/request.rst index 2ef2c615..49834794 100644 --- a/docs/source/request.rst +++ b/docs/source/request.rst @@ -45,3 +45,21 @@ LogsCancelRequest Class :members: :undoc-members: :show-inheritance: + +EventsRequest Class +------------------- + +.. autoclass:: ankaios_sdk._components.request.EventsRequest + :special-members: __init__ + :members: + :undoc-members: + :show-inheritance: + +EventsCancelRequest Class +------------------------- + +.. autoclass:: ankaios_sdk._components.request.EventsCancelRequest + :special-members: __init__ + :members: + :undoc-members: + :show-inheritance: diff --git a/tests/response/test_response.py b/tests/response/test_response.py index 260810b0..84eb407a 100644 --- a/tests/response/test_response.py +++ b/tests/response/test_response.py @@ -26,6 +26,7 @@ ResponseException, LogEntry, LogsStopResponse, + EventEntry, ) from tests.response.test_log_response import generate_test_log_entry @@ -112,9 +113,44 @@ ) ).SerializeToString() -MESSAGE_BUFFER_INVALID_RESPONSE = _control_api.FromAnkaios( +MESSAGE_BUFFER_LOGS_CANCEL_ACCEPTED_RESPONSE = _control_api.FromAnkaios( + response=_ank_base.Response( + requestId="4455", + logsCancelAccepted=_ank_base.LogsCancelAccepted(), + ) +).SerializeToString() + +MESSAGE_BUFFER_EVENT_ENTRY_RESPONSE = _control_api.FromAnkaios( + response=_ank_base.Response( + requestId="5566", + completeStateResponse=_ank_base.CompleteStateResponse( + completeState=_ank_base.CompleteState( + desiredState=_ank_base.State( + apiVersion="v0.1", + workloads=_ank_base.WorkloadMap( + workloads={}, + ), + ), + ), + alteredFields=_ank_base.AlteredFields( + addedFields=["field1"], + updatedFields=["field2"], + removedFields=["field3"], + ), + ), + ) +).SerializeToString() + +MESSAGE_BUFFER_EVENTS_CANCEL_ACCEPTED_RESPONSE = _control_api.FromAnkaios( response=_ank_base.Response( requestId="5566", + eventsCancelAccepted=_ank_base.EventsCancelAccepted(), + ) +).SerializeToString() + +MESSAGE_BUFFER_INVALID_RESPONSE = _control_api.FromAnkaios( + response=_ank_base.Response( + requestId="6677", ) ).SerializeToString() @@ -181,6 +217,21 @@ def test_initialisation(): assert len(response.content) == 1 assert isinstance(response.content[0], LogsStopResponse) + # Test logs cancel accepted response + response = Response(MESSAGE_BUFFER_LOGS_CANCEL_ACCEPTED_RESPONSE) + assert response.content_type == ResponseType.LOGS_CANCEL_ACCEPTED + assert response.content is None + + # Test events entry response + response = Response(MESSAGE_BUFFER_EVENT_ENTRY_RESPONSE) + assert response.content_type == ResponseType.EVENT_RESPONSE + assert isinstance(response.content, EventEntry) + + # Test events cancel accepted response + response = Response(MESSAGE_BUFFER_EVENTS_CANCEL_ACCEPTED_RESPONSE) + assert response.content_type == ResponseType.EVENT_CANCEL_ACCEPTED + assert response.content is None + # Test connection closed response = Response(MESSAGE_BUFFER_CONNECTION_CLOSED) assert response.content_type == ResponseType.CONNECTION_CLOSED diff --git a/tests/test_ankaios.py b/tests/test_ankaios.py index 1691c204..ef4d5fe8 100644 --- a/tests/test_ankaios.py +++ b/tests/test_ankaios.py @@ -16,6 +16,8 @@ This module contains unit tests for the Ankaios class in the ankaios_sdk. """ +# pylint: disable=too-many-lines + from io import StringIO import logging from unittest.mock import patch, MagicMock, PropertyMock @@ -36,6 +38,7 @@ AnkaiosResponseError, ConnectionClosedException, LogCampaignResponse, + EventQueue, ) from ankaios_sdk.utils import WORKLOADS_PREFIX from tests.workload.test_workload import generate_test_workload @@ -48,11 +51,13 @@ MESSAGE_BUFFER_CONNECTION_CLOSED, MESSAGE_BUFFER_LOGS_REQUEST_ACCEPTED, MESSAGE_BUFFER_LOGS_CANCEL_REQUEST_ACCEPTED, + MESSAGE_BUFFER_EVENTS_CANCEL_ACCEPTED_RESPONSE, ) from tests.test_manifest import MANIFEST_DICT from tests.workload_state.test_workload_state import ( generate_test_workload_state, ) +from tests.test_event_campaign import generate_test_event_entry def generate_test_ankaios() -> Ankaios: @@ -157,18 +162,40 @@ def test_add_logs(): LogEntry._from_entries(generate_test_log_entry(name="nginx_A")), LogEntry._from_entries(generate_test_log_entry(name="nginx_B")), ] - put_mock = MagicMock() + add_logs_cb_mock = MagicMock() ankaios = generate_test_ankaios() ankaios.logger = MagicMock() assert len(ankaios._logs_callbacks) == 0 - ankaios._logs_callbacks = {"correct_id": put_mock} + ankaios._logs_callbacks = {"correct_id": add_logs_cb_mock} ankaios._add_logs("correct_id", log_entries) - assert put_mock.call_count == 2 + assert add_logs_cb_mock.call_count == 2 ankaios.logger.warning.assert_not_called() ankaios._add_logs("wrong_id", log_entries) - assert put_mock.call_count == 2 + assert add_logs_cb_mock.call_count == 2 + ankaios.logger.warning.assert_called_once() + + +def test_add_events(): + """ + Test the _add_events method of the Ankaios class. + This method is called from the ControlInterface when a response + of type EventEntry is received. + """ + event_entry = generate_test_event_entry() + add_event_cb_mock = MagicMock() + + ankaios = generate_test_ankaios() + ankaios.logger = MagicMock() + assert len(ankaios._events_callbacks) == 0 + ankaios._events_callbacks = {"correct_id": add_event_cb_mock} + ankaios._add_events("correct_id", event_entry) + assert add_event_cb_mock.call_count == 1 + ankaios.logger.warning.assert_not_called() + + ankaios._add_events("wrong_id", event_entry) + assert add_event_cb_mock.call_count == 1 ankaios.logger.warning.assert_called_once() @@ -932,3 +959,112 @@ def test_stop_receiving_logs(): ankaios.stop_receiving_logs(log_campaign) mock_send_request.assert_called_once() ankaios.logger.error.assert_called() + + +def test_register_events(): + """ + Test the register_event method of the Ankaios class. + """ + ankaios = generate_test_ankaios() + ankaios.logger = MagicMock() + assert len(ankaios._events_callbacks) == 0 + + # Test success + with patch("ankaios_sdk.Ankaios._send_request") as mock_send_request: + mock_send_request.return_value = Response( + MESSAGE_BUFFER_COMPLETE_STATE + ) + events_queue = ankaios.register_event(field_masks=["field1"]) + request = mock_send_request.call_args[0][0] + assert isinstance(events_queue, EventQueue) + assert len(ankaios._events_callbacks) == 1 + # pylint: disable=comparison-with-callable + assert ( + ankaios._events_callbacks[request.get_id()] + == events_queue.add_event + ) + mock_send_request.assert_called_once() + ankaios.logger.info.assert_called() + + # Test error + with patch("ankaios_sdk.Ankaios._send_request") as mock_send_request: + mock_send_request.return_value = Response(MESSAGE_BUFFER_ERROR) + with pytest.raises(AnkaiosResponseError): + ankaios.register_event(field_masks=["field1"]) + mock_send_request.assert_called_once() + ankaios.logger.error.assert_called() + + # Test timeout + with patch("ankaios_sdk.Ankaios._send_request") as mock_send_request: + mock_send_request.side_effect = TimeoutError() + with pytest.raises(TimeoutError): + ankaios.register_event(field_masks=["field1"]) + mock_send_request.assert_called_once() + ankaios.logger.error.assert_called() + + # Test invalid content type + with patch("ankaios_sdk.Ankaios._send_request") as mock_send_request: + mock_send_request.return_value = Response( + MESSAGE_BUFFER_UPDATE_SUCCESS + ) + with pytest.raises(AnkaiosProtocolException): + ankaios.register_event(field_masks=["field1"]) + mock_send_request.assert_called_once() + ankaios.logger.error.assert_called() + + +def test_unregister_event(): + """ + Test the unregister_event method of the Ankaios class. + """ + ankaios = generate_test_ankaios() + ankaios.logger = MagicMock() + assert len(ankaios._events_callbacks) == 0 + + # Populate the events callback with a campaign + with patch("ankaios_sdk.Ankaios._send_request") as mock_send_request: + mock_send_request.return_value = Response( + MESSAGE_BUFFER_COMPLETE_STATE + ) + events_queue = ankaios.register_event(field_masks=["field1"]) + assert isinstance(events_queue, EventQueue) + assert len(ankaios._events_callbacks) == 1 + + # Test success + with patch("ankaios_sdk.Ankaios._send_request") as mock_send_request: + mock_send_request.return_value = Response( + MESSAGE_BUFFER_EVENTS_CANCEL_ACCEPTED_RESPONSE + ) + cancel_request = events_queue._get_cancel_request() + ankaios.unregister_event(events_queue) + request = mock_send_request.call_args[0][0] + assert cancel_request._to_proto() == request._to_proto() + assert len(ankaios._events_callbacks) == 0 + mock_send_request.assert_called_once() + ankaios.logger.info.assert_called() + + # Test error + with patch("ankaios_sdk.Ankaios._send_request") as mock_send_request: + mock_send_request.return_value = Response(MESSAGE_BUFFER_ERROR) + with pytest.raises(AnkaiosResponseError): + ankaios.unregister_event(events_queue) + mock_send_request.assert_called_once() + ankaios.logger.error.assert_called() + + # Test timeout + with patch("ankaios_sdk.Ankaios._send_request") as mock_send_request: + mock_send_request.side_effect = TimeoutError() + with pytest.raises(TimeoutError): + ankaios.unregister_event(events_queue) + mock_send_request.assert_called_once() + ankaios.logger.error.assert_called() + + # Test invalid content type + with patch("ankaios_sdk.Ankaios._send_request") as mock_send_request: + mock_send_request.return_value = Response( + MESSAGE_BUFFER_UPDATE_SUCCESS + ) + with pytest.raises(AnkaiosProtocolException): + ankaios.unregister_event(events_queue) + mock_send_request.assert_called_once() + ankaios.logger.error.assert_called() diff --git a/tests/test_complete_state.py b/tests/test_complete_state.py index 0d780feb..78ddf82e 100644 --- a/tests/test_complete_state.py +++ b/tests/test_complete_state.py @@ -70,7 +70,7 @@ ) -COMPLETE_PROTO = proto_msg = _ank_base.CompleteState( +COMPLETE_STATE_PROTO = _ank_base.CompleteState( desiredState=_ank_base.State( apiVersion="v0.1", workloads=WORKLOAD_PROTO, configs=CONFIGS_PROTO ), @@ -176,7 +176,7 @@ def test_to_dict(): """ Test converting the CompleteState to a dictionary. """ - complete_state = CompleteState(_proto=COMPLETE_PROTO) + complete_state = CompleteState(_proto=COMPLETE_STATE_PROTO) complete_state_dict = complete_state.to_dict() assert complete_state_dict == { @@ -272,7 +272,7 @@ def test_proto(): """ Test converting the CompleteState instance to and from a protobuf message. """ - complete_state = CompleteState(_proto=COMPLETE_PROTO) + complete_state = CompleteState(_proto=COMPLETE_STATE_PROTO) new_proto = complete_state._to_proto() - assert new_proto == COMPLETE_PROTO + assert new_proto == COMPLETE_STATE_PROTO diff --git a/tests/test_control_interface.py b/tests/test_control_interface.py index 4917ccab..c91d2a53 100644 --- a/tests/test_control_interface.py +++ b/tests/test_control_interface.py @@ -38,6 +38,7 @@ MESSAGE_BUFFER_CONNECTION_CLOSED, MESSAGE_BUFFER_CONNECTION_CLOSED_LENGTH, MESSAGE_BUFFER_LOGS_ENTRIES_RESPONSE, + MESSAGE_BUFFER_EVENT_ENTRY_RESPONSE, ) @@ -443,6 +444,28 @@ def test_handle_response_logs(): logs_callback.assert_called_once() +def test_handle_response_events(): + """ + Test the _handle_response method of the Ankaios class. + Test the events callback. + """ + events_response = Response(MESSAGE_BUFFER_EVENT_ENTRY_RESPONSE) + response_callback = MagicMock() + events_callback = MagicMock() + + ci = ControlInterface( + add_response_callback=response_callback, + add_log_callback=lambda _: None, + add_event_callback=events_callback, + ) + ci._state = ControlInterfaceState.CONNECTED + + ci._handle_response(events_response) + + response_callback.assert_not_called() + events_callback.assert_called_once() + + def test_agent_gone_routine(): """ Test the _agent_gone_routine method of the ControlInterface class. diff --git a/tests/test_event_campaign.py b/tests/test_event_campaign.py new file mode 100644 index 00000000..3dc31450 --- /dev/null +++ b/tests/test_event_campaign.py @@ -0,0 +1,81 @@ +# Copyright (c) 2025 Elektrobit Automotive GmbH +# +# This program and the accompanying materials are made available under the +# terms of the Apache License, Version 2.0 which is available at +# https://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. +# +# SPDX-License-Identifier: Apache-2.0 + +""" +This module contains unit tests for the EventQueue class in the ankaios_sdk. + +Helper Functions: + generate_test_event_entry: Helper function to generate an EventEntry + instance for testing. +""" + +from ankaios_sdk import ( + EventQueue, + EventEntry, + EventsRequest, + EventsCancelRequest, + CompleteState, +) +from tests.test_complete_state import COMPLETE_STATE_PROTO + + +def generate_test_event_entry(complete_state=None): + """ + Generate a test EventEntry object. + """ + if complete_state is None: + complete_state = CompleteState(_proto=COMPLETE_STATE_PROTO) + return EventEntry( + complete_state=complete_state, + added_fields=["field1"], + updated_fields=["field2"], + removed_fields=["field3"], + ) + + +def test_event_queue_requests(): + """ + Test the events queue. + """ + events_queue = EventQueue(EventsRequest(masks=["field1"])) + assert events_queue is not None + + request = events_queue._get_request() + request_id = request.get_id() + assert isinstance(request, EventsRequest) + + cancel_request = events_queue._get_cancel_request() + assert isinstance(cancel_request, EventsCancelRequest) + assert cancel_request.get_id() == request_id + + +def test_events_queue(): + """ + Test the queue functionality. + """ + events_queue = EventQueue(EventsRequest(masks=["field1"])) + complete_state = CompleteState(_proto=COMPLETE_STATE_PROTO) + + event_entry = generate_test_event_entry(complete_state=complete_state) + events_queue.add_event(event_entry) + assert events_queue.empty() is False + entry = events_queue.get() + assert isinstance(entry, EventEntry) + assert entry == event_entry + assert events_queue.complete_state == complete_state + + assert ( + str(entry) == "Event notification:\n Added fields: ['field1']\n " + "Updated fields: ['field2']\n Deleted fields: ['field3']\n" + ) diff --git a/tests/test_log_queue.py b/tests/test_log_campaign.py similarity index 100% rename from tests/test_log_queue.py rename to tests/test_log_campaign.py From 1f2949bbb4a5c124977121b3d4bbb5613c113076 Mon Sep 17 00:00:00 2001 From: Tomuta Gabriel Date: Tue, 25 Nov 2025 11:39:41 +0200 Subject: [PATCH 3/8] Update the ankaios_api to match the core --- ankaios_sdk/_components/complete_state.py | 5 +++-- setup.py | 2 +- tests/test_complete_state.py | 6 ++++-- 3 files changed, 8 insertions(+), 5 deletions(-) diff --git a/ankaios_sdk/_components/complete_state.py b/ankaios_sdk/_components/complete_state.py index 063cfa00..a3695f8e 100644 --- a/ankaios_sdk/_components/complete_state.py +++ b/ankaios_sdk/_components/complete_state.py @@ -214,9 +214,10 @@ def get_agents(self) -> dict[str, dict]: """ agents = {} for name, attributes in self._complete_state.agents.agents.items(): + status = attributes.status agents[name] = { - "cpu_usage": int(attributes.cpu_usage.cpu_usage), - "free_memory": attributes.free_memory.free_memory, + "cpu_usage": int(status.cpu_usage.cpu_usage), + "free_memory": status.free_memory.free_memory, } return agents diff --git a/setup.py b/setup.py index 5151e5a0..d5ebc4aa 100644 --- a/setup.py +++ b/setup.py @@ -18,7 +18,7 @@ PROJECT_DIR = "ankaios_sdk" ANKAIOS_RELEASE_LINK = "https://github.com/eclipse-ankaios/ankaios/releases/download/v{version}/{file}" -ANKAIOS_MAIN_LINK = "https://raw.githubusercontent.com/eclipse-ankaios/ankaios/refs/heads/main/api/proto/{file}" +ANKAIOS_MAIN_LINK = "https://raw.githubusercontent.com/eclipse-ankaios/ankaios/refs/heads/main/ankaios_api/proto/{file}" PROTO_FILES = ["ank_base.proto", "control_api.proto"] config = configparser.ConfigParser() diff --git a/tests/test_complete_state.py b/tests/test_complete_state.py index 832e04fc..5a44535b 100644 --- a/tests/test_complete_state.py +++ b/tests/test_complete_state.py @@ -63,8 +63,10 @@ AGENTS_PROTO = _ank_base.AgentMap( agents={ "agent_A": _ank_base.AgentAttributes( - cpu_usage=_ank_base.CpuUsage(cpu_usage=50), - free_memory=_ank_base.FreeMemory(free_memory=1024), + status=_ank_base.AgentStatus( + cpu_usage=_ank_base.CpuUsage(cpu_usage=50), + free_memory=_ank_base.FreeMemory(free_memory=1024), + ) ) } ) From 09444edb1ef62cde523e0b7868f605cf420ed61a Mon Sep 17 00:00:00 2001 From: Tomuta Gabriel Date: Wed, 21 Jan 2026 12:52:25 +0200 Subject: [PATCH 4/8] Fix findings --- ankaios_sdk/_components/event_campaign.py | 2 +- ankaios_sdk/_components/request.py | 4 ++-- ankaios_sdk/_components/response.py | 6 +++--- docs/source/getting_started.rst | 2 +- examples/apps/state_event.py | 9 +++------ examples/manifest.yaml | 2 +- tests/response/test_response.py | 4 ++-- tests/test_complete_state.py | 6 +++--- tests/test_event_campaign.py | 2 +- tests/test_manifest.py | 10 +++++----- tools/update_version.sh | 2 +- 11 files changed, 23 insertions(+), 26 deletions(-) diff --git a/ankaios_sdk/_components/event_campaign.py b/ankaios_sdk/_components/event_campaign.py index 44ea1a10..ba05841f 100644 --- a/ankaios_sdk/_components/event_campaign.py +++ b/ankaios_sdk/_components/event_campaign.py @@ -71,7 +71,7 @@ def __init__( self.complete_state: CompleteState = None self._request = request - def add_event(self, event: "EventEntry") -> None: + def add_event(self, event: EventEntry) -> None: """ Adds an event to the queue and updates the known complete state. diff --git a/ankaios_sdk/_components/request.py b/ankaios_sdk/_components/request.py index 0a8a1409..83c1425e 100644 --- a/ankaios_sdk/_components/request.py +++ b/ankaios_sdk/_components/request.py @@ -35,7 +35,7 @@ class that should be subclassed for specific request types. Represents a request to register for events from the Ankaios system. - EventsCancelRequest: Represents a request to unregister from the event stream of a specific - events campaign in the Ankaios system. + event campaign in the Ankaios system. Usage ----- @@ -320,7 +320,7 @@ def __init__(self, masks: list = []) -> None: class EventsCancelRequest(Request): """ Represents a request for unregistering from the event stream - of a specific events campaign in the Ankaios system. + of a specific event campaign in the Ankaios system. """ def __init__(self, request_id: str) -> None: diff --git a/ankaios_sdk/_components/response.py b/ankaios_sdk/_components/response.py index 6d4d6533..a483d832 100644 --- a/ankaios_sdk/_components/response.py +++ b/ankaios_sdk/_components/response.py @@ -30,7 +30,7 @@ Represents a response for marking the end of the log stream from a workload instance. - EventEntry: - Represents an event notification. + Represents an event. Enums ----- @@ -424,7 +424,7 @@ class EventEntry: """ complete_state: CompleteState - """The complete state of the event.""" + """The complete state of the event indicating the state changes.""" added_fields: list[str] """The list of added fields of the state.""" updated_fields: list[str] @@ -439,7 +439,7 @@ def __str__(self) -> str: Returns: str: The string representation of the EventEntry. """ - ret = "Event notification:\n" + ret = "Event:\n" if self.added_fields: ret += f" Added fields: {self.added_fields}\n" if self.updated_fields: diff --git a/docs/source/getting_started.rst b/docs/source/getting_started.rst index a0b2c15e..370c6b3e 100644 --- a/docs/source/getting_started.rst +++ b/docs/source/getting_started.rst @@ -25,7 +25,7 @@ Considering we have the following manifest file with a workload called ``nginx`` .. code-block:: yaml :caption: my_manifest.yaml - apiVersion: v0.1 + apiVersion: v1 workloads: nginx: runtime: podman diff --git a/examples/apps/state_event.py b/examples/apps/state_event.py index f1988171..99d1953f 100644 --- a/examples/apps/state_event.py +++ b/examples/apps/state_event.py @@ -48,12 +48,6 @@ def signal_handler(sig, frame): ) try: - # Run the workload - update_response = ankaios.apply_workload(workload) - - # Get the WorkloadInstanceName to check later if the workload is running - workload_instance_name = update_response.added_workloads[0] - # Register to events to the workload try: event_queue = ankaios.register_event( @@ -63,6 +57,9 @@ def signal_handler(sig, frame): print("Ankaios Exception occurred during event registration: ", e) sys.exit(1) + # Run the workload + update_response = ankaios.apply_workload(workload) + while True: # Get the events out of the queue event: EventEntry = event_queue.get() diff --git a/examples/manifest.yaml b/examples/manifest.yaml index f9cc0f49..35473edf 100644 --- a/examples/manifest.yaml +++ b/examples/manifest.yaml @@ -1,4 +1,4 @@ -apiVersion: v0.1 +apiVersion: v1 workloads: example: runtime: podman diff --git a/tests/response/test_response.py b/tests/response/test_response.py index 84eb407a..e1ac57e0 100644 --- a/tests/response/test_response.py +++ b/tests/response/test_response.py @@ -46,7 +46,7 @@ completeStateResponse=_ank_base.CompleteStateResponse( completeState=_ank_base.CompleteState( desiredState=_ank_base.State( - apiVersion="v0.1", + apiVersion="v1", workloads=_ank_base.WorkloadMap( workloads={}, ), @@ -126,7 +126,7 @@ completeStateResponse=_ank_base.CompleteStateResponse( completeState=_ank_base.CompleteState( desiredState=_ank_base.State( - apiVersion="v0.1", + apiVersion="v1", workloads=_ank_base.WorkloadMap( workloads={}, ), diff --git a/tests/test_complete_state.py b/tests/test_complete_state.py index 5a44535b..bec9c95e 100644 --- a/tests/test_complete_state.py +++ b/tests/test_complete_state.py @@ -74,7 +74,7 @@ COMPLETE_STATE_PROTO = _ank_base.CompleteState( desiredState=_ank_base.State( - apiVersion="v0.1", workloads=WORKLOAD_PROTO, configs=CONFIGS_PROTO + apiVersion="v1", workloads=WORKLOAD_PROTO, configs=CONFIGS_PROTO ), workloadStates=WORKLOAD_STATES_PROTO, agents=AGENTS_PROTO, @@ -167,7 +167,7 @@ def test_from_manifest(): """ manifest = Manifest.from_dict(MANIFEST_DICT) complete_state = CompleteState(manifest=manifest) - assert complete_state.get_api_version() == "v0.1" + assert complete_state.get_api_version() == "v1" workloads = complete_state.get_workloads() assert len(workloads) == 1 assert workloads[0].name == "nginx_test" @@ -183,7 +183,7 @@ def test_to_dict(): complete_state_dict = complete_state.to_dict() assert complete_state_dict == { "desired_state": { - "api_version": "v0.1", + "api_version": "v1", "workloads": { "dynamic_nginx": { "agent": "agent_A", diff --git a/tests/test_event_campaign.py b/tests/test_event_campaign.py index 3dc31450..57fa3ce6 100644 --- a/tests/test_event_campaign.py +++ b/tests/test_event_campaign.py @@ -76,6 +76,6 @@ def test_events_queue(): assert events_queue.complete_state == complete_state assert ( - str(entry) == "Event notification:\n Added fields: ['field1']\n " + str(entry) == "Event:\n Added fields: ['field1']\n " "Updated fields: ['field2']\n Deleted fields: ['field3']\n" ) diff --git a/tests/test_manifest.py b/tests/test_manifest.py index e45ef517..4d13779b 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -23,7 +23,7 @@ from ankaios_sdk.utils import WORKLOADS_PREFIX, CONFIGS_PREFIX -MANIFEST_CONTENT = """apiVersion: v0.1 +MANIFEST_CONTENT = """apiVersion: v1 workloads: nginx_test: runtime: podman @@ -38,7 +38,7 @@ port: \"8081\"""" MANIFEST_DICT = { - "apiVersion": "v0.1", + "apiVersion": "v1", "workloads": { "nginx_test": { "runtime": "podman", @@ -51,7 +51,7 @@ "configs": {"test_ports": {"port": "8081"}}, } MANIFEST_PROTO = _ank_base.State( - apiVersion="v0.1", + apiVersion="v1", workloads=_ank_base.WorkloadMap( workloads={ "nginx_test": _ank_base.Workload( @@ -125,13 +125,13 @@ def test_from_dict(): with pytest.raises(InvalidManifestException): _ = Manifest.from_dict( - {"apiVersion": "v0.1", "workloads": {"nginx_test": {}}} + {"apiVersion": "v1", "workloads": {"nginx_test": {}}} ) with pytest.raises(InvalidManifestException): _ = Manifest.from_dict( { - "apiVersion": "v0.1", + "apiVersion": "v1", "workloads": {"nginx_test": {"invalid_key": ""}}, } ) diff --git a/tools/update_version.sh b/tools/update_version.sh index 34a3e4a7..3b14a3a0 100755 --- a/tools/update_version.sh +++ b/tools/update_version.sh @@ -32,7 +32,7 @@ usage() { echo " --help Display this help message and exit." echo "" echo "Example:" - echo " $0 --sdk 0.1.0 --ank 0.1.0 --api v0.1" + echo " $0 --sdk 0.1.0 --ank 0.1.0 --api v1" exit 1 } From e0e4e3ee233d228c44becc9b3b2901449fc74108 Mon Sep 17 00:00:00 2001 From: Tomuta Gabriel Date: Wed, 21 Jan 2026 13:08:50 +0200 Subject: [PATCH 5/8] Event and Log queues now remember only the request id --- ankaios_sdk/_components/event_campaign.py | 29 ++++------------------- ankaios_sdk/_components/log_campaign.py | 28 ++++------------------ ankaios_sdk/ankaios.py | 22 ++++++++--------- tests/test_ankaios.py | 8 +++++-- tests/test_event_campaign.py | 17 ------------- tests/test_log_campaign.py | 22 ----------------- 6 files changed, 25 insertions(+), 101 deletions(-) diff --git a/ankaios_sdk/_components/event_campaign.py b/ankaios_sdk/_components/event_campaign.py index ba05841f..70ee7515 100644 --- a/ankaios_sdk/_components/event_campaign.py +++ b/ankaios_sdk/_components/event_campaign.py @@ -43,7 +43,7 @@ from queue import Queue -from .request import EventsRequest, EventsCancelRequest +from .request import EventsRequest from .response import EventEntry from .complete_state import CompleteState @@ -55,21 +55,20 @@ class EventQueue(Queue): All objects in this queue are of type :py:type:`EventEntry`. """ - # pylint: disable=too-many-arguments + # -- # pylint: disable=too-many-arguments def __init__( self, - request: EventsRequest, + request_id: EventsRequest, ) -> None: """ Initializes the EventQueue with the given parameters. Args: - request (EventRequest): The request object containing the event - campaign parameters. + request_id (str): The request id of the event campaign. """ super().__init__() self.complete_state: CompleteState = None - self._request = request + self._request_id = request_id def add_event(self, event: EventEntry) -> None: """ @@ -80,21 +79,3 @@ def add_event(self, event: EventEntry) -> None: """ self.complete_state = event.complete_state self.put(event) - - def _get_request(self) -> EventsRequest: - """ - Returns the EventRequest object. - - Returns: - EventRequest: The EventRequest object. - """ - return self._request - - def _get_cancel_request(self) -> EventsCancelRequest: - """ - Returns the EventCancelRequest object. - - Returns: - EventCancelRequest: The EventCancelRequest object. - """ - return EventsCancelRequest(request_id=self._request.get_id()) diff --git a/ankaios_sdk/_components/log_campaign.py b/ankaios_sdk/_components/log_campaign.py index 25d9b804..f7d58d54 100644 --- a/ankaios_sdk/_components/log_campaign.py +++ b/ankaios_sdk/_components/log_campaign.py @@ -44,7 +44,6 @@ from queue import Queue from .workload_state import WorkloadInstanceName -from .request import LogsRequest, LogsCancelRequest # pylint: disable=too-few-public-methods @@ -79,35 +78,16 @@ class LogQueue(Queue): All objects in this queue are of type :py:type:`LogResponse`. """ - # pylint: disable=too-many-arguments + # --- # pylint: disable=too-many-arguments def __init__( self, - request: LogsRequest, + request_id: str, ) -> None: """ Initializes the LogQueue with the given parameters. Args: - request (LogsRequest): The request object containing the log - campaign parameters. + request_id (str): The request id of the logs campaign. """ super().__init__() - self._request = request - - def _get_request(self) -> LogsRequest: - """ - Returns the LogsRequest object. - - Returns: - LogsRequest: The LogsRequest object. - """ - return self._request - - def _get_cancel_request(self) -> LogsCancelRequest: - """ - Returns the LogsCancelRequest object. - - Returns: - LogsCancelRequest: The LogsCancelRequest object. - """ - return LogsCancelRequest(request_id=self._request.get_id()) + self._request_id = request_id diff --git a/ankaios_sdk/ankaios.py b/ankaios_sdk/ankaios.py index 184fade2..1f7e585e 100644 --- a/ankaios_sdk/ankaios.py +++ b/ankaios_sdk/ankaios.py @@ -154,7 +154,9 @@ LogQueue, LogResponse, LogsRequest, + LogsCancelRequest, EventsRequest, + EventsCancelRequest, EventQueue, EventEntry, ) @@ -1049,7 +1051,7 @@ def request_logs( ConnectionClosedException: If the connection is closed. """ - logs_request = LogsRequest( + request = LogsRequest( workload_names=workload_names, follow=follow, tail=tail, @@ -1057,9 +1059,8 @@ def request_logs( until=until, ) - # Create the logs queue and get the request - log_queue = LogQueue(logs_request) - request = log_queue._get_request() + # Create the logs queue and get the request id + log_queue = LogQueue(request.get_id()) try: response = self._send_request(request, timeout) @@ -1099,8 +1100,7 @@ def stop_receiving_logs( ControlInterfaceException: If not connected. ConnectionClosedException: If the connection is closed. """ - # Get the cancel request - request = log_campaign.queue._get_cancel_request() + request = LogsCancelRequest(request_id=log_campaign.queue._request_id) try: response = self._send_request(request, timeout) @@ -1140,11 +1140,10 @@ def register_event( content type. ConnectionClosedException: If the connection is closed. """ - event_request = EventsRequest(masks=field_masks) + request = EventsRequest(masks=field_masks) - # Create the event queue and get the request - event_queue = EventQueue(event_request) - request = event_queue._get_request() + # Create the event queue and get the request id + event_queue = EventQueue(request.get_id()) try: response = self._send_request(request, timeout) @@ -1184,8 +1183,7 @@ def unregister_event( content type. ConnectionClosedException: If the connection is closed. """ - # Get the cancel request - request = event_queue._get_cancel_request() + request = EventsCancelRequest(request_id=event_queue._request_id) try: response = self._send_request(request, timeout) diff --git a/tests/test_ankaios.py b/tests/test_ankaios.py index ef4d5fe8..3be2affd 100644 --- a/tests/test_ankaios.py +++ b/tests/test_ankaios.py @@ -38,7 +38,9 @@ AnkaiosResponseError, ConnectionClosedException, LogCampaignResponse, + LogsCancelRequest, EventQueue, + EventsCancelRequest, ) from ankaios_sdk.utils import WORKLOADS_PREFIX from tests.workload.test_workload import generate_test_workload @@ -926,7 +928,7 @@ def test_stop_receiving_logs(): mock_send_request.return_value = Response( MESSAGE_BUFFER_LOGS_CANCEL_REQUEST_ACCEPTED ) - cancel_request = log_campaign.queue._get_cancel_request() + cancel_request = LogsCancelRequest(log_campaign.queue._request_id) ankaios.stop_receiving_logs(log_campaign) request = mock_send_request.call_args[0][0] assert cancel_request._to_proto() == request._to_proto() @@ -1035,7 +1037,9 @@ def test_unregister_event(): mock_send_request.return_value = Response( MESSAGE_BUFFER_EVENTS_CANCEL_ACCEPTED_RESPONSE ) - cancel_request = events_queue._get_cancel_request() + cancel_request = EventsCancelRequest( + request_id=events_queue._request_id + ) ankaios.unregister_event(events_queue) request = mock_send_request.call_args[0][0] assert cancel_request._to_proto() == request._to_proto() diff --git a/tests/test_event_campaign.py b/tests/test_event_campaign.py index 57fa3ce6..42f0525f 100644 --- a/tests/test_event_campaign.py +++ b/tests/test_event_campaign.py @@ -24,7 +24,6 @@ EventQueue, EventEntry, EventsRequest, - EventsCancelRequest, CompleteState, ) from tests.test_complete_state import COMPLETE_STATE_PROTO @@ -44,22 +43,6 @@ def generate_test_event_entry(complete_state=None): ) -def test_event_queue_requests(): - """ - Test the events queue. - """ - events_queue = EventQueue(EventsRequest(masks=["field1"])) - assert events_queue is not None - - request = events_queue._get_request() - request_id = request.get_id() - assert isinstance(request, EventsRequest) - - cancel_request = events_queue._get_cancel_request() - assert isinstance(cancel_request, EventsCancelRequest) - assert cancel_request.get_id() == request_id - - def test_events_queue(): """ Test the queue functionality. diff --git a/tests/test_log_campaign.py b/tests/test_log_campaign.py index 618a28d0..3b0f01da 100644 --- a/tests/test_log_campaign.py +++ b/tests/test_log_campaign.py @@ -20,32 +20,10 @@ LogQueue, WorkloadInstanceName, LogEntry, - LogsRequest, - LogsCancelRequest, ) from tests.response.test_log_response import generate_test_log_entry -def test_log_queue_requests(): - """ - Test the log queue requests generation. - """ - workload_name = WorkloadInstanceName( - workload_name="nginx", agent_name="agent_A", workload_id="1234" - ) - - log_queue = LogQueue(LogsRequest(workload_names=[workload_name])) - assert log_queue is not None - - request = log_queue._get_request() - request_id = request.get_id() - assert isinstance(request, LogsRequest) - - cancel_request = log_queue._get_cancel_request() - assert isinstance(cancel_request, LogsCancelRequest) - assert cancel_request.get_id() == request_id - - def test_log_queue(): """ Test the queue functionality. From 221cc0998e878a3b6741c25e36be4d635535ed62 Mon Sep 17 00:00:00 2001 From: Tomuta Gabriel Date: Wed, 21 Jan 2026 13:14:56 +0200 Subject: [PATCH 6/8] Make proto event entry valid --- tests/response/test_response.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/tests/response/test_response.py b/tests/response/test_response.py index e1ac57e0..d25c31ff 100644 --- a/tests/response/test_response.py +++ b/tests/response/test_response.py @@ -28,6 +28,7 @@ LogsStopResponse, EventEntry, ) +from tests.test_complete_state import CONFIGS_PROTO from tests.response.test_log_response import generate_test_log_entry @@ -130,12 +131,16 @@ workloads=_ank_base.WorkloadMap( workloads={}, ), + configs=CONFIGS_PROTO, ), ), alteredFields=_ank_base.AlteredFields( - addedFields=["field1"], - updatedFields=["field2"], - removedFields=["field3"], + addedFields=[ + "desiredState.configs.config2", + "desiredState.configs.config3", + ], + updatedFields=["desiredState.configs.config1"], + removedFields=["desiredState.configs.config4"], ), ), ) From 556a32153dac4390408916e2b62ed62e647ddd38 Mon Sep 17 00:00:00 2001 From: Tomuta Gabriel Date: Wed, 21 Jan 2026 13:32:50 +0200 Subject: [PATCH 7/8] Fix examples and add doc for state_event --- ankaios_sdk/_components/complete_state.py | 1 + ankaios_sdk/_components/event_campaign.py | 1 - ankaios_sdk/_components/log_campaign.py | 1 - examples/README.md | 4 ++++ 4 files changed, 5 insertions(+), 2 deletions(-) diff --git a/ankaios_sdk/_components/complete_state.py b/ankaios_sdk/_components/complete_state.py index a3695f8e..8e1ace7b 100644 --- a/ankaios_sdk/_components/complete_state.py +++ b/ankaios_sdk/_components/complete_state.py @@ -99,6 +99,7 @@ def __init__( the complete state. """ self._complete_state = _ank_base.CompleteState() + self._complete_state.desiredState.workloads.workloads.clear() self._set_api_version(SUPPORTED_API_VERSION) if _proto: self._complete_state = _proto diff --git a/ankaios_sdk/_components/event_campaign.py b/ankaios_sdk/_components/event_campaign.py index 70ee7515..a4b36846 100644 --- a/ankaios_sdk/_components/event_campaign.py +++ b/ankaios_sdk/_components/event_campaign.py @@ -55,7 +55,6 @@ class EventQueue(Queue): All objects in this queue are of type :py:type:`EventEntry`. """ - # -- # pylint: disable=too-many-arguments def __init__( self, request_id: EventsRequest, diff --git a/ankaios_sdk/_components/log_campaign.py b/ankaios_sdk/_components/log_campaign.py index f7d58d54..1d568f72 100644 --- a/ankaios_sdk/_components/log_campaign.py +++ b/ankaios_sdk/_components/log_campaign.py @@ -78,7 +78,6 @@ class LogQueue(Queue): All objects in this queue are of type :py:type:`LogResponse`. """ - # --- # pylint: disable=too-many-arguments def __init__( self, request_id: str, diff --git a/examples/README.md b/examples/README.md index 457aa694..1c4a70d4 100644 --- a/examples/README.md +++ b/examples/README.md @@ -78,6 +78,10 @@ This example has the Python SDK installed and allows to execute an interactive s The `get_logs` example workload starts another workload that prints 5 log lines and then it requests them and prints them. +### `state_event` + +The `state_event` example workload registers on the events of a specific workload, and then it starts it, showcasing the basic functionality of the events and their format. + ## Enter a container To `exec` into the example container, a convenience script is provided: From 66cfc8b9db3e2b9d486df09595b46f9f2ca5bbf0 Mon Sep 17 00:00:00 2001 From: Tomuta Gabriel Date: Wed, 21 Jan 2026 15:08:47 +0200 Subject: [PATCH 8/8] Add branch proto fetching --- setup.py | 24 ++++++++++++++++++++++-- tests/test_complete_state.py | 5 ++++- 2 files changed, 26 insertions(+), 3 deletions(-) diff --git a/setup.py b/setup.py index d5ebc4aa..1bcc9274 100644 --- a/setup.py +++ b/setup.py @@ -19,6 +19,7 @@ PROJECT_DIR = "ankaios_sdk" ANKAIOS_RELEASE_LINK = "https://github.com/eclipse-ankaios/ankaios/releases/download/v{version}/{file}" ANKAIOS_MAIN_LINK = "https://raw.githubusercontent.com/eclipse-ankaios/ankaios/refs/heads/main/ankaios_api/proto/{file}" +ANKAIOS_BRANCH_LINK = "https://raw.githubusercontent.com/eclipse-ankaios/ankaios/refs/heads/{branch}/ankaios_api/proto/{file}" PROTO_FILES = ["ank_base.proto", "control_api.proto"] config = configparser.ConfigParser() @@ -26,16 +27,35 @@ def extract_the_proto_files(): - """Download the proto files from the ankaios release branch.""" + """ + Download the proto files from the ankaios repository. + + The script supports three ways to fetch proto files: + 1. Default functionality, fetch the protos from a release version of Ankaios. + + 2. If the `ankaios_version` ends with '-pre', the proto files + should be fetched from the main branch of Ankaios. + + 3. If the `ANKAIOS_PROTO_BRANCH` env variable is set, fetch the protos from tha specific Ankaios branch. + Example: ANKAIOS_PROTO_BRANCH=my-feature-branch pip install . + + Priority: custom branch > main branch (for -pre) > release version + """ import requests ankaios_version = config["metadata"]["ankaios_version"] + custom_branch = os.environ.get("ANKAIOS_PROTO_BRANCH") if not os.path.exists(f"{PROJECT_DIR}/_protos/{ankaios_version}"): os.makedirs(f"{PROJECT_DIR}/_protos/{ankaios_version}") for file in PROTO_FILES: - if ankaios_version.endswith("-pre"): + if custom_branch: + file_url = ANKAIOS_BRANCH_LINK.format( + branch=custom_branch, file=file + ) + print(f"Using custom branch '{custom_branch}' for proto files.") + elif ankaios_version.endswith("-pre"): file_url = ANKAIOS_MAIN_LINK.format(file=file) else: file_url = ANKAIOS_RELEASE_LINK.format( diff --git a/tests/test_complete_state.py b/tests/test_complete_state.py index bec9c95e..bead1df8 100644 --- a/tests/test_complete_state.py +++ b/tests/test_complete_state.py @@ -89,7 +89,10 @@ def test_general_functionality(): assert complete_state.get_api_version() == SUPPORTED_API_VERSION complete_state._set_api_version("v0.2") assert complete_state.get_api_version() == "v0.2" - assert str(complete_state) == 'desiredState {\n apiVersion: "v0.2"\n}\n' + assert ( + str(complete_state) + == 'desiredState {\n apiVersion: "v0.2"\n workloads {\n }\n}\n' + ) def test_workload_functionality():