MRG: Merge pull request #639 from octue/use-single-topic-per-workspace

Use single topic per workspace

Author: cortadocodes

octue/__init__.py

@@ -6,6 +6,7 @@
 
 __all__ = ("Runner",)
 
+
 REPOSITORY_ROOT = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
 
 

octue/cli.py

@@ -357,35 +357,16 @@ def deploy():
     help="The service revision tag (e.g. 1.0.7). If this option isn't given, a random 'cool name' tag is generated e.g"
     ". 'curious-capybara'.",
 )
-@click.option(
-    "--filter",
-    is_flag=False,
-    default='attributes.sender_type = "PARENT"',
-    show_default=True,
-    help="An optional filter to apply to the subscription (see "
-    "https://cloud.google.com/pubsub/docs/subscription-message-filter). If not provided, the default filter is applied."
-    " To disable filtering, provide an empty string.",
-)
-@click.option(
-    "--subscription-suffix",
-    is_flag=False,
-    default=None,
-    show_default=True,
-    help="An optional suffix to add to the end of the subscription name. This is useful when needing to create "
-    "multiple subscriptions for the same topic (subscription names are unique).",
-)
 def create_push_subscription(
     project_name,
     service_namespace,
     service_name,
     push_endpoint,
     expiration_time,
     revision_tag,
-    filter,
-    subscription_suffix,
 ):
-    """Create a Google Pub/Sub push subscription for an Octue service for it to receive questions from parents. If a
-    corresponding topic doesn't exist, it will be created first. The subscription name is printed on completion.
+    """Create a Google Pub/Sub push subscription for an Octue service for it to receive questions from parents. The
+    subscription name is printed on completion.
 
     PROJECT_NAME is the name of the Google Cloud project in which the subscription will be created
 
@@ -403,8 +384,7 @@ def create_push_subscription(
         sruid,
         push_endpoint,
         expiration_time=expiration_time,
-        subscription_filter=filter or None,
-        subscription_suffix=subscription_suffix,
+        subscription_filter=f'attributes.recipient = "{sruid}" AND attributes.sender_type = "PARENT"',
     )
 
     click.echo(sruid)

octue/cloud/deployment/google/answer_pub_sub_question.py

@@ -1,8 +1,7 @@
 import logging
 
-from octue.cloud.pub_sub import Topic
 from octue.cloud.pub_sub.service import Service
-from octue.cloud.service_id import convert_service_id_to_pub_sub_form, create_sruid, get_sruid_parts
+from octue.cloud.service_id import create_sruid, get_sruid_parts
 from octue.configuration import load_service_and_app_configuration
 from octue.resources.service_backends import GCPPubSubBackend
 from octue.runner import Runner
@@ -47,9 +46,5 @@ def answer_question(question, project_name):
         logger.info("Analysis successfully run and response sent for question %r.", question_uuid)
 
     except BaseException as error:  # noqa
-        service.send_exception(
-            topic=Topic(name=convert_service_id_to_pub_sub_form(service_sruid), project_name=project_name),
-            question_uuid=question_uuid,
-        )
-
+        service.send_exception(question_uuid=question_uuid, originator="UNKNOWN")
         logger.exception(error)

octue/cloud/emulators/_pub_sub.py

@@ -1,21 +1,22 @@
 import importlib.metadata
 import json
 import logging
+from collections import defaultdict
 
 import google.api_core
 
 from octue.cloud.pub_sub import Subscription, Topic
-from octue.cloud.pub_sub.service import ANSWERS_NAMESPACE, PARENT_SENDER_TYPE, Service
-from octue.cloud.service_id import convert_service_id_to_pub_sub_form
+from octue.cloud.pub_sub.service import PARENT_SENDER_TYPE, Service
 from octue.resources import Manifest
 from octue.utils.dictionaries import make_minimal_dictionary
 from octue.utils.encoders import OctueJSONEncoder
 
 
 logger = logging.getLogger(__name__)
 
