Publish answers to question topic

.github/workflows/add-issues-to-octue-board.yml

@@ -6,6 +6,10 @@ on:
 
 jobs:
   add-issues-to-octue-board:
-    uses: octue/.github/.github/workflows/reusable-add-issues-to-octue-board.yml@main
-    secrets:
-      github-token: ${{ secrets.PROJECT_AUTOMATION_GITHUB_TOKEN_2 }}
+    runs-on: ubuntu-latest
+    steps:
+      - name: Add to Board
+        uses: actions/[email protected]
+        with:
+          project-url: https://github.com/orgs/octue/projects/22
+          github-token: ${{ secrets.github-token }}

.github/workflows/python-ci.yml

@@ -77,6 +77,9 @@ jobs:
     if: "!contains(github.event.head_commit.message, 'skipci')"
     runs-on: ubuntu-latest
     needs: [check-semantic-version, run-tests]
+    permissions:
+      id-token: write
+      contents: read
     steps:
       - name: Checkout Repository
         uses: actions/checkout@v3

.github/workflows/release.yml

@@ -101,24 +101,3 @@ jobs:
 
       - name: Publish package distributions to PyPI
         uses: pypa/[email protected]
-
-  docker:
-    runs-on: ubuntu-latest
-    needs: [run-tests, release]
-    timeout-minutes: 300
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v3
-
-      - name: Log in to DockerHub
-        uses: docker/login-action@v1
-        with:
-          username: ${{ secrets.DOCKERHUB_USERNAME }}
-          password: ${{ secrets.DOCKERHUB_TOKEN }}
-
-      - name: Build and push
-        uses: docker/[email protected]
-        with:
-          context: .
-          push: true
-          tags: octue/octue-sdk-python:${{ needs.run-tests.outputs.package_version }}-slim,octue/octue-sdk-python:latest

.readthedocs.yaml

@@ -0,0 +1,16 @@
+# Read the Docs configuration file for Sphinx projects
+# See https://docs.readthedocs.io/en/stable/config-file/v2.html for details
+
+version: 2
+
+build:
+  os: ubuntu-22.04
+  tools:
+    python: "3.10"
+
+sphinx:
+  configuration: docs/source/conf.py
+
+python:
+  install:
+    - requirements: docs/requirements.txt

docs/source/asking_questions.rst

@@ -48,7 +48,7 @@ You can also set the following options when you call :mod:`Child.ask <octue.reso
 - ``allow_local_files`` - if true, local files/datasets are allowed in any input manifest you supply
 - ``handle_monitor_message`` - if provided a function, it will be called on any monitor messages from the child
 - ``record_messages_to`` – if given a path to a JSON file, messages received from the parent while it processes the question are saved to it
-- ``allow_save_diagnostics_data_on_crash`` – if true, the input values and input manifest (including its datasets) will be saved by the child for future crash diagnostics if it fails while processing them
+- ``debug`` – must be one of {"DEBUG_OFF", "DEBUG_ON_CRASH", "DEBUG_ON"}; if turned on, allow the input values and manifest (and its datasets) to be saved by the child either all the time or just if the analysis fails
 - ``question_uuid`` - if provided, the question will use this UUID instead of a generated one
 - ``timeout`` - how long in seconds to wait for an answer (``None`` by default - i.e. don't time out)
 

docs/source/inter_service_compatibility.rst

@@ -2,9 +2,14 @@
 Inter-service compatibility
 ===========================
 
-Parents and children running nearly all versions of ``octue`` can communicate with each other compatibly, although a
-small number can't. The table below shows which parent SDK versions (rows) send questions that can be processed by each
-child SDK version (columns).
+Octue services acting as parents and children communicate with each other according to the `services communication
+schema <https://strands.octue.com/octue/service-communication>`_. Up until version ``0.51.0``, services running nearly
+all versions of ``octue`` could communicate with each other compatibly. To allow a significant infrastructure upgrade,
+version ``0.51.0`` introduced a number of breaking changes to the standard meaning services running ``0.51.0`` or
+greater are only able to communicate with other services running ``0.51.0`` or greater. The table below shows which
+``octue`` versions parents can run (rows) to send questions compatible with versions children are running (columns).
+Note that this table does not display whether children's responses are compatible with the parent, just that a child is
+able to accept a question.
 
 **Key**
 

