MRG: Merge pull request #673 from octue/improve-event-filtering

Improve event filtering

Author: cortadocodes

.devcontainer/Dockerfile

@@ -16,5 +16,5 @@ ENV HISTFILE="/workspaces/octue-sdk-python/.devcontainer/.zsh_history"
 USER vscode
 ENV POETRY_HOME=/home/vscode/.poetry
 RUN curl -sSL https://install.python-poetry.org | python3 -
-ENV PATH "$POETRY_HOME/bin:$PATH"
+ENV PATH="$POETRY_HOME/bin:$PATH"
 RUN poetry config virtualenvs.create false

Dockerfile

@@ -4,7 +4,7 @@ RUN apt-get update && apt-get install -y git curl
 
 ENV POETRY_HOME=/etc/poetry
 RUN curl -sSL https://install.python-poetry.org | python3 -
-ENV PATH "$POETRY_HOME/bin:$PATH"
+ENV PATH="$POETRY_HOME/bin:$PATH"
 RUN poetry config virtualenvs.create false
 
 # Install python dependencies. Note that poetry installs any root packages by default, but this is not available at this

docs/source/inter_service_compatibility.rst

@@ -1,7 +1,7 @@
 FROM windpioneers/gdal-python:little-gecko-gdal-2.4.1-python-3.10-slim
 
 # Ensure print statements and log messages appear promptly in Cloud Logging.
-ENV PYTHONUNBUFFERED True
+ENV PYTHONUNBUFFERED=True
 
 ENV PROJECT_ROOT=/workspace
 WORKDIR $PROJECT_ROOT
@@ -10,7 +10,7 @@ RUN apt-get update -y && apt-get install -y --fix-missing build-essential && rm
 
 # Install poetry.
 ENV POETRY_HOME=/root/.poetry
-ENV PATH "$POETRY_HOME/bin:$PATH"
+ENV PATH="$POETRY_HOME/bin:$PATH"
 RUN curl -sSL https://install.python-poetry.org | python3 - && poetry config virtualenvs.create false;
 
 # Copy in the dependencies file(s) for caching. One or more of `requirements.txt`, `setup.py`, and `pyproject.toml and

octue/cloud/deployment/google/cloud_run/Dockerfile-python310

@@ -1,7 +1,7 @@
 FROM windpioneers/gdal-python:modest-heron-gdal-2.4.1-python-3.11-slim
 
 # Ensure print statements and log messages appear promptly in Cloud Logging.
-ENV PYTHONUNBUFFERED True
+ENV PYTHONUNBUFFERED=True
 
 ENV PROJECT_ROOT=/workspace
 WORKDIR $PROJECT_ROOT
@@ -10,7 +10,7 @@ RUN apt-get update -y && apt-get install -y --fix-missing build-essential && rm
 
 # Install poetry.
 ENV POETRY_HOME=/root/.poetry
-ENV PATH "$POETRY_HOME/bin:$PATH"
+ENV PATH="$POETRY_HOME/bin:$PATH"
 RUN curl -sSL https://install.python-poetry.org | python3 - && poetry config virtualenvs.create false;
 
 # Copy in the dependencies file(s) for caching. One or more of `requirements.txt`, `setup.py`, and `pyproject.toml and

octue/cloud/deployment/google/cloud_run/Dockerfile-python311

@@ -1,7 +1,7 @@
 FROM windpioneers/gdal-python:little-gecko-gdal-2.4.1-python-3.9-slim
 
 # Ensure print statements and log messages appear promptly in Cloud Logging.
-ENV PYTHONUNBUFFERED True
+ENV PYTHONUNBUFFERED=True
 
 ENV PROJECT_ROOT=/workspace
 WORKDIR $PROJECT_ROOT
@@ -10,7 +10,7 @@ RUN apt-get update -y && apt-get install -y --fix-missing build-essential && rm
 
 # Install poetry.
 ENV POETRY_HOME=/root/.poetry
-ENV PATH "$POETRY_HOME/bin:$PATH"
+ENV PATH="$POETRY_HOME/bin:$PATH"
 RUN curl -sSL https://install.python-poetry.org | python3 - && poetry config virtualenvs.create false;
 
 # Copy in the dependencies file(s) for caching. One or more of `requirements.txt`, `setup.py`, and `pyproject.toml and

octue/cloud/deployment/google/cloud_run/Dockerfile-python39

@@ -81,7 +81,7 @@ def _check_if_should_drop_question(question_uuid, retry_count):
         previous_question_attempts = get_events(
             table_id=service_configuration.event_store_table_id,
             question_uuid=question_uuid,
-            kind="delivery_acknowledgement",
+            exclude_kinds=["question"],
         )
 
     except google.api_core.exceptions.NotFound:

octue/cloud/deployment/google/cloud_run/flask_app.py

