Merge pull request #144 from cloudblue/lite-26666_add_transformation_execution

LITE-26666 Add execution of transformations to extension runner

Author: ffaraone

connect/eaas/runner/config.py

@@ -130,6 +130,9 @@ def get_tfnapp_ws_url(self):
     def get_api_url(self):
         return f'https://{self.env["api_address"]}/public/v1'
 
+    def get_api_address(self):
+        return f'https://{self.env["api_address"]}'
+
     def get_user_agent(self):
         version = get_version()
         pimpl = platform.python_implementation()

connect/eaas/runner/constants.py

@@ -98,8 +98,13 @@
 BACKGROUND_TASK_MAX_EXECUTION_TIME = 300
 INTERACTIVE_TASK_MAX_EXECUTION_TIME = 120
 SCHEDULED_TASK_MAX_EXECUTION_TIME = 60 * 60 * 12
+TRANSFORMATION_TASK_MAX_EXECUTION_TIME = 300
 RESULT_SENDER_MAX_RETRIES = 5
 RESULT_SENDER_WAIT_GRACE_SECONDS = 90
+TRANSFORMATION_TASK_MAX_PARALLEL_LINES = 20
+DOWNLOAD_CHUNK_SIZE = 1024
+UPLOAD_CHUNK_SIZE = 65535
+TRANSFORMATION_WRITE_QUEUE_TIMEOUT = 600
 
 MAX_RETRY_TIME_GENERIC_SECONDS = 15 * 60
 MAX_RETRY_TIME_MAINTENANCE_SECONDS = 3 * 60 * 60

connect/eaas/runner/handlers/base.py

@@ -1,14 +1,21 @@
 import importlib
 import inspect
 import logging
+import os
 import sys
 from abc import (
     ABC,
     abstractmethod,
 )
 
