From ad3ef532e4113e56f9d68dc2cfca0185496f09df Mon Sep 17 00:00:00 2001 From: James Paul Turner Date: Thu, 11 Jun 2026 16:05:04 +0100 Subject: [PATCH 01/10] Temp rename of broadcast dir. --- src/drunc/{broadcast => REMOVE_ME_broadcast}/__init__.py | 0 src/drunc/{broadcast => REMOVE_ME_broadcast}/client/__init__.py | 0 .../client/broadcast_handler.py | 0 .../client/broadcast_handler_implementation.py | 0 .../{broadcast => REMOVE_ME_broadcast}/client/configuration.py | 0 .../client/grpc_stdout_broadcast_handler.py | 0 .../client/kafka_stdout_broadcast_handler.py | 0 src/drunc/{broadcast => REMOVE_ME_broadcast}/server/__init__.py | 0 .../server/broadcast_sender.py | 0 .../server/broadcast_sender_implementation.py | 0 .../{broadcast => REMOVE_ME_broadcast}/server/configuration.py | 0 .../{broadcast => REMOVE_ME_broadcast}/server/decorators.py | 2 -- .../{broadcast => REMOVE_ME_broadcast}/server/grpc_servicer.py | 0 .../{broadcast => REMOVE_ME_broadcast}/server/kafka_sender.py | 0 src/drunc/{broadcast => REMOVE_ME_broadcast}/types.py | 0 src/drunc/{broadcast => REMOVE_ME_broadcast}/utils.py | 0 16 files changed, 2 deletions(-) rename src/drunc/{broadcast => REMOVE_ME_broadcast}/__init__.py (100%) rename src/drunc/{broadcast => REMOVE_ME_broadcast}/client/__init__.py (100%) rename src/drunc/{broadcast => REMOVE_ME_broadcast}/client/broadcast_handler.py (100%) rename src/drunc/{broadcast => REMOVE_ME_broadcast}/client/broadcast_handler_implementation.py (100%) rename src/drunc/{broadcast => REMOVE_ME_broadcast}/client/configuration.py (100%) rename src/drunc/{broadcast => REMOVE_ME_broadcast}/client/grpc_stdout_broadcast_handler.py (100%) rename src/drunc/{broadcast => REMOVE_ME_broadcast}/client/kafka_stdout_broadcast_handler.py (100%) rename src/drunc/{broadcast => REMOVE_ME_broadcast}/server/__init__.py (100%) rename src/drunc/{broadcast => REMOVE_ME_broadcast}/server/broadcast_sender.py (100%) rename src/drunc/{broadcast => REMOVE_ME_broadcast}/server/broadcast_sender_implementation.py (100%) rename src/drunc/{broadcast => REMOVE_ME_broadcast}/server/configuration.py (100%) rename src/drunc/{broadcast => REMOVE_ME_broadcast}/server/decorators.py (99%) rename src/drunc/{broadcast => REMOVE_ME_broadcast}/server/grpc_servicer.py (100%) rename src/drunc/{broadcast => REMOVE_ME_broadcast}/server/kafka_sender.py (100%) rename src/drunc/{broadcast => REMOVE_ME_broadcast}/types.py (100%) rename src/drunc/{broadcast => REMOVE_ME_broadcast}/utils.py (100%) diff --git a/src/drunc/broadcast/__init__.py b/src/drunc/REMOVE_ME_broadcast/__init__.py similarity index 100% rename from src/drunc/broadcast/__init__.py rename to src/drunc/REMOVE_ME_broadcast/__init__.py diff --git a/src/drunc/broadcast/client/__init__.py b/src/drunc/REMOVE_ME_broadcast/client/__init__.py similarity index 100% rename from src/drunc/broadcast/client/__init__.py rename to src/drunc/REMOVE_ME_broadcast/client/__init__.py diff --git a/src/drunc/broadcast/client/broadcast_handler.py b/src/drunc/REMOVE_ME_broadcast/client/broadcast_handler.py similarity index 100% rename from src/drunc/broadcast/client/broadcast_handler.py rename to src/drunc/REMOVE_ME_broadcast/client/broadcast_handler.py diff --git a/src/drunc/broadcast/client/broadcast_handler_implementation.py b/src/drunc/REMOVE_ME_broadcast/client/broadcast_handler_implementation.py similarity index 100% rename from src/drunc/broadcast/client/broadcast_handler_implementation.py rename to src/drunc/REMOVE_ME_broadcast/client/broadcast_handler_implementation.py diff --git a/src/drunc/broadcast/client/configuration.py b/src/drunc/REMOVE_ME_broadcast/client/configuration.py similarity index 100% rename from src/drunc/broadcast/client/configuration.py rename to src/drunc/REMOVE_ME_broadcast/client/configuration.py diff --git a/src/drunc/broadcast/client/grpc_stdout_broadcast_handler.py b/src/drunc/REMOVE_ME_broadcast/client/grpc_stdout_broadcast_handler.py similarity index 100% rename from src/drunc/broadcast/client/grpc_stdout_broadcast_handler.py rename to src/drunc/REMOVE_ME_broadcast/client/grpc_stdout_broadcast_handler.py diff --git a/src/drunc/broadcast/client/kafka_stdout_broadcast_handler.py b/src/drunc/REMOVE_ME_broadcast/client/kafka_stdout_broadcast_handler.py similarity index 100% rename from src/drunc/broadcast/client/kafka_stdout_broadcast_handler.py rename to src/drunc/REMOVE_ME_broadcast/client/kafka_stdout_broadcast_handler.py diff --git a/src/drunc/broadcast/server/__init__.py b/src/drunc/REMOVE_ME_broadcast/server/__init__.py similarity index 100% rename from src/drunc/broadcast/server/__init__.py rename to src/drunc/REMOVE_ME_broadcast/server/__init__.py diff --git a/src/drunc/broadcast/server/broadcast_sender.py b/src/drunc/REMOVE_ME_broadcast/server/broadcast_sender.py similarity index 100% rename from src/drunc/broadcast/server/broadcast_sender.py rename to src/drunc/REMOVE_ME_broadcast/server/broadcast_sender.py diff --git a/src/drunc/broadcast/server/broadcast_sender_implementation.py b/src/drunc/REMOVE_ME_broadcast/server/broadcast_sender_implementation.py similarity index 100% rename from src/drunc/broadcast/server/broadcast_sender_implementation.py rename to src/drunc/REMOVE_ME_broadcast/server/broadcast_sender_implementation.py diff --git a/src/drunc/broadcast/server/configuration.py b/src/drunc/REMOVE_ME_broadcast/server/configuration.py similarity index 100% rename from src/drunc/broadcast/server/configuration.py rename to src/drunc/REMOVE_ME_broadcast/server/configuration.py diff --git a/src/drunc/broadcast/server/decorators.py b/src/drunc/REMOVE_ME_broadcast/server/decorators.py similarity index 99% rename from src/drunc/broadcast/server/decorators.py rename to src/drunc/REMOVE_ME_broadcast/server/decorators.py index ece3f95dd..e8407748c 100644 --- a/src/drunc/broadcast/server/decorators.py +++ b/src/drunc/REMOVE_ME_broadcast/server/decorators.py @@ -1,5 +1,3 @@ - - from drunc.utils.utils import get_logger diff --git a/src/drunc/broadcast/server/grpc_servicer.py b/src/drunc/REMOVE_ME_broadcast/server/grpc_servicer.py similarity index 100% rename from src/drunc/broadcast/server/grpc_servicer.py rename to src/drunc/REMOVE_ME_broadcast/server/grpc_servicer.py diff --git a/src/drunc/broadcast/server/kafka_sender.py b/src/drunc/REMOVE_ME_broadcast/server/kafka_sender.py similarity index 100% rename from src/drunc/broadcast/server/kafka_sender.py rename to src/drunc/REMOVE_ME_broadcast/server/kafka_sender.py diff --git a/src/drunc/broadcast/types.py b/src/drunc/REMOVE_ME_broadcast/types.py similarity index 100% rename from src/drunc/broadcast/types.py rename to src/drunc/REMOVE_ME_broadcast/types.py diff --git a/src/drunc/broadcast/utils.py b/src/drunc/REMOVE_ME_broadcast/utils.py similarity index 100% rename from src/drunc/broadcast/utils.py rename to src/drunc/REMOVE_ME_broadcast/utils.py From 054df89bed500019c10c950611c5e43af58ed3f0 Mon Sep 17 00:00:00 2001 From: James Paul Turner Date: Thu, 11 Jun 2026 16:28:42 +0100 Subject: [PATCH 02/10] Remove broadcasting from controller. --- src/drunc/controller/controller.py | 53 ------------------------------ 1 file changed, 53 deletions(-) diff --git a/src/drunc/controller/controller.py b/src/drunc/controller/controller.py index c73f8174b..69ffef1fa 100644 --- a/src/drunc/controller/controller.py +++ b/src/drunc/controller/controller.py @@ -6,7 +6,6 @@ from daqpytools.logging import LogHandlerConf, setup_daq_ers_logger from druncschema.authoriser_pb2 import ActionType, SystemType -from druncschema.broadcast_pb2 import BroadcastType from druncschema.controller_pb2 import ( DescribeFSMRequest, DescribeFSMResponse, @@ -47,9 +46,6 @@ from drunc.authoriser.configuration import DummyAuthoriserConfHandler from drunc.authoriser.decorators import authentified_and_authorised from drunc.authoriser.dummy_authoriser import DummyAuthoriser -from drunc.broadcast.server.broadcast_sender import BroadcastSender -from drunc.broadcast.server.configuration import BroadcastSenderConfHandler -from drunc.broadcast.server.decorators import broadcasted from drunc.connectivity_service.client import ConnectivityServiceClient from drunc.connectivity_service.exceptions import ApplicationLookupUnsuccessful from drunc.controller.children_interface.child_node import ChildNode @@ -85,7 +81,6 @@ def __init__(self, configuration, name: str, session: str, token: Token): self._previous_error_state = False self.name = name self.session = session - self.broadcast_service = None self.monitoring_metrics = ControllerMonitoringMetrics() self.handlerconf = LogHandlerConf(init_ers=True) self.log = get_logger(f"controller.core.{name}_ctrl") @@ -109,15 +104,6 @@ def __init__(self, configuration, name: str, session: str, token: Token): self.opmon_publisher = getattr(self.configuration, "opmon_publisher", None) self.stop_event: threading.Event | None = None self.thread: threading.Thread | None = None - bsch = BroadcastSenderConfHandler( - data=self.configuration.data.controller.broadcaster, - ) - - self.broadcast_service = BroadcastSender( - name=name, - session=session, - configuration=bsch, - ) self.fsm_config = FSMConfHandler( data=self.configuration.data.controller.fsm, @@ -256,28 +242,9 @@ def init_controller(self) -> None: ) self.thread.start() - self.broadcast(message="ready", btype=BroadcastType.SERVER_READY) self.stateful_node.set_ready_state(True) log_init_controller.info("Controller ready") - """ - A couple of simple pass-through functions to the broadcasting service - """ - - def broadcast(self, *args, **kwargs): - return self.broadcast_service.broadcast(*args, **kwargs) - - def can_broadcast(self, *args, **kwargs): - if self.broadcast_service: - return self.broadcast_service.can_broadcast(*args, **kwargs) - return False - - def describe_broadcast(self, *args, **kwargs): - return self.broadcast_service.describe_broadcast(*args, **kwargs) - - def interrupt_with_exception(self, *args, **kwargs): - return self.broadcast_service._interrupt_with_exception(*args, **kwargs) - def controller_publisher(self, message, custom_origin: dict | None = None): if isinstance(message, FSMStatus) and message.in_error: if message.in_error and not self._previous_error_state: @@ -394,12 +361,6 @@ def terminate(self): self.log.info("Unregistering from the connectivity service") self.connectivity_service.retract(self.name + "_control", fail_quickly=True) - if self.can_broadcast(): - self.broadcast( - btype=BroadcastType.SERVER_SHUTDOWN, - message="over_and_out", - ) - self.log.info("Stopping children") for child in self.children_nodes: self.log.debug(f"Stopping {child.name}") @@ -625,7 +586,6 @@ def _partition_connected_children( ############# Status, description commands ############# ######################################################## - @broadcasted @authentified_and_authorised(action=ActionType.READ, system=SystemType.CONTROLLER) @publish_command_time def status( @@ -688,7 +648,6 @@ def status( return response - @broadcasted @authentified_and_authorised(action=ActionType.READ, system=SystemType.CONTROLLER) @publish_command_time def describe( @@ -717,8 +676,6 @@ def describe( session=self.session, commands=None, ) - if broadcast_description := self.describe_broadcast(): - description.broadcast.Pack(broadcast_description) response.description.CopyFrom(description) # Children nodes (ignore exclusion). @@ -754,7 +711,6 @@ def describe( return response - @broadcasted @authentified_and_authorised(action=ActionType.READ, system=SystemType.CONTROLLER) @publish_command_time def describe_fsm( @@ -837,7 +793,6 @@ def describe_fsm( ############# FSM commands ############# ######################################## - @broadcasted @authentified_and_authorised(action=ActionType.UPDATE, system=SystemType.CONTROLLER) @in_control @publish_command_time @@ -1016,7 +971,6 @@ def execute_fsm_command( return response - @broadcasted @authentified_and_authorised(action=ActionType.EXPERT, system=SystemType.CONTROLLER) @in_control @publish_command_time @@ -1076,7 +1030,6 @@ def execute_expert_command( return response - @broadcasted @authentified_and_authorised(action=ActionType.UPDATE, system=SystemType.CONTROLLER) @in_control @publish_command_time @@ -1142,7 +1095,6 @@ def include( return response - @broadcasted @authentified_and_authorised(action=ActionType.UPDATE, system=SystemType.CONTROLLER) @in_control @publish_command_time @@ -1208,7 +1160,6 @@ def exclude( return response - @broadcasted @authentified_and_authorised(action=ActionType.UPDATE, system=SystemType.CONTROLLER) @in_control @publish_command_time @@ -1337,7 +1288,6 @@ def recompute_status( ############# Actor commands ############# ########################################## - @broadcasted @authentified_and_authorised(action=ActionType.UPDATE, system=SystemType.CONTROLLER) @publish_command_time def take_control( @@ -1413,7 +1363,6 @@ def take_control( return response - @broadcasted @authentified_and_authorised(action=ActionType.UPDATE, system=SystemType.CONTROLLER) @in_control @publish_command_time @@ -1490,7 +1439,6 @@ def surrender_control( return response - @broadcasted @authentified_and_authorised(action=ActionType.READ, system=SystemType.CONTROLLER) @publish_command_time def who_is_in_charge( @@ -1552,7 +1500,6 @@ def who_is_in_charge( ####### Integration test commands ######## ########################################## - @broadcasted @authentified_and_authorised(action=ActionType.UPDATE, system=SystemType.CONTROLLER) @in_control @publish_command_time From 7ad16d41d0013115ccc5fefda755669bc9f91891 Mon Sep 17 00:00:00 2001 From: James Paul Turner Date: Thu, 11 Jun 2026 16:49:19 +0100 Subject: [PATCH 03/10] Remove broadcasting from rest of controller package. --- .../controller/children_interface/grpc_child.py | 17 ++--------------- src/drunc/controller/configuration.py | 1 - src/drunc/controller/interface/context.py | 9 --------- src/drunc/controller/interface/shell_utils.py | 3 --- src/drunc/controller/stateful_node.py | 8 -------- 5 files changed, 2 insertions(+), 36 deletions(-) diff --git a/src/drunc/controller/children_interface/grpc_child.py b/src/drunc/controller/children_interface/grpc_child.py index 54c7f6f0d..c5f9910ff 100644 --- a/src/drunc/controller/children_interface/grpc_child.py +++ b/src/drunc/controller/children_interface/grpc_child.py @@ -35,15 +35,13 @@ from druncschema.token_pb2 import Token from grpc_status import rpc_status -from drunc.broadcast.client.broadcast_handler import BroadcastHandler -from drunc.broadcast.client.configuration import BroadcastClientConfHandler from drunc.connectivity_service.exceptions import ( ApplicationLookupUnsuccessful, ) from drunc.controller.children_interface.child_node import ChildNode from drunc.exceptions import DruncSetupException from drunc.grpc_settings import CONTROLLER_CLIENT_GRPC_CONFIG -from drunc.utils.configuration import ConfHandler, ConfTypes +from drunc.utils.configuration import ConfHandler from drunc.utils.grpc_utils import ( ServerUnreachable, rethrow_if_unreachable_server, @@ -116,7 +114,7 @@ def _setup_connection(self): tries_remaining -= 1 try: - response = self.stub.describe(request) + self.stub.describe(request) except grpc.RpcError as error: if tries_remaining == 0: @@ -131,7 +129,6 @@ def _setup_connection(self): else: self.log.info(f"Connected to the controller ({self.uri})!") - self.start_listening(response.description.broadcast) break def _attempt_reconnection(self, retry_call): @@ -185,9 +182,7 @@ def terminate(self) -> None: del self.channel if self.stub: del self.stub - self.channel = None - self.broadcast.stop() def check_connection(self) -> bool: """Probe child connectivity and retry once after reconnecting if needed. @@ -233,14 +228,6 @@ def check_connection(self) -> bool: return False - def start_listening(self, bdesc): - self.broadcast = BroadcastHandler( - BroadcastClientConfHandler( - data=bdesc, - type=ConfTypes.ProtobufAny, - ) - ) - def status( self, target: str = "", diff --git a/src/drunc/controller/configuration.py b/src/drunc/controller/configuration.py index 64944fa8c..18f4fab12 100644 --- a/src/drunc/controller/configuration.py +++ b/src/drunc/controller/configuration.py @@ -39,7 +39,6 @@ class cler: pass self.controller = cler() - self.controller.broadcaster = id_able() self.controller.fsm = id_able() diff --git a/src/drunc/controller/interface/context.py b/src/drunc/controller/interface/context.py index 082c42a00..8da69f38e 100644 --- a/src/drunc/controller/interface/context.py +++ b/src/drunc/controller/interface/context.py @@ -2,10 +2,7 @@ from druncschema.token_pb2 import Token -from drunc.broadcast.client.broadcast_handler import BroadcastHandler -from drunc.broadcast.client.configuration import BroadcastClientConfHandler from drunc.controller.controller_driver import ControllerDriver -from drunc.utils.configuration import ConfTypes from drunc.utils.shell_utils import ( ShellContext, create_dummy_token_from_uname, @@ -33,12 +30,6 @@ def create_drivers(self, **kwargs) -> Mapping[str, object]: def create_token(self, **kwargs) -> Token: return create_dummy_token_from_uname() - def start_listening_controller(self, broadcaster_conf): - bcch = BroadcastClientConfHandler( - data=broadcaster_conf, type=ConfTypes.ProtobufAny - ) - self.status_receiver = BroadcastHandler(broadcast_configuration=bcch) - def terminate(self): if self.status_receiver: self.status_receiver.stop() diff --git a/src/drunc/controller/interface/shell_utils.py b/src/drunc/controller/interface/shell_utils.py index 3ffba7455..190a7c305 100644 --- a/src/drunc/controller/interface/shell_utils.py +++ b/src/drunc/controller/interface/shell_utils.py @@ -210,7 +210,6 @@ def get_renderable(self) -> ConsoleRenderable | RichCast | str: def controller_cleanup_wrapper(ctx): def controller_cleanup(): log = logging.getLogger("controller.shell_utils") - # remove the shell from the controller broadcast list dead = False who = "" @@ -291,8 +290,6 @@ def controller_setup(ctx, controller_address): f"{controller_address} is '{desc.name}.{desc.session}' (name.session), starting listening..." ) ctx.get_driver("controller").name = f"{desc.name}.{desc.session}" - if desc.HasField("broadcast"): - ctx.start_listening_controller(desc.broadcast) log.debug("Connected to the controller") diff --git a/src/drunc/controller/stateful_node.py b/src/drunc/controller/stateful_node.py index 007bd200f..d503c63fa 100644 --- a/src/drunc/controller/stateful_node.py +++ b/src/drunc/controller/stateful_node.py @@ -20,14 +20,6 @@ def value(self): @value.setter def value(self, value): - # if self._broadcast_on_change is None or self._broadcast_key is None: - # self._value = value - # return - - # self._broadcast_on_change.broadcast( - # message = f'Changing {self._name} from {self._value} to {value}', - # btype = self._broadcast_key, - # ) self._value = value if self.stateful_node: self.stateful_node.log.info(f"{self._name} changed to {value}") From c4a49b58961cd10e4dcf19925f61c86bab9b4fb4 Mon Sep 17 00:00:00 2001 From: James Paul Turner Date: Thu, 11 Jun 2026 16:50:12 +0100 Subject: [PATCH 04/10] Remove all of broadcast package. --- src/drunc/REMOVE_ME_broadcast/__init__.py | 0 .../REMOVE_ME_broadcast/client/__init__.py | 0 .../client/broadcast_handler.py | 38 --- .../broadcast_handler_implementation.py | 7 - .../client/configuration.py | 43 --- .../client/grpc_stdout_broadcast_handler.py | 109 ------- .../client/kafka_stdout_broadcast_handler.py | 106 ------- .../REMOVE_ME_broadcast/server/__init__.py | 0 .../server/broadcast_sender.py | 128 -------- .../server/broadcast_sender_implementation.py | 17 -- .../server/configuration.py | 34 --- .../REMOVE_ME_broadcast/server/decorators.py | 48 --- .../server/grpc_servicer.py | 282 ------------------ .../server/kafka_sender.py | 69 ----- src/drunc/REMOVE_ME_broadcast/types.py | 15 - src/drunc/REMOVE_ME_broadcast/utils.py | 30 -- 16 files changed, 926 deletions(-) delete mode 100644 src/drunc/REMOVE_ME_broadcast/__init__.py delete mode 100644 src/drunc/REMOVE_ME_broadcast/client/__init__.py delete mode 100644 src/drunc/REMOVE_ME_broadcast/client/broadcast_handler.py delete mode 100644 src/drunc/REMOVE_ME_broadcast/client/broadcast_handler_implementation.py delete mode 100644 src/drunc/REMOVE_ME_broadcast/client/configuration.py delete mode 100644 src/drunc/REMOVE_ME_broadcast/client/grpc_stdout_broadcast_handler.py delete mode 100644 src/drunc/REMOVE_ME_broadcast/client/kafka_stdout_broadcast_handler.py delete mode 100644 src/drunc/REMOVE_ME_broadcast/server/__init__.py delete mode 100644 src/drunc/REMOVE_ME_broadcast/server/broadcast_sender.py delete mode 100644 src/drunc/REMOVE_ME_broadcast/server/broadcast_sender_implementation.py delete mode 100644 src/drunc/REMOVE_ME_broadcast/server/configuration.py delete mode 100644 src/drunc/REMOVE_ME_broadcast/server/decorators.py delete mode 100644 src/drunc/REMOVE_ME_broadcast/server/grpc_servicer.py delete mode 100644 src/drunc/REMOVE_ME_broadcast/server/kafka_sender.py delete mode 100644 src/drunc/REMOVE_ME_broadcast/types.py delete mode 100644 src/drunc/REMOVE_ME_broadcast/utils.py diff --git a/src/drunc/REMOVE_ME_broadcast/__init__.py b/src/drunc/REMOVE_ME_broadcast/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/src/drunc/REMOVE_ME_broadcast/client/__init__.py b/src/drunc/REMOVE_ME_broadcast/client/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/src/drunc/REMOVE_ME_broadcast/client/broadcast_handler.py b/src/drunc/REMOVE_ME_broadcast/client/broadcast_handler.py deleted file mode 100644 index 1f144f582..000000000 --- a/src/drunc/REMOVE_ME_broadcast/client/broadcast_handler.py +++ /dev/null @@ -1,38 +0,0 @@ -from drunc.broadcast.client.configuration import BroadcastClientConfHandler -from drunc.broadcast.types import BroadcastTypes - - -class BroadcastHandler: - def __init__(self, broadcast_configuration: BroadcastClientConfHandler): - super().__init__() - - from logging import getLogger - - self.log = getLogger("BroadcastHandler") - - self.configuration = broadcast_configuration - self.implementation = None - - match self.configuration.data.type: - # Being a bit sloppy here, having a Kafka sender doesn't mean we want to dump everything to stdout - # There could be cases where we want to do other things. - # For now, 1 server type <-> 1 client type... - # Maybe in the future some sort of callback-based functionality would be preferable. - case BroadcastTypes.Kafka: - from druncschema.broadcast_pb2 import BroadcastMessage - - from drunc.broadcast.client.kafka_stdout_broadcast_handler import ( - KafkaStdoutBroadcastHandler, - ) - - self.implementation = KafkaStdoutBroadcastHandler( - message_format=BroadcastMessage, conf=self.configuration - ) - case _: - self.log.info( - "Could not understand the BroadcastHandler technology you want to use, you will get no broadcast!" - ) - - def stop(self): - if self.implementation: - self.implementation.stop() diff --git a/src/drunc/REMOVE_ME_broadcast/client/broadcast_handler_implementation.py b/src/drunc/REMOVE_ME_broadcast/client/broadcast_handler_implementation.py deleted file mode 100644 index 99d0fa95f..000000000 --- a/src/drunc/REMOVE_ME_broadcast/client/broadcast_handler_implementation.py +++ /dev/null @@ -1,7 +0,0 @@ -import abc - - -class BroadcastHandlerImplementation(abc.ABC): - @abc.abstractmethod - def stop(self): - pass diff --git a/src/drunc/REMOVE_ME_broadcast/client/configuration.py b/src/drunc/REMOVE_ME_broadcast/client/configuration.py deleted file mode 100644 index 4b5cdd7f1..000000000 --- a/src/drunc/REMOVE_ME_broadcast/client/configuration.py +++ /dev/null @@ -1,43 +0,0 @@ -from drunc.broadcast.types import BroadcastTypes -from drunc.utils.configuration import ConfHandler - - -class BroadcastClientConfData: # OKSeroo - def __init__(self, type: BroadcastTypes, address: str, topic: str): - self.type = type - self.address = address - self.topic = topic - - -class BroadcastClientConfHandler(ConfHandler): - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - - def get_impl_technology(self): - return self.impl_technology - - def _parse_pbany(self, data): - # potentially do something more complicated with different implementation technology here - # match data.format(): - # case KafkaBroadcastHandlerConfiguration - # ... - - from druncschema.broadcast_pb2 import KafkaBroadcastHandlerConfiguration - - from drunc.utils.grpc_utils import UnpackingError, unpack_any - - if not data.ByteSize(): - return BroadcastClientConfData(type=None, address=None, topic=None) - try: - data = unpack_any(data, KafkaBroadcastHandlerConfiguration) - return BroadcastClientConfData( - type=BroadcastTypes.Kafka, address=data.kafka_address, topic=data.topic - ) - - except UnpackingError as e: - from drunc.exceptions import DruncSetupException - - raise DruncSetupException( - f"Input configuration to configure the broadcast was not understood, could not setup the broadcast handler: {e}", - e, - ) diff --git a/src/drunc/REMOVE_ME_broadcast/client/grpc_stdout_broadcast_handler.py b/src/drunc/REMOVE_ME_broadcast/client/grpc_stdout_broadcast_handler.py deleted file mode 100644 index e0faa89b5..000000000 --- a/src/drunc/REMOVE_ME_broadcast/client/grpc_stdout_broadcast_handler.py +++ /dev/null @@ -1,109 +0,0 @@ -import grpc -from druncschema.broadcast_pb2 import BroadcastMessage, BroadcastType -from druncschema.broadcast_pb2_grpc import BroadcastReceiverServicer -from druncschema.generic_pb2 import Empty - -from drunc.controller.configuration import ControllerConfHandler - - -class gRPCStdoutBroadcastHandler(BroadcastReceiverServicer): - def __init__(self, conf: ControllerConfHandler, token, **kwargs) -> None: - super(gRPCStdoutBroadcastHandler, self).__init__(**kwargs) - from drunc.exceptions import DruncSetupException - - raise DruncSetupException( - "gRPCStdoutBroadcastHandler is not handled it needs to be reworked!" - ) - self.ready = False - self.stub = None - self.token = token - from logging import getLogger - - self._log = getLogger("BroadcastReceiver") - self._address = None # f'[::]:{port}' - self._log.debug("Broadcast receiver initialised") - - def stop_receiving(self) -> None: - self._server.stop(0) - self._log.debug("Broadcast receiver stopped") - - def connect(self) -> None: - from druncschema.broadcast_pb2 import BroadcastRequest - - from drunc.utils.grpc_utils import send_command - - self._log.info(f"Connecting to {self.stub}") - try: - send_command( - controller=self.stub, - token=self.token, - command="add_to_broadcast_list", - data=BroadcastRequest(broadcast_receiver_address=self._address), - rethrow=True, - ) - - except Exception as e: - self._log.error("Could not connect to service to receive broadcast") - self._log.error(str(e)) - raise e - - def disconnect(self) -> None: - from druncschema.broadcast_receiver_pb2 import BroadcastRequest - - from drunc.utils.grpc_utils import send_command - - try: - send_command( - controller=self.stub, - token=self.token, - command="remove_from_broadcast_list", - data=BroadcastRequest(broadcast_receiver_address=self._address), - rethrow=True, - ) - - except Exception as e: - self._log.error( - "Could not disconnect from broadcaster (maybe it' dead and you won't receive any broadcast anyway)" - ) - self._log.error(str(e)) - - def terminate(self) -> None: - self.disconnect() - self.stop_receiving() - - def serve(self) -> None: - from concurrent import futures - - self._server = grpc.server(futures.ThreadPoolExecutor(max_workers=1)) - - from druncschema.broadcast_pb2_grpc import ( - add_BroadcastReceiverServicer_to_server, - ) - - add_BroadcastReceiverServicer_to_server(self, self._server) - - self._server.add_insecure_port(self._address) - - self._server.start() - self.ready = True - self._log.debug("Broadcast receiver server started") - self._server.wait_for_termination() - self.ready = False - - def handle_broadcast( - self, bm: BroadcastMessage, context: grpc.aio.ServicerContext = None - ) -> Empty: - from druncschema.generic_pb2 import PlainText - - from drunc.utils.grpc_utils import unpack_any - - type = bm.type - if type == BroadcastType.TEXT_MESSAGE: - pt = unpack_any(bm.data, PlainText) - self._log.info(f"{bm.emitter}: {pt}") - elif type == BroadcastType.ACK: - self._log.info(f"{bm.emitter}: Ack") - elif type == BroadcastType.SERVER_SHUTDOWN: - self._log.info(f"{bm.emitter} is shutting down") - - return Empty() diff --git a/src/drunc/REMOVE_ME_broadcast/client/kafka_stdout_broadcast_handler.py b/src/drunc/REMOVE_ME_broadcast/client/kafka_stdout_broadcast_handler.py deleted file mode 100644 index 36022ee96..000000000 --- a/src/drunc/REMOVE_ME_broadcast/client/kafka_stdout_broadcast_handler.py +++ /dev/null @@ -1,106 +0,0 @@ -from drunc.broadcast.client.broadcast_handler_implementation import ( - BroadcastHandlerImplementation, -) - - -class KafkaStdoutBroadcastHandler(BroadcastHandlerImplementation): - def __init__(self, message_format, conf): - from drunc.broadcast.utils import broadcast_types_loglevels - - self.broadcast_types_loglevels = ( - broadcast_types_loglevels # in this case, we stick with default - ) - self.conf = conf - # import os - # drunc_shell_conf = os.getenv('DRUNC_SHELL_CONF', None) - # if drunc_shell_conf is not None: - - # with open(drunc_shell_conf) as f: - # import json - # self.global_kafka_stdout_conf = json.load(f).get('kafka_broadcast_handler', {}) - # if 'broadcast_types_loglevels' in self.global_kafka_stdout_conf: - # self.broadcast_types_loglevels.update(self.global_kafka_stdout_conf['broadcast_types_loglevels']) - - self.kafka_address = self.conf.data.address - self.topic = self.conf.data.topic - - # self.broadcast_types_loglevels.update(conf.data.get('broadcast_types_loglevels', {})) - - self.message_format = message_format - - import logging - - self._log = logging.getLogger("Broadcast") - - import getpass - - from drunc.utils.utils import get_random_string, now_str - - group_id = f"drunc-stdout-broadcasthandler-{getpass.getuser()}-{now_str(True)}-{get_random_string(5)}" - - from kafka import KafkaConsumer - - self.consumer = KafkaConsumer( - self.topic, - client_id="run_control", - bootstrap_servers=[self.kafka_address], - group_id=group_id, - ) - - self.run = True - import threading - - self.thread = threading.Thread(target=self.consume) - self.thread.start() - - def stop(self): - self._log.info(f"Stopping listening to '{self.topic}'") - self.run = False - self.thread.join() - - def consume(self): - from druncschema.broadcast_pb2 import BroadcastType - from druncschema.generic_pb2 import PlainText - from google.protobuf import text_format - - from drunc.utils.grpc_utils import unpack_any - - while self.run: - for messages in self.consumer.poll(timeout_ms=500).values(): - for message in messages: - decoded = "" - try: - decoded = self.message_format() - decoded.ParseFromString(message.value) - self._log.debug(f"{decoded=}, {type(decoded)=}") - except Exception as e: - self._log.error( - f"Unhandled broadcast message: {message} (error: {e!s})" - ) - pass - - try: - if decoded.data.Is(PlainText.DESCRIPTOR): - txt = unpack_any(decoded.data, PlainText).text - else: - txt = decoded.data - - from druncschema.broadcast_pb2 import BroadcastType - - from drunc.broadcast.utils import ( - get_broadcast_level_from_broadcast_type, - ) - - bt = BroadcastType.Name(decoded.type) - - get_broadcast_level_from_broadcast_type( - decoded.type, self._log, self.broadcast_types_loglevels - )(f"'{bt}' {txt}") - - except Exception as e: - self._log.error( - f"Weird broadcast message: {message} (error: {e!s})" - ) - text_proto = text_format.MessageToString(decoded) - self._log.info(text_proto) - pass diff --git a/src/drunc/REMOVE_ME_broadcast/server/__init__.py b/src/drunc/REMOVE_ME_broadcast/server/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/src/drunc/REMOVE_ME_broadcast/server/broadcast_sender.py b/src/drunc/REMOVE_ME_broadcast/server/broadcast_sender.py deleted file mode 100644 index f138a3f5a..000000000 --- a/src/drunc/REMOVE_ME_broadcast/server/broadcast_sender.py +++ /dev/null @@ -1,128 +0,0 @@ -from drunc.broadcast.server.configuration import BroadcastSenderConfHandler - - -class BroadcastSender: - implementation = None - - def __init__( - self, - name: str, - configuration: BroadcastSenderConfHandler, - session: str = "no_session", - ): - super().__init__() - - self.configuration = configuration - - self.name = name - self.session = session - self.identifier = f"{self.session}.{self.name}" - - from logging import getLogger - - self.logger = getLogger("Broadcast") - - self.logger.info("Initialising broadcast") - - from drunc.broadcast.utils import broadcast_types_loglevels - - self.broadcast_types_loglevels = broadcast_types_loglevels - - # TODO - # self.broadcast_types_loglevels.update(self.configuration.get_raw('broadcast_types_loglevels', {})) - - self.impl_technology = self.configuration.get_impl_technology() - - self.implementation = None - - if self.impl_technology is None: - self.logger.info("There is no broadcasting service!") - return - - from drunc.broadcast.types import BroadcastTypes - - match self.impl_technology: - case BroadcastTypes.Kafka: - from drunc.broadcast.server.kafka_sender import KafkaSender - - self.implementation = KafkaSender( - self.configuration.data.address, - self.configuration.data.publish_timeout, - topic=f"control.{self.identifier}", - ) - case _: - from drunc.exceptions import DruncSetupException - - raise DruncSetupException( - f"Broadcaster cannot be {self.impl_technology}" - ) - - def describe_broadcast(self): - if self.implementation: - return self.implementation.describe_broadcast() - else: - return None - - def can_broadcast(self): - if not self.implementation: - return False - - return self.implementation.can_broadcast() - - def broadcast(self, message, btype): - if self.logger: - from drunc.broadcast.utils import get_broadcast_level_from_broadcast_type - - get_broadcast_level_from_broadcast_type( - btype, self.logger, self.broadcast_types_loglevels - )(message) - - if self.implementation is None: - # nice and easy case - return - - from druncschema.broadcast_pb2 import BroadcastMessage, Emitter - from druncschema.generic_pb2 import PlainText - - from drunc.utils.grpc_utils import pack_to_any - - any = pack_to_any(PlainText(text=message)) - emitter = Emitter( - process=self.name, - session=self.session, - ) - bm = BroadcastMessage( - emitter=emitter, - type=btype, - data=any, - ) - bm.type = btype - - self.implementation._send(bm) - - def _interrupt_with_exception(self, exception, context, stack=""): - from druncschema.broadcast_pb2 import BroadcastType - - txt = f"'{exception.__class__.__name__}' exception thrown: {exception}" - - from drunc.exceptions import DruncException - - self.broadcast( - btype=( - BroadcastType.DRUNC_EXCEPTION_RAISED - if isinstance(exception, DruncException) - else BroadcastType.UNHANDLED_EXCEPTION_RAISED - ), - message=txt, - ) - - if stack: - txt += "\n\n" + stack - - from google.rpc import code_pb2 - - error_code = getattr(exception, "grpc_error_code", code_pb2.INTERNAL) - context.abort( - code=error_code, - details=txt, - ) diff --git a/src/drunc/REMOVE_ME_broadcast/server/broadcast_sender_implementation.py b/src/drunc/REMOVE_ME_broadcast/server/broadcast_sender_implementation.py deleted file mode 100644 index f6ace3a2b..000000000 --- a/src/drunc/REMOVE_ME_broadcast/server/broadcast_sender_implementation.py +++ /dev/null @@ -1,17 +0,0 @@ -import abc - -from druncschema.broadcast_pb2 import BroadcastMessage - - -class BroadcastSenderImplementation(abc.ABC): - @abc.abstractmethod - def _send(self, bm: BroadcastMessage): - pass - - @abc.abstractmethod - def describe_broadcast(self): - pass - - @abc.abstractmethod - def can_broadcast(self): - pass diff --git a/src/drunc/REMOVE_ME_broadcast/server/configuration.py b/src/drunc/REMOVE_ME_broadcast/server/configuration.py deleted file mode 100644 index fe485e4fa..000000000 --- a/src/drunc/REMOVE_ME_broadcast/server/configuration.py +++ /dev/null @@ -1,34 +0,0 @@ -from drunc.utils.configuration import ConfHandler - - -class KafkaBroadcastSenderConfData: - def __init__(self, address=None, publish_timeout=None): - self.address = address - self.publish_timeout = publish_timeout - - @staticmethod - def from_dict(data: dict): - address = data.get("address") - if address is None: - address = data["kafka_address"] - - return KafkaBroadcastSenderConfData( - address=address, publish_timeout=data["publish_timeout"] - ) - - -class BroadcastSenderConfHandler(ConfHandler): - def _post_process_oks(self): - from drunc.broadcast.types import BroadcastTypes - - self.impl_technology = BroadcastTypes.Kafka if self.data else None - self.log.debug(self.data) - - def get_impl_technology(self): - return self.impl_technology - - def _parse_dict(self, data): - if data == {}: - self.impl_technology = None - return None - return KafkaBroadcastSenderConfData.from_dict(data) diff --git a/src/drunc/REMOVE_ME_broadcast/server/decorators.py b/src/drunc/REMOVE_ME_broadcast/server/decorators.py deleted file mode 100644 index e8407748c..000000000 --- a/src/drunc/REMOVE_ME_broadcast/server/decorators.py +++ /dev/null @@ -1,48 +0,0 @@ -from drunc.utils.utils import get_logger - - -def broadcasted(cmd): - import functools - - @functools.wraps( - cmd - ) # this nifty decorator of decorator (!) is nicely preserving the cmd.__name__ (i.e. signature) - def wrap(obj, request, context): - log = get_logger("broadcasted_decorator", rich_handler=True) - - # hummmm I feel like creating a level myself, but... - # https://docs.python.org/3/howto/logging.html#custom-levels - # lets not - log.debug("Entering") - from druncschema.broadcast_pb2 import BroadcastType - - msg = f"User '{request.token.user_name}' executing '{cmd.__name__}'" - - log.debug(msg) - - obj.broadcast(message=msg, btype=BroadcastType.ACK) - - ret = None - try: - log.debug("Executing wrapped function") - ret = cmd(obj, request, context) - - except Exception as e: - log.exception(e) - - obj.broadcast( - message=f"Command '{cmd.__name__}' failed", - btype=BroadcastType.UNHANDLED_EXCEPTION_RAISED, - ) - # raise the exception to the client so the interceptor can handle it - raise e - - msg = f"User '{request.token.user_name}' successfully executed '{cmd.__name__}'" - - obj.broadcast(message=msg, btype=BroadcastType.COMMAND_EXECUTION_SUCCESS) - log.debug(msg) - - log.debug("Exiting") - return ret - - return wrap diff --git a/src/drunc/REMOVE_ME_broadcast/server/grpc_servicer.py b/src/drunc/REMOVE_ME_broadcast/server/grpc_servicer.py deleted file mode 100644 index eabed6282..000000000 --- a/src/drunc/REMOVE_ME_broadcast/server/grpc_servicer.py +++ /dev/null @@ -1,282 +0,0 @@ -from queue import Queue -from threading import Lock, Thread - -import grpc -from druncschema.authoriser_pb2 import ActionType -from druncschema.broadcast_pb2 import BroadcastMessage, BroadcastRequest, BroadcastType -from druncschema.broadcast_pb2_grpc import BroadcastSenderServicer -from druncschema.generic_pb2 import PlainText, StringStringMap -from druncschema.request_response_pb2 import Request, Response -from google.protobuf.any_pb2 import Any - -import drunc.controller.exceptions as ctler_excpt -from drunc.utils.grpc_utils import unpack_any - - -class ListenerRepresentation: - def __init__(self, configuration): - self.address = configuration["address"] - self.channel = grpc.insecure_channel(self.address) - from druncschema.broadcast_pb2_grpc import BroadcastReceiverStub - - self.stub = BroadcastReceiverStub(self.channel) - - def handle_broadcast(self, message): - return self.stub.handle_broadcast(message) - - -class GRCPBroadcastSender(BroadcastSenderServicer): - def __init__(self): - from drunc.exceptions import DruncSetupException - - raise DruncSetupException("GRCPBroadcastSender not supported") - - from logging import getLogger - - self.name = "broadcast_sender" - self._log = getLogger("Broadcast Sender") - self._listeners = {} - self._listener_lock = Lock() - self._message_queue = Queue() - self._consumer_thread = Thread(target=self._consumer, name="broadcast_consumer") - self._consumer_thread.start() - self._log.info("Broadcaster started") - - def _send(self, bm: BroadcastMessage): - pass - - def get_listeners(self): - import copy as cp - - with self._listener_lock: - ret = cp.deepcopy(list(self._listeners.keys())) - return ret - - def broadcast(self, txt, type=BroadcastType.TEXT_MESSAGE): - from druncschema.broadcast_pb2 import BroadcastMessage - from google.protobuf import any_pb2 - - message = PlainText(text=txt) - data_detail = any_pb2.Any() - data_detail.Pack(message) - - bm = BroadcastMessage( - emitter=self.name, type=BroadcastType.TEXT_MESSAGE, data=data_detail - ) - - return self._message_queue.put(bm) - - def broadcast_exception(self, exception): - from druncschema.broadcast_pb2 import BroadcastMessage - from google.protobuf import any_pb2 - - message = PlainText(text=str(exception)) - data_detail = any_pb2.Any() - data_detail.Pack(message) - - bm = BroadcastMessage( - emitter=self.name, type=BroadcastType.EXCEPTION_RAISED, data=data_detail - ) - return bm - - def add_to_bl_logic(self, request: BroadcastRequest): - self.add_listener(request.broadcast_receiver_address) - - def execute_command(self, request, format, logic, action): - uname = request.token.user_name - if self.authoriser: - if not self.authoriser.is_authorised(request.token, action): - from drunc.authoriser.exceptions import Unauthorised - - raise Unauthorised(uname, action) - from drunc.utils.grpc_utils import unpack_any - - data = unpack_any(request.data, format) - - self.broadcast( - f"Executing {action} (user: {uname})", BroadcastType.COMMAND_EXECUTION_START - ) - ret = logic(data) - self.broadcast( - f"Finshed executing {action} (user: {uname})", - BroadcastType.COMMAND_EXECUTION_SUCCESS, - ) - - response = Response() - response.token.CopyFrom(request.token) - data = Any() - if ret: - data.Pack(ret) - response.data.CopyFrom(data) - - return response - - def add_to_broadcast_list(self, request: Request, context) -> Response: - try: - return self.execute_command( - request=request, - format=BroadcastRequest, - logic=self.add_to_bl_logic, - action=ActionType.CREATE, - ) - except Exception as e: - self.broadcast_exception(e) - - def remove_from_broadcast_list(self, request: Request, context) -> Response: - r = unpack_any(request, BroadcastRequest) - if not self.broadcaster.rm_listener(r.broadcast_receiver_address): - raise ctler_excpt.ControllerException( - f"Failed to remove {r.broadcast_receiver_address} from broadcast list" - ) - return PlainText( - text=f"Removed {r.broadcast_receiver_address} to broadcast list" - ) - - def get_broadcast_list(self, request: Request, context) -> Response: - # return self._generic_user_command(request, '_get_broadcast_list', context) - ret = StringStringMap() - listeners = self.broadcaster.get_listeners() - for k, v in listeners.items(): - ret[k] = v - return - - def ack(self, address): - if address not in self._listeners: - raise RuntimeError(f"Cannot send ack to {address}") - - stub = self._listeners[address] - self._log.debug(f"Ack to {address}") - - from druncschema.broadcast_pb2 import BroadcastMessage, BroadcastType - - message = BroadcastMessage(emitter=self.name, type=BroadcastType.ACK) - - try: - stub.handle_broadcast(message) - except Exception as e: - self._log.error(f"Could not Ack to {address}: {e!s}") - - def shutdown(self): - from druncschema.broadcast_pb2 import BroadcastMessage, BroadcastType - - bm = BroadcastMessage(emitter=self.name, type=BroadcastType.SERVER_SHUTDOWN) - - return self._message_queue.put(bm) - - def add_listener(self, address): - with self._listener_lock: - if address in self._listeners.keys(): - self._log.error(f"Listener {address} already exists") - self._listener_lock.release() - return False - self._log.info(f"Adding listener {address}") - self._listeners[address] = ListenerRepresentation(address) - self.ack(address) - - return True - - def rm_listener(self, address): - with self._listener_lock: - if address not in self._listeners.keys(): - self._log.error(f"Listener {address} does not exist") - self._listener_lock.release() - return False - self._log.info(f"Removing listener {address}") - del self._listeners[address] - return True - - def join(self): - return self._consumer_thread.join() - - def _consumer(self): - from druncschema.broadcast_pb2 import BroadcastType - - while True: - message = ( - self._message_queue.get() - ) # Wait for a message from the controller - self._log.debug("Received broadcast message: " + str(message)) - self._listener_lock.acquire() - for address, listener in self._listeners.items(): - self._log.debug(f"Broadcasting {message} to {address}") - try: - response = listener.handle_broadcast(message) - except Exception as e: - self._log.error(f"Could not broadcast to {address}: {e}") - self._log.debug(f"Received response from {address}: {response}") - - self._listener_lock.release() - - if message.type == BroadcastType.SERVER_SHUTDOWN: - break - - -def main(): - from druncschema.broadcast_pb2 import BroadcastMessage, BroadcastType - from druncschema.broadcast_pb2_grpc import BroadcastReceiver - from druncschema.generic_pb2 import Empty - - class StatusReceiver(BroadcastReceiver): - def __init__(self, port): - super(StatusReceiver, self).__init__() - self.port = port - - def handle_broadcast( - self, bm: BroadcastMessage, context: grpc.aio.ServicerContext = None - ): - from drunc.utils.grpc_utils import unpack_any - - if bm.type == BroadcastType.SERVER_SHUTDOWN: - print("End of broadcast") - - elif bm.type == BroadcastType.TEXT_MESSAGE: - pt = unpack_any(bm.data, PlainText) - print(f'Got broadcasted message: "{pt.text}" on port {self.port}') - - return Empty() - - def serve(port: int) -> None: - from concurrent import futures - - server = grpc.server(futures.ThreadPoolExecutor(max_workers=1)) - from druncschema.broadcast_pb2_grpc import ( - add_BroadcastReceiverServicer_to_server, - ) - - add_BroadcastReceiverServicer_to_server(StatusReceiver(port), server) - server.add_insecure_port(f"[::]:{port}") - server.start() - print(f"Status receiver started on {port}") - server.wait_for_termination() - - receiver_threads = [] - port_list = range(15000, 15010) - for port in port_list: - try: - server_thread = Thread( - target=serve, kwargs={"port": port}, name=f"serve_thread_{port}" - ) - server_thread.start() - receiver_threads.append(server_thread) - except: - pass - from drunc.broadcast.server.broadcast_sender import BroadcastSender - - broadcaster = BroadcastSender() - for port in port_list: - broadcaster.add_listener(f"[::]:{port}") - broadcaster.broadcast("Test message") - broadcaster.broadcast("How do you do?") - broadcaster.broadcast("Let's all go for coffee!") - broadcaster.broadcast("End of broadcast") - broadcaster.shutdown() - broadcaster.join() - print("Broadcasting done") - print("\n\nYou can now ctrl-c\n\n") - - for thread in receiver_threads: - thread.join() - - -if __name__ == "__main__": - main() diff --git a/src/drunc/REMOVE_ME_broadcast/server/kafka_sender.py b/src/drunc/REMOVE_ME_broadcast/server/kafka_sender.py deleted file mode 100644 index c1a861a71..000000000 --- a/src/drunc/REMOVE_ME_broadcast/server/kafka_sender.py +++ /dev/null @@ -1,69 +0,0 @@ -from druncschema.broadcast_pb2 import BroadcastMessage - -from drunc.broadcast.server.broadcast_sender_implementation import ( - BroadcastSenderImplementation, -) - - -class KafkaSender(BroadcastSenderImplementation): - def __init__(self, kafka_address: str, publish_timeout: int, topic: str, **kwargs): - super(KafkaSender, self).__init__(**kwargs) - - import logging - - self._log = logging.getLogger(f"{topic}.KafkaSender") - - from kafka import KafkaProducer - from kafka import errors as Errors - - self.topic = topic - self._can_broadcast = False - - self.kafka_address = kafka_address - self.publish_timeout = publish_timeout - - try: - self.kafka = KafkaProducer( - bootstrap_servers=[self.kafka_address], - client_id="run_control", - ) - except Errors.NoBrokersAvailable as e: - t = f"{self.kafka_address} does not seem to point to a kafka broker." - self._log.critical(t) - from drunc.exceptions import DruncSetupException - - raise DruncSetupException(t) from e - - self._log.info( - f'Broadcasting to Kafka ({self.kafka_address}) client_id: "run_control", topic: "{self.topic}"' - ) - self._can_broadcast = True - - def can_broadcast(self): - return self._can_broadcast - - def _send(self, bm: BroadcastMessage): - from kafka.errors import KafkaError - - future = self.kafka.send(self.topic, bm.SerializeToString()) - - record_metadata = None - - try: - record_metadata = future.get(timeout=self.publish_timeout) - except KafkaError as e: - # Decide what to do if produce request failed... - self._log.error(f"Kafka exception sending message {bm}: {e!s}") - except Exception as e: - # Decide what to do if produce request failed... - self._log.error(f"Unhandled exception sending message {bm}: {e!s}") - else: - self._log.debug(f"{record_metadata} published") - - def describe_broadcast(self): - from druncschema.broadcast_pb2 import KafkaBroadcastHandlerConfiguration - - return KafkaBroadcastHandlerConfiguration( - topic=self.topic, - kafka_address=self.kafka_address, - ) diff --git a/src/drunc/REMOVE_ME_broadcast/types.py b/src/drunc/REMOVE_ME_broadcast/types.py deleted file mode 100644 index d91ce34be..000000000 --- a/src/drunc/REMOVE_ME_broadcast/types.py +++ /dev/null @@ -1,15 +0,0 @@ -from enum import Enum - -from drunc.exceptions import DruncSetupException - - -class BroadcastTypes(Enum): - Unknown = 0 - Kafka = 1 - ERS = 2 - - -class BroadcastTypeNotHandled(DruncSetupException): - def __init__(self, btype): - message = f"{btype} not handled" - super(BroadcastTypeNotHandled, self).__init__(message) diff --git a/src/drunc/REMOVE_ME_broadcast/utils.py b/src/drunc/REMOVE_ME_broadcast/utils.py deleted file mode 100644 index 1492996ef..000000000 --- a/src/drunc/REMOVE_ME_broadcast/utils.py +++ /dev/null @@ -1,30 +0,0 @@ -broadcast_types_loglevels = { - "ACK": "debug", - "RECEIVER_REMOVED": "info", - "RECEIVER_ADDED": "info", - "SERVER_READY": "info", - "SERVER_SHUTDOWN": "info", - "TEXT_MESSAGE": "info", - "COMMAND_EXECUTION_START": "info", - "COMMAND_EXECUTION_SUCCESS": "info", - "EXCEPTION_RAISED": "error", - "UNHANDLED_EXCEPTION_RAISED": "critical", - "STATUS_UPDATE": "info", - "SUBPROCESS_STATUS_UPDATE": "info", - "DEBUG": "debug", - "CHILD_COMMAND_EXECUTION_START": "info", - "CHILD_COMMAND_EXECUTION_SUCCESS": "info", - "CHILD_COMMAND_EXECUTION_FAILED": "error", -} - - -def get_broadcast_level_from_broadcast_type( - btype, logger, levels=broadcast_types_loglevels -): - from druncschema.broadcast_pb2 import BroadcastType - - bt = BroadcastType.Name(btype) - if bt not in levels: - return logger.info - else: - return getattr(logger, levels[bt].lower()) From 749ade1a8d46851c56960b4a89598072f28d3dcb Mon Sep 17 00:00:00 2001 From: James Paul Turner Date: Thu, 11 Jun 2026 18:00:18 +0100 Subject: [PATCH 05/10] Fix authoriser types. --- src/drunc/authoriser/dummy_authoriser.py | 23 +++++++---------------- src/drunc/controller/controller.py | 1 - 2 files changed, 7 insertions(+), 17 deletions(-) diff --git a/src/drunc/authoriser/dummy_authoriser.py b/src/drunc/authoriser/dummy_authoriser.py index c093e8fb5..a7098f687 100644 --- a/src/drunc/authoriser/dummy_authoriser.py +++ b/src/drunc/authoriser/dummy_authoriser.py @@ -5,22 +5,21 @@ from drunc.utils.utils import get_logger -# TODO: Should be communicating over network -# The Rolls Royce of the authoriser systems class DummyAuthoriser: def __init__( - self, - system: SystemType, - configuration_handler: DummyAuthoriserConfHandler = None, + self, configuration: DummyAuthoriserConfHandler, system: SystemType.ValueType ): self.log = get_logger("utils.authorizer") self.log.debug("DummyAuthoriser ready") - self.configuration = configuration_handler - self.command_actions = {} # Dict[str, ActionType] + self.configuration = configuration self.system = system def is_authorised( - self, token: Token, action: ActionType, system: SystemType, cmd_name: str = "" + self, + token: Token, + action: ActionType.ValueType, + system: SystemType.ValueType, + cmd_name: str, ) -> bool: self.log.debug( f"Authorising {token.user_name} to {ActionType.Name(action)} ({cmd_name}) on {SystemType.Name(system)}" @@ -30,11 +29,3 @@ def is_authorised( def authorised_actions(self, token: Token) -> list[str]: self.log.info(f"Grabbing authorisations for {token.token}") return [] - - -def main(): - DummyAuthoriser() - - -if __name__ == "__main__": - main() diff --git a/src/drunc/controller/controller.py b/src/drunc/controller/controller.py index 69ffef1fa..daaa5ace3 100644 --- a/src/drunc/controller/controller.py +++ b/src/drunc/controller/controller.py @@ -121,7 +121,6 @@ def __init__(self, configuration, name: str, session: str, token: Token): dach = DummyAuthoriserConfHandler( data=self.configuration.authoriser, ) - self.authoriser = DummyAuthoriser(dach, SystemType.CONTROLLER) self.actor = ControllerActor(token) From 11b90cabbcdbb0bc5472426864b2d33d740fd2d3 Mon Sep 17 00:00:00 2001 From: James Paul Turner Date: Thu, 11 Jun 2026 18:05:34 +0100 Subject: [PATCH 06/10] Remove broadcasting from the main PM class. --- src/drunc/process_manager/process_manager.py | 93 ++------------------ 1 file changed, 5 insertions(+), 88 deletions(-) diff --git a/src/drunc/process_manager/process_manager.py b/src/drunc/process_manager/process_manager.py index 07afa8b86..b4e979608 100644 --- a/src/drunc/process_manager/process_manager.py +++ b/src/drunc/process_manager/process_manager.py @@ -6,7 +6,6 @@ from daqpytools.logging import LogHandlerConf, exceptions, setup_daq_ers_logger from druncschema.authoriser_pb2 import ActionType, SystemType -from druncschema.broadcast_pb2 import BroadcastType from druncschema.description_pb2 import CommandDescription, Description from druncschema.opmon.process_manager_pb2 import ProcessStatus from druncschema.process_manager_pb2 import ( @@ -28,9 +27,6 @@ from drunc.authoriser.configuration import DummyAuthoriserConfHandler from drunc.authoriser.decorators import authentified_and_authorised from drunc.authoriser.dummy_authoriser import DummyAuthoriser -from drunc.broadcast.server.broadcast_sender import BroadcastSender -from drunc.broadcast.server.configuration import BroadcastSenderConfHandler -from drunc.broadcast.server.decorators import broadcasted from drunc.exceptions import ( DruncCommandException, DruncNotImplementedException, @@ -50,11 +46,7 @@ def __init__(self, txt): class ProcessManager(abc.ABC, ProcessManagerServicer): def __init__( - self, - configuration: ProcessManagerConfHandler, - name: str, - session: str = None, - **kwargs, + self, configuration: ProcessManagerConfHandler, name: str, session: str ): """C'tor. Note that this takes the ERS env variables from the json files defined in data/process_manager!""" @@ -80,17 +72,14 @@ def __init__( self.configuration = configuration self.name = name self.session = session - - self._create_broadcast_service(self.name, self.session) - - dach = DummyAuthoriserConfHandler( - data=self.configuration.get_data_authoriser(), type=ConfTypes.PyObject - ) - self.opmon_publisher = getattr( self.configuration.get_data(), "opmon_publisher", None ) interval_s = getattr(self.configuration.get_data(), "interval_s", 10.0) + + dach = DummyAuthoriserConfHandler( + data=self.configuration.get_data_authoriser(), type=ConfTypes.PyObject + ) self.authoriser = DummyAuthoriser(dach, SystemType.PROCESS_MANAGER) self.process_store = {} # dict[str, sh.RunningCommand] # str = uuid @@ -154,8 +143,6 @@ def __init__( ), ] - self.broadcast(message="ready", btype=BroadcastType.SERVER_READY) - if self.opmon_publisher is not None: self.stop_event = threading.Event() self.thread = threading.Thread( @@ -168,21 +155,6 @@ def __init__( def get_log_path(self): return self.configuration.get_log_path() - def _create_broadcast_service(self, name, session): - bsch = BroadcastSenderConfHandler( - data=self.configuration.get_data_broadcaster(), type=ConfTypes.PyObject - ) - - self.broadcast_service = ( - BroadcastSender( - name=name, - session=session, - configuration=bsch, - ) - if bsch.data - else None - ) - def __del__(self): if hasattr(self, "opmon_publisher") and self.opmon_publisher is not None: self.stop_event.set() @@ -247,48 +219,10 @@ def find_by_uuid(pi_list, target_uuid: str): time.sleep(interval_s) - """ - A couple of simple pass-through functions to the broadcasting service - """ - - def broadcast(self, *args, **kwargs): - self.log.debug(f"{self.name} broadcasting") - return ( - self.broadcast_service.broadcast(*args, **kwargs) - if self.broadcast_service - else None - ) - - def can_broadcast(self, *args, **kwargs): - self.log.debug(f"Checking if {self.name} can broadcast") - return ( - self.broadcast_service.can_broadcast(*args, **kwargs) - if self.broadcast_service - else False - ) - - def describe_broadcast(self, *args, **kwargs): - self.log.debug(f"Describing {self.name} broadcast") - return ( - self.broadcast_service.describe_broadcast(*args, **kwargs) - if self.broadcast_service - else None - ) - - def interrupt_with_exception(self, *args, **kwargs): - self.log.debug(f"Interrupting {self.name} broadcast with exception") - return ( - self.broadcast_service._interrupt_with_exception(*args, **kwargs) - if self.broadcast_service - else None - ) - @abc.abstractmethod def _boot_impl(self, boot_request: BootRequest) -> ProcessInstanceList: raise NotImplementedError - # ORDER MATTERS! - @broadcasted # outer most wrapper 1st step @authentified_and_authorised( action=ActionType.CREATE, system=SystemType.PROCESS_MANAGER ) # 2nd step @@ -321,8 +255,6 @@ def boot( def _terminate_impl(self) -> ProcessInstanceList: raise NotImplementedError - # ORDER MATTERS! - @broadcasted # outer most wrapper 1st step @authentified_and_authorised( action=ActionType.DELETE, system=SystemType.PROCESS_MANAGER ) # 2nd step @@ -356,8 +288,6 @@ def terminate( def _restart_impl(self, query: ProcessQuery) -> ProcessInstanceList: raise NotImplementedError - # ORDER MATTERS! - @broadcasted # outer most wrapper 1st step @authentified_and_authorised( action=ActionType.DELETE, system=SystemType.PROCESS_MANAGER ) # 2nd step @@ -388,8 +318,6 @@ def restart( def _kill_impl(self, query: ProcessQuery) -> ProcessInstanceList: raise NotImplementedError - # ORDER MATTERS! - @broadcasted # outer most wrapper 1st step @authentified_and_authorised( action=ActionType.DELETE, system=SystemType.PROCESS_MANAGER ) # 2nd step @@ -420,8 +348,6 @@ def kill( def _ps_impl(self, query: ProcessQuery) -> ProcessInstanceList: raise NotImplementedError - # ORDER MATTERS! - @broadcasted # outer most wrapper 1st step @authentified_and_authorised( action=ActionType.READ, system=SystemType.PROCESS_MANAGER ) # 2nd step @@ -452,8 +378,6 @@ def ps( def _flush_impl(self, query: ProcessQuery) -> ProcessInstanceList: raise NotImplementedError - # ORDER MATTERS! - @broadcasted # outer most wrapper 1st step @authentified_and_authorised( action=ActionType.DELETE, system=SystemType.PROCESS_MANAGER ) # 2nd step @@ -493,8 +417,6 @@ def flush( return response - # ORDER MATTERS! - @broadcasted # outer most wrapper 1st step @authentified_and_authorised( action=ActionType.READ, system=SystemType.PROCESS_MANAGER ) # 2nd step @@ -511,17 +433,12 @@ def describe(self, request: Request, context: ServicerContext) -> Description: token=None, ) - if broadcast_description := self.describe_broadcast(): - response.broadcast.Pack(broadcast_description) - return response @abc.abstractmethod def _logs_impl(self, log_request: LogRequest) -> LogLines: raise NotImplementedError - # ORDER MATTERS! - @broadcasted # outer most wrapper 1st step @authentified_and_authorised( action=ActionType.READ, system=SystemType.PROCESS_MANAGER ) # 2nd step From 4840f9f637766c06b8e0f0170fa631f07829f1ed Mon Sep 17 00:00:00 2001 From: James Paul Turner Date: Thu, 11 Jun 2026 18:11:04 +0100 Subject: [PATCH 07/10] Remove broadcasting from PM subclasses. --- src/drunc/process_manager/k8s_process_manager.py | 6 ++---- src/drunc/process_manager/ssh_process_manager.py | 3 --- 2 files changed, 2 insertions(+), 7 deletions(-) diff --git a/src/drunc/process_manager/k8s_process_manager.py b/src/drunc/process_manager/k8s_process_manager.py index 04116cd61..358804062 100644 --- a/src/drunc/process_manager/k8s_process_manager.py +++ b/src/drunc/process_manager/k8s_process_manager.py @@ -12,7 +12,6 @@ from time import sleep, time # Local Application Imports -from druncschema.broadcast_pb2 import BroadcastType from druncschema.process_manager_pb2 import ( BootRequest, LogLines, @@ -337,8 +336,8 @@ def notify_termination( """ Callback for when a pod terminates. - Updates the final exit code, broadcasts a status update, and signals - the termination_complete_event when all pending deletions are confirmed. + Updates the final exit code and signals the termination_complete_event + when all pending deletions are confirmed. Args: proc_uuid: The UUID string of the terminated process. @@ -359,7 +358,6 @@ def notify_termination( # Publish this information self.log.info(end_str) - self.broadcast(end_str, BroadcastType.SUBPROCESS_STATUS_UPDATE) # Clear the list of processes being removed if proc_uuid in self.uuids_pending_deletion: diff --git a/src/drunc/process_manager/ssh_process_manager.py b/src/drunc/process_manager/ssh_process_manager.py index e915ecd73..e0bfb7aa4 100644 --- a/src/drunc/process_manager/ssh_process_manager.py +++ b/src/drunc/process_manager/ssh_process_manager.py @@ -3,7 +3,6 @@ import uuid from typing import List, Optional -from druncschema.broadcast_pb2 import BroadcastType from druncschema.process_manager_pb2 import ( BootRequest, LogLines, @@ -317,10 +316,8 @@ def _logs_impl(self, log_request: LogRequest) -> LogLines: ) def notify_join(self, name, session, user, exit_status: ExitStatus): - self.log.debug(f"{self.name} sending broadcast after ssh process exit") end_str = exit_status.get_process_manager_log_message(name, session, user) self.log.info(end_str) - self.broadcast(end_str, BroadcastType.SUBPROCESS_STATUS_UPDATE) def __boot(self, boot_request: BootRequest, uuid: str) -> ProcessInstance: """ From 28f7c5ed64621cbbc5aa893a5b067a8b40594944 Mon Sep 17 00:00:00 2001 From: James Paul Turner Date: Thu, 11 Jun 2026 19:44:22 +0100 Subject: [PATCH 08/10] Remove broadcast from config and schema. --- src/drunc/data/process_manager/k8s-CERN.json | 6 ----- src/drunc/data/process_manager/k8s.json | 5 ---- .../process-manager-k8s-pocket.json | 6 ----- .../schema/process_manager.schema.json | 10 -------- .../data/process_manager/ssh-CERN-kafka.json | 6 ----- .../process_manager/ssh-pocket-kafka.json | 9 +------- src/drunc/process_manager/configuration.py | 6 ----- .../process_manager/interface/context.py | 23 ++++--------------- src/drunc/process_manager/interface/shell.py | 2 -- 9 files changed, 6 insertions(+), 67 deletions(-) diff --git a/src/drunc/data/process_manager/k8s-CERN.json b/src/drunc/data/process_manager/k8s-CERN.json index 100dc2851..b59b10fb1 100644 --- a/src/drunc/data/process_manager/k8s-CERN.json +++ b/src/drunc/data/process_manager/k8s-CERN.json @@ -7,12 +7,6 @@ "authoriser": { "type": "dummy" }, - - "broadcaster": { - "type": "kafka", - "kafka_address": "monkafka.cern.ch:30092", - "publish_timeout": 2 - }, "environment": { "DUNEDAQ_ERS_ERROR": "erstrace,throttle,lstdout,protobufstream(monkafka.cern.ch:30092)", "DUNEDAQ_ERS_FATAL": "erstrace,lstdout,protobufstream(monkafka.cern.ch:30092)", diff --git a/src/drunc/data/process_manager/k8s.json b/src/drunc/data/process_manager/k8s.json index fa94a98ea..c46412eb6 100644 --- a/src/drunc/data/process_manager/k8s.json +++ b/src/drunc/data/process_manager/k8s.json @@ -6,11 +6,6 @@ "authoriser": { "type": "dummy" }, - "broadcaster": { - "type": "kafka", - "kafka_address": "monkafka.cern.ch:30092", - "publish_timeout": 2 - }, "environment": { "DUNEDAQ_ERS_ERROR": "erstrace,throttle,lstdout", "DUNEDAQ_ERS_FATAL": "erstrace,lstdout", diff --git a/src/drunc/data/process_manager/process-manager-k8s-pocket.json b/src/drunc/data/process_manager/process-manager-k8s-pocket.json index 54c295afd..069ec89d8 100644 --- a/src/drunc/data/process_manager/process-manager-k8s-pocket.json +++ b/src/drunc/data/process_manager/process-manager-k8s-pocket.json @@ -6,12 +6,6 @@ "authoriser": { "type": "dummy" }, - - "broadcaster": { - "type": "kafka", - "kafka_address": "localhost:30092", - "publish_timeout": 2 - }, "environment": { "DUNEDAQ_ERS_ERROR": "erstrace,throttle,lstdout,protobufstream(monkafka.cern.ch:30092)", "DUNEDAQ_ERS_FATAL": "erstrace,lstdout,protobufstream(monkafka.cern.ch:30092)", diff --git a/src/drunc/data/process_manager/schema/process_manager.schema.json b/src/drunc/data/process_manager/schema/process_manager.schema.json index 0ad199b8e..bcb906a7e 100644 --- a/src/drunc/data/process_manager/schema/process_manager.schema.json +++ b/src/drunc/data/process_manager/schema/process_manager.schema.json @@ -33,16 +33,6 @@ }, "additionalProperties": true }, - "broadcaster": { - "type": "object", - "properties": { - "type": {"type": "string"}, - "kafka_address": {"type": "string"}, - "publish_timeout": {"type": ["integer", "number"]} - }, - "required": ["type", "kafka_address", "publish_timeout"], - "additionalProperties": true - }, "command_address": {"type": "string"} }, "required": ["type", "name"], diff --git a/src/drunc/data/process_manager/ssh-CERN-kafka.json b/src/drunc/data/process_manager/ssh-CERN-kafka.json index 1158be0d9..0f62572e7 100644 --- a/src/drunc/data/process_manager/ssh-CERN-kafka.json +++ b/src/drunc/data/process_manager/ssh-CERN-kafka.json @@ -7,12 +7,6 @@ "authoriser": { "type": "dummy" }, - - "broadcaster": { - "type": "kafka", - "kafka_address": "monkafka.cern.ch:30092", - "publish_timeout": 2 - }, "environment": { "DUNEDAQ_ERS_ERROR": "erstrace,throttle,lstdout,protobufstream(monkafka.cern.ch:30092)", "DUNEDAQ_ERS_FATAL": "erstrace,lstdout,protobufstream(monkafka.cern.ch:30092)", diff --git a/src/drunc/data/process_manager/ssh-pocket-kafka.json b/src/drunc/data/process_manager/ssh-pocket-kafka.json index 9208772da..950083f7b 100644 --- a/src/drunc/data/process_manager/ssh-pocket-kafka.json +++ b/src/drunc/data/process_manager/ssh-pocket-kafka.json @@ -7,13 +7,6 @@ "authoriser": { "type": "dummy" }, - - "broadcaster": { - "type": "kafka", - "kafka_address": "localhost:31014", - "publish_timeout": 2 - }, - "environment": { "GRPC_ENABLE_FORK_SUPPORT": "false", "DUNEDAQ_ERS_ERROR": "erstrace,throttle,lstdout,protobufstream(monkafka.cern.ch:30092)", @@ -30,4 +23,4 @@ "level": "debug", "interval_s": 10.0 } -} \ No newline at end of file +} diff --git a/src/drunc/process_manager/configuration.py b/src/drunc/process_manager/configuration.py index da9198722..5172fa219 100644 --- a/src/drunc/process_manager/configuration.py +++ b/src/drunc/process_manager/configuration.py @@ -12,7 +12,6 @@ from opmonlib.publisher import OpMonPublisher from opmonlib.utils import parse_opmon_conf -from drunc.broadcast.server.configuration import KafkaBroadcastSenderConfData from drunc.exceptions import DruncCommandException from drunc.process_manager.exceptions import UnknownProcessManagerType from drunc.utils.configuration import ConfHandler @@ -40,7 +39,6 @@ class ProcessManagerTypes(Enum): class ProcessManagerConfData: def __init__(self): - self.broadcaster = None self.authoriser = None self.type = ProcessManagerTypes.Unknown self.command_address = "" @@ -61,10 +59,6 @@ def get_log_path(self): def _parse_dict(self, data): new_data = ProcessManagerConfData() - if data.get("broadcaster"): - new_data.broadcaster = KafkaBroadcastSenderConfData.from_dict( - data.get("broadcaster") - ) new_data.environment = data.get("environment", {}) new_data.settings = data.get("settings", {}) diff --git a/src/drunc/process_manager/interface/context.py b/src/drunc/process_manager/interface/context.py index a81396130..d6a30f592 100644 --- a/src/drunc/process_manager/interface/context.py +++ b/src/drunc/process_manager/interface/context.py @@ -1,24 +1,21 @@ -from collections.abc import Mapping +from collections.abc import MutableMapping from druncschema.token_pb2 import Token -from drunc.broadcast.client.broadcast_handler import BroadcastHandler -from drunc.broadcast.client.configuration import BroadcastClientConfHandler from drunc.process_manager.process_manager_driver import ProcessManagerDriver -from drunc.utils.configuration import ConfTypes from drunc.utils.shell_utils import ( ShellContext, create_dummy_token_from_uname, ) -from drunc.utils.utils import get_logger, resolve_localhost_to_hostname +from drunc.utils.utils import resolve_localhost_to_hostname -class ProcessManagerContext(ShellContext): # boilerplatefest +class ProcessManagerContext(ShellContext): def __init__(self, *args, **kwargs): self.status_receiver = None super(ProcessManagerContext, self).__init__(*args, **kwargs) - def reset(self, address: str = None): + def reset(self, address: str = "", **kwargs): self.address = resolve_localhost_to_hostname(address) super(ProcessManagerContext, self)._reset( name="process_manager_context", @@ -26,7 +23,7 @@ def reset(self, address: str = None): driver_args={}, ) - def create_drivers(self, **kwargs) -> Mapping[str, object]: + def create_drivers(self, **kwargs) -> MutableMapping[str, object]: if not self.address: return {} return { @@ -39,16 +36,6 @@ def create_drivers(self, **kwargs) -> Mapping[str, object]: def create_token(self, **kwargs) -> Token: return create_dummy_token_from_uname() - def start_listening(self, broadcaster_conf): - bcch = BroadcastClientConfHandler( - data=broadcaster_conf, - type=ConfTypes.ProtobufAny, - ) - self.status_receiver = BroadcastHandler(bcch) - get_logger("process_manager.shell").info( - f":ear: Listening to the Process Manager at {self.address}" - ) - def terminate(self): if self.status_receiver: self.status_receiver.stop() diff --git a/src/drunc/process_manager/interface/shell.py b/src/drunc/process_manager/interface/shell.py index 5da7434c5..18f7962e2 100644 --- a/src/drunc/process_manager/interface/shell.py +++ b/src/drunc/process_manager/interface/shell.py @@ -71,8 +71,6 @@ def process_manager_shell(ctx, process_manager_address: str, log_level: str) -> process_manager_shell_log.info( f"Connected to {process_manager_address}, running '{desc.name}.{desc.session}' (name.session), starting listening..." ) - if desc.HasField("broadcast"): - ctx.obj.start_listening(desc.broadcast) def cleanup(): ctx.obj.terminate() From 2bd6a5b30eefcab6f8667d17486295b3c21d5bd8 Mon Sep 17 00:00:00 2001 From: James Paul Turner Date: Thu, 11 Jun 2026 20:06:10 +0100 Subject: [PATCH 09/10] Remove broadcast from unified shell. --- src/drunc/unified_shell/context.py | 25 ------------------------- src/drunc/unified_shell/shell.py | 17 ++++------------- src/drunc/utils/configuration.py | 9 --------- 3 files changed, 4 insertions(+), 47 deletions(-) diff --git a/src/drunc/unified_shell/context.py b/src/drunc/unified_shell/context.py index f5c9c759e..73ff03fde 100644 --- a/src/drunc/unified_shell/context.py +++ b/src/drunc/unified_shell/context.py @@ -66,37 +66,12 @@ def set_controller_driver(self, address_controller, **kwargs) -> None: self._token, ) - # This will raise an exception if the driver already exists - # self.set_driver("controller", driver) - def create_token(self, **kwargs) -> Token: from drunc.utils.shell_utils import create_dummy_token_from_uname token = create_dummy_token_from_uname() return token - def start_listening_pm(self, broadcaster_conf) -> None: - from drunc.broadcast.client.broadcast_handler import BroadcastHandler - from drunc.broadcast.client.configuration import BroadcastClientConfHandler - from drunc.utils.configuration import ConfTypes - - bcch = BroadcastClientConfHandler( - type=ConfTypes.ProtobufAny, - data=broadcaster_conf, - ) - self.status_receiver_pm = BroadcastHandler(broadcast_configuration=bcch) - - def start_listening_controller(self, broadcaster_conf) -> None: - from drunc.broadcast.client.broadcast_handler import BroadcastHandler - from drunc.broadcast.client.configuration import BroadcastClientConfHandler - from drunc.utils.configuration import ConfTypes - - bcch = BroadcastClientConfHandler( - type=ConfTypes.ProtobufAny, - data=broadcaster_conf, - ) - self.status_receiver_controller = BroadcastHandler(broadcast_configuration=bcch) - def terminate(self) -> None: if self.status_receiver_pm: self.status_receiver_pm.stop() diff --git a/src/drunc/unified_shell/shell.py b/src/drunc/unified_shell/shell.py index af8bedbec..576201d3c 100644 --- a/src/drunc/unified_shell/shell.py +++ b/src/drunc/unified_shell/shell.py @@ -12,7 +12,6 @@ import click_shell import conffwk from daqpytools.logging import logging_log_levels -from druncschema.description_pb2 import Description from druncschema.process_manager_pb2 import ProcessQuery from drunc.connectivity_service.client import ConnectivityServiceClient @@ -277,9 +276,8 @@ def unified_shell( ctx.obj.reset(address_pm=process_manager_address) # Run a simple command (describe) to check the connection with the process manager - desc: Description | None = None try: - desc = ctx.obj.get_driver().describe() + ctx.obj.get_driver().describe() except Exception as e: ctx.obj.log.error( f"[red]Could not connect to the process manager at the address: [/red]" @@ -305,13 +303,6 @@ def unified_shell( sys.exit(1) - # Broadcasting configuration if requested - if desc.HasField("broadcast"): - ctx.obj.log.debug("Broadcasting") - ctx.obj.start_listening_pm( - broadcaster_conf=desc.broadcast, - ) - # Add the unified shell Click commands to the CLI ctx.obj.log.debug("Adding [green]unified_shell[/green] commands") ctx.command.add_command(boot, "boot") @@ -521,9 +512,9 @@ def cleanup(): ctx.obj.log.debug("Process manager terminated") ctx.obj.log.info("[green]unified_shell exited successfully[/green]") - logging.shutdown() # Shutdown logging - ctx.obj.terminate() # Terminate the broadcasters in the context - ctx.exit() # Close the click context + logging.shutdown() + ctx.obj.terminate() + ctx.exit() ctx.call_on_close(cleanup) diff --git a/src/drunc/utils/configuration.py b/src/drunc/utils/configuration.py index d5747a7e6..2883d8ed5 100644 --- a/src/drunc/utils/configuration.py +++ b/src/drunc/utils/configuration.py @@ -116,7 +116,6 @@ class _DataTypeName(Protocol): class _ConfigurationData(Protocol): type: _DataTypeName - broadcaster: object authoriser: object @@ -179,14 +178,6 @@ def get_data_type_name(self) -> str: """ return str(cast(_ConfigurationData, self.get_data()).type._name_) - def get_data_broadcaster(self) -> object: - """Get the broadcaster from the configuration data. - - Returns: - Any: The broadcaster object. - """ - return cast(_ConfigurationData, self.get_data()).broadcaster - def get_data_authoriser(self) -> object: """Get the authoriser from the configuration data. From 4dca1875e7285e8e64974487eb22addc38edea74 Mon Sep 17 00:00:00 2001 From: James Paul Turner Date: Thu, 11 Jun 2026 20:21:38 +0100 Subject: [PATCH 10/10] Remove broadcast from tests. --- tests/broadcast/__init__.py | 0 tests/broadcast/client/__init__.py | 0 .../client/test_broadcast_handler.py | 0 .../test_broadcast_handler_implementation.py | 0 tests/broadcast/client/test_configuration.py | 0 .../test_grpc_stdout_broadcast_handler.py | 0 .../test_kafka_stdout_broadcast_handler.py | 0 tests/broadcast/server/__init__.py | 0 .../broadcast/server/test_broadcast_sender.py | 0 .../test_broadcast_sender_implementation.py | 0 tests/broadcast/server/test_configuration.py | 0 tests/broadcast/server/test_decorators.py | 72 -------------- tests/broadcast/server/test_grpc_servicer.py | 0 tests/broadcast/server/test_kafka_sender.py | 93 ------------------- tests/broadcast/test_types.py | 0 tests/broadcast/test_utils.py | 0 .../process_manager_mock_impls.py | 3 - 17 files changed, 168 deletions(-) delete mode 100644 tests/broadcast/__init__.py delete mode 100644 tests/broadcast/client/__init__.py delete mode 100644 tests/broadcast/client/test_broadcast_handler.py delete mode 100644 tests/broadcast/client/test_broadcast_handler_implementation.py delete mode 100644 tests/broadcast/client/test_configuration.py delete mode 100644 tests/broadcast/client/test_grpc_stdout_broadcast_handler.py delete mode 100644 tests/broadcast/client/test_kafka_stdout_broadcast_handler.py delete mode 100644 tests/broadcast/server/__init__.py delete mode 100644 tests/broadcast/server/test_broadcast_sender.py delete mode 100644 tests/broadcast/server/test_broadcast_sender_implementation.py delete mode 100644 tests/broadcast/server/test_configuration.py delete mode 100644 tests/broadcast/server/test_decorators.py delete mode 100644 tests/broadcast/server/test_grpc_servicer.py delete mode 100644 tests/broadcast/server/test_kafka_sender.py delete mode 100644 tests/broadcast/test_types.py delete mode 100644 tests/broadcast/test_utils.py diff --git a/tests/broadcast/__init__.py b/tests/broadcast/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/broadcast/client/__init__.py b/tests/broadcast/client/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/broadcast/client/test_broadcast_handler.py b/tests/broadcast/client/test_broadcast_handler.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/broadcast/client/test_broadcast_handler_implementation.py b/tests/broadcast/client/test_broadcast_handler_implementation.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/broadcast/client/test_configuration.py b/tests/broadcast/client/test_configuration.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/broadcast/client/test_grpc_stdout_broadcast_handler.py b/tests/broadcast/client/test_grpc_stdout_broadcast_handler.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/broadcast/client/test_kafka_stdout_broadcast_handler.py b/tests/broadcast/client/test_kafka_stdout_broadcast_handler.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/broadcast/server/__init__.py b/tests/broadcast/server/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/broadcast/server/test_broadcast_sender.py b/tests/broadcast/server/test_broadcast_sender.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/broadcast/server/test_broadcast_sender_implementation.py b/tests/broadcast/server/test_broadcast_sender_implementation.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/broadcast/server/test_configuration.py b/tests/broadcast/server/test_configuration.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/broadcast/server/test_decorators.py b/tests/broadcast/server/test_decorators.py deleted file mode 100644 index 6dfbde783..000000000 --- a/tests/broadcast/server/test_decorators.py +++ /dev/null @@ -1,72 +0,0 @@ -from unittest.mock import MagicMock - -import pytest -from druncschema.broadcast_pb2 import BroadcastType -from druncschema.request_response_pb2 import Request -from druncschema.token_pb2 import Token - -from drunc.broadcast.server.decorators import broadcasted - - -class MockException(Exception): - pass - -@pytest.fixture -def mock_obj(): - """Mock the object that has the .broadcast() method.""" - obj = MagicMock() - obj.name = "test-node" - return obj - - -@pytest.fixture(scope="function") -def mock_request(): - return Request(token=Token(user_name="test", token="tets-token")) - - -@pytest.fixture -def mock_context(): - return MagicMock() - - -def test_broadcasted_success(mock_obj, mock_request, mock_context): - - @broadcasted - def dummy_command(obj, request, context): - return "Success" - - result = dummy_command(mock_obj, mock_request, mock_context) - - assert result == "Success" - - assert mock_obj.broadcast.call_count == 2 # ACK and COMMAND_EXECUTION_SUCCESS - - # first call - ACK - args, kwargs = mock_obj.broadcast.call_args_list[0] - assert kwargs['message'] == "User 'test' executing 'dummy_command'" - assert kwargs['btype'] == BroadcastType.ACK - - # second call - COMMAND_EXECUTION_SUCCESS - # check no missing or additional arguments - mock_obj.broadcast.assert_called_with( - message="User 'test' successfully executed 'dummy_command'", - btype=BroadcastType.COMMAND_EXECUTION_SUCCESS) - - -def test_broadcasted_failure(mock_obj, mock_request, mock_context): - - # command that raises an error - @broadcasted - def dummy_command(obj, request, context): - raise MockException("Test exception") - - with pytest.raises(MockException): - dummy_command(mock_obj, mock_request, mock_context) - - assert mock_obj.broadcast.call_count == 2 # ACK and Exception - - # check no missing or additional arguments passed to broadcast - mock_obj.broadcast.assert_called_with( - message="Command 'dummy_command' failed", - btype=BroadcastType.UNHANDLED_EXCEPTION_RAISED) - \ No newline at end of file diff --git a/tests/broadcast/server/test_grpc_servicer.py b/tests/broadcast/server/test_grpc_servicer.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/broadcast/server/test_kafka_sender.py b/tests/broadcast/server/test_kafka_sender.py deleted file mode 100644 index f2e4df400..000000000 --- a/tests/broadcast/server/test_kafka_sender.py +++ /dev/null @@ -1,93 +0,0 @@ -from unittest.mock import MagicMock, patch - -import pytest -from kafka.errors import KafkaError, NoBrokersAvailable - -from drunc.broadcast.server.kafka_sender import KafkaSender -from drunc.exceptions import DruncSetupException - - -@pytest.fixture -def mock_kafka_producer(): - """Fixture to mock a KafkaProducer class and its instance.""" - with patch("kafka.KafkaProducer") as mock_class: - mock_instance = mock_class.return_value - yield mock_class, mock_instance - - -@pytest.fixture -def sender_with_mocked_producer(mock_kafka_producer): - """Fixture to create a KafkaSender instance with a mocked KafkaProducer.""" - kafka_sender = KafkaSender( - kafka_address="test-kafka-address", publish_timeout=5, topic="test-topic" - ) - kafka_sender._log = MagicMock() - return kafka_sender - - -def test_init_raises_drunc_setup_exception(): - """Test that KafkaSender raises DruncSetupException when KafkaProducer cannot connect to a broker.""" - mock_logger = MagicMock() - kafka_address = "test-address" - expected_exc_msg = f"{kafka_address} does not seem to point to a kafka broker." - - with ( - patch("logging.getLogger", return_value=mock_logger), - patch("kafka.KafkaProducer", side_effect=NoBrokersAvailable), - ): - with pytest.raises(DruncSetupException) as exc_info: - KafkaSender( - kafka_address=kafka_address, publish_timeout=5, topic="test-topic" - ) - - mock_logger.critical.assert_called_once() - log_call_args = mock_logger.critical.call_args[0][0] - assert expected_exc_msg in log_call_args - assert expected_exc_msg in str(exc_info.value) - - -def test_send_success(sender_with_mocked_producer, mock_kafka_producer): - """Test that KafkaSender._send successfully sends a message and logs the metadata.""" - _, mock_instance = mock_kafka_producer - sender = sender_with_mocked_producer - - mock_future = MagicMock() - mock_future.get.return_value = "test-metadata" - mock_instance.send.return_value = mock_future - - mock_broadcast_msg = MagicMock() - mock_broadcast_msg.SerializeToString.return_value = b"test-msg" - sender._send(mock_broadcast_msg) - - mock_instance.send.assert_called_with("test-topic", b"test-msg") - sender._log.debug.assert_called_with("test-metadata published") - - -def test_send_handle_exception(sender_with_mocked_producer, mock_kafka_producer): - """Test that KafkaSender._send handles KafkaError exceptions and logs the error.""" - _, mock_instance = mock_kafka_producer - sender = sender_with_mocked_producer - - mock_future = MagicMock() - mock_future.get.side_effect = KafkaError("Connection lost") - mock_instance.send.return_value = mock_future - - mock_broadcast_msg = MagicMock() - mock_broadcast_msg.SerializeToString.return_value = b"test-broadcast-msg" - - sender._log = MagicMock() - - sender._send(mock_broadcast_msg) - - sender._log.error.assert_called() - log_message = sender._log.error.call_args[0][0] - assert "Connection lost" in log_message - - -def test_describe_broadcast(sender_with_mocked_producer): - """Test that KafkaSender.describe_broadcast returns the correct information.""" - - result = sender_with_mocked_producer.describe_broadcast() - - assert result.topic == "test-topic" - assert result.kafka_address == "test-kafka-address" diff --git a/tests/broadcast/test_types.py b/tests/broadcast/test_types.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/broadcast/test_utils.py b/tests/broadcast/test_utils.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/process_manager/process_manager_mock_impls.py b/tests/process_manager/process_manager_mock_impls.py index b9b84cf23..d7f83c9c8 100644 --- a/tests/process_manager/process_manager_mock_impls.py +++ b/tests/process_manager/process_manager_mock_impls.py @@ -55,9 +55,6 @@ def _not_implemented_response(self): flag=ResponseFlag.NOT_EXECUTED_NOT_IMPLEMENTED, ) - def _create_broadcast_service(self, name, session): - self.broadcast_service = None - def _boot_impl(self, boot_request: BootRequest) -> ProcessInstanceList: """ Returns default not implemented response to indicate communication is working