-TOPICS = {}
-SUBSCRIPTIONS = {}
+TOPICS = set()
+SUBSCRIPTIONS = set()
+MESSAGES = defaultdict(list)
 
 
 class MockTopic(Topic):
@@ -33,7 +34,7 @@ def create(self, allow_existing=False):
                 raise google.api_core.exceptions.AlreadyExists(f"Topic {self.path!r} already exists.")
 
         if not self.exists():
-            TOPICS[self.name] = []
+            TOPICS.add(self.name)
             self._created = True
 
     def delete(self):
@@ -42,7 +43,7 @@ def delete(self):
         :return None:
         """
         try:
-            del TOPICS[self.name]
+            TOPICS.remove(self.name)
         except KeyError:
             pass
 
@@ -73,7 +74,7 @@ def create(self, allow_existing=False):
                 raise google.api_core.exceptions.AlreadyExists(f"Subscription {self.path!r} already exists.")
 
         if not self.exists():
-            SUBSCRIPTIONS[self.name] = []
+            SUBSCRIPTIONS.add(self.name)
             self._created = True
 
     def delete(self):
@@ -146,8 +147,7 @@ def publish(self, topic, data, retry=None, **attributes):
         :param google.api_core.retry.Retry|None retry:
         :return MockFuture:
         """
-        subscription_name = ".".join((get_pub_sub_resource_name(topic), ANSWERS_NAMESPACE, attributes["question_uuid"]))
-        SUBSCRIPTIONS[subscription_name].append(MockMessage(data=data, attributes=attributes))
+        MESSAGES[attributes["question_uuid"]].append(MockMessage(data=data, attributes=attributes))
         return MockFuture()
 
 
@@ -191,12 +191,12 @@ def pull(self, request, timeout=None, retry=None):
         if self.closed:
             raise ValueError("ValueError: Cannot invoke RPC: Channel closed!")
 
-        subscription_name = get_pub_sub_resource_name(request["subscription"])
+        question_uuid = request["subscription"].split(".")[-1]
 
         try:
             return MockPullResponse(
                 received_messages=[
-                    MockMessageWrapper(message=SUBSCRIPTIONS[subscription_name].pop(0)),
+                    MockMessageWrapper(message=MESSAGES[question_uuid].pop(0)),
                 ]
             )
 
@@ -242,6 +242,13 @@ def __init__(self, message):
         self.message = message
         self.ack_id = None
 
+    def __repr__(self):
+        """Represent the mock message as a string.
+
+        :return str:
+        """
+        return f"<{type(self).__name__}(message={self.message})>"
+
 
 class MockMessage:
     """A mock Pub/Sub message containing serialised data and any number of attributes.
@@ -356,8 +363,7 @@ def ask(
 
         # Delete question from messages sent to subscription so the parent doesn't pick it up as a response message. We
         # do this as subscription filtering isn't implemented in this set of mocks.
-        subscription_name = ".".join((convert_service_id_to_pub_sub_form(service_id), ANSWERS_NAMESPACE, question_uuid))
-        SUBSCRIPTIONS["octue.services." + subscription_name].pop(0)
+        MESSAGES[question_uuid].pop(0)
 
         question = make_minimal_dictionary(kind="question", input_values=input_values, children=children)
 
@@ -371,13 +377,15 @@ def ask(
                 MockMessage.from_primitive(
                     data=question,
                     attributes={
-                        "sender_type": PARENT_SENDER_TYPE,
                         "question_uuid": question_uuid,
                         "forward_logs": subscribe_to_logs,
-                        "version": parent_sdk_version,
                         "save_diagnostics": save_diagnostics,
-                        "message_number": 0,
-                        "sender": service_id,
+                        "order": 0,
+                        "originator": self.id,
+                        "sender": self.id,
+                        "sender_type": PARENT_SENDER_TYPE,
+                        "sender_sdk_version": parent_sdk_version,
+                        "recipient": service_id,
                     },
                 )
             )

octue/cloud/events/__init__.py

@@ -0,0 +1 @@
+OCTUE_SERVICES_PREFIX = "octue.services"

octue/cloud/events/counter.py

@@ -0,0 +1,35 @@
+class EventCounter:
+    """A mutable counter for keeping track of the emission order of events. This is used in the `Service` class instead
+    of an integer because it is mutable and can be passed to the `Service._emit_event` method and incremented as
+    events are emitted.
+
+    :return None:
+    """
+
+    def __init__(self):
+        self.count = 0
+
+    def __iadd__(self, other):
+        """Increment the counter by an integer.
+
+        :return octue.cloud.events.counter.EventCounter: the event counter with its count updated
+        """
+        if not isinstance(other, int):
+            raise ValueError(f"Event counters can only be incremented by an integer; received {other!r}.")
+
+        self.count += other
+        return self
+
+    def __int__(self):
+        """Get the counter as an integer.
+
+        :return int: the counter as an integer
+        """
+        return int(self.count)
+
+    def __repr__(self):
+        """Represent the counter as a string.
+
+        :return str: the counter represented as a string.
+        """
+        return f"<{type(self).__name__}(count={self.count})"

octue/cloud/events/handler.py

@@ -31,7 +31,7 @@ class AbstractEventHandler:
     """An abstract event handler. Inherit from this and add the `handle_events` and `_extract_event_and_attributes`
     methods to handle events from a specific source synchronously or asynchronously.
 