docs/source/services.rst

@@ -88,3 +88,12 @@ They look like ``namespace/name:tag`` where the tag is often a semantic version
         name. It can be used to ask a question to a service without specifying a specific revision of it. This enables
         asking questions to, for example, the service ``octue/my-service`` and automatically having them routed to its
         default (usually latest) revision. :ref:`See here for more info<using_default_revision_tag>`.
+
+
+Service communication standard
+==============================
+
+Octue services communicate according to the service communication standard. The JSON schema defining this can be found
+`here <https://strands.octue.com/octue/service-communication>`_. Messages received by services are validated against it
+and invalid messages are rejected. The schema is in beta, so (rare) breaking changes are reflected in the minor version
+number.

docs/source/troubleshooting_services.rst

@@ -111,7 +111,7 @@ your service to fail.
 Disabling crash diagnostics
 ===========================
 When asking a question to a child, parents can disable crash diagnostics upload in the child on a question-by-question
-basis by setting ``allow_save_diagnostics_data_on_crash`` to ``False`` in :mod:`Child.ask <octue.resources.child.Child.ask>`.
+basis by setting ``debug`` to ``"DEBUG_OFF"`` in :mod:`Child.ask <octue.resources.child.Child.ask>`.
 For example:
 
 .. code-block:: python
@@ -123,5 +123,5 @@ For example:
 
     answer = child.ask(
         input_values={"height": 32, "width": 3},
-        allow_save_diagnostics_data_on_crash=False,
+        debug="DEBUG_OFF",
     )

octue/__init__.py

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

octue/cli.py

@@ -12,7 +12,7 @@
 
 from octue.cloud import storage
 from octue.cloud.pub_sub import Subscription, Topic
-from octue.cloud.pub_sub.service import Service
+from octue.cloud.pub_sub.service import PARENT_SENDER_TYPE, Service
 from octue.cloud.service_id import convert_service_id_to_pub_sub_form, create_sruid, get_sruid_parts
 from octue.cloud.storage import GoogleCloudStorageClient
 from octue.configuration import load_service_and_app_configuration
@@ -409,6 +409,7 @@ def create_push_subscription(
         name=pub_sub_sruid,
         topic=topic,
         project_name=project_name,
+        filter=f'attributes.sender_type = "{PARENT_SENDER_TYPE}"',
         expiration_time=expiration_time,
         push_endpoint=push_endpoint,
     )

octue/cloud/deployment/google/answer_pub_sub_question.py

@@ -1,7 +1,8 @@
 import logging
 
+from octue.cloud.pub_sub import Topic
 from octue.cloud.pub_sub.service import Service
-from octue.cloud.service_id import create_sruid, get_sruid_parts
+from octue.cloud.service_id import convert_service_id_to_pub_sub_form, 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
@@ -17,7 +18,7 @@
 def answer_question(question, project_name):
     """Answer a question sent to an app deployed in Google Cloud.
 
-    :param dict|tuple|apache_beam.io.gcp.pubsub.PubsubMessage question:
+    :param dict|tuple question:
     :param str project_name:
     :return None:
     """
@@ -31,9 +32,7 @@ def answer_question(question, project_name):
     )
 
     service = Service(service_id=service_sruid, backend=GCPPubSubBackend(project_name=project_name))
-
     question_uuid = get_nested_attribute(question, "attributes.question_uuid")
-    answer_topic = service.instantiate_answer_topic(question_uuid)
 
     try:
         runner = Runner(
@@ -50,11 +49,13 @@ def answer_question(question, project_name):
         )
 
         service.run_function = runner.run
-
-        service.answer(question, answer_topic=answer_topic)
+        service.answer(question)
         logger.info("Analysis successfully run and response sent for question %r.", question_uuid)
 
-    # Forward any errors in the deployment configuration (errors in the analysis are already forwarded by the service).
     except BaseException as error:  # noqa
-        service.send_exception(topic=answer_topic)
+        service.send_exception(
+            topic=Topic(name=convert_service_id_to_pub_sub_form(service_sruid), project_name=project_name),
+            question_uuid=question_uuid,
+        )
+
         logger.exception(error)