chore: remove wait_for_image_built (#819)

* chore: remove `wait_for_image_built` Signed-off-by: Guilhem Barthés <[email protected]> * fix: change queue when saving image and add logging Signed-off-by: Guilhem Barthés <[email protected]> * chore: change import orders Signed-off-by: Guilhem Barthés <[email protected]> * docs: changelog Signed-off-by: Guilhem Barthés <[email protected]> --------- Signed-off-by: Guilhem Barthés <[email protected]>
Substra · Feb 13, 2024 · 9906d7d · 9906d7d
1 parent e7aa3e7
commit 9906d7d
Show file tree

Hide file tree

Showing 5 changed files with 7 additions and 30 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -31,6 +31,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Remove mention of `chaincode` after `distributed mode` deprecation ([#795](https://github.com/Substra/substra-backend/pull/795))
 - BREAKING: remove `distributed` Skaffold profile [#768](https://github.com/Substra/substra-backend/pull/768)
+- Remove `wait_for_image_built` as he logic for changing status to `TODO` has been moved to the orchestrator and set only when the function is built ([#819](https://github.com/Substra/substra-backend/pull/819))
 
 ### Fixed
 

diff --git a/backend/substrapp/compute_tasks/image_builder.py b/backend/substrapp/compute_tasks/image_builder.py
@@ -1,15 +1,12 @@
 import os
 import pathlib
-import time
 from tempfile import TemporaryDirectory
 
 import structlog
 from django.conf import settings
 
 import orchestrator
 import substrapp.clients.organization as organization_client
-from api.models import Function as ApiFunction
-from builder import exceptions
 from image_transfer import push_payload
 from substrapp.compute_tasks import utils
 
@@ -21,24 +18,6 @@
 SUBTUPLE_TMP_DIR = settings.SUBTUPLE_TMP_DIR
 
 
-def wait_for_image_built(function_key: str, channel: str) -> None:
-    api_function = ApiFunction.objects.get(key=function_key)
-
-    attempt = 0
-    # with 60 attempts we wait max 2 min with a pending pod
-    max_attempts = IMAGE_BUILD_TIMEOUT / IMAGE_BUILD_CHECK_DELAY
-    while attempt < max_attempts:
-        if api_function.status == ApiFunction.Status.FUNCTION_STATUS_READY:
-            return
-        attempt += 1
-        time.sleep(IMAGE_BUILD_CHECK_DELAY)
-        api_function.refresh_from_db()
-
-    raise exceptions.PodTimeoutError(
-        f"Build for function {function_key} didn't complete after {IMAGE_BUILD_TIMEOUT} seconds"
-    )
-
-
 def load_remote_function_image(function: orchestrator.Function, channel: str) -> None:
     # Ask the backend owner of the function if it's available
     container_image_tag = utils.container_image_tag_from_function(function)

diff --git a/backend/substrapp/events/reactor.py b/backend/substrapp/events/reactor.py
@@ -16,8 +16,8 @@
 from substrapp.events import health
 from substrapp.models import WorkerLastEvent
 from substrapp.orchestrator import get_orchestrator_client
-from substrapp.task_routing import WORKER_QUEUE
 from substrapp.task_routing import get_builder_queue
+from substrapp.task_routing import get_generic_worker_queue
 from substrapp.tasks.tasks_compute_plan import queue_delete_cp_pod_and_dirs_and_optionally_images
 from substrapp.tasks.tasks_compute_task import queue_compute_task
 from substrapp.tasks.tasks_save_image import save_image_task
@@ -91,10 +91,12 @@ def on_function_event(payload):
         if orc_function.owner == _MY_ORGANIZATION:
             function_key = orc_function.key
             builder_queue = get_builder_queue()
+            worker_queue = get_generic_worker_queue()
             logger.info(
-                "Assigned function to builder queue",
+                "Assigned function to queues",
                 asset_key=function_key,
-                queue=builder_queue,
+                builder_queue=builder_queue,
+                worker_queue=worker_queue,
             )
 
             building_params = {
@@ -103,7 +105,7 @@ def on_function_event(payload):
             }
             (
                 build_image.si(**building_params).set(queue=builder_queue)
-                | save_image_task.si(**building_params).set(queue=WORKER_QUEUE)
+                | save_image_task.si(**building_params).set(queue=worker_queue)
             ).apply_async()
 
         else:

diff --git a/backend/substrapp/tasks/tasks_compute_task.py b/backend/substrapp/tasks/tasks_compute_task.py
@@ -186,8 +186,6 @@ def _run(
     dirs = None
 
     try:
-        image_builder.wait_for_image_built(task.function_key, channel_name)
-
         # Create context
         ctx = Context.from_task(channel_name, task)
         dirs = ctx.directories

diff --git a/backend/substrapp/tests/tasks/test_compute_task.py b/backend/substrapp/tests/tasks/test_compute_task.py
@@ -30,7 +30,6 @@ def test_compute_task_exception(mocker: MockerFixture):
     mock_add_asset_to_buffer = mocker.patch("substrapp.tasks.tasks_compute_task.add_task_assets_to_buffer")
     mock_add_asset_to_task_dir = mocker.patch("substrapp.tasks.tasks_compute_task.add_assets_to_taskdir")
     mock_load_remote_function_image = mocker.patch("substrapp.compute_tasks.image_builder.load_remote_function_image")
-    mock_wait_for_image_built = mocker.patch("substrapp.compute_tasks.image_builder.wait_for_image_built")
     mock_execute_compute_task = mocker.patch("substrapp.tasks.tasks_compute_task.execute_compute_task")
     saver = mocker.MagicMock()
     mock_output_saver = mocker.patch("substrapp.tasks.tasks_compute_task.OutputSaver", return_value=saver)
@@ -66,7 +65,6 @@ class FakeDirectories:
     mock_add_asset_to_buffer.assert_called_once()
     mock_add_asset_to_task_dir.assert_called_once()
     mock_load_remote_function_image.assert_called_once()
-    mock_wait_for_image_built.assert_called_once()
     mock_execute_compute_task.assert_called_once()
     saver.save_outputs.assert_called_once()
     mock_output_saver.assert_called_once()
@@ -136,7 +134,6 @@ def test_celery_retry(mocker: MockerFixture):
     mocker.patch("substrapp.tasks.tasks_compute_task.add_assets_to_taskdir")
     mocker.patch("substrapp.tasks.tasks_compute_task.restore_dir")
     mocker.patch("substrapp.compute_tasks.image_builder.load_remote_function_image")
-    mocker.patch("substrapp.compute_tasks.image_builder.wait_for_image_built")
     mock_execute_compute_task = mocker.patch("substrapp.tasks.tasks_compute_task.execute_compute_task")
     mocker.patch("substrapp.tasks.tasks_compute_task.teardown_task_dirs")
     mock_retry = mocker.patch("substrapp.tasks.tasks_compute_task.ComputeTask.retry")