-    :param octue.cloud.pub_sub.service.Service receiving_service: the service that's receiving the events
+    :param octue.cloud.pub_sub.service.Service recipient: the service that's receiving the events
     :param callable|None handle_monitor_message: a function to handle monitor messages (e.g. send them to an endpoint for plotting or displaying) - this function should take a single JSON-compatible python primitive
     :param bool record_events: if `True`, record received events in the `received_events` attribute
     :param dict|None event_handlers: a mapping of event type names to callables that handle each type of event. The handlers should not mutate the events.
@@ -43,15 +43,15 @@ class AbstractEventHandler:
 
     def __init__(
         self,
-        receiving_service,
+        recipient,
         handle_monitor_message=None,
         record_events=True,
         event_handlers=None,
         schema=SERVICE_COMMUNICATION_SCHEMA,
         skip_missing_events_after=10,
         only_handle_result=False,
     ):
-        self.receiving_service = receiving_service
+        self.recipient = recipient
         self.handle_monitor_message = handle_monitor_message
         self.record_events = record_events
         self.schema = schema
@@ -122,33 +122,31 @@ def _extract_and_enqueue_event(self, container):
         if not is_event_valid(
             event=event,
             attributes=attributes,
-            receiving_service=self.receiving_service,
+            recipient=self.recipient,
             parent_sdk_version=PARENT_SDK_VERSION,
-            child_sdk_version=attributes.get("version"),
+            child_sdk_version=attributes["sender_sdk_version"],
             schema=self.schema,
         ):
             return
 
         # Get the child's SRUID and Octue SDK version from the first event.
         if not self._child_sdk_version:
-            self.question_uuid = attributes.get("question_uuid")
-            self._child_sdk_version = attributes["version"]
+            self.child_sruid = attributes["sender"]
+            self.question_uuid = attributes["question_uuid"]
+            self._child_sdk_version = attributes["sender_sdk_version"]
 
-            # Backwards-compatible with previous event schema versions.
-            self.child_sruid = attributes.get("sender", "REMOTE")
+        logger.debug("%r received an event related to question %r.", self.recipient, self.question_uuid)
+        order = attributes["order"]
 
-        logger.debug("%r received an event related to question %r.", self.receiving_service, self.question_uuid)
-        event_number = attributes["message_number"]
-
-        if event_number in self.waiting_events:
+        if order in self.waiting_events:
             logger.warning(
-                "%r: Event with duplicate event number %d received for question %s - overwriting original event.",
-                self.receiving_service,
-                event_number,
+                "%r: Event with duplicate order %d received for question %s - overwriting original event.",
+                self.recipient,
+                order,
                 self.question_uuid,
             )
 