@@ -39,6 +39,7 @@ class AbstractEventHandler:
     :param dict|None event_handlers: a mapping of event type names to callables that handle each type of event. The handlers must not mutate the events.
     :param dict schema: the JSON schema to validate events against
     :param bool include_service_metadata_in_logs: if `True`, include the SRUIDs and question UUIDs of the service revisions involved in the question to the start of the log message
+    :param str|None exclude_logs_containing: if provided, skip handling log messages containing this string
     :param bool only_handle_result: if `True`, skip handling non-result events and only handle the "result" event when received (turning this on speeds up event handling)
     :param bool validate_events: if `True`, validate events before attempting to handle them (turning this off speeds up event handling)
     :return None:
@@ -51,13 +52,15 @@ def __init__(
         event_handlers=None,
         schema=SERVICE_COMMUNICATION_SCHEMA,
         include_service_metadata_in_logs=True,
+        exclude_logs_containing=None,
         only_handle_result=False,
         validate_events=True,
     ):
         self.handle_monitor_message = handle_monitor_message
         self.record_events = record_events
         self.schema = schema
         self.include_service_metadata_in_logs = include_service_metadata_in_logs
+        self.exclude_logs_containing = exclude_logs_containing
         self.only_handle_result = only_handle_result
         self.validate_events = validate_events
 
@@ -201,6 +204,9 @@ def _handle_log_message(self, event, attributes):
         :param dict attributes: the event's attributes
         :return None:
         """
+        if self.exclude_logs_containing and self.exclude_logs_containing in event["log_record"]["msg"]:
+            return
+
         record = logging.makeLogRecord(event["log_record"])
 
         # Split the log message into its parts.

octue/cloud/events/handler.py

@@ -8,15 +8,17 @@
 
 
 class EventReplayer(AbstractEventHandler):
-    """A replayer for events retrieved asynchronously from storage.
+    """A replayer for events retrieved asynchronously from storage. Note that events aren't validated by default as the
+    main use case for this class is to replay already-validated events from the event store.
 
     :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 must not mutate the events.
     :param dict|str schema: the JSON schema to validate events against
     :param bool include_service_metadata_in_logs: if `True`, include the SRUIDs and question UUIDs of the service revisions involved in the question to the start of the log message
+    :param str|None exclude_logs_containing: if provided, skip handling log messages containing this string
     :param bool only_handle_result: if `True`, skip non-result events and only handle the "result" event if present (turning this on speeds up event handling)
-    :param bool validate_events: if `True`, validate events before attempting to handle them (turning this off speeds up event handling)
+    :param bool validate_events: if `True`, validate events before attempting to handle them (this is off by default to speed up event handling)
     :return None:
     """
 
