14 forward error from worker to sdk

dev-requirements.txt

@@ -111,7 +111,7 @@ omotes-sdk-protocol==0.0.8
     # via
     #   -c requirements.txt
     #   omotes-sdk-python
-omotes-sdk-python==0.0.11
+omotes-sdk-python==0.0.12
     # via
     #   -c requirements.txt
     #   orchestrator (pyproject.toml)

pyproject.toml

@@ -26,7 +26,7 @@ dependencies = [
     "sqlalchemy ~= 2.0.27",
     "psycopg2-binary ~= 2.9.9",
     "celery ~= 5.3.6",
-    "omotes-sdk-python ~= 0.0.11",
+    "omotes-sdk-python ~= 0.0.12",
 ]
 
 [project.optional-dependencies]

requirements.txt

@@ -46,7 +46,7 @@ multidict==6.0.5
     # via yarl
 omotes-sdk-protocol==0.0.8
     # via omotes-sdk-python
-omotes-sdk-python==0.0.11
+omotes-sdk-python==0.0.12
     # via orchestrator (pyproject.toml)
 pamqp==3.2.1
     # via aiormq

src/omotes_orchestrator/__init__.py

@@ -13,8 +13,12 @@
 #  You should have received a copy of the GNU General Public License
 #  along with this program.  If not, see <https://www.gnu.org/licenses/>.
 import logging
-import os
-from omotes_sdk.internal.common.app_logging import setup_logging, LogLevel
+from dotenv import load_dotenv
 
-setup_logging(LogLevel.parse(os.environ.get("LOG_LEVEL", "DEBUG")), "omotes_orchestrator")
+load_dotenv(verbose=True)
+
+import os  # noqa: E402
+from omotes_sdk.internal.common.app_logging import setup_logging, LogLevel  # noqa: E402
+
+setup_logging(LogLevel.parse(os.environ.get("LOG_LEVEL", "INFO")), "omotes_orchestrator")
 setup_logging(LogLevel.parse(os.environ.get("LOG_LEVEL_SQL", "WARNING")), "sqlalchemy.engine")

src/omotes_orchestrator/main.py

@@ -8,7 +8,6 @@
 from types import FrameType
 from typing import Any, Union
 
-from dotenv import load_dotenv
 from omotes_orchestrator.postgres_interface import PostgresInterface
 from omotes_sdk.internal.orchestrator_worker_events.messages.task_pb2 import (
     TaskResult,
@@ -31,7 +30,6 @@
 from omotes_orchestrator.config import OrchestratorConfig
 from omotes_orchestrator.db_models.job import JobStatus as JobStatusDB
 
-load_dotenv(verbose=True)
 logger = logging.getLogger("omotes_orchestrator")
 
 
@@ -267,6 +265,27 @@ def task_result_received(self, serialized_message: bytes) -> None:
                     ),
                 )
                 self.postgresql_if.delete_job(job.id)
+            elif task_result.result_type == TaskResult.ResultType.ERROR:
+                logger.info(
+                    "Received error result for job %s through task %s",
+                    task_result.job_id,
+                    task_result.celery_task_id,
+                )
+                self.omotes_if.send_job_result(
+                    job=job,
+                    result=JobResult(
+                        uuid=str(job.id),
+                        result_type=JobResult.ResultType.ERROR,
+                        output_esdl=task_result.output_esdl,
+                        logs=task_result.logs,
+                    ),
+                )
+                self.postgresql_if.delete_job(job.id)
+            else:
+                logger.error(
+                    "Unknown task result %s. Please report and/or implement.",
+                    task_result.result_type,
+                )
 
     def task_progress_update(self, serialized_message: bytes) -> None:
         """When a task event is received from a worker through RabbitMQ, Celery side.