MRG: Merge pull request #545 from octue/fix/fix-crash-diagnostics-manifest-upload

Fix manifest upload in crash diagnostics

Author: cortadocodes

docs/source/inter_service_compatibility.rst

@@ -17,6 +17,7 @@
 from octue.cloud.pub_sub.message_handler import OrderedMessageHandler
 from octue.cloud.service_id import convert_service_id_to_pub_sub_form, create_service_sruid, validate_service_sruid
 from octue.compatibility import warn_if_incompatible
+from octue.utils.decoders import OctueJSONDecoder
 from octue.utils.encoders import OctueJSONEncoder
 from octue.utils.exceptions import convert_exception_to_primitives
 from octue.utils.objects import get_nested_attribute
@@ -286,14 +287,12 @@ def ask(
         )
         answer_subscription.create(allow_existing=False)
 
-        serialised_input_manifest = None
-
         if input_manifest is not None:
             input_manifest.use_signed_urls_for_datasets()
-            serialised_input_manifest = input_manifest.serialise()
+            input_manifest = input_manifest.to_primitive()
 
         self._send_message(
-            {"input_values": input_values, "input_manifest": serialised_input_manifest, "children": children},
+            {"input_values": input_values, "input_manifest": input_manifest, "children": children},
             topic=question_topic,
             question_uuid=question_uuid,
             forward_logs=subscribe_to_logs,
@@ -474,6 +473,8 @@ def _parse_question(self, question):
         :param dict|Message question:
         :return (dict, str, bool, str|None, bool):
         """
+        logger.info("%r received a question.", self)
+
         try:
             # Parse question directly from Pub/Sub or Dataflow.
             data = json.loads(question.data.decode())
@@ -486,7 +487,9 @@ def _parse_question(self, question):
             # Parse question from Google Cloud Run.
             data = json.loads(base64.b64decode(question["data"]).decode("utf-8").strip())
 
-        logger.info("%r received a question.", self)
+        # Keep backwards compatibility with questions from Octue services running `octue<0.41.1`.
+        if isinstance(data["input_manifest"], str):
+            data["input_manifest"] = json.loads(data["input_manifest"], cls=OctueJSONDecoder)
 
         question_uuid = get_nested_attribute(question, "attributes.question_uuid")
         forward_logs = bool(int(get_nested_attribute(question, "attributes.forward_logs")))
@@ -504,4 +507,5 @@ def _parse_question(self, question):
         except AttributeError:
             allow_save_diagnostics_data_on_crash = False
 
+        logger.info("%r parsed the question successfully.", self)
         return data, question_uuid, forward_logs, parent_sdk_version, allow_save_diagnostics_data_on_crash

octue/cloud/pub_sub/service.py

@@ -56,3 +56,4 @@
 {"parent_sdk_version": "0.40.1", "question": {"data": "{\"input_values\": {\"height\": 4, \"width\": 72}, \"input_manifest\": \"{\\n    \\\"datasets\\\": {\\n        \\\"my_dataset\\\": \\\"/var/folders/sk/hf5fbp616c77tsys9lz55qn40000gp/T/tmppnbc6rcc\\\"\\n    },\\n    \\\"id\\\": \\\"31440887-beef-41a3-935f-08adbe8dbbdc\\\",\\n    \\\"name\\\": null\\n}\", \"children\": null}", "attributes": {"question_uuid": "f66edd6e-517c-4c26-9d60-16cb508e5fe1", "forward_logs": "1", "allow_save_diagnostics_data_on_crash": "1", "octue_sdk_version": "0.40.1"}}}
 {"parent_sdk_version": "0.40.2", "question": {"data": "{\"input_values\": {\"height\": 4, \"width\": 72}, \"input_manifest\": \"{\\n    \\\"datasets\\\": {\\n        \\\"my_dataset\\\": \\\"/var/folders/sk/hf5fbp616c77tsys9lz55qn40000gp/T/tmprj5sl5al\\\"\\n    },\\n    \\\"id\\\": \\\"6ce744fb-fce6-4346-949a-109675ca7a5a\\\",\\n    \\\"name\\\": null\\n}\", \"children\": null}", "attributes": {"question_uuid": "42d9a880-8672-4be6-a3ae-b8518a9e58db", "forward_logs": "1", "allow_save_diagnostics_data_on_crash": "1", "octue_sdk_version": "0.40.2"}}}
 {"parent_sdk_version": "0.41.0", "question": {"data": "{\"input_values\": {\"height\": 4, \"width\": 72}, \"input_manifest\": \"{\\n    \\\"datasets\\\": {\\n        \\\"my_dataset\\\": \\\"/var/folders/sk/hf5fbp616c77tsys9lz55qn40000gp/T/tmpnc9s8hwm\\\"\\n    },\\n    \\\"id\\\": \\\"b21ca765-9dac-4291-a301-daa4d282f1fd\\\",\\n    \\\"name\\\": null\\n}\", \"children\": null}", "attributes": {"question_uuid": "76ae0850-50b7-4c7d-a22b-c8bcd0ab2bb3", "forward_logs": "1", "allow_save_diagnostics_data_on_crash": "1", "octue_sdk_version": "0.41.0"}}}
+{"parent_sdk_version": "0.41.1", "question": {"data": "{\"input_values\": {\"height\": 4, \"width\": 72}, \"input_manifest\": {\"id\": \"6e57125c-3e32-407c-b0f8-6403d2db6d8a\", \"name\": null, \"datasets\": {\"my_dataset\": \"/var/folders/sk/hf5fbp616c77tsys9lz55qn40000gp/T/tmpsri5p05d\"}}, \"children\": null}", "attributes": {"question_uuid": "07bfa681-a93c-44e4-9ee7-a33e9518a77c", "forward_logs": "1", "allow_save_diagnostics_data_on_crash": "1", "octue_sdk_version": "0.41.1"}}}