Install extension from docker entrypoint.

Create new extension instance for each task so the instance will be thread safe.
For interactive tasks, call handler with task data instead of retrieving the object through API.

Author: Francesco Faraone

Dockerfile

@@ -18,6 +18,9 @@ RUN poetry build
 
 RUN pip install dist/*.whl
 
-WORKDIR /
-
 RUN rm -rf /install_temp
+
+COPY ./entrypoint.sh /entrypoint.sh
+RUN chmod 755 /entrypoint.sh
+
+ENTRYPOINT [ "/entrypoint.sh" ]

connect/eaas/helpers.py

@@ -13,19 +13,6 @@
 from connect.eaas.exceptions import EaaSError
 
 
-def install_extension(root_dir):
-    result = subprocess.run(
-        ['poetry', 'install'],
-        cwd=root_dir,
-        stdin=subprocess.DEVNULL,
-        start_new_session=True,
-    )
-    try:
-        result.check_returncode()
-    except subprocess.CalledProcessError:
-        raise EaaSError(result.stderr.decode())
-
-
 def get_container_id():
     result = subprocess.run(
         ['cat', '/proc/1/cpuset'],

connect/eaas/main.py

@@ -10,7 +10,6 @@
 import pathlib
 import signal
 
-from connect.eaas.helpers import install_extension
 from connect.eaas.worker import Worker
 
 
@@ -42,18 +41,20 @@ def configure_logger(debug):
                     'handlers': ['console'],
                     'level': 'DEBUG' if debug else 'INFO',
                 },
+                'eaas.extension': {
+                    'handlers': ['console'],
+                    'level': 'DEBUG' if debug else 'INFO',
+                },
             },
         },
     )
 
 
 def start(data):
     logger.info('Starting Connect EaaS runtime....')
-    logger.info('Installing the extension package...')
-    install_extension(data.extension_dir)
-    logger.info('The extension has been installed, starting the control worker...')
     if data.unsecure:
         logger.warning('Websocket connections will be established using unsecure protocol (ws).')
+
     worker = Worker(secure=not data.unsecure)
     loop = asyncio.get_event_loop()
     loop.add_signal_handler(

connect/eaas/manager.py

@@ -41,9 +41,8 @@ class TasksManager:
     Once completed, it will push the result to a queue from which, the
     result_sender task will pick it and send the result to the backend.
     """
-    def __init__(self, worker, extension):
+    def __init__(self, worker):
         self.worker = worker
-        self.extension = extension
         self.background_executor = ThreadPoolExecutor()
         self.interactive_executor = ThreadPoolExecutor()
         self.run_event = asyncio.Event()
@@ -87,17 +86,22 @@ async def submit_task(self, data):
         object_id = data.object_id
         task_type = data.task_type
         method_name = TASK_TYPE_EXT_METHOD_MAP[task_type]
-        method = getattr(self.extension, method_name)
+        extension = self.worker.get_extension()
+        method = getattr(extension, method_name)
         logger.debug(f'invoke {method_name}')
         self.running_tasks += 1
-        try:
-            request = await self.get_request(object_id, task_type)
-        except ClientError as e:
-            logger.warning(f'Cannot retrieve object {data.object_id} for task {data.task_id}')
-            future = Future()
-            future.set_exception(e)
-            asyncio.create_task(self.enqueue_result(data, future))
-            return
+        request = None
+        if data.task_category == TaskCategory.BACKGROUND:
+            try:
+                request = await self.get_request(object_id, task_type)
+            except ClientError as e:
+                logger.warning(f'Cannot retrieve object {data.object_id} for task {data.task_id}')
+                future = Future()
+                future.set_exception(e)
+                asyncio.create_task(self.enqueue_result(data, future))
+                return
+        else:
+            request = data.data
 
         if inspect.iscoroutinefunction(method):
             future = asyncio.create_task(method(request))

connect/eaas/worker.py

@@ -118,12 +118,7 @@ async def stop_tasks_manager(self):
 
     def start_tasks_manager(self):
         logger.info('Starting tasks worker...')
-        extension = self.extension_class(
-            self.get_client(),
-            self.get_extension_logger(self.logging_api_key),
-            self.extension_config,
-        )
-        self.tasks_manager = TasksManager(self, extension)
+        self.tasks_manager = TasksManager(self)
         self.tasks_manager.start()
         logger.info('Task worker started')
 
@@ -134,6 +129,13 @@ def ensure_tasks_manager_running(self):
     def get_url(self):
         return f'{self.base_ws_url}/{self.environment_id}/{self.instance_id}'
 
+    def get_extension(self):
+        return self.extension_class(
+            self.get_client(),
+            self.get_extension_logger(self.logging_api_key),
+            self.extension_config,
+        )
+
     async def run(self):  # noqa: CCR001
         """
         Main loop for the websocket connection.
@@ -195,12 +197,8 @@ async def configuration(self, data):
         It will stop the tasks manager so the extension can be
         reconfigured, then restart the tasks manager.
         """
-        self.paused = True
-        await self.stop_tasks_manager()
         self.extension_config = data.configuration
         self.logging_api_key = data.logging_api_key
-        self.paused = False
-        self.start_tasks_manager()
 
     async def pause(self):
         """

entrypoint.sh

@@ -0,0 +1,24 @@
+#!/bin/bash
+set -e
+
+EXTENSION_DIR=${EXTENSION_DIR:-'/extension'}
+
+DIST_DIR=$EXTENSION_DIR/dist
+
+echo "Installing the extension package from $EXTENSION_DIR..."
+
+pushd .
+
+cd $EXTENSION_DIR
+
+if [ -d "$DIST_DIR" ]; then rm -Rf $DIST_DIR; fi
+
+poetry build
+
+pip install -U pip && pip install $DIST_DIR/*.whl
+
+popd
+
+echo "Extension installed."
+
+exec "$@"

tests/test_helpers.py

@@ -15,33 +15,9 @@
     get_environment,
     get_extension_class,
     get_extension_type,
-    install_extension,
 )
 
 
-def test_install_extension(mocker):
-    result = mocker.MagicMock()
-    result.returnvalue = 0
-    mocked = mocker.patch('connect.eaas.helpers.subprocess.run', return_value=result)
-    install_extension('/root_dir')
-
-    assert mocked.mock_calls[0].args[0] == ['poetry', 'install']
-    assert mocked.mock_calls[0].kwargs['cwd'] == '/root_dir'
-
-
-def test_install_extension_ko(mocker):
-    result = mocker.MagicMock()
-    result.check_returncode = mocker.MagicMock(
-        side_effect=subprocess.CalledProcessError(128, cmd=[]),
-    )
-    result.stderr = 'error message'.encode('utf-8')
-    mocker.patch('connect.eaas.helpers.subprocess.run', return_value=result)
-    with pytest.raises(EaaSError) as cv:
-        install_extension('/root_dir')
-
-    assert str(cv.value) == 'error message'
-
-
 def test_get_container_id(mocker):
     result = mocker.MagicMock()
     result.returnvalue = 0

tests/test_main.py

@@ -31,7 +31,7 @@ def process_asset_purchase_request(self, request):
         def validate_asset_purchase_request(self, request):
             pass
     run_mock = mocker.AsyncMock()
-    mocker.patch('connect.eaas.main.install_extension')
+
     mocker.patch('connect.eaas.worker.get_extension_class', return_value=MyExtension)
     mocker.patch.object(Worker, 'run', run_mock)
     parsed_args = namedtuple('_Args', ('unsecure', 'extension_dir'))

tests/test_manager.py

@@ -22,7 +22,7 @@
 
 @pytest.mark.asyncio
 async def test_start_stop_is_running(mocker, caplog):
-    manager = TasksManager(mocker.MagicMock(), mocker.MagicMock())
+    manager = TasksManager(mocker.MagicMock())
     manager.start()
     assert manager.run_event.is_set() is True
     assert manager.is_running() is True
@@ -43,9 +43,10 @@ async def test_background_task_sync(mocker, extension_cls, task_type):
     extension = extension_class(None, None, None)
 
     worker = mocker.MagicMock()
+    worker.get_extension.return_value = extension
     worker.send = mocker.AsyncMock()
 
-    manager = TasksManager(worker, extension)
+    manager = TasksManager(worker)
     manager.get_request = mocker.AsyncMock(return_value={'id': 'PR-000'})
 
     manager.start()
@@ -77,9 +78,10 @@ async def test_background_task_async(mocker, extension_cls, task_type):
     extension = extension_class(None, None, None)
 
     worker = mocker.MagicMock()
+    worker.get_extension.return_value = extension
     worker.send = mocker.AsyncMock()
 
-    manager = TasksManager(worker, extension)
+    manager = TasksManager(worker)
     manager.get_request = mocker.AsyncMock(return_value={'id': 'PR-000'})
 
     manager.start()
@@ -111,9 +113,10 @@ async def test_interactive_task_sync(mocker, extension_cls, task_type):
     extension = extension_class(None, None, None)
 
     worker = mocker.MagicMock()
+    worker.get_extension.return_value = extension
     worker.send = mocker.AsyncMock()
 
-    manager = TasksManager(worker, extension)
+    manager = TasksManager(worker)
     manager.get_request = mocker.AsyncMock(return_value={'id': 'PR-000'})
 
     manager.start()
@@ -145,9 +148,10 @@ async def test_interactive_task_async(mocker, extension_cls, task_type):
     extension = extension_class(None, None, None)
 
     worker = mocker.MagicMock()
+    worker.get_extension.return_value = extension
     worker.send = mocker.AsyncMock()
 
-    manager = TasksManager(worker, extension)
+    manager = TasksManager(worker)
     manager.get_request = mocker.AsyncMock(return_value={'id': 'PR-000'})
 
     manager.start()
@@ -175,9 +179,10 @@ async def test_background_task_request_error(mocker, extension_cls):
     extension = extension_class(None, None, None)
 
     worker = mocker.MagicMock()
+    worker.get_extension.return_value = extension
     worker.send = mocker.AsyncMock()
 
-    manager = TasksManager(worker, extension)
+    manager = TasksManager(worker)
     manager.get_request = mocker.AsyncMock(side_effect=ClientError('Request not found', 404))
 
     manager.start()
@@ -199,45 +204,16 @@ async def test_background_task_request_error(mocker, extension_cls):
     worker.send.assert_awaited_once_with(json_msg)
 
 
-@pytest.mark.asyncio
-async def test_interactive_task_request_error(mocker, extension_cls):
-    extension_class = extension_cls('validate_asset_purchase_request')
-    extension = extension_class(None, None, None)
-
-    worker = mocker.MagicMock()
-    worker.send = mocker.AsyncMock()
-
-    manager = TasksManager(worker, extension)
-    manager.get_request = mocker.AsyncMock(side_effect=ClientError('Request not found', 404))
-
-    manager.start()
-
-    task = TaskPayload(
-        'TQ-000',
-        TaskCategory.INTERACTIVE,
-        TaskType.ASSET_PURCHASE_REQUEST_VALIDATION,
-        'PR-000',
-    )
-
-    await manager.submit_task(task)
-    await asyncio.sleep(.1)
-    await manager.stop()
-    message = Message(message_type=MessageType.TASK, data=task)
-    json_msg = message.to_json()
-    json_msg['data']['result'] = 'failed'
-    json_msg['data']['failure_output'] = 'Request not found'
-    worker.send.assert_awaited_once_with(json_msg)
-
-
 @pytest.mark.asyncio
 async def test_result_sender_retries(mocker, extension_cls):
     extension_class = extension_cls('process_asset_purchase_request')
     extension = extension_class(None, None, None)
 
     worker = mocker.MagicMock()
+    worker.get_extension.return_value = extension
     worker.send = mocker.AsyncMock(side_effect=[Exception('retry'), None])
 
-    manager = TasksManager(worker, extension)
+    manager = TasksManager(worker)
     manager.get_request = mocker.AsyncMock(return_value={'id': 'PR-000'})
 
     manager.start()
@@ -264,11 +240,12 @@ async def test_result_sender_max_retries_exceeded(mocker, extension_cls, caplog)
     extension = extension_class(None, None, None)
 
     worker = mocker.MagicMock()
+    worker.get_extension.return_value = extension
     worker.send = mocker.AsyncMock(
         side_effect=[Exception('retry') for _ in range(RESULT_SENDER_MAX_RETRIES)],
     )
 
-    manager = TasksManager(worker, extension)
+    manager = TasksManager(worker)
     manager.get_request = mocker.AsyncMock(return_value={'id': 'PR-000'})
 
     manager.start()