diff --git a/CHANGELOG.md b/CHANGELOG.md index ee0a5442d..d5459f9fc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,7 @@ - [Azure Funcions] Updated default funcions plan to Flex Consumption - [AWS EC2] Updated default Ubuntu Image to Ubuntu 24 - [Azure VMS] Updated default Ubuntu Image to Ubuntu 24 +- [Aliyun FC] Updated backend to Function Compute 3.0 (FC3 API) and added custom-container deploy mode support ### Fixed - [K8s] Fixed default runtime builds impacted by Debian Buster end-of-life. diff --git a/config/config_template.yaml b/config/config_template.yaml index 4565cc3de..088faf801 100644 --- a/config/config_template.yaml +++ b/config/config_template.yaml @@ -354,7 +354,10 @@ #aliyun_fc: #role_arn: # Mandatory #region: # Falls back to aliyun.region - #service: + #deploy_mode: runtime # runtime | custom-container + #docker_server: docker.io # Default registry for custom-container + #docker_user: # Required for custom-container + #docker_password: #runtime: #runtime_memory: 256 # MB #runtime_timeout: 300 # Seconds diff --git a/docs/source/compute_config/aliyun_functions.md b/docs/source/compute_config/aliyun_functions.md index a86ebbb05..0d6a61153 100644 --- a/docs/source/compute_config/aliyun_functions.md +++ b/docs/source/compute_config/aliyun_functions.md @@ -36,7 +36,7 @@ aliyun_fc: role_arn: ``` -5. **(optional)** By default Lithops will automatically create a new **service** in your *Function Compute* account. For this purpose your user must have **List** and **Create** permissions for *Function Compute*. Alternatively, you can create a new service through the dashboard (or use one already created), assign the Role created in the previous step (under *Service Configuration* --> *Modify Configuration* --> *Role Config*), and configure the *service* entry in the *aliyun_fc* config section. +5. Your RAM user needs permissions to create, list, invoke, and delete **FC 3.0 functions** (for example `AliyunFCFullAccess`). The `role_arn` is attached to each Lithops worker function so it can access OSS and other services. ## Summary of configuration keys for Aliyun @@ -54,14 +54,17 @@ aliyun_fc: |Group|Key|Default|Mandatory|Additional info| |---|---|---|---|---| |aliyun_fc | role_arn | |yes | Role ARN. For example: `acs:ram::5244532493961771:role/aliyunfclogexecutionrole` | -|aliyun_fc | region | |no | Region name. For example: `eu-west-1`. Lithops will use the region set under the `aliyun` section if it is not set here | -|aliyun_fc | service | |no | Service name | +|aliyun_fc | region | |no | Region name. For example: `us-east-1`. Lithops will use the region set under the `aliyun` section if it is not set here | |aliyun_fc | max_workers | 300 | no | Max number of workers. Alibaba limits the number of parallel workers to 300| |aliyun_fc | worker_processes | 1 | no | Number of Lithops processes within a given worker. This can be used to parallelize function activations within a worker | |aliyun_fc | runtime | |no | Runtime name you built and deployed using the lithops client| |aliyun_fc | runtime_memory | 256 |no | Memory limit in MB. Default 256MB | |aliyun_fc | runtime_timeout | 300 |no | Runtime timeout in seconds. Default 5 minutes | |aliyun_fc | invoke_pool_threads | 300 |no | Number of concurrent threads used for invocation | +|aliyun_fc | deploy_mode | runtime | no | `runtime` (zip + managed Python) or `custom-container` (Docker image) | +|aliyun_fc | docker_server | docker.io | no | Container registry host (Docker Hub by default) | +|aliyun_fc | docker_user | | no | Docker Hub user/namespace; **required** for `custom-container` | +|aliyun_fc | docker_password | | no | Registry password or token used when pushing images | ## Test Lithops diff --git a/lithops/serverless/backends/aliyun_fc/aliyun_fc.py b/lithops/serverless/backends/aliyun_fc/aliyun_fc.py index 5592b4a8e..8cf25b671 100644 --- a/lithops/serverless/backends/aliyun_fc/aliyun_fc.py +++ b/lithops/serverless/backends/aliyun_fc/aliyun_fc.py @@ -14,14 +14,24 @@ # limitations under the License. # +import base64 import hashlib -import os -import sys +import io +import json import logging +import os +import re import shutil -import json +import subprocess as sp +import sys +import time + import lithops -import fc2 +from alibabacloud_fc20230330 import models as fc_models +from alibabacloud_fc20230330.client import Client as FC3Client +from alibabacloud_tea_openapi import models as open_api_models +from alibabacloud_tea_openapi.exceptions import ClientException as FCClientException +from darabonba.runtime import RuntimeOptions from lithops import utils from lithops.version import __version__ @@ -31,37 +41,82 @@ logger = logging.getLogger(__name__) +# Managed-Python (zip) deploy: GetFunction often omits state/lastUpdateStatus. +ZIP_METADATA_MAX_ATTEMPTS = 24 +ZIP_METADATA_RETRY_SLEEP = 5 +CONTAINER_MAX_WAIT = 900 +CONTAINER_METADATA_MAX_ATTEMPTS = 60 +CONTAINER_METADATA_RETRY_SLEEP = 10 +FC_INVOKE_RUNTIME = RuntimeOptions(read_timeout=120000, connect_timeout=30000) + + +def _init_fc3_client(access_key_id, access_key_secret, region, endpoint): + cfg = open_api_models.Config( + access_key_id=access_key_id, + access_key_secret=access_key_secret, + region_id=region, + endpoint=endpoint, + ) + return FC3Client(cfg) + + +def _is_not_found_error(exc): + if isinstance(exc, FCClientException): + return exc.status_code == 404 or exc.code in ( + 'FunctionNotFound', 'ServiceNotFound', 'NotFound', + ) + return False + + +def _is_pending_error(exc): + if isinstance(exc, FCClientException): + return exc.status_code == 412 or exc.code == 'PreconditionFailed' + return False + + +def _read_invoke_body(response): + body = response.body + if body is None: + return b'' + if hasattr(body, 'read'): + return body.read() + if isinstance(body, bytes): + return body + return str(body).encode('utf-8') + + +def _get_response_header(response, name): + headers = response.headers or {} + for key, value in headers.items(): + if key.lower() == name.lower(): + return value + return None + class AliyunFunctionComputeBackend: """ - A wrap-up around Aliyun Function Compute backend. + Lithops backend for Alibaba Cloud Function Compute 3.0 (API 2023-03-30). """ def __init__(self, afc_config, internal_storage): - logger.debug("Creating Aliyun Function Compute client") + logger.debug("Creating Aliyun Function Compute 3.0 client") self.name = 'aliyun_fc' self.type = utils.BackendType.FAAS.value self.config = afc_config - self.user_agent = afc_config['user_agent'] - - self.endpoint = afc_config['public_endpoint'] - self.access_key_id = afc_config['access_key_id'] - self.access_key_secret = afc_config['access_key_secret'] self.role_arn = afc_config['role_arn'] self.region = afc_config['region'] - self.default_service_name = f'{config.SERVICE_NAME}_{self.access_key_id[0:4].lower()}' - self.service_name = afc_config.get('service', self.default_service_name) - - logger.debug(f"Set Aliyun FC Service to {self.service_name}") - logger.debug(f"Set Aliyun FC Endpoint to {self.endpoint}") + self.fc_client = _init_fc3_client( + afc_config['access_key_id'], + afc_config['access_key_secret'], + self.region, + afc_config['public_endpoint'], + ) - self.fc_client = fc2.Client(endpoint=self.endpoint, - accessKeyID=self.access_key_id, - accessKeySecret=self.access_key_secret) + self.deploy_mode = self.config.get('deploy_mode', 'runtime') - msg = COMPUTE_CLI_MSG.format('Aliyun Function Compute') - logger.info(f"{msg} - Region: {self.region}") + msg = COMPUTE_CLI_MSG.format('Aliyun Function Compute 3.0') + logger.info(f"{msg} - Region: {self.region} - Deploy mode: {self.deploy_mode}") def _format_function_name(self, runtime_name, runtime_memory, version=__version__): name = f'{runtime_name}-{runtime_memory}-{version}' @@ -70,7 +125,7 @@ def _format_function_name(self, runtime_name, runtime_memory, version=__version_ return f'lithops-worker-{runtime_name}-v{version.replace(".", "-")}-{name_hash}' def _unformat_function_name(self, function_name): - runtime_name, hash = function_name.rsplit('-', 1) + runtime_name, _hash = function_name.rsplit('-', 1) runtime_name = runtime_name.replace('lithops-worker-', '') runtime_name, version = runtime_name.rsplit('-v', 1) version = version.replace('-', '.') @@ -78,9 +133,211 @@ def _unformat_function_name(self, function_name): def _get_default_runtime_name(self): py_version = utils.CURRENT_PY_VERSION.replace('.', '') - return f'default-runtime-v{py_version}' + if self._use_custom_container(): + return f'default-container-runtime-v{py_version}' + return f'default-python-runtime-v{py_version}' + + def _use_custom_container(self): + return self.deploy_mode == 'custom-container' + + def _format_image_name(self, runtime_name, version=__version__): + if '/' in runtime_name and ':' in runtime_name: + return runtime_name + + docker_server = self.config.get('docker_server', 'docker.io') + docker_user = self.config['docker_user'] + tag = f"v{version.replace('.', '-')}" + image = f'lithops-aliyunfc-{runtime_name}:{tag}' + + if docker_server == 'docker.io': + return f'{docker_user}/{image}' + return f'{docker_server}/{docker_user}/{image}' + + def _list_functions(self, prefix='lithops-worker'): + functions = [] + next_token = None + + while True: + req = fc_models.ListFunctionsRequest( + prefix=prefix, + fc_version='v3', + limit=100, + next_token=next_token, + ) + res = self.fc_client.list_functions(req) + output = res.body + if output and output.functions: + functions.extend(output.functions) + next_token = output.next_token if output else None + if not next_token: + break + + return functions + + def _get_function(self, function_name): + res = self.fc_client.get_function( + function_name, fc_models.GetFunctionRequest() + ) + return res.body + + def _function_exists(self, function_name): + try: + self._get_function(function_name) + return True + except FCClientException as e: + if _is_not_found_error(e): + return False + raise + + def _check_function_failed(self, fn): + state = (fn.state or '').strip().lower() + update_status = (fn.last_update_status or '').strip().lower() + + if state in ('failed', 'inactive'): + raise Exception( + f'Function {fn.function_name} is {fn.state}: ' + f'{fn.state_reason or ""} ({fn.state_reason_code or ""})' + ) + if update_status == 'failed': + raise Exception( + f'Function {fn.function_name} update failed: ' + f'{fn.last_update_status_reason or ""} ' + f'({fn.last_update_status_reason_code or ""})' + ) + + def _wait_for_function_active(self, function_name): + """ + Wait until a custom-container function state is Active (image pull). + """ + poll_interval = 5 + elapsed = 0 + + while elapsed < CONTAINER_MAX_WAIT: + fn = self._get_function(function_name) + self._check_function_failed(fn) + state = (fn.state or '').strip().lower() + + if state in ('active', 'ready'): + logger.debug('Function %s is ready (state=%s)', function_name, fn.state) + return + + logger.info( + 'Waiting for function %s to become active (state=%s, elapsed=%ss)...', + function_name, fn.state or 'pending', elapsed, + ) + time.sleep(poll_interval) + elapsed += poll_interval + + raise Exception( + f'Timed out after {CONTAINER_MAX_WAIT}s waiting for function {function_name} ' + 'to become active. Check the FC console (FC 3.0 functions list).' + ) + + def _delete_function_if_exists(self, function_name): + if self._function_exists(function_name): + logger.debug('Deleting function %s', function_name) + self.fc_client.delete_function(function_name) + + def _zip_code_location(self, zip_path): + with open(zip_path, 'rb') as zipf: + encoded = base64.b64encode(zipf.read()).decode('utf-8') + return fc_models.InputCodeLocation(zip_file=encoded) + + def _build_container_image(self, runtime_name, dockerfile, extra_args=None, + context_files=None): + if extra_args is None: + extra_args = [] + image_name = self._format_image_name(runtime_name) + docker_path = utils.get_docker_path() + if not docker_path: + raise Exception('Docker is required to build custom-container runtimes for Aliyun FC') + + assert os.path.isfile(dockerfile), f'Cannot locate "{dockerfile}"' + + build_dir = os.path.join(config.BUILD_DIR, runtime_name) + shutil.rmtree(build_dir, ignore_errors=True) + os.makedirs(build_dir) + + if context_files: + for src, dst in context_files.items(): + shutil.copy(src, os.path.join(build_dir, dst)) + + current_location = os.path.dirname(os.path.abspath(__file__)) + handler_file = os.path.join(current_location, 'container_entry_point.py') + utils.create_handler_zip( + config.FH_ZIP_LOCATION, handler_file, 'container_entry_point.py' + ) + shutil.copy(config.FH_ZIP_LOCATION, os.path.join(build_dir, 'lithops_aliyun_fc.zip')) + shutil.copy(dockerfile, os.path.join(build_dir, 'Dockerfile')) + + cmd = ( + f'{docker_path} build --platform=linux/amd64 ' + f'-t {image_name} -f {os.path.join(build_dir, "Dockerfile")} {build_dir}' + ) + cmd = f'{cmd} {" ".join(extra_args)}'.strip() + logger.info('Building container image %s', image_name) + utils.run_command(cmd) + + docker_password = self.config.get('docker_password') + docker_server = self.config.get('docker_server', 'docker.io') + if docker_password: + docker_user = self.config['docker_user'] + logger.debug('Logging in to container registry %s', docker_server) + utils.docker_login(docker_user, docker_password, docker_server) + + if utils.is_podman(docker_path): + push_cmd = f'{docker_path} push {image_name} --format docker --remove-signatures' + else: + push_cmd = f'{docker_path} push {image_name}' + logger.info('Pushing container image %s', image_name) + push_output = sp.check_output( + push_cmd, shell=True, stderr=sp.STDOUT, text=True + ) + digest_match = re.search(r'digest:\s*(sha256:[a-f0-9]+)', push_output) + if digest_match: + image_name = f'{image_name}@{digest_match.group(1)}' + logger.info('Using pinned container image %s', image_name) + else: + logger.warning( + 'Could not parse image digest from push output; ' + 'FC may reuse a cached image layer' + ) + + if os.path.exists(config.FH_ZIP_LOCATION): + os.remove(config.FH_ZIP_LOCATION) + + self._container_image = image_name + return image_name + + def build_runtime(self, runtime_name, requirements_file, extra_args=None): + if extra_args is None: + extra_args = [] + if self._use_custom_container(): + if not requirements_file: + raise Exception( + 'Please provide a Dockerfile path or a requirements.txt file ' + 'for custom-container runtime builds' + ) + basename = os.path.basename(requirements_file).lower() + if 'dockerfile' in basename: + return self._build_container_image(runtime_name, requirements_file, extra_args) + + dockerfile = os.path.join(config.BUILD_DIR, f'{runtime_name}.Dockerfile') + os.makedirs(config.BUILD_DIR, exist_ok=True) + with open(dockerfile, 'w') as df: + df.write(f'FROM python:{utils.CURRENT_PY_VERSION}-slim-bookworm\n') + df.write(config.DEFAULT_DOCKERFILE) + df.write('COPY requirements.txt .\n') + df.write('RUN pip install --no-cache-dir -r requirements.txt\n') + try: + return self._build_container_image( + runtime_name, dockerfile, extra_args, + context_files={requirements_file: 'requirements.txt'} + ) + finally: + if os.path.exists(dockerfile): + os.remove(dockerfile) - def build_runtime(self, runtime_name, requirements_file, extra_args=[]): if not requirements_file: raise Exception('Please provide a "requirements.txt" file with the necessary modules') @@ -91,7 +348,6 @@ def build_runtime(self, runtime_name, requirements_file, extra_args=[]): shutil.rmtree(build_dir, ignore_errors=True) os.makedirs(build_dir) - # Add lithops base modules logger.debug("Downloading base modules (via pip install)") req_file = os.path.join(build_dir, 'requirements.txt') with open(req_file, 'w') as reqf: @@ -101,26 +357,28 @@ def download_requirements(): cmd = f'{sys.executable} -m pip install -t {build_dir} -r {req_file} --no-deps' utils.run_command(cmd) - if utils.is_linux_system(): + docker_path = utils.get_docker_path() + if docker_path and not utils.is_linux_system(): + cmd = 'python3 -m pip install -U -t . -r requirements.txt' + cmd = ( + f'{docker_path} run --platform linux/amd64 -w /tmp ' + f'-v {build_dir}:/tmp ' + f'python:{utils.CURRENT_PY_VERSION}-slim-bookworm {cmd}' + ) + utils.run_command(cmd) + elif utils.is_linux_system(): download_requirements() else: - docker_path = utils.get_docker_path() - if docker_path: - # Build the runtime in a docker - cmd = 'python3 -m pip install -U -t . -r requirements.txt' - cmd = f'docker run -w /tmp -v {build_dir}:/tmp python:{utils.CURRENT_PY_VERSION}-slim-bookworm {cmd}' - utils.run_command(cmd) - else: - logger.warning('Aliyun Functions use a Linux environment. Building' - 'a runtime from a non-Linux environemnt might cause issues') - download_requirements() - - # Add function handlerd + logger.warning( + 'Docker is not available. Building a zip runtime on a non-Linux ' + 'host may produce incompatible binaries for Aliyun FC.' + ) + download_requirements() + current_location = os.path.dirname(os.path.abspath(__file__)) handler_file = os.path.join(current_location, 'entry_point.py') shutil.copy(handler_file, build_dir) - # Add lithops module module_location = os.path.dirname(os.path.abspath(lithops.__file__)) dst_location = os.path.join(build_dir, 'lithops') @@ -130,7 +388,6 @@ def download_requirements(): else: shutil.copytree(module_location, dst_location, ignore=shutil.ignore_patterns('__pycache__')) - # Create zip file os.chdir(build_dir) runtime_zip = f'{config.BUILD_DIR}/{runtime_name}.zip' if os.path.exists(runtime_zip): @@ -138,20 +395,19 @@ def download_requirements(): utils.run_command(f'zip -r {runtime_zip} .') shutil.rmtree(build_dir, ignore_errors=True) - def _service_exists(self, service_name): - """ - Checks if a given service exists - """ - services = self.fc_client.list_services(prefix=service_name).data['services'] - for serv in services: - if serv['serviceName'] == service_name: - return True - return False - def _build_default_runtime(self, runtime_name): - """ - Builds the default runtime - """ + if self._use_custom_container(): + dockerfile = os.path.join(TEMP_DIR, 'aliyun_default.Dockerfile') + with open(dockerfile, 'w') as df: + df.write(f'FROM python:{utils.CURRENT_PY_VERSION}-slim-bookworm\n') + df.write(config.DEFAULT_DOCKERFILE) + try: + self.build_runtime(runtime_name, dockerfile) + finally: + if os.path.exists(dockerfile): + os.remove(dockerfile) + return + requirements_file = os.path.join(TEMP_DIR, 'aliyun_default_requirements.txt') with open(requirements_file, 'w') as reqf: reqf.write(config.REQUIREMENTS_FILE) @@ -160,125 +416,168 @@ def _build_default_runtime(self, runtime_name): finally: os.remove(requirements_file) + def _container_image_uri(self, runtime_name): + return getattr(self, '_container_image', None) or self._format_image_name(runtime_name) + + def _create_function(self, function_name, memory, timeout, runtime_name): + if self._use_custom_container(): + image_name = self._container_image_uri(runtime_name) + function_input = fc_models.CreateFunctionInput( + function_name=function_name, + runtime='custom-container', + handler='container_entry_point.invoke', + custom_container_config=fc_models.CustomContainerConfig( + image=image_name, + acceleration_type='Default', + port=config.CA_PORT, + ), + memory_size=memory, + timeout=timeout, + role=self.role_arn, + internet_access=True, + ) + else: + zip_path = f'{config.BUILD_DIR}/{runtime_name}.zip' + function_input = fc_models.CreateFunctionInput( + function_name=function_name, + runtime=config.AVAILABLE_PY_RUNTIMES[utils.CURRENT_PY_VERSION], + handler='entry_point.main', + code=self._zip_code_location(zip_path), + memory_size=memory, + timeout=timeout, + role=self.role_arn, + internet_access=True, + ) + + res = self.fc_client.create_function( + fc_models.CreateFunctionRequest(body=function_input) + ) + if res.body and res.body.function_name: + logger.info( + 'Created function %s (state=%s)', + res.body.function_name, + res.body.state or 'pending', + ) + def deploy_runtime(self, runtime_name, memory, timeout): - """ - Deploys a new runtime into Aliyun Function Compute - with the custom modules for lithops - """ logger.info(f"Deploying runtime: {runtime_name} - Memory: {memory} Timeout: {timeout}") - if not self._service_exists(self.service_name): - logger.debug(f"creating service {self.service_name}") - self.fc_client.create_service(self.service_name, role=self.role_arn) - + self._container_image = None if runtime_name == self._get_default_runtime_name(): self._build_default_runtime(runtime_name) function_name = self._format_function_name(runtime_name, memory) - logger.debug(f'Crating function {function_name}') - functions = self.fc_client.list_functions(self.service_name).data['functions'] - for function in functions: - if function['functionName'] == function_name: - logger.debug(f'Function {function_name} already exists. Deleting it') - self.delete_runtime(runtime_name, memory) - - self.fc_client.create_function( - serviceName=self.service_name, - functionName=function_name, - runtime=config.AVAILABLE_PY_RUNTIMES[utils.CURRENT_PY_VERSION], - handler='entry_point.main', - codeZipFile=f'{config.BUILD_DIR}/{runtime_name}.zip', - memorySize=memory, - timeout=timeout - ) + self._delete_function_if_exists(function_name) - metadata = self._generate_runtime_meta(function_name) + logger.debug('Creating function %s', function_name) + self._create_function(function_name, memory, timeout, runtime_name) + if self._use_custom_container(): + self._wait_for_function_active(function_name) + time.sleep(10) + max_attempts = CONTAINER_METADATA_MAX_ATTEMPTS + retry_sleep = CONTAINER_METADATA_RETRY_SLEEP + else: + max_attempts = ZIP_METADATA_MAX_ATTEMPTS + retry_sleep = ZIP_METADATA_RETRY_SLEEP - return metadata + return self._generate_runtime_meta( + function_name, max_attempts=max_attempts, retry_sleep=retry_sleep + ) def delete_runtime(self, runtime_name, memory, version=__version__): - """ - Deletes a runtime - """ logger.info(f'Deleting runtime: {runtime_name} - {memory}MB') function_name = self._format_function_name(runtime_name, memory, version) - self.fc_client.delete_function(self.service_name, function_name) + self._delete_function_if_exists(function_name) def clean(self, **kwargs): - """" - Deletes all runtimes from the current service - """ - logger.debug('Going to delete all deployed runtimes') - if not self._service_exists(self.service_name): - return - - functions = self.fc_client.list_functions(self.service_name).data['functions'] - - for function in functions: - function_name = function['functionName'] - if function_name.startswith('lithops-worker'): - logger.info(f'Going to delete runtime {function_name}') - self.fc_client.delete_function(self.service_name, function_name) - - self.fc_client.delete_service(self.service_name) + logger.debug('Going to delete all deployed Lithops runtimes') + for function in self._list_functions(): + if function.function_name.startswith('lithops-worker'): + logger.info(f'Going to delete runtime {function.function_name}') + self.fc_client.delete_function(function.function_name) def list_runtimes(self, runtime_name='all'): - """ - List all the runtimes deployed in the Aliyun FC service - return: list of tuples (container_image_name, memory, version) - """ logger.debug('Listing deployed runtimes') runtimes = [] - if not self._service_exists(self.service_name): - return runtimes - - functions = self.fc_client.list_functions(self.service_name).data['functions'] - - for function in functions: - if function['functionName'].startswith('lithops-worker'): - memory = function['memorySize'] - version, img_name = self._unformat_function_name(function['functionName']) + for function in self._list_functions(): + if function.function_name.startswith('lithops-worker'): + memory = function.memory_size + version, img_name = self._unformat_function_name(function.function_name) if runtime_name == img_name or runtime_name == 'all': - runtimes.append((img_name, memory, version, function['functionName'])) + runtimes.append((img_name, memory, version, function.function_name)) return runtimes - def invoke(self, runtime_name, memory, payload={}): - """ - Invoke function - """ + def invoke(self, runtime_name, memory, payload=None): + if payload is None: + payload = {} function_name = self._format_function_name(runtime_name, memory) + payload_bytes = json.dumps(payload, default=str).encode('utf-8') - try: - res = self.fc_client.invoke_function( - serviceName=self.service_name, - functionName=function_name, - payload=json.dumps(payload, default=str), - headers={'x-fc-invocation-type': 'Async'} - ) - except fc2.fc_exceptions.FcError as e: - raise (e) + headers = fc_models.InvokeFunctionHeaders( + x_fc_invocation_type='Async', + ) + request = fc_models.InvokeFunctionRequest(body=io.BytesIO(payload_bytes)) - return res.headers['X-Fc-Request-Id'] + res = self.fc_client.invoke_function_with_options( + function_name, request, headers, FC_INVOKE_RUNTIME + ) - def _generate_runtime_meta(self, function_name): - """ - Extract installed Python modules from Aliyun runtime - """ + request_id = _get_response_header(res, 'x-fc-request-id') + if not request_id: + raise Exception(f'Aliyun FC invoke did not return a request ID: {res.headers}') + return request_id + + def _generate_runtime_meta(self, function_name, max_attempts, retry_sleep): logger.info(f'Extracting runtime metadata from: {function_name}') payload = {'log_level': logger.getEffectiveLevel(), 'get_metadata': True} + payload_bytes = json.dumps(payload, default=str).encode('utf-8') - try: - res = self.fc_client.invoke_function( - self.service_name, function_name, - payload=json.dumps(payload, default=str), - headers={'x-fc-invocation-type': 'Sync'} + headers = fc_models.InvokeFunctionHeaders( + x_fc_invocation_type='Sync', + ) + request = fc_models.InvokeFunctionRequest(body=io.BytesIO(payload_bytes)) + + last_error = None + runtime_meta = None + max_wait = max_attempts * retry_sleep + for attempt in range(max_attempts): + if self._use_custom_container(): + try: + self._check_function_failed(self._get_function(function_name)) + except FCClientException: + pass + + try: + res = self.fc_client.invoke_function_with_options( + function_name, request, headers, FC_INVOKE_RUNTIME + ) + runtime_meta = json.loads(_read_invoke_body(res)) + if isinstance(runtime_meta, dict) and runtime_meta.get('errorMessage'): + last_error = runtime_meta['errorMessage'] + logger.warning( + 'Metadata invoke failed (attempt %s/%s): %s', + attempt + 1, max_attempts, last_error, + ) + time.sleep(retry_sleep) + continue + break + except FCClientException as e: + last_error = e + if _is_pending_error(e): + logger.debug( + 'Function %s not ready for invoke (attempt %s/%s), retrying...', + function_name, attempt + 1, max_attempts, + ) + time.sleep(retry_sleep) + continue + raise Exception(f'Unable to extract runtime metadata: {e}') from e + else: + raise Exception( + f'Unable to extract runtime metadata from {function_name} after ' + f'{max_wait}s: {last_error}' ) - runtime_meta = json.loads(res.data) - - except Exception: - raise Exception("Unable to extract runtime metadata") if not runtime_meta or 'preinstalls' not in runtime_meta: raise Exception(runtime_meta) @@ -287,26 +586,19 @@ def _generate_runtime_meta(self, function_name): return runtime_meta def get_runtime_key(self, runtime_name, runtime_memory, version=__version__): - """ - Method that creates and returns the runtime key. - Runtime keys are used to uniquely identify runtimes within the storage, - in order to know which runtimes are installed and which not. - """ function_name = self._format_function_name(runtime_name, runtime_memory, version) - runtime_key = os.path.join(self.name, version, self.region, self.service_name, function_name) + runtime_key = os.path.join(self.name, version, self.region, function_name) return runtime_key def get_runtime_info(self): - """ - Method that returns all the relevant information about the runtime set - in config - """ - if utils.CURRENT_PY_VERSION not in config.AVAILABLE_PY_RUNTIMES: - raise Exception( - f'Python {utils.CURRENT_PY_VERSION} is not available for Aliyun ' - f'Functions. Please use one of {list(config.AVAILABLE_PY_RUNTIMES.keys())}' - ) + if not self._use_custom_container(): + if utils.CURRENT_PY_VERSION not in config.AVAILABLE_PY_RUNTIMES: + raise Exception( + f'Python {utils.CURRENT_PY_VERSION} is not available for Aliyun ' + f'Functions. Please use one of {list(config.AVAILABLE_PY_RUNTIMES.keys())}, ' + "or set aliyun_fc.deploy_mode to 'custom-container'" + ) if 'runtime' not in self.config or self.config['runtime'] == 'default': self.config['runtime'] = self._get_default_runtime_name() diff --git a/lithops/serverless/backends/aliyun_fc/config.py b/lithops/serverless/backends/aliyun_fc/config.py index 4e4a56df8..ec5d21093 100644 --- a/lithops/serverless/backends/aliyun_fc/config.py +++ b/lithops/serverless/backends/aliyun_fc/config.py @@ -25,19 +25,42 @@ 'max_workers': 300, 'worker_processes': 1, 'invoke_pool_threads': 64, + 'deploy_mode': 'runtime', # 'runtime' or 'custom-container' + 'docker_server': 'docker.io', } -CONNECTION_POOL_SIZE = 300 +# FC custom-container HTTP port (fixed by Alibaba Cloud; do not change). +CA_PORT = 9000 -SERVICE_NAME = 'lithops' BUILD_DIR = os.path.join(TEMP_DIR, 'AliyunRuntimeBuild') +FH_ZIP_LOCATION = os.path.join(TEMP_DIR, 'lithops_aliyun_fc.zip') + +DEFAULT_DOCKERFILE = """ +RUN apt-get update && apt-get install -y zip && rm -rf /var/lib/apt/lists/* + +RUN pip install --upgrade pip setuptools \ + && pip install --no-cache-dir \ + gunicorn flask gevent six pika redis requests PyYAML oss2 \ + cloudpickle ps-mem tblib psutil + +ENV APP_HOME=/function +WORKDIR ${APP_HOME} + +COPY lithops_aliyun_fc.zip . +RUN unzip lithops_aliyun_fc.zip && rm lithops_aliyun_fc.zip + +ENV CAPort=9000 +CMD exec gunicorn --bind 0.0.0.0:${CAPort} --workers 1 --timeout 600 --keep-alive 95 container_entry_point:app +""" AVAILABLE_PY_RUNTIMES = { + '3.9': 'python3.9', '3.10': 'python3.10', '3.12': 'python3.12', } REQUIREMENTS_FILE = """ +six pika tblib cloudpickle @@ -85,3 +108,16 @@ def load_config(config_data=None): account_id = config_data['aliyun_fc']['account_id'] region = config_data['aliyun_fc']['region'] config_data['aliyun_fc']['public_endpoint'] = ENDPOINT.format(account_id, region) + + deploy_mode = config_data['aliyun_fc'].get('deploy_mode', 'runtime') + if deploy_mode not in ('runtime', 'custom-container'): + raise Exception( + "aliyun_fc.deploy_mode must be 'runtime' or 'custom-container'" + ) + config_data['aliyun_fc']['deploy_mode'] = deploy_mode + + if deploy_mode == 'custom-container' and not config_data['aliyun_fc'].get('docker_user'): + raise Exception( + "aliyun_fc.docker_user is required when deploy_mode is 'custom-container' " + '(Docker Hub namespace for pushing images to docker.io)' + ) diff --git a/lithops/serverless/backends/aliyun_fc/container_entry_point.py b/lithops/serverless/backends/aliyun_fc/container_entry_point.py new file mode 100644 index 000000000..ff1585ec5 --- /dev/null +++ b/lithops/serverless/backends/aliyun_fc/container_entry_point.py @@ -0,0 +1,73 @@ +# +# Copyright Cloudlab URV 2020 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# HTTP entry point for Aliyun FC custom-container runtimes. +# See: https://www.alibabacloud.com/help/en/functioncompute/fc/user-guide/custom-container/ + +import json +import os +import logging + +from flask import Flask, request + +from lithops.version import __version__ +from lithops.utils import setup_lithops_logger +from lithops.worker import function_handler +from lithops.worker import function_invoker +from lithops.worker.utils import get_runtime_metadata + +logger = logging.getLogger('lithops.worker') + +app = Flask(__name__) + + +@app.route('/initialize', methods=['POST']) +def initialize(): + request_id = request.headers.get('x-fc-request-id', '') + logger.info('FC Initialize Start RequestId: %s', request_id) + logger.info('FC Initialize End RequestId: %s', request_id) + return f'Function is initialized, request_id: {request_id}\n' + + +@app.route('/invoke', methods=['POST']) +def invoke(): + request_id = request.headers.get('x-fc-request-id', '') + os.environ['__LITHOPS_ACTIVATION_ID'] = request_id + os.environ['__LITHOPS_BACKEND'] = 'Aliyun Function Compute' + + raw = request.get_data() + if not raw: + return 'Empty event\n', 400 + + args = json.loads(raw.decode('utf-8') if isinstance(raw, bytes) else raw) + setup_lithops_logger(args.get('log_level', logging.INFO)) + + if args.get('get_metadata'): + logger.info('Lithops v%s - Generating metadata', __version__) + return json.dumps(get_runtime_metadata()), 200, {'Content-Type': 'application/json'} + + if 'remote_invoker' in args: + logger.info('Lithops v%s - Starting Aliyun Function Compute invoker', __version__) + function_invoker(args) + else: + logger.info('Lithops v%s - Starting Aliyun Function Compute execution', __version__) + function_handler(args) + + return json.dumps({'Execution': 'Finished'}), 200, {'Content-Type': 'application/json'} + + +if __name__ == '__main__': + port = int(os.getenv('CAPort', '9000')) + app.run(host='0.0.0.0', port=port) diff --git a/lithops/storage/backends/aliyun_oss/aliyun_oss.py b/lithops/storage/backends/aliyun_oss/aliyun_oss.py index 584e413cc..c6c0b0b4a 100644 --- a/lithops/storage/backends/aliyun_oss/aliyun_oss.py +++ b/lithops/storage/backends/aliyun_oss/aliyun_oss.py @@ -15,6 +15,7 @@ # import os +import hashlib import oss2 import shutil import logging @@ -59,21 +60,66 @@ def _connect_bucket(self, bucket_name): def get_client(self): return self - def generate_bucket_name(self): + def _bucket_name_suffix(self, salt=''): """ - Generates a unique bucket name + OSS bucket names are globally unique; use a stable hash instead of + a short access-key prefix to avoid collisions with other accounts. """ key = self.config['access_key_id'] - self.config['storage_bucket'] = f'lithops-{self.region}-{key[:6].lower()}' + account_id = self.config.get('account_id', '') + digest = hashlib.sha1(f'{key}{account_id}{salt}'.encode()).hexdigest() + return digest[:12] + def generate_bucket_name(self): + """ + Generates a unique bucket name + """ + self.config['storage_bucket'] = ( + f'lithops-{self.region}-{self._bucket_name_suffix()}' + ) return self.config['storage_bucket'] + def _bucket_exists(self, bucket): + try: + bucket.get_bucket_info() + return True + except oss2.exceptions.NoSuchBucket: + return False + def create_bucket(self, bucket_name): """ Create a bucket if it doesn't exist """ bucket = self._connect_bucket(bucket_name) - bucket.create_bucket() + if self._bucket_exists(bucket): + logger.debug(f'Bucket {bucket_name} already exists in Aliyun OSS') + self.config['storage_bucket'] = bucket_name + return bucket_name + + logger.debug(f'Creating bucket {bucket_name} in Aliyun OSS') + for attempt in range(5): + try: + bucket.create_bucket() + self.config['storage_bucket'] = bucket_name + return bucket_name + except oss2.exceptions.ServerError as e: + code = (e.details or {}).get('Code', '') + if e.status == 409 and code == 'BucketAlreadyExists': + bucket_name = ( + f'lithops-{self.region}-{self._bucket_name_suffix(str(attempt))}' + ) + logger.debug( + f'OSS bucket name unavailable, retrying with {bucket_name}' + ) + bucket = self._connect_bucket(bucket_name) + continue + raise + + raise Exception( + 'Unable to create an Aliyun OSS bucket: all generated names are ' + 'already taken. Set storage_bucket in the aliyun_oss section of ' + 'your Lithops config to a globally unique name.' + ) def put_object(self, bucket_name, key, data): """ diff --git a/runtime/aliyun_fc/Dockerfile b/runtime/aliyun_fc/Dockerfile new file mode 100644 index 000000000..7125fde85 --- /dev/null +++ b/runtime/aliyun_fc/Dockerfile @@ -0,0 +1,23 @@ +# Reference Dockerfile for Aliyun FC custom-container mode. +# Lithops generates this automatically when deploy_mode is custom-container. +FROM python:3.12-slim-bookworm + +RUN apt-get update && apt-get install -y zip && rm -rf /var/lib/apt/lists/* + +RUN pip install --upgrade pip setuptools \ + && pip install --no-cache-dir \ + gunicorn flask gevent six pika redis requests PyYAML oss2 \ + cloudpickle ps-mem tblib psutil + +ENV APP_HOME=/function +WORKDIR ${APP_HOME} + +COPY lithops_aliyun_fc.zip . +RUN unzip lithops_aliyun_fc.zip && rm lithops_aliyun_fc.zip + +# Optional: add extra dependencies +# COPY requirements.txt . +# RUN pip install --no-cache-dir -r requirements.txt + +ENV CAPort=9000 +CMD exec gunicorn --bind 0.0.0.0:${CAPort} --workers 1 --timeout 600 container_entry_point:app diff --git a/runtime/aliyun_fc/README.md b/runtime/aliyun_fc/README.md index 05b444c8b..b800d94e6 100644 --- a/runtime/aliyun_fc/README.md +++ b/runtime/aliyun_fc/README.md @@ -1,8 +1,36 @@ -# Lithops runtime for Aliyun Functions Compute +# Lithops runtime for Aliyun Functions Compute 3.0 The runtime is the place where your functions are executed. The default runtime is automatically created the first time you execute a function. Lithops automatically detects the Python version of your environment and deploys the default runtime based on it. -Currently, Aliyun Function Compute supports Python 3.10 and 3.12 (3.12 is in public preview). You can find the list of pre-installed modules [here](https://www.alibabacloud.com/help/en/function-compute/latest/python-event-functions). In addition, the Lithops default runtimes are built with the packages included in this [requirements.txt](requirements.txt) file: +Aliyun FC supports two Lithops deploy modes (set `deploy_mode` under `aliyun_fc` in config): + +- **`runtime`** (default): zip package + managed Python runtime (`python3.10`, `python3.12`, region-dependent). +- **`custom-container`**: Docker image; see [Custom container mode](#custom-container-mode) below. + +For managed runtimes, see pre-installed modules [here](https://www.alibabacloud.com/help/en/functioncompute/fc/user-guide/python/). Lithops default zip runtimes use [requirements.txt](requirements.txt): + +## Custom container mode + +Use this mode to run any Python version (e.g. 3.12) without relying on managed FC runtimes: + +```yaml +aliyun_fc: + role_arn: + deploy_mode: custom-container + docker_user: + docker_password: + docker_server: docker.io +``` + +Lithops builds a Linux/amd64 image, pushes it to **Docker Hub** (`docker.io`) by default, and creates an FC function with `runtime: custom-container`. The container must expose an HTTP server on `0.0.0.0:9000` (see [FC custom container docs](https://www.alibabacloud.com/help/en/functioncompute/fc/user-guide/custom-container/)). The service role must be able to pull the image. + +Reference Dockerfile: [Dockerfile](Dockerfile). Example template: [start-fc custom-container/python](https://github.com/devsapp/start-fc/tree/V3/custom-container/python). + +Build a custom container runtime from `requirements.txt`: + +``` +$ lithops runtime build -b aliyun_fc -f requirements.txt my_container_runtime +``` To run a function with the default runtime you don't need to specify anything in the code, since everything is handled internally by Lithops: diff --git a/runtime/aliyun_fc/requirements.txt b/runtime/aliyun_fc/requirements.txt index 0e9ce45f7..48e71167f 100644 --- a/runtime/aliyun_fc/requirements.txt +++ b/runtime/aliyun_fc/requirements.txt @@ -1,7 +1,6 @@ # Requirements.txt contains a list of dependencies for the Python Application # # Mandatory Lithops packages -aliyun-fc2 oss2 pika flask diff --git a/setup.py b/setup.py index 0110e6cd0..2c8b772ab 100644 --- a/setup.py +++ b/setup.py @@ -45,8 +45,8 @@ 'azure-storage-queue' ], 'aliyun': [ - 'aliyun-fc2', - 'oss2' + 'alibabacloud-fc20230330>=4.7.0', + 'oss2', ], 'ceph': [ 'boto3'