@@ -27,8 +29,9 @@ def __init__(
         event_handlers=None,
         schema=SERVICE_COMMUNICATION_SCHEMA,
         include_service_metadata_in_logs=True,
+        exclude_logs_containing=None,
         only_handle_result=False,
-        validate_events=True,
+        validate_events=False,
     ):
         event_handlers = event_handlers or {
             "question": self._handle_question,
@@ -46,6 +49,7 @@ def __init__(
             event_handlers=event_handlers,
             schema=schema,
             include_service_metadata_in_logs=include_service_metadata_in_logs,
+            exclude_logs_containing=exclude_logs_containing,
             only_handle_result=only_handle_result,
             validate_events=validate_events,
         )

octue/cloud/events/replayer.py

@@ -28,7 +28,8 @@ def get_events(
     question_uuid=None,
     parent_question_uuid=None,
     originator_question_uuid=None,
-    kind=None,
+    kinds=None,
+    exclude_kinds=None,
     include_backend_metadata=False,
     tail=True,
     limit=1000,
@@ -47,13 +48,14 @@ def get_events(
     :param str|None question_uuid: the UUID of a question to get events for
     :param str|None parent_question_uuid: the UUID of a parent question to get the sub-question events for
     :param str|None originator_question_uuid: the UUID of an originator question get the full tree of events for
-    :param str|None kind: the kind of event to get; if `None`, all event kinds are returned
+    :param iter(str)|None kinds: the kinds of event to get; if `None`, all event kinds are returned
+    :param iter(str)|None exclude_kinds: the kind of events to exclude; if `None`, all event kinds are returned
     :param bool include_backend_metadata: if `True`, include the service backend metadata
     :param bool tail: if `True`, return the most recent events (where a limit applies); e.g. return the most recent 100 log records
     :param int limit: the maximum number of events to return
     :return list(dict): the events for the question; this will be empty if there are no events for the question
     """
-    _validate_inputs(question_uuid, parent_question_uuid, originator_question_uuid, kind)
+    _validate_inputs(question_uuid, parent_question_uuid, originator_question_uuid, kinds, exclude_kinds)
 
     if question_uuid:
         question_uuid_condition = "WHERE question_uuid=@relevant_question_uuid"
@@ -62,10 +64,17 @@ def get_events(
     elif originator_question_uuid:
         question_uuid_condition = "WHERE originator_question_uuid=@relevant_question_uuid"
 
-    if kind:
-        event_kind_condition = [f"AND kind={kind!r}"]
+    if kinds:
+        kinds_string = ", ".join([f"{kind!r}" for kind in kinds])
+        event_kinds_condition = [f"AND kind IN ({kinds_string})"]
     else:
-        event_kind_condition = []
+        event_kinds_condition = []
+
+    if exclude_kinds:
+        exclude_kinds_string = ", ".join([f"{kind!r}" for kind in exclude_kinds])
+        exclude_kinds_condition = [f"AND kind NOT IN ({exclude_kinds_string})"]
+    else:
+        exclude_kinds_condition = []
 
     # Make a shallow copy of the fields to query.
     fields = list(DEFAULT_FIELDS)
@@ -77,7 +86,8 @@ def get_events(
         [
             f"SELECT {', '.join(fields)} FROM `{table_id}`",
             question_uuid_condition,
-            *event_kind_condition,
+            *event_kinds_condition,
+            *exclude_kinds_condition,
         ]
     )
 
@@ -116,14 +126,15 @@ def get_events(
     return _unflatten_events(events)
 
 
-def _validate_inputs(question_uuid, parent_question_uuid, originator_question_uuid, kind):
+def _validate_inputs(question_uuid, parent_question_uuid, originator_question_uuid, kinds, exclude_kinds):
     """Check that only one of `question_uuid`, `parent_question_uuid`, or `originator_question_uuid` are provided and
     that the `kind` parameter is a valid event kind.
 
     :param str|None question_uuid: the UUID of a question to get events for
     :param str|None parent_question_uuid: the UUID of a parent question to get the sub-question events for
     :param str|None originator_question_uuid: the UUID of an originator question get the full tree of events for
-    :param str|None kind: the kind of event to get; if `None`, all event kinds are returned
+    :param iter(str)|None kinds: the kinds of event to get; if `None`, all event kinds are returned
+    :param iter(str)|None exclude_kinds: the kind of events to exclude; if `None`, all event kinds are returned
     :raise ValueError: if more than one of `question_uuid`, `parent_question_uuid`, or `originator_question_uuid` are provided or the `kind` parameter is invalid
     :return None:
     """
@@ -135,8 +146,25 @@ def _validate_inputs(question_uuid, parent_question_uuid, originator_question_uu
             "provided."
         )
 
-    if kind and kind not in VALID_EVENT_KINDS:
-        raise ValueError(f"`kind` must be one of {VALID_EVENT_KINDS!r}; received {kind!r}.")
+    kinds_inputs = (bool(kinds), bool(exclude_kinds))
+
+    if sum(kinds_inputs) > 1:
+        raise ValueError(
+            f"Only one of `kinds` and `exclude_kinds` can be provided at once; received kinds={kinds!r} and "
+            f"exclude_kinds={exclude_kinds!r}."
+        )
+
+    if kinds:
+        for kind in kinds:
+            if kind not in VALID_EVENT_KINDS:
+                raise ValueError(f"All items in `kinds` must be one of {VALID_EVENT_KINDS!r}; received {kind!r}.")
+
+    if exclude_kinds:
+        for kind in exclude_kinds:
+            if kind not in VALID_EVENT_KINDS:
+                raise ValueError(
+                    f"All items in `exclude_kinds` must be one of {VALID_EVENT_KINDS!r}; received {kind!r}."
+                )
 
 
 def _deserialise_event(event):

octue/cloud/pub_sub/bigquery.py

@@ -70,6 +70,9 @@ class GoogleCloudPubSubEventHandler(AbstractEventHandler):
     :param dict|None event_handlers: a mapping of event type names to callables that handle each type of event. The handlers must not mutate the events.
     :param dict|str schema: the JSON schema to validate events against
     :param bool include_service_metadata_in_logs: if `True`, include the SRUIDs and question UUIDs of the service revisions involved in the question to the start of the log message
+    :param str|None exclude_logs_containing: if provided, skip handling log messages containing this string
+    :param bool only_handle_result: if `True`, skip non-result events and only handle the "result" event if present (turning this on speeds up event handling)
+    :param bool validate_events: if `True`, validate events before attempting to handle them (turn this off to speed up event handling at risk of failure if an invalid event is received)
     :return None:
     """
 
@@ -81,6 +84,9 @@ def __init__(
         event_handlers=None,
         schema=SERVICE_COMMUNICATION_SCHEMA,
         include_service_metadata_in_logs=True,
+        exclude_logs_containing=None,
+        only_handle_result=False,
+        validate_events=True,
     ):
         self.subscription = subscription
 
@@ -90,6 +96,9 @@ def __init__(
             event_handlers=event_handlers,
             schema=schema,
             include_service_metadata_in_logs=include_service_metadata_in_logs,
+            exclude_logs_containing=exclude_logs_containing,
+            only_handle_result=only_handle_result,
+            validate_events=validate_events,
         )
 
         self._heartbeat_checker = None