+from connect.client import (
+    AsyncConnectClient,
+    ConnectClient,
+)
+
 from connect.eaas.core.logging import (
     ExtensionLogHandler,
+    RequestLogger,
 )
 from connect.eaas.runner.helpers import (
     iter_entry_points,
@@ -115,3 +122,37 @@ def get_application(self):
     @abstractmethod
     def get_features(self):
         raise NotImplementedError()
+
+    def _create_client(self, event_type, task_id, method_name, api_key, connect_correlation_id):
+        """
+        Get an instance of Connect Openapi Client. Returns an instance of the AsyncConnectClient
+        or the ConnectClient depending on method type.
+        """
+        method = getattr(self.get_application(), method_name)
+
+        Client = ConnectClient if not inspect.iscoroutinefunction(method) else AsyncConnectClient
+
+        default_headers = {
+            'EAAS_EXT_ID': self._config.service_id,
+            'EAAS_TASK_ID': task_id,
+            'EAAS_TASK_TYPE': event_type,
+        }
+
+        default_headers.update(self._config.get_user_agent())
+
+        if connect_correlation_id:
+            operation_id = connect_correlation_id[3:34]
+            span_id = os.urandom(8).hex()
+            correlation_id = f'00-{operation_id}-{span_id}-01'
+            default_headers['ext-traceparent'] = correlation_id
+
+        return Client(
+            api_key,
+            endpoint=self._config.get_api_url(),
+            use_specs=False,
+            max_retries=3,
+            logger=RequestLogger(
+                self.get_logger(extra={'task_id': task_id}),
+            ),
+            default_headers=default_headers,
+        )

connect/eaas/runner/handlers/events.py

@@ -1,15 +1,6 @@
 import inspect
 import logging
-import os
 
-from connect.client import (
-    AsyncConnectClient,
-    ConnectClient,
-)
-
-from connect.eaas.core.logging import (
-    RequestLogger,
-)
 from connect.eaas.core.models import (
     Context,
 )
@@ -95,40 +86,6 @@ def get_method(
 
         return getattr(ext, method_name, None)
 
-    def _create_client(self, event_type, task_id, method_name, api_key, connect_correlation_id):
-        """
-        Get an instance of the Connect Openapi Client. If the extension is asyncrhonous
-        it returns an instance of the AsyncConnectClient otherwise the ConnectClient.
-        """
-        method = getattr(self.get_application(), method_name)
-
-        Client = ConnectClient if not inspect.iscoroutinefunction(method) else AsyncConnectClient
-
-        default_headers = {
-            'EAAS_EXT_ID': self._config.service_id,
-            'EAAS_TASK_ID': task_id,
-            'EAAS_TASK_TYPE': event_type,
-        }
-
-        default_headers.update(self._config.get_user_agent())
-
-        if connect_correlation_id:
-            operation_id = connect_correlation_id[3:34]
-            span_id = os.urandom(8).hex()
-            correlation_id = f'00-{operation_id}-{span_id}-01'
-            default_headers['ext-traceparent'] = correlation_id
-
-        return Client(
-            api_key,
-            endpoint=self._config.get_api_url(),
-            use_specs=False,
-            max_retries=3,
-            logger=RequestLogger(
-                self.get_logger(extra={'task_id': task_id}),
-            ),
-            default_headers=default_headers,
-        )
-
     def get_events(self):
         if 'capabilities' in self.get_descriptor():
             logger.warning(

connect/eaas/runner/handlers/transformations.py

@@ -3,8 +3,12 @@
 #
 # Copyright (c) 2022 Ingram Micro. All Rights Reserved.
 #
+import inspect
 import logging
 
+from connect.eaas.core.models import (
+    Context,
+)
 from connect.eaas.runner.config import (
     ConfigHelper,
 )
@@ -20,8 +24,11 @@ class TfnApp(ApplicationHandlerBase):
     """
     Handle the lifecycle of a Transformation extension.
     """
+    LOGGER_NAME = 'eaas.tfnapp'
+
     def __init__(self, config: ConfigHelper):
         super().__init__(config)
+        self._config = config
         self._transformations = None
 
     def get_application(self):
@@ -35,3 +42,50 @@ def get_features(self):
         return {
             'transformations': self.transformations,
         }
+
+    def get_method(
+            self,
+            event_type,
+            task_id,
+            method_name,
+            transformation_request=None,
+            installation=None,
+            api_key=None,
+            connect_correlation_id=None,
+    ):
+        if not method_name:  # pragma: no cover
+            return
+
+        args = (
+            self._create_client(
+                event_type, task_id, method_name, self._config.api_key, connect_correlation_id,
+            ),
+            self.get_logger(extra={'task_id': task_id}),
+            self._config.variables,
+        )
+
+        kwargs = {}
+        if installation:
+            kwargs['installation'] = installation
+            kwargs['installation_client'] = self._create_client(
+                event_type,
+                task_id,
+                method_name,
+                api_key,
+                connect_correlation_id,
+            )
+        if transformation_request:
+            kwargs['transformation_request'] = transformation_request
+
+        app_class = self.get_application()
+
+        if 'context' in inspect.signature(app_class.__init__).parameters:
+            kwargs['context'] = Context(
+                extension_id=self.config.service_id,
+                environment_id=self.config.environment_id,
+                environment_type=self.config.environment_type,
+            )
+
+        ext = app_class(*args, **kwargs)
+
+        return getattr(ext, method_name, None)

connect/eaas/runner/helpers.py

@@ -52,6 +52,8 @@
     ORDINAL_SUFFIX,
     PYPI_EXTENSION_RUNNER_URL,
     SCHEDULED_TASK_MAX_EXECUTION_TIME,
+    TRANSFORMATION_TASK_MAX_EXECUTION_TIME,
+    TRANSFORMATION_WRITE_QUEUE_TIMEOUT,
 )
 
 
@@ -110,6 +112,12 @@ def get_environment():
         'scheduled_task_max_execution_time': int(os.getenv(
             'SCHEDULED_TASK_MAX_EXECUTION_TIME', SCHEDULED_TASK_MAX_EXECUTION_TIME,
         )),
+        'transformation_task_max_execution_time': int(os.getenv(
+            'TRANSFORMATION_TASK_MAX_EXECUTION_TIME', TRANSFORMATION_TASK_MAX_EXECUTION_TIME,
+        )),
+        'transformation_write_queue_timeout': int(os.getenv(
+            'TRANSFORMATION_WRITE_QUEUE_TIMEOUT', TRANSFORMATION_WRITE_QUEUE_TIMEOUT,
+        )),
     }
 
 

connect/eaas/runner/managers/__init__.py

@@ -1,3 +1,4 @@
 from connect.eaas.runner.managers.background import BackgroundTasksManager  # noqa
 from connect.eaas.runner.managers.interactive import InteractiveTasksManager  # noqa
 from connect.eaas.runner.managers.scheduled import ScheduledTasksManager  # noqa
+from connect.eaas.runner.managers.transformation import TransformationTasksManager  # noqa

connect/eaas/runner/managers/background.py

@@ -7,15 +7,11 @@
 import logging
 import time
 import traceback
-from string import (
-    Template,
-)
 
 from connect.client import (
     AsyncConnectClient,
 )
 from connect.client.models import (
-    AsyncCollection,
     AsyncResource,
 )
 
@@ -27,7 +23,7 @@
     TaskOutput,
 )
 from connect.eaas.core.responses import (
-    ProcessingResponse,
+    BackgroundResponse,
 )
 from connect.eaas.runner.managers.base import (
     TasksManagerBase,
@@ -42,6 +38,11 @@ class BackgroundTasksManager(TasksManagerBase):
     def get_method_name(self, task_data, argument):
         return self.handler.events[task_data.input.event_type]['method']
 
+    def send_skip_response(self, data, output):
+        future = asyncio.Future()
+        future.set_result(BackgroundResponse.skip(output))
+        asyncio.create_task(self.enqueue_result(data, future))
+
     async def get_argument(self, task_data):
         """
         Get the request object through the Connect public API
@@ -56,28 +57,14 @@ async def get_argument(self, task_data):
                 default_headers=self.config.get_user_agent(),
             )
 
-        definition = self.config.event_definitions[task_data.input.event_type]
-        supported_statuses = self.handler.events[task_data.input.event_type]['statuses']
-        rql_filter = Template(definition.api_collection_filter).substitute(
-            {
-                '_statuses_': f'({",".join(supported_statuses)})',
-                '_object_id_': task_data.input.object_id,
-            },
+        object_exists = await self.filter_collection_by_event_definition(
+            client,
+            task_data,
         )
-
-        collection = AsyncCollection(client, definition.api_collection_endpoint)
-        if await collection.filter(rql_filter).count() == 0:
-            logger.info(
-                f'Send skip response for {task_data.options.task_id} since '
-                'the current request status is not supported.',
-            )
-            self.send_skip_response(
-                task_data,
-                'The request status does not match the '
-                f'supported statuses: {",".join(supported_statuses)}.',
-            )
+        if not object_exists:
             return
 
+        definition = self.config.event_definitions[task_data.input.event_type]
         url = definition.api_resource_endpoint.format(pk=task_data.input.object_id)
         resource = AsyncResource(client, url)
 
@@ -112,8 +99,3 @@ async def build_response(self, task_data, future):
             result_message.output.message = traceback.format_exc()[:4000]
 
         return result_message
-
-    def send_skip_response(self, data, output):
-        future = asyncio.Future()
-        future.set_result(ProcessingResponse.skip(output))
-        asyncio.create_task(self.enqueue_result(data, future))