-        self.waiting_events[event_number] = event
+        self.waiting_events[order] = event
 
     def _attempt_to_handle_waiting_events(self):
         """Attempt to handle events waiting in `self.waiting_events`. If these events aren't consecutive to the
@@ -204,7 +202,7 @@ def _skip_to_earliest_waiting_event(self):
         logger.warning(
             "%r: %d consecutive events missing for question %r after %ds - skipping to next earliest waiting event "
             "(event %d).",
-            self.receiving_service,
+            self.recipient,
             number_of_missing_events,
             self.question_uuid,
             self.skip_missing_events_after,
@@ -236,7 +234,7 @@ def _handle_delivery_acknowledgement(self, event):
         :param dict event:
         :return None:
         """
-        logger.info("%r's question was delivered at %s.", self.receiving_service, event["datetime"])
+        logger.info("%r's question was delivered at %s.", self.recipient, event["datetime"])
 
     def _handle_heartbeat(self, event):
         """Record the time the heartbeat was received.
@@ -253,7 +251,7 @@ def _handle_monitor_message(self, event):
         :param dict event:
         :return None:
         """
-        logger.debug("%r received a monitor message.", self.receiving_service)
+        logger.debug("%r received a monitor message.", self.recipient)
 
         if self.handle_monitor_message is not None:
             self.handle_monitor_message(event["data"])
@@ -317,7 +315,7 @@ def _handle_result(self, event):
         :param dict event:
         :return dict:
         """
-        logger.info("%r received an answer to question %r.", self.receiving_service, self.question_uuid)
+        logger.info("%r received an answer to question %r.", self.recipient, self.question_uuid)
 
         if event.get("output_manifest"):
             output_manifest = Manifest.deserialise(event["output_manifest"])

octue/cloud/events/replayer.py

@@ -12,7 +12,7 @@
 class EventReplayer(AbstractEventHandler):
     """A replayer for events retrieved asynchronously from some kind of storage.
 
-    :param octue.cloud.pub_sub.service.Service receiving_service: the service that's receiving the events
+    :param octue.cloud.pub_sub.service.Service recipient: the service that's receiving the events
     :param callable|None handle_monitor_message: a function to handle monitor messages (e.g. send them to an endpoint for plotting or displaying) - this function should take a single JSON-compatible python primitive
     :param bool record_events: if `True`, record received events in the `received_events` attribute
     :param dict|None event_handlers: a mapping of event type names to callables that handle each type of event. The handlers should not mutate the events.
@@ -23,15 +23,15 @@ class EventReplayer(AbstractEventHandler):
 
     def __init__(
         self,
-        receiving_service=None,
+        recipient=None,
         handle_monitor_message=None,
         record_events=True,
         event_handlers=None,
         schema=SERVICE_COMMUNICATION_SCHEMA,
         only_handle_result=False,
     ):
         super().__init__(
-            receiving_service or Service(backend=ServiceBackend(), service_id="local/local:local"),
+            recipient or Service(backend=ServiceBackend(), service_id="local/local:local"),
             handle_monitor_message=handle_monitor_message,
             record_events=record_events,
             event_handlers=event_handlers,
@@ -51,14 +51,16 @@ def handle_events(self, events):
         for event in events:
             self._extract_and_enqueue_event(event)
 
-        self._earliest_waiting_event_number = min(self.waiting_events.keys())
-        return self._attempt_to_handle_waiting_events()
+        # Handle the case where no events (or no valid events) have been received.
+        if self.waiting_events:
+            self._earliest_waiting_event_number = min(self.waiting_events.keys())
+            return self._attempt_to_handle_waiting_events()
 
     def _extract_event_and_attributes(self, container):
         """Extract an event and its attributes from the event container.
 
         :param dict container: the container of the event
         :return (any, dict): the event and its attributes
         """
-        container["attributes"]["message_number"] = int(container["attributes"]["message_number"])
+        container["attributes"]["order"] = int(container["attributes"]["order"])
         return container["event"], container["attributes"]