From 6b5aaf4e179de2ca36d6635c2b218ecabcf9e987 Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Tue, 16 Jun 2026 18:11:59 +0200 Subject: [PATCH 1/5] Update code_engine backend --- CHANGELOG.md | 1 + lithops/scripts/cli.py | 28 +- .../backends/code_engine/code_engine.py | 752 ++++++++---------- .../serverless/backends/code_engine/config.py | 77 +- .../backends/code_engine/entry_point.py | 4 +- 5 files changed, 346 insertions(+), 516 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d5459f9fc..2b9251eb6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,7 @@ - [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 +- [Code Engine] Rewrote backend to use the IBM Code Engine SDK v2 (`CodeEngineV2`) instead of the Kubernetes API ### Fixed - [K8s] Fixed default runtime builds impacted by Debian Buster end-of-life. diff --git a/lithops/scripts/cli.py b/lithops/scripts/cli.py index fac28f962..7fe835f6c 100644 --- a/lithops/scripts/cli.py +++ b/lithops/scripts/cli.py @@ -188,7 +188,9 @@ def test(test, config, backend, storage, debug, region, exitfirst): @click.option('--storage', '-s', default=None, help='storage backend') @click.option('--debug', '-d', is_flag=True, help='debug mode') @click.option('--region', '-r', default=None, help='compute backend region') -def hello(config, backend, storage, debug, region): +@click.option('--map', 'map_count', '-m', default=None, type=click.IntRange(min=1), + help='number of map invocations to run instead of a single call_async') +def hello(config, backend, storage, debug, region, map_count): config = load_yaml_config(config) if config else None log_level = logging.INFO if not debug else logging.DEBUG @@ -207,13 +209,25 @@ def hello(name): config=config, backend=backend, storage=storage, region=region ) - fexec.call_async(hello, username) - result = fexec.get_result() - print() - if result == f'Hello {username}!': - print(result, 'Lithops is working as expected :)') + expected = f'Hello {username}!' + + if map_count: + fexec.map(hello, [username] * map_count) + results = fexec.get_result() + print() + if all(result == expected for result in results): + print(f'All {map_count} map activations returned: {expected}') + print('Lithops is working as expected :)') + else: + print(results, 'Something went wrong :(') else: - print(result, 'Something went wrong :(') + fexec.call_async(hello, username) + result = fexec.get_result() + print() + if result == expected: + print(result, 'Lithops is working as expected :)') + else: + print(result, 'Something went wrong :(') print() diff --git a/lithops/serverless/backends/code_engine/code_engine.py b/lithops/serverless/backends/code_engine/code_engine.py index 56f4251c6..0bdf41abf 100644 --- a/lithops/serverless/backends/code_engine/code_engine.py +++ b/lithops/serverless/backends/code_engine/code_engine.py @@ -17,64 +17,68 @@ import os import re -import base64 import hashlib import json import time import logging -import urllib3 import copy -import yaml -from kubernetes import client, watch -from kubernetes.config import load_incluster_config -from kubernetes.client.rest import ApiException + +from ibm_cloud_sdk_core import ApiException +from ibm_code_engine_sdk.code_engine_v2 import ( + CodeEngineV2, + EnvVarPrototype, + SecretDataRegistrySecretData, +) from lithops import utils from lithops.config import dump_yaml_config, load_yaml_config from lithops.version import __version__ from lithops.constants import CACHE_DIR, COMPUTE_CLI_MSG, JOBS_PREFIX -from lithops.util.ibm_token_manager import IAMTokenManager from . import config -urllib3.disable_warnings() - logger = logging.getLogger(__name__) +REGISTRY_SECRET_NAME = 'lithops-regcred' +LITHOPS_RUNTIME_TYPE = 'lithops-runtime' +ENTRYPOINT_SCRIPT = '/lithops/lithopsentry.py' +PYTHON_BIN = '/usr/local/bin/python' + -# Decorator to wrap a function to reinit clients and retry on except. def retry_on_except(func): def decorated_func(*args, **kwargs): _self = args[0] connection_retries = _self.config.get('connection_retries') if not connection_retries: return func(*args, **kwargs) - else: - ex = None - for retry in range(connection_retries): - try: - return func(*args, **kwargs) - except ApiException as e: - if e.status == 409: - ex = e - body = json.loads(e.body) - if body.get('reason') in {'AlreadyExists', 'Conflict'} or 'already exists' in body.get('message'): - logger.debug("Encountered conflict error {}, ignoring".format(body.get('message'))) - elif e.status == 500: - ex = e - logger.exception((f'Got exception {e}, retrying for the {retry} time, left retries {connection_retries - 1 - retry}')) - else: - logger.debug((f'Got exception {e} when trying to invoke {func.__name__}, raising')) - raise e - time.sleep(5) - # we got run out of retries, now raising - raise ex + + ex = None + for retry in range(connection_retries): + try: + return func(*args, **kwargs) + except ApiException as e: + if e.status_code == 409: + ex = e + logger.debug(f"Encountered conflict error {e.message}, ignoring") + elif e.status_code == 500: + ex = e + logger.exception( + f'Got exception {e}, retrying for the {retry} time, ' + f'left retries {connection_retries - 1 - retry}' + ) + else: + logger.debug( + f'Got exception {e} when trying to invoke {func.__name__}, raising' + ) + raise e + time.sleep(5) + raise ex return decorated_func class CodeEngineBackend: """ - A wrap-up around Code Engine backend. + IBM Code Engine backend using the official Code Engine SDK. """ def __init__(self, ce_config, internal_storage): @@ -83,114 +87,69 @@ def __init__(self, ce_config, internal_storage): self.type = utils.BackendType.BATCH.value self.config = ce_config self.internal_storage = internal_storage - self.is_lithops_worker = utils.is_lithops_worker() - self.user_agent = ce_config['user_agent'] self.iam_api_key = ce_config['iam_api_key'] self.namespace = ce_config.get('namespace') self.region = ce_config['region'] - self.user_key = re.sub(r'[^a-z0-9\-\.]', '0', self.iam_api_key[:4].lower()) # RFC 1123 compliant + self.user_key = re.sub(r'[^a-z0-9\-\.]', '0', self.iam_api_key[:4].lower()) self.project_name = ce_config.get('project_name', f'lithops-{self.region}-{self.user_key}') - self.project_id = None + self.project_id = ce_config.get('project_id') self.config['project_name'] = self.project_name - self.token_manager = None - self.code_engine_service_v1 = None - self.code_engine_service_v2 = None - - self.cache_dir = os.path.join(CACHE_DIR, self.name) - self.cache_file = os.path.join(self.cache_dir, self.project_name + '_data') - - self.cluster = config.CLUSTER_URL.format(self.region) - - if self.is_lithops_worker: - logger.debug('Loading incluster kubecfg') - load_incluster_config() - self.custom_api = client.CustomObjectsApi() - self.core_api = client.CoreV1Api() - else: - self._create_k8s_iam_client() - - self.jobs = [] # list to store executed jobs (job_keys) + self.ce_client = None + self.cache_file = os.path.join(CACHE_DIR, self.name, self.project_name + '_data') + self.jobs = [] msg = COMPUTE_CLI_MSG.format('IBM Code Engine') logger.info(f"{msg} - Project: {self.project_name} - Region: {self.region}") def _create_code_engine_client(self): - """ - Creates new code engine clients - """ - if self.code_engine_service_v1 and self.code_engine_service_v2: + if self.ce_client: return from ibm_cloud_sdk_core.authenticators import IAMAuthenticator - from ibm_code_engine_sdk.code_engine_v2 import CodeEngineV2 - from ibm_code_engine_sdk.ibm_cloud_code_engine_v1 import IbmCloudCodeEngineV1 authenticator = IAMAuthenticator(self.iam_api_key) - self.code_engine_service_v1 = IbmCloudCodeEngineV1(authenticator=authenticator) - self.code_engine_service_v1.set_service_url(config.BASE_URL_V1.format(self.region)) - self.code_engine_service_v2 = CodeEngineV2(authenticator=authenticator) - self.code_engine_service_v2.set_service_url(config.BASE_URL_V2.format(self.region)) + self.ce_client = CodeEngineV2(authenticator=authenticator) + self.ce_client.set_service_url(config.BASE_URL_V2.format(self.region)) def _get_or_create_namespace(self, create=True): """ - Gets or creates a new namespace + Resolves the Code Engine project. Namespace is kept for runtime key compatibility. """ - if self.namespace or self.is_lithops_worker: - return self.namespace - ce_data = load_yaml_config(self.cache_file) - self.namespace = ce_data.get('namespace') - self.project_id = ce_data.get('project_id') - self.config['project_id'] = self.project_id - self.config['namespace'] = self.namespace + if not self.project_id: + self.project_id = ce_data.get('project_id') + if not self.namespace: + self.namespace = ce_data.get('namespace') - if self.namespace: + if self.project_id: + self._sync_project_config() return self.namespace self._create_code_engine_client() - def get_k8s_namespace(project_id): - delegated_refresh_token_payload = { - 'grant_type': 'urn:ibm:params:oauth:grant-type:apikey', - 'apikey': self.iam_api_key, - 'response_type': 'delegated_refresh_token', - 'receiver_client_ids': 'ce', - 'delegated_refresh_token_expiry': '3600' - } - token_manager = self.code_engine_service_v2.authenticator.token_manager - request_payload = token_manager.request_payload - token_manager.request_payload = delegated_refresh_token_payload - iam_response = token_manager.request_token() - token_manager.request_payload = request_payload - delegated_refresh_token = iam_response['delegated_refresh_token'] - kc_resp = self.code_engine_service_v1.get_kubeconfig(delegated_refresh_token, project_id) - return kc_resp.get_result()['contexts'][0]['context']['namespace'] - - response = self.code_engine_service_v2.list_projects().get_result() - if 'projects' in response: - for project in response['projects']: - if project['name'] == self.project_name: - logger.debug(f"Found Code Engine project: {self.project_name}") - self.project_id = project['id'] - self.namespace = get_k8s_namespace(self.project_id) - self.config['project_id'] = self.project_id - self.config['namespace'] = self.namespace - - if not self.namespace and create: + response = self.ce_client.list_projects().get_result() + for project in response.get('projects', []): + if project['name'] == self.project_name: + logger.debug(f"Found Code Engine project: {self.project_name}") + self.project_id = project['id'] + break + + if not self.project_id and create: logger.debug(f"Creating new Code Engine project: {self.project_name}") - response = self.code_engine_service_v2.create_project( + response = self.ce_client.create_project( name=self.project_name, resource_group_id=self.config['resource_group_id'] - ) - project = response.get_result() - self.project_id = project['id'] - self.namespace = get_k8s_namespace(self.project_id) - self.config['project_id'] = self.project_id - self.config['namespace'] = self.namespace + ).get_result() + self.project_id = response['id'] + + if not self.project_id: + return None + + self._sync_project_config() ce_data['project_name'] = self.project_name ce_data['project_id'] = self.project_id @@ -199,47 +158,76 @@ def get_k8s_namespace(project_id): return self.namespace - def _create_k8s_iam_client(self): - """ - Creates the k8s client with the IAM token - """ - if self.is_lithops_worker: - return - - if not self.token_manager: - self.token_manager = IAMTokenManager(self.iam_api_key) - - token, expiry_time = self.token_manager.get_token() - - if expiry_time != self.config.get('token_expiry_time'): - self.config['token_expiry_time'] = expiry_time + def _sync_project_config(self): + self.namespace = self.namespace or self.project_id + self.config['project_id'] = self.project_id + self.config['namespace'] = self.namespace + self._create_code_engine_client() - configuration = client.Configuration.get_default_copy() - configuration.host = self.cluster - configuration.api_key = {"authorization": "Bearer " + token} - client.Configuration.set_default(configuration) + @staticmethod + def _format_memory(memory_mb): + """ + Code Engine expects memory in G/M units using supported GB values + (e.g. 256 MB -> 0.25G, not 256M). + """ + gb = memory_mb / 1024 + if gb == int(gb): + return f'{int(gb)}G' + text = f'{gb:.3f}'.rstrip('0').rstrip('.') + return f'{text}G' + + @staticmethod + def _format_cpu(cpu): + text = f'{cpu:.3f}'.rstrip('0').rstrip('.') + return text + + @staticmethod + def _parse_memory(memory_limit): + if not memory_limit: + return 0 + if memory_limit.endswith('G'): + mb = round(float(memory_limit[:-1]) * 1024) + elif memory_limit.endswith('M'): + mb = int(float(memory_limit[:-1])) + else: + mb = int(float(memory_limit)) + return CodeEngineBackend._normalize_memory_mb(mb) - self.custom_api = client.CustomObjectsApi() - self.core_api = client.CoreV1Api() + @staticmethod + def _normalize_memory_mb(memory_mb): + return min(config.VALID_MEMORY_VALUES, key=lambda v: abs(v - memory_mb)) def _format_jobdef_name(self, runtime_name, runtime_memory, version=__version__): name = f'{runtime_name}-{runtime_memory}-{version}' name_hash = hashlib.sha1(name.encode("utf-8")).hexdigest()[:10] - return f'lithops-worker-{self.user_key}-{version.replace(".", "")}-{name_hash}' def _get_default_runtime_image_name(self): - """ - Generates the default runtime image name - """ return utils.get_default_container_name( self.name, self.config, 'lithops-codeenigne-default' ) + def _has_registry_credentials(self): + return all(key in self.config for key in ["docker_user", "docker_password"]) + + def _build_job_env_variables(self, action, payload_config_map_name=None): + env_variables = [ + EnvVarPrototype(name='ACTION', type='literal', value=action), + EnvVarPrototype(name='LITHOPS_RUNTIME_TYPE', type='literal', value=LITHOPS_RUNTIME_TYPE), + EnvVarPrototype(name='LITHOPS_VERSION', type='literal', value=__version__), + ] + if payload_config_map_name: + env_variables.append( + EnvVarPrototype( + name='PAYLOAD', + type='config_map_key_reference', + reference=payload_config_map_name, + key='lithops.payload', + ) + ) + return env_variables + def build_runtime(self, docker_image_name, dockerfile, extra_args=[]): - """ - Builds a new runtime from a Docker file and pushes it to the Docker hub - """ logger.info(f'Building runtime {docker_image_name} from {dockerfile}') docker_path = utils.get_docker_path() @@ -277,10 +265,6 @@ def build_runtime(self, docker_image_name, dockerfile, extra_args=[]): logger.debug('Building done!') def _build_default_runtime(self, default_runtime_img_name): - """ - Builds the default runtime - """ - # Build default runtime using local dokcer dockerfile = "Dockefile.default-ce-runtime" python_version = utils.CURRENT_PY_VERSION base_image = "slim-bookworm" @@ -293,9 +277,6 @@ def _build_default_runtime(self, default_runtime_img_name): os.remove(dockerfile) def deploy_runtime(self, docker_image_name, memory, timeout): - """ - Deploys a new runtime from an already built Docker image - """ self._get_or_create_namespace() try: @@ -307,40 +288,62 @@ def deploy_runtime(self, docker_image_name, memory, timeout): self._build_default_runtime(docker_image_name) logger.debug(f"Deploying runtime: {docker_image_name} - Memory: {memory} Timeout: {timeout}") - self._create_k8s_iam_client() self._create_job_definition(docker_image_name, memory, timeout) runtime_meta = self._generate_runtime_meta(docker_image_name, memory) return runtime_meta - def delete_runtime(self, runtime_name, memory, version=__version__): - """ - Deletes a runtime - We need to delete job definition - """ + def _resolve_jobdef_name(self, runtime_name, memory, version=__version__): + jobdef_name = self._format_jobdef_name(runtime_name, memory, version) + if self._job_def_exists(jobdef_name): + return jobdef_name + + logger.debug( + f"Job definition {jobdef_name} not found, searching by image {runtime_name}" + ) + try: + jobs = self.ce_client.list_jobs(self.project_id).get_result() + except ApiException as e: + logger.debug(f"List all jobs failed with {e.status_code} {e.message}") + return jobdef_name + + for job in jobs.get('jobs', []): + fn_name = job.get('name') + if not fn_name or not fn_name.startswith(f'lithops-worker-{self.user_key}'): + continue + + image_name = job.get('image_reference') or '' + if runtime_name not in image_name and image_name not in runtime_name: + continue + + job_version = __version__ + for env_var in job.get('run_env_variables', []): + if env_var.get('name') == 'LITHOPS_VERSION': + job_version = env_var.get('value', job_version) + + if job_version != version: + continue + + logger.debug(f"Resolved job definition {fn_name} for runtime {runtime_name}") + return fn_name + + return jobdef_name + + def delete_runtime(self, runtime_name, memory, version=__version__, jobdef_name=None): if not self._get_or_create_namespace(create=False): logger.info(f"Project {self.project_name} does not exist") return + if not jobdef_name: + jobdef_name = self._resolve_jobdef_name(runtime_name, memory, version) + logger.info(f'Deleting runtime: {runtime_name} - {memory}MB') - self._create_k8s_iam_client() try: - jobdef_id = self._format_jobdef_name(runtime_name, memory, version) - self.custom_api.delete_namespaced_custom_object( - group=config.DEFAULT_GROUP, - version=config.DEFAULT_VERSION, - name=jobdef_id, - namespace=self.namespace, - plural="jobdefinitions", - body=client.V1DeleteOptions(), - ) + self.ce_client.delete_job(self.project_id, jobdef_name) except ApiException as e: - logger.debug(f"Deleting a jobdef failed with {e.status} {e.reason}") + logger.debug(f"Deleting a job failed with {e.status_code} {e.message}") def clean(self, all=False): - """ - Deletes all runtimes from all packages - """ logger.info(f'Cleaning project {self.project_name}') if not self._get_or_create_namespace(create=False): logger.info(f"Project {self.project_name} does not exist") @@ -348,65 +351,62 @@ def clean(self, all=False): os.remove(self.cache_file) return - self._create_k8s_iam_client() self.clear() runtimes = self.list_runtimes() - for image_name, memory, version, fn_name in runtimes: - self.delete_runtime(image_name, memory, version) - - logger.debug('Deleting all lithops configmaps') - configmaps = self.core_api.list_namespaced_config_map(namespace=self.namespace) - for configmap in configmaps.items: - config_name = configmap.metadata.name - if config_name.startswith('lithops'): - logger.debug(f'Deleting configmap {config_name}') - self.core_api.delete_namespaced_config_map( - name=config_name, - namespace=self.namespace, - grace_period_seconds=0) + for image_name, memory, version, jobdef_name in runtimes: + self.delete_runtime(image_name, memory, version, jobdef_name=jobdef_name) + + logger.debug('Deleting all lithops config maps') + try: + configmaps = self.ce_client.list_config_maps(self.project_id).get_result() + for configmap in configmaps.get('config_maps', []): + config_name = configmap.get('name') + if config_name and config_name.startswith('lithops'): + logger.debug(f'Deleting config map {config_name}') + try: + self.ce_client.delete_config_map(self.project_id, config_name) + except ApiException as e: + logger.debug(f"Deleting config map failed with {e.status_code} {e.message}") + except ApiException as e: + logger.debug(f"Listing config maps failed with {e.status_code} {e.message}") if all and os.path.exists(self.cache_file): - self._create_code_engine_client() logger.info(f"Deleting Code Engine project: {self.project_name}") - self.code_engine_service_v2.delete_project(id=self.project_id) + self.ce_client.delete_project(id=self.project_id) os.remove(self.cache_file) def list_runtimes(self, docker_image_name='all'): - """ - List all the runtimes - return: list of tuples (docker_image_name, memory) - """ runtimes = [] if not self._get_or_create_namespace(create=False): logger.info(f"Project {self.project_name} does not exist") return runtimes - self._create_k8s_iam_client() - try: - jobdefs = self.custom_api.list_namespaced_custom_object( - group=config.DEFAULT_GROUP, - version=config.DEFAULT_VERSION, - namespace=self.namespace, - plural="jobdefinitions" - ) + jobs = self.ce_client.list_jobs(self.project_id).get_result() except ApiException as e: - logger.debug(f"List all jobdefinitions failed with {e.status} {e.reason}") + logger.debug(f"List all jobs failed with {e.status_code} {e.message}") return runtimes - for jobdef in jobdefs['items']: + for job in jobs.get('jobs', []): try: - if not jobdef['metadata']['name'].startswith(f'lithops-worker-{self.user_key}'): + fn_name = job.get('name') + if not fn_name or not fn_name.startswith(f'lithops-worker-{self.user_key}'): continue - if not jobdef['metadata']['labels']['type'] == 'lithops-runtime': + + runtime_type = None + version = __version__ + for env_var in job.get('run_env_variables', []): + if env_var.get('name') == 'LITHOPS_RUNTIME_TYPE': + runtime_type = env_var.get('value') + elif env_var.get('name') == 'LITHOPS_VERSION': + version = env_var.get('value', version) + + if runtime_type != LITHOPS_RUNTIME_TYPE: continue - fn_name = jobdef['metadata']['name'] - version = jobdef['metadata']['labels']['version'].replace('lithops_v', '') - container = jobdef['spec']['template']['containers'][0] - image_name = container['image'] - memory = container['resources']['requests']['memory'].replace('M', '') - memory = int(int(memory) / 1000 * 1024) + + image_name = job.get('image_reference') + memory = self._parse_memory(job.get('scale_memory_limit')) if docker_image_name in image_name or docker_image_name == 'all': runtimes.append((image_name, memory, version, fn_name)) except Exception: @@ -415,45 +415,28 @@ def list_runtimes(self, docker_image_name='all'): return runtimes def clear(self, job_keys=None): - """ - Clean all completed jobruns in the current executor - """ if not self._get_or_create_namespace(create=False): logger.info(f"Project {self.project_name} does not exist") return - self._create_k8s_iam_client() jobs_to_delete = job_keys or self.jobs for job_key in jobs_to_delete: + jobrun_name = f'lithops-{job_key.lower()}' try: - jobrun_name = f'lithops-{job_key.lower()}' - self.custom_api.delete_namespaced_custom_object( - group=config.DEFAULT_GROUP, - version=config.DEFAULT_VERSION, - name=jobrun_name, - namespace=self.namespace, - plural="jobruns", - body=client.V1DeleteOptions(), - ) + self.ce_client.delete_job_run(self.project_id, jobrun_name) self._delete_config_map(jobrun_name) except ApiException as e: - logger.debug(f"Deleting a jobrun failed with {e.status} {e.reason}") + logger.debug(f"Deleting a job run failed with {e.status_code} {e.message}") try: self.jobs.remove(job_key) except ValueError: pass def invoke(self, docker_image_name, runtime_memory, job_payload): - """ - Invoke -- return information about this invocation - For array jobs only remote_invocator is allowed - """ self._get_or_create_namespace() - self._create_k8s_iam_client() executor_id = job_payload['executor_id'] job_id = job_payload['job_id'] - job_key = job_payload['job_key'] self.jobs.append(job_key) @@ -461,7 +444,6 @@ def invoke(self, docker_image_name, runtime_memory, job_payload): chunksize = job_payload['chunksize'] max_workers = job_payload['max_workers'] - # Make sure only max_workers are started total_workers = total_calls // chunksize + (total_calls % chunksize > 0) if max_workers < total_workers: chunksize = total_calls // max_workers + (total_calls % max_workers > 0) @@ -473,56 +455,39 @@ def invoke(self, docker_image_name, runtime_memory, job_payload): ) jobdef_name = self._format_jobdef_name(docker_image_name, runtime_memory) - if not self._job_def_exists(jobdef_name): - jobdef_name = self._create_job_definition(docker_image_name, runtime_memory) - - jobrun_res = yaml.safe_load(config.JOBRUN_DEFAULT) + self._create_job_definition(docker_image_name, runtime_memory) activation_id = f'lithops-{job_key.lower()}' - - jobrun_res['metadata']['name'] = activation_id - jobrun_res['metadata']['namespace'] = self.namespace - - jobrun_res['spec']['jobDefinitionRef'] = str(jobdef_name) - jobrun_res['spec']['jobDefinitionSpec']['arraySpec'] = '0-' + str(total_workers - 1) - jobrun_res['spec']['jobDefinitionSpec']['maxExecutionTime'] = self.config['runtime_timeout'] - - container = jobrun_res['spec']['jobDefinitionSpec']['template']['containers'][0] - container['name'] = str(jobdef_name) - container['env'][0]['value'] = 'run' - - config_map = self._create_config_map(activation_id, job_payload) - container['env'][1]['valueFrom']['configMapKeyRef']['name'] = config_map - - container['resources']['requests']['memory'] = f'{runtime_memory / 1024}G' - container['resources']['requests']['cpu'] = str(self.config['runtime_cpu']) - - logger.debug('ExecutorID {} | JobID {} - Going to run {} activations ' - '{} workers'.format(executor_id, job_id, total_calls, total_workers)) - - self._run_job(jobrun_res) - - # logger.debug(f"response - {res}") + config_map_name = self._create_config_map(activation_id, job_payload) + + self._run_job( + jobdef_name=jobdef_name, + jobrun_name=activation_id, + total_workers=total_workers, + runtime_memory=runtime_memory, + action='run', + payload_config_map_name=config_map_name, + ) return activation_id @retry_on_except - def _run_job(self, jobrun_res): - self.custom_api.create_namespaced_custom_object( - group=config.DEFAULT_GROUP, - version=config.DEFAULT_VERSION, - namespace=self.namespace, - plural="jobruns", - body=jobrun_res, + def _run_job(self, jobdef_name, jobrun_name, total_workers, runtime_memory, + action, payload_config_map_name): + self.ce_client.create_job_run( + self.project_id, + job_name=jobdef_name, + name=jobrun_name, + scale_array_spec=f'0-{total_workers - 1}', + scale_max_execution_time=self.config['runtime_timeout'], + scale_memory_limit=self._format_memory(runtime_memory), + scale_cpu_limit=self._format_cpu(self.config['runtime_cpu']), + run_env_variables=self._build_job_env_variables(action, payload_config_map_name), ) def _create_container_registry_secret(self): - """ - Create the container registry secret in the cluster - (only if credentials are present in config) - """ - if not all(key in self.config for key in ["docker_user", "docker_password"]): + if not self._has_registry_credentials(): return logger.debug('Creating container registry secret') @@ -530,119 +495,82 @@ def _create_container_registry_secret(self): docker_user = self.config['docker_user'] docker_password = self.config['docker_password'] - cred_payload = { - "auths": { - docker_server: { - "Username": docker_user, - "Password": docker_password - } - } - } - - data = { - ".dockerconfigjson": base64.b64encode( - json.dumps(cred_payload).encode() - ).decode() - } - - secret = client.V1Secret( - api_version="v1", - data=data, - kind="Secret", - metadata=dict(name="lithops-regcred", namespace=self.namespace), - type="kubernetes.io/dockerconfigjson", + secret_data = SecretDataRegistrySecretData( + server=docker_server, + username=docker_user, + password=docker_password, ) try: - self.core_api.delete_namespaced_secret("lithops-regcred", self.namespace) - except ApiException: - pass + self.ce_client.delete_secret(self.project_id, REGISTRY_SECRET_NAME) + except ApiException as e: + if e.status_code != 404: + raise e try: - self.core_api.create_namespaced_secret(self.namespace, secret) + self.ce_client.create_secret( + self.project_id, + format='registry', + name=REGISTRY_SECRET_NAME, + data=secret_data, + ) except ApiException as e: - if e.status != 409: + if e.status_code != 409: raise e @retry_on_except def _create_job_definition(self, docker_image_name, runtime_memory, timeout=None): - """ - Creates a Job definition - """ self._create_container_registry_secret() jobdef_name = self._format_jobdef_name(docker_image_name, runtime_memory) logger.debug(f"Creating job definition {jobdef_name}") - jobdef_res = yaml.safe_load(config.JOBDEF_DEFAULT) - - jobdef_res['metadata']['name'] = jobdef_name - jobdef_res['metadata']['labels']['version'] = 'lithops_v' + __version__ - jobdef_res['spec']['maxExecutionTime'] = timeout or self.config['runtime_timeout'] - - container = jobdef_res['spec']['template']['containers'][0] - container['image'] = docker_image_name - container['name'] = jobdef_name - container['env'][0]['value'] = 'run' - container['resources']['requests']['memory'] = f'{runtime_memory / 1024}G' - container['resources']['requests']['cpu'] = str(self.config['runtime_cpu']) - - if not all(key in self.config for key in ["docker_user", "docker_password"]): - del jobdef_res['spec']['template']['imagePullSecrets'] - try: - self.custom_api.delete_namespaced_custom_object( - group=config.DEFAULT_GROUP, - version=config.DEFAULT_VERSION, - namespace=self.namespace, - plural="jobdefinitions", - name=jobdef_name, - ) + self.ce_client.delete_job(self.project_id, jobdef_name) except ApiException as e: - if e.status == 404: - pass - else: + if e.status_code != 404: raise e while self._job_def_exists(jobdef_name): time.sleep(1) - try: - self.custom_api.create_namespaced_custom_object( - group=config.DEFAULT_GROUP, - version=config.DEFAULT_VERSION, - namespace=self.namespace, - plural="jobdefinitions", - body=jobdef_res, - ) - except ApiException as e: - raise e + kwargs = { + 'run_commands': [PYTHON_BIN], + 'run_arguments': [ENTRYPOINT_SCRIPT], + 'run_env_variables': self._build_job_env_variables('run'), + 'run_mode': 'task', + 'scale_array_spec': '0', + 'scale_max_execution_time': timeout or self.config['runtime_timeout'], + 'scale_memory_limit': self._format_memory(runtime_memory), + 'scale_cpu_limit': self._format_cpu(self.config['runtime_cpu']), + 'scale_retry_limit': 3, + } + if self._has_registry_credentials(): + kwargs['image_secret'] = REGISTRY_SECRET_NAME + + self.ce_client.create_job( + self.project_id, + name=jobdef_name, + image_reference=docker_image_name, + **kwargs, + ) logger.debug(f'Job Definition {jobdef_name} created') - return jobdef_name def get_runtime_key(self, docker_image_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. - """ self._get_or_create_namespace() jobdef_name = self._format_jobdef_name(docker_image_name, 256, version) - runtime_key = os.path.join(self.name, version, self.region, self.namespace, jobdef_name) - + runtime_key = os.path.join( + self.name, version, self.region, self.namespace, jobdef_name + ) return runtime_key def get_runtime_info(self): - """ - Method that returns all the relevant information about the runtime set - in config - """ if 'runtime' not in self.config or self.config['runtime'] == 'default': self.config['runtime'] = self._get_default_runtime_image_name() - runtime_info = { + return { 'runtime_name': self.config['runtime'], 'runtime_cpu': self.config['runtime_cpu'], 'runtime_memory': self.config['runtime_memory'], @@ -650,34 +578,22 @@ def get_runtime_info(self): 'max_workers': self.config['max_workers'], } - return runtime_info - @retry_on_except def _job_def_exists(self, jobdef_name): logger.debug(f"Checking if job definition {jobdef_name} already exists") try: - self.custom_api.get_namespaced_custom_object( - group=config.DEFAULT_GROUP, - version=config.DEFAULT_VERSION, - namespace=self.namespace, - plural="jobdefinitions", - name=jobdef_name - ) + self.ce_client.get_job(self.project_id, jobdef_name) except ApiException as e: - if e.status == 404: + if e.status_code == 404: logger.debug(f"Job definition {jobdef_name} not found (404)") return False - else: - raise e + raise e logger.debug(f"Job definition {jobdef_name} found") return True def _generate_runtime_meta(self, docker_image_name, memory): - logger.info(f"Extracting metadata from: {docker_image_name}") - jobrun_res = yaml.safe_load(config.JOBRUN_DEFAULT) - jobdef_name = self._format_jobdef_name(docker_image_name, memory) jobrun_name = 'lithops-runtime-metadata' @@ -685,87 +601,64 @@ def _generate_runtime_meta(self, docker_image_name, memory): job_payload['log_level'] = logger.getEffectiveLevel() job_payload['runtime_name'] = jobdef_name - jobrun_res['metadata']['name'] = jobrun_name - jobrun_res['metadata']['namespace'] = self.namespace - jobrun_res['spec']['jobDefinitionRef'] = str(jobdef_name) - container = jobrun_res['spec']['jobDefinitionSpec']['template']['containers'][0] - container['name'] = str(jobdef_name) - container['env'][0]['value'] = 'metadata' - config_map_name = f'lithops-{jobdef_name}-metadata' config_map_name = self._create_config_map(config_map_name, job_payload) - container['env'][1]['valueFrom']['configMapKeyRef']['name'] = config_map_name try: - self.custom_api.delete_namespaced_custom_object( - group=config.DEFAULT_GROUP, - version=config.DEFAULT_VERSION, - namespace=self.namespace, - plural="jobruns", - name=jobrun_name - ) - except Exception: + self.ce_client.delete_job_run(self.project_id, jobrun_name) + except ApiException: pass - try: - self.custom_api.create_namespaced_custom_object( - group=config.DEFAULT_GROUP, - version=config.DEFAULT_VERSION, - namespace=self.namespace, - plural="jobruns", - body=jobrun_res, - ) - except Exception: - pass + self._run_job( + jobdef_name=jobdef_name, + jobrun_name=jobrun_name, + total_workers=1, + runtime_memory=memory, + action='metadata', + payload_config_map_name=config_map_name, + ) logger.debug("Waiting for runtime metadata") - done = False failed = False failed_message = "" while not done and not failed: try: - w = watch.Watch() - for event in w.stream(self.custom_api.list_namespaced_custom_object, - namespace=self.namespace, group=config.DEFAULT_GROUP, - version=config.DEFAULT_VERSION, plural="jobruns", - field_selector=f"metadata.name={jobrun_name}", - timeout_seconds=10): - failed = int(event['object'].get('status')['failed']) - done = int(event['object'].get('status')['succeeded']) - logger.debug('...') - if failed: - try: - pod_description = self.core_api.read_namespaced_pod( - name=f'{jobrun_name}-1-0', namespace=self.namespace - ) - failed_message = pod_description.status.container_statuses[0].state.terminated.message - except Exception: - pass - if done or failed: - w.stop() + job_run = self.ce_client.get_job_run(self.project_id, jobrun_name).get_result() + status = job_run.get('status') + status_details = job_run.get('status_details') or {} + succeeded = status_details.get('succeeded') or 0 + failed_count = status_details.get('failed') or 0 + requested = status_details.get('requested') or 1 + + if status == 'completed' or succeeded >= requested: + done = True + elif status == 'failed' or failed_count > 0: + failed = True + indices_details = status_details.get('indices_details') or {} + for details in indices_details.values(): + if isinstance(details, dict): + failed_message = details.get('message') or details.get('reason') or failed_message + else: + time.sleep(2) except Exception: - pass + time.sleep(2) if done: logger.debug("Runtime metadata generated successfully") try: - self.custom_api.delete_namespaced_custom_object( - group=config.DEFAULT_GROUP, - version=config.DEFAULT_VERSION, - namespace=self.namespace, - plural="jobruns", - name=jobrun_name - ) - except Exception: + self.ce_client.delete_job_run(self.project_id, jobrun_name) + except ApiException: pass self._delete_config_map(config_map_name) if failed: - raise Exception(f"Unable to extract Python preinstalled modules from the runtime: {failed_message}") + raise Exception( + f"Unable to extract Python preinstalled modules from the runtime: {failed_message}" + ) data_key = '/'.join([JOBS_PREFIX, jobdef_name + '.meta']) json_str = self.internal_storage.get_data(key=data_key) @@ -776,35 +669,30 @@ def _generate_runtime_meta(self, docker_image_name, memory): @retry_on_except def _create_config_map(self, config_map_name, payload): - """ - Creates a configmap - """ - cmap = client.V1ConfigMap() - cmap.metadata = client.V1ObjectMeta(name=config_map_name) - cmap.data = {} - cmap.data["lithops.payload"] = utils.dict_to_b64str(payload) - + data = {'lithops.payload': utils.dict_to_b64str(payload)} logger.debug(f"Creating ConfigMap {config_map_name}") - self.core_api.create_namespaced_config_map( - namespace=self.namespace, - body=cmap, - field_manager='lithops' - ) + try: + self.ce_client.create_config_map( + self.project_id, + config_map_name, + data=data, + ) + except ApiException as e: + if e.status_code == 409: + self.ce_client.replace_config_map( + self.project_id, + config_map_name, + data=data, + ) + else: + raise e return config_map_name def _delete_config_map(self, config_map_name): - """ - Deletes a configmap - """ - grace_period_seconds = 0 try: logger.debug(f"Deleting ConfigMap {config_map_name}") - self.core_api.delete_namespaced_config_map( - name=config_map_name, - namespace=self.namespace, - grace_period_seconds=grace_period_seconds - ) + self.ce_client.delete_config_map(self.project_id, config_map_name) except ApiException as e: - logger.debug(f"Deleting a configmap failed with {e.status} {e.reason}") + logger.debug(f"Deleting a configmap failed with {e.status_code} {e.message}") diff --git a/lithops/serverless/backends/code_engine/config.py b/lithops/serverless/backends/code_engine/config.py index 4d65abefa..34f7d9fc6 100644 --- a/lithops/serverless/backends/code_engine/config.py +++ b/lithops/serverless/backends/code_engine/config.py @@ -26,17 +26,12 @@ 'docker_server': 'docker.io' } -DEFAULT_GROUP = "codeengine.cloud.ibm.com" -DEFAULT_VERSION = "v1beta1" - FH_ZIP_LOCATION = os.path.join(os.getcwd(), 'lithops_codeengine.zip') VALID_CPU_VALUES = [0.125, 0.25, 0.5, 1, 2, 4, 6, 8] VALID_MEMORY_VALUES = [256, 512, 1024, 2048, 4096, 8192, 12288, 16384, 24576, 32768] VALID_REGIONS = ['us-south', 'us-east', 'ca-tor', 'eu-de', 'eu-gb', 'jp-osa', 'jp-tok', 'br-sao', 'au-syd'] -CLUSTER_URL = 'https://proxy.{}.codeengine.cloud.ibm.com' -BASE_URL_V1 = 'https://api.{}.codeengine.cloud.ibm.com/api/v1' BASE_URL_V2 = 'https://api.{}.codeengine.cloud.ibm.com/v2' REQ_PARAMS = ('iam_api_key',) @@ -53,12 +48,9 @@ flask \ gevent \ ibm-cos-sdk \ - ibm-vpc \ - ibm-code-engine-sdk \ redis \ requests \ PyYAML \ - kubernetes \ numpy \ cloudpickle \ ps-mem \ @@ -77,74 +69,7 @@ COPY lithops_codeengine.zip . RUN unzip lithops_codeengine.zip && rm lithops_codeengine.zip -CMD exec gunicorn --bind :$PORT --workers $CONCURRENCY --timeout $TIMEOUT lithopsentry:proxy -""" - -JOBDEF_DEFAULT = """ -apiVersion: codeengine.cloud.ibm.com/v1beta1 -kind: JobDefinition -metadata: - name: lithops-runtime-name - labels: - type: lithops-runtime - version: lithops_vX.X.X -spec: - arraySpec: '0' - maxExecutionTime: 600 - retryLimit: 3 - template: - containers: - - image: "" - name: "" - command: - - "/usr/local/bin/python" - args: - - "/lithops/lithopsentry.py" - - "$(ACTION)" - - "$(PAYLOAD)" - env: - - name: ACTION - value: '' - - name: PAYLOAD - valueFrom: - configMapKeyRef: - key: 'lithops.payload' - name : NAME - resources: - requests: - cpu: '1' - memory: 128Mi - imagePullSecrets: - - name: lithops-regcred -""" - - -JOBRUN_DEFAULT = """ -apiVersion: codeengine.cloud.ibm.com/v1beta1 -kind: JobRun -metadata: - name: "" -spec: - jobDefinitionRef: "" - jobDefinitionSpec: - arraySpec: '1' - maxExecutionTime: 600 - retryLimit: 2 - template: - containers: - - name: "" - env: - - name: ACTION - value: '' - - name: PAYLOAD - valueFrom: - configMapKeyRef: - key: 'lithops.payload' - name : '' - resources: - requests: - cpu: '1' - memory: 128Mi +CMD ["sh", "-c", "exec gunicorn --bind :$PORT --workers $CONCURRENCY --timeout $TIMEOUT lithopsentry:proxy"] """ diff --git a/lithops/serverless/backends/code_engine/entry_point.py b/lithops/serverless/backends/code_engine/entry_point.py index 5fd14228c..efd58f03a 100644 --- a/lithops/serverless/backends/code_engine/entry_point.py +++ b/lithops/serverless/backends/code_engine/entry_point.py @@ -136,6 +136,8 @@ def run_ce_job(action, encoded_payload): if __name__ == '__main__': if 'JOB_INDEX' in os.environ: - run_ce_job(sys.argv[1:][0], sys.argv[1:][1]) + action = os.environ['ACTION'] + payload = os.environ['PAYLOAD'] + run_ce_job(action, payload) else: run_knative_server() From aeda12b8909c8840c4ee5921b604b202efd2c93a Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Tue, 16 Jun 2026 18:19:29 +0200 Subject: [PATCH 2/5] Update --- .../serverless/backends/code_engine/code_engine.py | 11 +++++++++-- lithops/serverless/backends/code_engine/config.py | 1 + 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/lithops/serverless/backends/code_engine/code_engine.py b/lithops/serverless/backends/code_engine/code_engine.py index 0bdf41abf..9e32ece07 100644 --- a/lithops/serverless/backends/code_engine/code_engine.py +++ b/lithops/serverless/backends/code_engine/code_engine.py @@ -634,12 +634,19 @@ def _generate_runtime_meta(self, docker_image_name, memory): if status == 'completed' or succeeded >= requested: done = True - elif status == 'failed' or failed_count > 0: + elif status == 'failed' or ( + failed_count > 0 and status not in ('pending', 'running') + ): failed = True indices_details = status_details.get('indices_details') or {} for details in indices_details.values(): if isinstance(details, dict): - failed_message = details.get('message') or details.get('reason') or failed_message + failed_message = ( + details.get('last_failure_reason') + or details.get('message') + or details.get('reason') + or failed_message + ) else: time.sleep(2) except Exception: diff --git a/lithops/serverless/backends/code_engine/config.py b/lithops/serverless/backends/code_engine/config.py index 34f7d9fc6..16a01a4f7 100644 --- a/lithops/serverless/backends/code_engine/config.py +++ b/lithops/serverless/backends/code_engine/config.py @@ -48,6 +48,7 @@ flask \ gevent \ ibm-cos-sdk \ + ibm-cloud-sdk-core \ redis \ requests \ PyYAML \ From a8a44202ea1fd1e0163c100d4173e9c5653be121 Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Tue, 16 Jun 2026 18:23:33 +0200 Subject: [PATCH 3/5] Update --- .../backends/code_engine/code_engine.py | 583 +++++++++--------- 1 file changed, 301 insertions(+), 282 deletions(-) diff --git a/lithops/serverless/backends/code_engine/code_engine.py b/lithops/serverless/backends/code_engine/code_engine.py index 9e32ece07..04be8a64c 100644 --- a/lithops/serverless/backends/code_engine/code_engine.py +++ b/lithops/serverless/backends/code_engine/code_engine.py @@ -43,6 +43,8 @@ LITHOPS_RUNTIME_TYPE = 'lithops-runtime' ENTRYPOINT_SCRIPT = '/lithops/lithopsentry.py' PYTHON_BIN = '/usr/local/bin/python' +METADATA_JOBRUN_NAME = 'lithops-runtime-metadata' +JOB_RUN_POLL_INTERVAL = 2 def retry_on_except(func): @@ -105,6 +107,10 @@ def __init__(self, ce_config, internal_storage): msg = COMPUTE_CLI_MSG.format('IBM Code Engine') logger.info(f"{msg} - Project: {self.project_name} - Region: {self.region}") + # ------------------------------------------------------------------ # + # Project / client + # ------------------------------------------------------------------ # + def _create_code_engine_client(self): if self.ce_client: return @@ -164,6 +170,10 @@ def _sync_project_config(self): self.config['namespace'] = self.namespace self._create_code_engine_client() + # ------------------------------------------------------------------ # + # Formatting helpers + # ------------------------------------------------------------------ # + @staticmethod def _format_memory(memory_mb): """ @@ -227,6 +237,179 @@ def _build_job_env_variables(self, action, payload_config_map_name=None): ) return env_variables + # ------------------------------------------------------------------ # + # CE resource helpers + # ------------------------------------------------------------------ # + + def _list_jobs(self): + try: + return self.ce_client.list_jobs(self.project_id).get_result().get('jobs', []) + except ApiException as e: + logger.debug(f"List all jobs failed with {e.status_code} {e.message}") + return [] + + def _is_lithops_job(self, job): + fn_name = job.get('name') or '' + return fn_name.startswith(f'lithops-worker-{self.user_key}') + + @staticmethod + def _read_job_env(job): + runtime_type = None + version = __version__ + for env_var in job.get('run_env_variables', []): + if env_var.get('name') == 'LITHOPS_RUNTIME_TYPE': + runtime_type = env_var.get('value') + elif env_var.get('name') == 'LITHOPS_VERSION': + version = env_var.get('value', version) + return runtime_type, version + + def _iter_lithops_runtimes(self, docker_image_name='all'): + for job in self._list_jobs(): + if not self._is_lithops_job(job): + continue + + runtime_type, version = self._read_job_env(job) + if runtime_type != LITHOPS_RUNTIME_TYPE: + continue + + image_name = job.get('image_reference') or '' + if docker_image_name != 'all' and docker_image_name not in image_name: + continue + + memory = self._parse_memory(job.get('scale_memory_limit')) + yield image_name, memory, version, job['name'] + + def _delete_job_run(self, jobrun_name): + try: + self.ce_client.delete_job_run(self.project_id, jobrun_name) + except ApiException as e: + if e.status_code != 404: + logger.debug(f"Deleting job run {jobrun_name} failed with {e.status_code} {e.message}") + + def _delete_job_definition(self, jobdef_name): + try: + self.ce_client.delete_job(self.project_id, jobdef_name) + except ApiException as e: + if e.status_code != 404: + logger.debug(f"Deleting job {jobdef_name} failed with {e.status_code} {e.message}") + + def _delete_config_map(self, config_map_name): + try: + logger.debug(f"Deleting ConfigMap {config_map_name}") + self.ce_client.delete_config_map(self.project_id, config_map_name) + except ApiException as e: + logger.debug(f"Deleting config map {config_map_name} failed with {e.status_code} {e.message}") + + def _delete_lithops_config_maps(self): + try: + configmaps = self.ce_client.list_config_maps(self.project_id).get_result() + except ApiException as e: + logger.debug(f"Listing config maps failed with {e.status_code} {e.message}") + return + + lithops_configmaps = [ + configmap['name'] + for configmap in configmaps.get('config_maps', []) + if configmap.get('name', '').startswith('lithops') + ] + if not lithops_configmaps: + return + + logger.debug(f'Deleting {len(lithops_configmaps)} leftover lithops config map(s)') + for config_name in lithops_configmaps: + self._delete_config_map(config_name) + + @staticmethod + def _job_run_failure_message(status_details): + for details in (status_details.get('indices_details') or {}).values(): + if not isinstance(details, dict): + continue + message = ( + details.get('last_failure_reason') + or details.get('message') + or details.get('reason') + ) + if message: + return message + return '' + + @staticmethod + def _job_run_finished(job_run): + status = job_run.get('status') + status_details = job_run.get('status_details') or {} + succeeded = status_details.get('succeeded') or 0 + failed_count = status_details.get('failed') or 0 + requested = status_details.get('requested') or 1 + + if status == 'completed' or succeeded >= requested: + return 'completed', status_details + if status == 'failed' or (failed_count > 0 and status not in ('pending', 'running')): + return 'failed', status_details + return 'running', status_details + + def _wait_for_job_run(self, jobrun_name): + logger.debug(f"Waiting for job run {jobrun_name}") + while True: + try: + job_run = self.ce_client.get_job_run(self.project_id, jobrun_name).get_result() + except ApiException as e: + logger.debug(f"Polling job run {jobrun_name} failed with {e.status_code} {e.message}") + time.sleep(JOB_RUN_POLL_INTERVAL) + continue + + state, status_details = self._job_run_finished(job_run) + if state == 'completed': + logger.debug(f"Job run {jobrun_name} completed") + return + if state == 'failed': + reason = self._job_run_failure_message(status_details) + raise Exception( + f"Job run {jobrun_name} failed" + + (f": {reason}" if reason else '') + ) + time.sleep(JOB_RUN_POLL_INTERVAL) + + @retry_on_except + def _create_config_map(self, config_map_name, payload): + data = {'lithops.payload': utils.dict_to_b64str(payload)} + logger.debug(f"Creating ConfigMap {config_map_name}") + + try: + self.ce_client.create_config_map( + self.project_id, + config_map_name, + data=data, + ) + except ApiException as e: + if e.status_code == 409: + self.ce_client.replace_config_map( + self.project_id, + config_map_name, + data=data, + ) + else: + raise e + + return config_map_name + + @retry_on_except + def _job_def_exists(self, jobdef_name): + logger.debug(f"Checking if job definition {jobdef_name} already exists") + try: + self.ce_client.get_job(self.project_id, jobdef_name) + except ApiException as e: + if e.status_code == 404: + logger.debug(f"Job definition {jobdef_name} not found (404)") + return False + raise e + + logger.debug(f"Job definition {jobdef_name} found") + return True + + # ------------------------------------------------------------------ # + # Runtime build / deploy + # ------------------------------------------------------------------ # + def build_runtime(self, docker_image_name, dockerfile, extra_args=[]): logger.info(f'Building runtime {docker_image_name} from {dockerfile}') @@ -289,43 +472,129 @@ def deploy_runtime(self, docker_image_name, memory, timeout): logger.debug(f"Deploying runtime: {docker_image_name} - Memory: {memory} Timeout: {timeout}") self._create_job_definition(docker_image_name, memory, timeout) - runtime_meta = self._generate_runtime_meta(docker_image_name, memory) + return self._generate_runtime_meta(docker_image_name, memory) + + def _generate_runtime_meta(self, docker_image_name, memory): + logger.info(f"Extracting metadata from: {docker_image_name}") + jobdef_name = self._format_jobdef_name(docker_image_name, memory) + + job_payload = copy.deepcopy(self.internal_storage.storage.config) + job_payload['log_level'] = logger.getEffectiveLevel() + job_payload['runtime_name'] = jobdef_name + + config_map_name = self._create_config_map(f'lithops-{jobdef_name}-metadata', job_payload) + try: + self._delete_job_run(METADATA_JOBRUN_NAME) + self._run_job( + jobdef_name=jobdef_name, + jobrun_name=METADATA_JOBRUN_NAME, + total_workers=1, + runtime_memory=memory, + action='metadata', + payload_config_map_name=config_map_name, + ) + self._wait_for_job_run(METADATA_JOBRUN_NAME) + except Exception as e: + raise Exception( + f"Unable to extract Python preinstalled modules from the runtime: {e}" + ) from e + finally: + self._delete_job_run(METADATA_JOBRUN_NAME) + self._delete_config_map(config_map_name) + + data_key = '/'.join([JOBS_PREFIX, jobdef_name + '.meta']) + json_str = self.internal_storage.get_data(key=data_key) + runtime_meta = json.loads(json_str.decode("ascii")) + self.internal_storage.del_data(key=data_key) return runtime_meta - def _resolve_jobdef_name(self, runtime_name, memory, version=__version__): - jobdef_name = self._format_jobdef_name(runtime_name, memory, version) - if self._job_def_exists(jobdef_name): - return jobdef_name + def _create_container_registry_secret(self): + if not self._has_registry_credentials(): + return - logger.debug( - f"Job definition {jobdef_name} not found, searching by image {runtime_name}" + logger.debug('Creating container registry secret') + secret_data = SecretDataRegistrySecretData( + server=self.config['docker_server'], + username=self.config['docker_user'], + password=self.config['docker_password'], ) + try: - jobs = self.ce_client.list_jobs(self.project_id).get_result() + self.ce_client.delete_secret(self.project_id, REGISTRY_SECRET_NAME) except ApiException as e: - logger.debug(f"List all jobs failed with {e.status_code} {e.message}") - return jobdef_name + if e.status_code != 404: + raise e - for job in jobs.get('jobs', []): - fn_name = job.get('name') - if not fn_name or not fn_name.startswith(f'lithops-worker-{self.user_key}'): - continue + try: + self.ce_client.create_secret( + self.project_id, + format='registry', + name=REGISTRY_SECRET_NAME, + data=secret_data, + ) + except ApiException as e: + if e.status_code != 409: + raise e - image_name = job.get('image_reference') or '' - if runtime_name not in image_name and image_name not in runtime_name: - continue + @retry_on_except + def _create_job_definition(self, docker_image_name, runtime_memory, timeout=None): + self._create_container_registry_secret() + + jobdef_name = self._format_jobdef_name(docker_image_name, runtime_memory) + logger.debug(f"Creating job definition {jobdef_name}") + + try: + self.ce_client.delete_job(self.project_id, jobdef_name) + except ApiException as e: + if e.status_code != 404: + raise e + + while self._job_def_exists(jobdef_name): + time.sleep(1) + + kwargs = { + 'run_commands': [PYTHON_BIN], + 'run_arguments': [ENTRYPOINT_SCRIPT], + 'run_env_variables': self._build_job_env_variables('run'), + 'run_mode': 'task', + 'scale_array_spec': '0', + 'scale_max_execution_time': timeout or self.config['runtime_timeout'], + 'scale_memory_limit': self._format_memory(runtime_memory), + 'scale_cpu_limit': self._format_cpu(self.config['runtime_cpu']), + 'scale_retry_limit': 3, + } + if self._has_registry_credentials(): + kwargs['image_secret'] = REGISTRY_SECRET_NAME + + self.ce_client.create_job( + self.project_id, + name=jobdef_name, + image_reference=docker_image_name, + **kwargs, + ) + + logger.debug(f'Job Definition {jobdef_name} created') + return jobdef_name - job_version = __version__ - for env_var in job.get('run_env_variables', []): - if env_var.get('name') == 'LITHOPS_VERSION': - job_version = env_var.get('value', job_version) + # ------------------------------------------------------------------ # + # Lifecycle + # ------------------------------------------------------------------ # + def _resolve_jobdef_name(self, runtime_name, memory, version=__version__): + jobdef_name = self._format_jobdef_name(runtime_name, memory, version) + if self._job_def_exists(jobdef_name): + return jobdef_name + + logger.debug( + f"Job definition {jobdef_name} not found, searching by image {runtime_name}" + ) + for image_name, _, job_version, fn_name in self._iter_lithops_runtimes('all'): if job_version != version: continue - - logger.debug(f"Resolved job definition {fn_name} for runtime {runtime_name}") - return fn_name + if runtime_name in image_name or image_name in runtime_name: + logger.debug(f"Resolved job definition {fn_name} for runtime {runtime_name}") + return fn_name return jobdef_name @@ -338,10 +607,7 @@ def delete_runtime(self, runtime_name, memory, version=__version__, jobdef_name= jobdef_name = self._resolve_jobdef_name(runtime_name, memory, version) logger.info(f'Deleting runtime: {runtime_name} - {memory}MB') - try: - self.ce_client.delete_job(self.project_id, jobdef_name) - except ApiException as e: - logger.debug(f"Deleting a job failed with {e.status_code} {e.message}") + self._delete_job_definition(jobdef_name) def clean(self, all=False): logger.info(f'Cleaning project {self.project_name}') @@ -352,23 +618,10 @@ def clean(self, all=False): return self.clear() - runtimes = self.list_runtimes() - for image_name, memory, version, jobdef_name in runtimes: + for image_name, memory, version, jobdef_name in self.list_runtimes(): self.delete_runtime(image_name, memory, version, jobdef_name=jobdef_name) - logger.debug('Deleting all lithops config maps') - try: - configmaps = self.ce_client.list_config_maps(self.project_id).get_result() - for configmap in configmaps.get('config_maps', []): - config_name = configmap.get('name') - if config_name and config_name.startswith('lithops'): - logger.debug(f'Deleting config map {config_name}') - try: - self.ce_client.delete_config_map(self.project_id, config_name) - except ApiException as e: - logger.debug(f"Deleting config map failed with {e.status_code} {e.message}") - except ApiException as e: - logger.debug(f"Listing config maps failed with {e.status_code} {e.message}") + self._delete_lithops_config_maps() if all and os.path.exists(self.cache_file): logger.info(f"Deleting Code Engine project: {self.project_name}") @@ -376,57 +629,21 @@ def clean(self, all=False): os.remove(self.cache_file) def list_runtimes(self, docker_image_name='all'): - runtimes = [] - if not self._get_or_create_namespace(create=False): logger.info(f"Project {self.project_name} does not exist") - return runtimes - - try: - jobs = self.ce_client.list_jobs(self.project_id).get_result() - except ApiException as e: - logger.debug(f"List all jobs failed with {e.status_code} {e.message}") - return runtimes - - for job in jobs.get('jobs', []): - try: - fn_name = job.get('name') - if not fn_name or not fn_name.startswith(f'lithops-worker-{self.user_key}'): - continue - - runtime_type = None - version = __version__ - for env_var in job.get('run_env_variables', []): - if env_var.get('name') == 'LITHOPS_RUNTIME_TYPE': - runtime_type = env_var.get('value') - elif env_var.get('name') == 'LITHOPS_VERSION': - version = env_var.get('value', version) - - if runtime_type != LITHOPS_RUNTIME_TYPE: - continue - - image_name = job.get('image_reference') - memory = self._parse_memory(job.get('scale_memory_limit')) - if docker_image_name in image_name or docker_image_name == 'all': - runtimes.append((image_name, memory, version, fn_name)) - except Exception: - pass + return [] - return runtimes + return list(self._iter_lithops_runtimes(docker_image_name)) def clear(self, job_keys=None): if not self._get_or_create_namespace(create=False): logger.info(f"Project {self.project_name} does not exist") return - jobs_to_delete = job_keys or self.jobs - for job_key in jobs_to_delete: + for job_key in job_keys or self.jobs: jobrun_name = f'lithops-{job_key.lower()}' - try: - self.ce_client.delete_job_run(self.project_id, jobrun_name) - self._delete_config_map(jobrun_name) - except ApiException as e: - logger.debug(f"Deleting a job run failed with {e.status_code} {e.message}") + self._delete_job_run(jobrun_name) + self._delete_config_map(jobrun_name) try: self.jobs.remove(job_key) except ValueError: @@ -486,78 +703,6 @@ def _run_job(self, jobdef_name, jobrun_name, total_workers, runtime_memory, run_env_variables=self._build_job_env_variables(action, payload_config_map_name), ) - def _create_container_registry_secret(self): - if not self._has_registry_credentials(): - return - - logger.debug('Creating container registry secret') - docker_server = self.config['docker_server'] - docker_user = self.config['docker_user'] - docker_password = self.config['docker_password'] - - secret_data = SecretDataRegistrySecretData( - server=docker_server, - username=docker_user, - password=docker_password, - ) - - try: - self.ce_client.delete_secret(self.project_id, REGISTRY_SECRET_NAME) - except ApiException as e: - if e.status_code != 404: - raise e - - try: - self.ce_client.create_secret( - self.project_id, - format='registry', - name=REGISTRY_SECRET_NAME, - data=secret_data, - ) - except ApiException as e: - if e.status_code != 409: - raise e - - @retry_on_except - def _create_job_definition(self, docker_image_name, runtime_memory, timeout=None): - self._create_container_registry_secret() - - jobdef_name = self._format_jobdef_name(docker_image_name, runtime_memory) - logger.debug(f"Creating job definition {jobdef_name}") - - try: - self.ce_client.delete_job(self.project_id, jobdef_name) - except ApiException as e: - if e.status_code != 404: - raise e - - while self._job_def_exists(jobdef_name): - time.sleep(1) - - kwargs = { - 'run_commands': [PYTHON_BIN], - 'run_arguments': [ENTRYPOINT_SCRIPT], - 'run_env_variables': self._build_job_env_variables('run'), - 'run_mode': 'task', - 'scale_array_spec': '0', - 'scale_max_execution_time': timeout or self.config['runtime_timeout'], - 'scale_memory_limit': self._format_memory(runtime_memory), - 'scale_cpu_limit': self._format_cpu(self.config['runtime_cpu']), - 'scale_retry_limit': 3, - } - if self._has_registry_credentials(): - kwargs['image_secret'] = REGISTRY_SECRET_NAME - - self.ce_client.create_job( - self.project_id, - name=jobdef_name, - image_reference=docker_image_name, - **kwargs, - ) - - logger.debug(f'Job Definition {jobdef_name} created') - return jobdef_name - def get_runtime_key(self, docker_image_name, runtime_memory, version=__version__): self._get_or_create_namespace() jobdef_name = self._format_jobdef_name(docker_image_name, 256, version) @@ -577,129 +722,3 @@ def get_runtime_info(self): 'runtime_timeout': self.config['runtime_timeout'], 'max_workers': self.config['max_workers'], } - - @retry_on_except - def _job_def_exists(self, jobdef_name): - logger.debug(f"Checking if job definition {jobdef_name} already exists") - try: - self.ce_client.get_job(self.project_id, jobdef_name) - except ApiException as e: - if e.status_code == 404: - logger.debug(f"Job definition {jobdef_name} not found (404)") - return False - raise e - - logger.debug(f"Job definition {jobdef_name} found") - return True - - def _generate_runtime_meta(self, docker_image_name, memory): - logger.info(f"Extracting metadata from: {docker_image_name}") - jobdef_name = self._format_jobdef_name(docker_image_name, memory) - jobrun_name = 'lithops-runtime-metadata' - - job_payload = copy.deepcopy(self.internal_storage.storage.config) - job_payload['log_level'] = logger.getEffectiveLevel() - job_payload['runtime_name'] = jobdef_name - - config_map_name = f'lithops-{jobdef_name}-metadata' - config_map_name = self._create_config_map(config_map_name, job_payload) - - try: - self.ce_client.delete_job_run(self.project_id, jobrun_name) - except ApiException: - pass - - self._run_job( - jobdef_name=jobdef_name, - jobrun_name=jobrun_name, - total_workers=1, - runtime_memory=memory, - action='metadata', - payload_config_map_name=config_map_name, - ) - - logger.debug("Waiting for runtime metadata") - done = False - failed = False - failed_message = "" - - while not done and not failed: - try: - job_run = self.ce_client.get_job_run(self.project_id, jobrun_name).get_result() - status = job_run.get('status') - status_details = job_run.get('status_details') or {} - succeeded = status_details.get('succeeded') or 0 - failed_count = status_details.get('failed') or 0 - requested = status_details.get('requested') or 1 - - if status == 'completed' or succeeded >= requested: - done = True - elif status == 'failed' or ( - failed_count > 0 and status not in ('pending', 'running') - ): - failed = True - indices_details = status_details.get('indices_details') or {} - for details in indices_details.values(): - if isinstance(details, dict): - failed_message = ( - details.get('last_failure_reason') - or details.get('message') - or details.get('reason') - or failed_message - ) - else: - time.sleep(2) - except Exception: - time.sleep(2) - - if done: - logger.debug("Runtime metadata generated successfully") - - try: - self.ce_client.delete_job_run(self.project_id, jobrun_name) - except ApiException: - pass - - self._delete_config_map(config_map_name) - - if failed: - raise Exception( - f"Unable to extract Python preinstalled modules from the runtime: {failed_message}" - ) - - data_key = '/'.join([JOBS_PREFIX, jobdef_name + '.meta']) - json_str = self.internal_storage.get_data(key=data_key) - runtime_meta = json.loads(json_str.decode("ascii")) - self.internal_storage.del_data(key=data_key) - - return runtime_meta - - @retry_on_except - def _create_config_map(self, config_map_name, payload): - data = {'lithops.payload': utils.dict_to_b64str(payload)} - logger.debug(f"Creating ConfigMap {config_map_name}") - - try: - self.ce_client.create_config_map( - self.project_id, - config_map_name, - data=data, - ) - except ApiException as e: - if e.status_code == 409: - self.ce_client.replace_config_map( - self.project_id, - config_map_name, - data=data, - ) - else: - raise e - - return config_map_name - - def _delete_config_map(self, config_map_name): - try: - logger.debug(f"Deleting ConfigMap {config_map_name}") - self.ce_client.delete_config_map(self.project_id, config_map_name) - except ApiException as e: - logger.debug(f"Deleting a configmap failed with {e.status_code} {e.message}") From dd27b1e324e98e0403d0094d7a9404463863772c Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Tue, 16 Jun 2026 18:24:55 +0200 Subject: [PATCH 4/5] Update config --- .../serverless/backends/code_engine/config.py | 60 +++++++++++++++---- 1 file changed, 48 insertions(+), 12 deletions(-) diff --git a/lithops/serverless/backends/code_engine/config.py b/lithops/serverless/backends/code_engine/config.py index 16a01a4f7..04da239ee 100644 --- a/lithops/serverless/backends/code_engine/config.py +++ b/lithops/serverless/backends/code_engine/config.py @@ -28,14 +28,33 @@ FH_ZIP_LOCATION = os.path.join(os.getcwd(), 'lithops_codeengine.zip') +# https://cloud.ibm.com/docs/codeengine?topic=codeengine-mem-cpu-combo VALID_CPU_VALUES = [0.125, 0.25, 0.5, 1, 2, 4, 6, 8] VALID_MEMORY_VALUES = [256, 512, 1024, 2048, 4096, 8192, 12288, 16384, 24576, 32768] -VALID_REGIONS = ['us-south', 'us-east', 'ca-tor', 'eu-de', 'eu-gb', 'jp-osa', 'jp-tok', 'br-sao', 'au-syd'] +VALID_CPU_MEMORY = { + 0.125: (256, 8192), + 0.25: (512, 16384), + 0.5: (1024, 32768), + 1: (4096, 32768), + 2: (4096, 32768), + 4: (4096, 32768), + 6: (4096, 32768), + 8: (4096, 32768), +} + +# https://cloud.ibm.com/docs/codeengine?topic=codeengine-regions +VALID_REGIONS = [ + 'us-south', 'us-east', 'ca-tor', 'eu-de', 'eu-gb', 'eu-es', + 'jp-osa', 'jp-tok', 'br-sao', 'au-syd', +] BASE_URL_V2 = 'https://api.{}.codeengine.cloud.ibm.com/v2' -REQ_PARAMS = ('iam_api_key',) +REQ_PARAMS = ('iam_api_key', 'resource_group_id') +# Default runtime image for auto-built runtimes. +# CE jobs override the container command; the gunicorn CMD is kept for custom +# runtimes that start the container without JOB_INDEX (legacy Knative path). DOCKERFILE_DEFAULT = """ RUN apt-get update && apt-get install -y \ zip \ @@ -44,7 +63,6 @@ RUN pip install --upgrade --ignore-installed setuptools six pip \ && pip install --upgrade --no-cache-dir --ignore-installed \ gunicorn \ - pika \ flask \ gevent \ ibm-cos-sdk \ @@ -63,7 +81,6 @@ ENV TIMEOUT=600 ENV PYTHONUNBUFFERED=TRUE -# Copy Lithops proxy and lib to the container image. ENV APP_HOME=/lithops WORKDIR $APP_HOME @@ -74,6 +91,16 @@ """ +def _validate_cpu_memory(runtime_cpu, runtime_memory): + min_memory, max_memory = VALID_CPU_MEMORY[runtime_cpu] + if not min_memory <= runtime_memory <= max_memory: + raise Exception( + f'{runtime_memory} MB is not a valid memory value for {runtime_cpu} vCPU. ' + f'Use a value between {min_memory} and {max_memory} MB. ' + 'See https://cloud.ibm.com/docs/codeengine?topic=codeengine-mem-cpu-combo' + ) + + def load_config(config_data): if 'ibm' not in config_data or config_data['ibm'] is None: @@ -84,7 +111,7 @@ def load_config(config_data): msg = f'"{param}" is mandatory in the "ibm" section of the configuration' raise Exception(msg) - if not config_data['code_engine']: + if not config_data.get('code_engine'): config_data['code_engine'] = {} temp = copy.deepcopy(config_data['code_engine']) @@ -92,13 +119,17 @@ def load_config(config_data): config_data['code_engine'].update(temp) if 'region' not in config_data['code_engine']: - msg = "'region' parameter is mandatory under the 'ibm' or 'code_engine' section of the configuration" + msg = ( + "'region' parameter is mandatory under the 'ibm' or " + "'code_engine' section of the configuration" + ) raise Exception(msg) region = config_data['code_engine']['region'] if region not in VALID_REGIONS: - raise Exception('{} is an invalid region name. Set one of: ' - '{}'.format(region, VALID_REGIONS)) + raise Exception( + f'{region} is an invalid region name. Set one of: {VALID_REGIONS}' + ) for key in DEFAULT_CONFIG_KEYS: if key not in config_data['code_engine']: @@ -106,13 +137,18 @@ def load_config(config_data): runtime_cpu = config_data['code_engine']['runtime_cpu'] if runtime_cpu not in VALID_CPU_VALUES: - raise Exception('{} is an invalid runtime cpu value. Set one of: ' - '{}'.format(runtime_cpu, VALID_CPU_VALUES)) + raise Exception( + f'{runtime_cpu} is an invalid runtime cpu value. Set one of: {VALID_CPU_VALUES}' + ) runtime_memory = config_data['code_engine']['runtime_memory'] if runtime_memory not in VALID_MEMORY_VALUES: - raise Exception('{} is an invalid runtime memory value in MB. Set one of: ' - '{}'.format(runtime_memory, VALID_MEMORY_VALUES)) + raise Exception( + f'{runtime_memory} is an invalid runtime memory value in MB. ' + f'Set one of: {VALID_MEMORY_VALUES}' + ) + + _validate_cpu_memory(runtime_cpu, runtime_memory) if 'runtime' in config_data['code_engine']: runtime = config_data['code_engine']['runtime'] From 46ff49613d7892ba8d73c881d1e536eb775486a0 Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Thu, 18 Jun 2026 15:15:03 +0200 Subject: [PATCH 5/5] Update --- config/config_template.yaml | 32 ++-- docs/source/compute_config/code_engine.md | 32 +++- docs/source/storage_config/ibm_cos.md | 78 ++++---- .../backends/code_engine/code_engine.py | 178 +++++++++++++----- .../serverless/backends/code_engine/config.py | 17 ++ runtime/code_engine/Dockerfile | 16 +- runtime/code_engine/requirements.txt | 11 +- 7 files changed, 246 insertions(+), 118 deletions(-) diff --git a/config/config_template.yaml b/config/config_template.yaml index 088faf801..19b958d16 100644 --- a/config/config_template.yaml +++ b/config/config_template.yaml @@ -259,9 +259,9 @@ # IBM Cloud – credentials shared by all ibm_* / code_engine sections # ============================================================================= #ibm: - #iam_api_key: - #region: # One of: eu-gb, eu-de, us-south, us-east, br-sao, ca-tor, jp-tok, jp-osa, au-syd - #resource_group_id: + #iam_api_key: # Mandatory for code_engine + #region: # One of: eu-gb, eu-de, eu-es, us-south, us-east, br-sao, ca-tor, jp-tok, jp-osa, au-syd + #resource_group_id: # Mandatory for code_engine (auto-create project) # IBM Cloud Functions – DEPRECATED serverless compute #ibm_cf: @@ -283,20 +283,17 @@ #code_engine: #region: # Falls back to ibm.region #project_name: # Existing Code Engine project (auto-created otherwise) - #namespace: # Alternative to project_name - #kubecfg_path: #docker_server: docker.io - #docker_user: + #docker_user: # Required for private container registries #docker_password: - #docker_namespace: - #runtime: - #runtime_cpu: 0.125 # See valid combinations in the IBM docs - #runtime_memory: 256 # MB + #docker_namespace: # Required for IBM Container Registry (icr.io) + #runtime: # Docker image name. Default: auto-built image + #runtime_cpu: 0.125 # vCPU. See valid combinations in the IBM docs + #runtime_memory: 256 # MB. See valid combinations in the IBM docs #runtime_timeout: 600 # Seconds #max_workers: 1000 #worker_processes: 1 - #connection_retries: - #runtime_include_function: False + #connection_retries: # API retries on HTTP 500 errors # IBM VPC – standalone compute #ibm_vpc: @@ -331,12 +328,13 @@ # IBM Cloud Object Storage #ibm_cos: - #region: # Or set `endpoint` directly + #region: # Recommended. Auto-sets endpoint and worker endpoint #endpoint: # https://s3..cloud-object-storage.appdomain.cloud - #private_endpoint: # Required for ibm_cf, code_engine, ibm_vpc compute backends - #api_key: # COS service API key - #service_instance_id: - #access_key_id: # HMAC credentials (alternative to api_key) + #private_endpoint: # Auto-set from region when backend is code_engine or ibm_vpc + # # https://s3.direct..cloud-object-storage.appdomain.cloud + #api_key: # COS service API key (Option 1) + #service_instance_id: # Required if not using HMAC credentials or api_key + #access_key_id: # HMAC credentials (Option 2). Bucket can be auto-created #secret_access_key: #storage_bucket: # Mandatory unless HMAC credentials allow auto-creation diff --git a/docs/source/compute_config/code_engine.md b/docs/source/compute_config/code_engine.md index 5c7d4a578..717316bd9 100644 --- a/docs/source/compute_config/code_engine.md +++ b/docs/source/compute_config/code_engine.md @@ -16,7 +16,7 @@ python3 -m pip install lithops[ibm] 2. Click `Create an IBM Cloud API Key` and provide the necessary information. -3. Copy the generated IAM API key (You can only see the key the first time you create it, so make sure to copy it). +3. Copy the generated IAM API key (you can only see the key the first time you create it, so make sure to copy it). 4. Navigate to the [resource groups dashboard](https://cloud.ibm.com/account/resource-groups), and copy the desired resource group ID. @@ -40,27 +40,25 @@ python3 -m pip install lithops[ibm] |Group|Key|Default|Mandatory|Additional info| |---|---|---|---|---| |ibm | iam_api_key | |yes | IBM Cloud IAM API key to authenticate against IBM services. Obtain the key [here](https://cloud.ibm.com/iam/apikeys) | -|ibm | region | |yes | IBM Region. One of: `eu-gb`, `eu-de`, `us-south`, `us-east`, `br-sao`, `ca-tor`, `jp-tok`, `jp-osa`, `au-syd` | -|ibm | resource_group_id | | yes | Resource group id from your IBM Cloud account. Get it from [here](https://cloud.ibm.com/account/resource-groups) | +|ibm | region | |yes | IBM Region. One of: `eu-gb`, `eu-de`, `eu-es`, `us-south`, `us-east`, `br-sao`, `ca-tor`, `jp-tok`, `jp-osa`, `au-syd` | +|ibm | resource_group_id | | yes | Resource group id from your IBM Cloud account. Get it from [here](https://cloud.ibm.com/account/resource-groups). Required to auto-create a Code Engine project | ### Code Engine: |Group|Key|Default|Mandatory|Additional info| |---|---|---|---|---| -|code_engine | project_name | |no | Project name that already exists in Code Engine. If not provided, Lithops will automatically create a new project| -|code_engine | namespace | |no | As an alternative to `project_name`, you can provide `namespace`. Get it from your Code Engine k8s config file.| -|code_engine | region | | no | Cluster region. One of: `eu-gb`, `eu-de`, `us-south`, `us-east`, `br-sao`, `ca-tor`, `jp-tok`, `jp-osa`, `au-syd`. Lithops will use the `region` set under the `ibm` section if it is not set here | +|code_engine | project_name | |no | Name of an existing Code Engine project. If not provided, Lithops creates `lithops--` automatically | +|code_engine | region | | no | Code Engine region. One of: `eu-gb`, `eu-de`, `eu-es`, `us-south`, `us-east`, `br-sao`, `ca-tor`, `jp-tok`, `jp-osa`, `au-syd`. Lithops uses the `region` set under the `ibm` section if not set here | |code_engine | docker_server | docker.io |no | Container registry URL | |code_engine | docker_user | |no | Container registry user name | |code_engine | docker_password | |no | Container registry password/token. For Docker Hub, log in to your Docker Hub account and generate a new access token [here](https://hub.docker.com/settings/security)| |code_engine | max_workers | 1000 | no | Max number of workers per `FunctionExecutor()`| |code_engine | worker_processes | 1 | no | Number of Lithops processes within a given worker. This can be used to parallelize function activations within a worker. It is recommended to set this value to the same number of CPUs as the container. | |code_engine | runtime | |no | Docker image name.| -|code_engine | runtime_cpu | 0.125 |no | CPU limit. Default 0.125vCPU. See [valid combinations](https://cloud.ibm.com/docs/codeengine?topic=codeengine-mem-cpu-combo) | -|code_engine | runtime_memory | 256 |no | Memory limit in MB. Default 256Mi. See [valid combinations](https://cloud.ibm.com/docs/codeengine?topic=codeengine-mem-cpu-combo) | +|code_engine | runtime_cpu | 0.125 |no | CPU limit. Default 0.125 vCPU. See [valid combinations](https://cloud.ibm.com/docs/codeengine?topic=codeengine-mem-cpu-combo) | +|code_engine | runtime_memory | 256 |no | Memory limit in MB. Default 256 MB. See [valid combinations](https://cloud.ibm.com/docs/codeengine?topic=codeengine-mem-cpu-combo) | |code_engine | runtime_timeout | 600 |no | Runtime timeout in seconds. Default 600 seconds | -|code_engine | connection_retries | |no | If specified, number of job invoke retries in case of connection failure with error code 500 | -|code_engine | runtime_include_function | False | no | If set to true, Lithops will automatically build a new runtime, including the function's code, instead of transferring it through the storage backend at invocation time. This is useful when the function's code size is large (on the order of tens of MB) and the code does not change frequently | +|code_engine | connection_retries | |no | If specified, number of API call retries in case of connection failure with error code 500 | ## Runtime @@ -121,3 +119,17 @@ You can view the function executions logs in your local machine using the *litho ```bash lithops logs poll ``` + +## Clean up resources + +Delete Lithops runtimes, job runs, and cached metadata: + +```bash +lithops clean -b code_engine -s ibm_cos +``` + +To also delete the Code Engine project and local cache: + +```bash +lithops clean -b code_engine -s ibm_cos --all +``` diff --git a/docs/source/storage_config/ibm_cos.md b/docs/source/storage_config/ibm_cos.md index a8648202b..5b81d9243 100644 --- a/docs/source/storage_config/ibm_cos.md +++ b/docs/source/storage_config/ibm_cos.md @@ -16,19 +16,19 @@ python3 -m pip install lithops[ibm] ## Configuration -1. Create the credentials to access to your COS account (Choose one option): - +Choose one authentication option below. + ### Option 1 (COS API Key): -2. In the side navigation, click `Service Credentials`. +1. In the side navigation, click `Service Credentials`. -3. Click `New credential +` and provide the necessary information. +2. Click `New credential +` and provide the necessary information. -4. Click `Add` to generate service credential. +3. Click `Add` to generate service credential. -5. Click `View credentials` and copy the *apikey* value. +4. Click `View credentials` and copy the *apikey* value. -6. Edit your Lithops config file and add the following keys: +5. Edit your Lithops config file and add the following keys: ```yaml lithops: @@ -42,19 +42,19 @@ python3 -m pip install lithops[ibm] ### Option 2 (COS HMAC credentials): -2. In the side navigation, click `Service Credentials`. +1. In the side navigation, click `Service Credentials`. -3. Click `New credential +`. +2. Click `New credential +`. -4. Click on advanced options and enable `Include HMAC Credential` button. +3. Click on advanced options and enable `Include HMAC Credential` button. -5. Click `Add` to generate service credential. +4. Click `Add` to generate service credential. -6. Click `View credentials` and copy the *access_key_id* and *secret_access_key* values. +5. Click `View credentials` and copy the *access_key_id* and *secret_access_key* values. -7. When using HMAC credentials, you can omit providing a storage bucket, since Lithops will be able to create it automatically. +6. When using HMAC credentials, you can omit providing a storage bucket, since Lithops will be able to create it automatically. -8. Edit your Lithops config file and add the following keys: +7. Edit your Lithops config file and add the following keys: ```yaml lithops: @@ -63,19 +63,19 @@ python3 -m pip install lithops[ibm] ibm_cos: region: access_key_id: - secret_access_key: + secret_access_key: ``` ### Option 3 (IBM IAM API Key): -2. If you don't have an IAM API key created, navigate to the [IBM IAM dashboard](https://cloud.ibm.com/iam/apikeys) +1. If you don't have an IAM API key created, navigate to the [IBM IAM dashboard](https://cloud.ibm.com/iam/apikeys) -3. Click `Create an IBM Cloud API Key` and provide the necessary information. +2. Click `Create an IBM Cloud API Key` and provide the necessary information. -4. Copy the generated IAM API key (You can only see the key the first time you create it, so make sure to copy it). +3. Copy the generated IAM API key (you can only see the key the first time you create it, so make sure to copy it). -5. Edit your Lithops config file and add the following keys: +4. Edit your Lithops config file and add the following keys: ```yaml lithops: @@ -91,30 +91,40 @@ python3 -m pip install lithops[ibm] ## Lithops COS Endpoint configuration -### Using region -The easiest approach is to let Lithops choose the right endpoint by itself. To enable this, just configure Lithops with the region name of your `storage_bucket`, as follows: +### Using region (recommended) + +The easiest approach is to let Lithops choose the right endpoint by itself. Configure Lithops with the region name of your `storage_bucket`: ```yaml ibm_cos: region : ``` -Valid region names are: `us-east`, `us-south`, `eu-gb`, `eu-de`, etc. +Valid region names include: `us-east`, `us-south`, `eu-gb`, `eu-de`, `eu-es`, `ca-tor`, `br-sao`, `jp-tok`, `jp-osa`, `au-syd`. + +When `region` is set, Lithops automatically configures: + +- **Public endpoint** (client): `https://s3..cloud-object-storage.appdomain.cloud` +- **Worker endpoint** (`code_engine` and `ibm_vpc`): `https://s3.direct..cloud-object-storage.appdomain.cloud` + +You do not need to set `endpoint` or `private_endpoint` manually when using `region` with these backends. ### Using endpoint paths -As an alternative to using `region`, you can configure the public and private endpoints as follows: +As an alternative to using `region`, you can configure the public and worker endpoints explicitly: 1. Login to IBM Cloud and open up your dashboard. Then navigate to your instance of Object Storage. -2. In the side navigation, click `Endpoints` to find your COS endpoints. You must copy both the `public` and `private` endpoints of the region where you created your bucket. +2. In the side navigation, click `Endpoints` to find your COS endpoints. Copy the endpoint for the region where you created your bucket. ```yaml ibm_cos: endpoint: https://s3..cloud-object-storage.appdomain.cloud - private_endpoint: https://s3.private..cloud-object-storage.appdomain.cloud + private_endpoint: https://s3.direct..cloud-object-storage.appdomain.cloud ``` +Use the **direct** endpoint (`s3.direct`) as `private_endpoint` when running Lithops with `code_engine` or `ibm_vpc`. + ## Summary of configuration keys for IBM Cloud: @@ -123,18 +133,18 @@ As an alternative to using `region`, you can configure the public and private en |Group|Key|Default|Mandatory|Additional info| |---|---|---|---|---| |ibm | iam_api_key | |no | IBM Cloud IAM API key to authenticate against IBM services. Obtain the key [here](https://cloud.ibm.com/iam/apikeys) | -|ibm | region | |no | IBM Region. One of: `eu-gb`, `eu-de`, `us-south`, `us-east`, `br-sao`, `ca-tor`, `jp-tok`, `jp-osa`, `au-syd` | +|ibm | region | |no | IBM Region. One of: `eu-gb`, `eu-de`, `eu-es`, `us-south`, `us-east`, `br-sao`, `ca-tor`, `jp-tok`, `jp-osa`, `au-syd` | |ibm | resource_group_id | | no | Resource group id from your IBM Cloud account. Get it from [here](https://cloud.ibm.com/account/resource-groups) | ### IBM Cloud Object Storage: | Group | Key |Default|Mandatory| Additional info | |---------|---------------------|---|---|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| ibm_cos | region | |yes | Region of your bucket. One of: `eu-gb`, `eu-de`, `us-south`, `us-east`, `br-sao`, `ca-tor`, `jp-tok`, `jp-osa`, `au-syd`. Lithops will use the region set under the `ibm` section if it is not set here | -| ibm_cos | api_key | |yes | API Key to your COS account. Not needed if using IAM API Key | -| ibm_cos | storage_bucket | | yes | The name of a bucket that exists in your account. This will be used by Lithops for intermediate data. You must provide HMAC credentials if you want the bucket to be automatically created | -| ibm_cos | service_instance_id | |no | The service instance (CRN format) of your COS instance. **Mandatory** if no HMAC Credentials provided. | | | | | -| ibm_cos | access_key_id | |no | HMAC Credentials. **Mandatory** if no api_key. Not needed if using IAM API Key | -| ibm_cos | secret_access_key | |no | HMAC Credentials. **Mandatory** if no api_key. Not needed if using IAM API Key | -| ibm_cos | endpoint | |no | Endpoint to your COS account. **Mandatory** if no region. Make sure to use the full path with 'https://' as prefix | -| ibm_cos | private_endpoint | |no | Private endpoint to your COS account. **Mandatory** if no region. Make sure to use the full path with 'https://' or 'http://' as prefix | +| ibm_cos | region | |yes* | Region of your bucket. One of: `eu-gb`, `eu-de`, `eu-es`, `us-south`, `us-east`, `br-sao`, `ca-tor`, `jp-tok`, `jp-osa`, `au-syd`. Lithops uses the region set under the `ibm` section if it is not set here. *Not required if `endpoint` is set | +| ibm_cos | api_key | |no | API Key to your COS account. Required for Option 1 | +| ibm_cos | storage_bucket | |yes* | Bucket used by Lithops for intermediate data. *Can be auto-created when using HMAC credentials | +| ibm_cos | service_instance_id | |no | The service instance (CRN format) of your COS instance. Required if neither HMAC credentials nor `api_key` are provided | +| ibm_cos | access_key_id | |no | HMAC credentials. Required for Option 2 | +| ibm_cos | secret_access_key | |no | HMAC credentials. Required for Option 2 | +| ibm_cos | endpoint | |no | Public endpoint to your COS account. Auto-set from `region` if not provided. Must start with `https://` | +| ibm_cos | private_endpoint | |no | Worker endpoint for compute backends. Auto-set from `region` for `code_engine` and `ibm_vpc`. Must start with `https://` or `http://` | diff --git a/lithops/serverless/backends/code_engine/code_engine.py b/lithops/serverless/backends/code_engine/code_engine.py index 04be8a64c..cf99c4a23 100644 --- a/lithops/serverless/backends/code_engine/code_engine.py +++ b/lithops/serverless/backends/code_engine/code_engine.py @@ -39,14 +39,8 @@ logger = logging.getLogger(__name__) -REGISTRY_SECRET_NAME = 'lithops-regcred' -LITHOPS_RUNTIME_TYPE = 'lithops-runtime' -ENTRYPOINT_SCRIPT = '/lithops/lithopsentry.py' -PYTHON_BIN = '/usr/local/bin/python' -METADATA_JOBRUN_NAME = 'lithops-runtime-metadata' -JOB_RUN_POLL_INTERVAL = 2 - +# Decorator to wrap a function to reinit clients and retry on except. def retry_on_except(func): def decorated_func(*args, **kwargs): _self = args[0] @@ -80,7 +74,7 @@ def decorated_func(*args, **kwargs): class CodeEngineBackend: """ - IBM Code Engine backend using the official Code Engine SDK. + A wrap-up around Code Engine backend. """ def __init__(self, ce_config, internal_storage): @@ -107,11 +101,10 @@ def __init__(self, ce_config, internal_storage): msg = COMPUTE_CLI_MSG.format('IBM Code Engine') logger.info(f"{msg} - Project: {self.project_name} - Region: {self.region}") - # ------------------------------------------------------------------ # - # Project / client - # ------------------------------------------------------------------ # - def _create_code_engine_client(self): + """ + Creates the Code Engine SDK client + """ if self.ce_client: return @@ -123,7 +116,8 @@ def _create_code_engine_client(self): def _get_or_create_namespace(self, create=True): """ - Resolves the Code Engine project. Namespace is kept for runtime key compatibility. + Gets or creates the Code Engine project. + Namespace is kept for runtime key compatibility. """ ce_data = load_yaml_config(self.cache_file) if not self.project_id: @@ -165,15 +159,14 @@ def _get_or_create_namespace(self, create=True): return self.namespace def _sync_project_config(self): + """ + Syncs project and namespace metadata into the backend config + """ self.namespace = self.namespace or self.project_id self.config['project_id'] = self.project_id self.config['namespace'] = self.namespace self._create_code_engine_client() - # ------------------------------------------------------------------ # - # Formatting helpers - # ------------------------------------------------------------------ # - @staticmethod def _format_memory(memory_mb): """ @@ -188,11 +181,17 @@ def _format_memory(memory_mb): @staticmethod def _format_cpu(cpu): + """ + Formats CPU value for the Code Engine API + """ text = f'{cpu:.3f}'.rstrip('0').rstrip('.') return text @staticmethod def _parse_memory(memory_limit): + """ + Parses a Code Engine memory limit back to MB + """ if not memory_limit: return 0 if memory_limit.endswith('G'): @@ -205,25 +204,40 @@ def _parse_memory(memory_limit): @staticmethod def _normalize_memory_mb(memory_mb): + """ + Snaps parsed memory to the nearest valid Lithops memory tier + """ return min(config.VALID_MEMORY_VALUES, key=lambda v: abs(v - memory_mb)) def _format_jobdef_name(self, runtime_name, runtime_memory, version=__version__): + """ + Formats the job definition name + """ name = f'{runtime_name}-{runtime_memory}-{version}' name_hash = hashlib.sha1(name.encode("utf-8")).hexdigest()[:10] return f'lithops-worker-{self.user_key}-{version.replace(".", "")}-{name_hash}' def _get_default_runtime_image_name(self): + """ + Generates the default runtime image name + """ return utils.get_default_container_name( self.name, self.config, 'lithops-codeenigne-default' ) def _has_registry_credentials(self): + """ + Checks whether container registry credentials are set in config + """ return all(key in self.config for key in ["docker_user", "docker_password"]) def _build_job_env_variables(self, action, payload_config_map_name=None): + """ + Builds the environment variables passed to a Code Engine job run + """ env_variables = [ EnvVarPrototype(name='ACTION', type='literal', value=action), - EnvVarPrototype(name='LITHOPS_RUNTIME_TYPE', type='literal', value=LITHOPS_RUNTIME_TYPE), + EnvVarPrototype(name='LITHOPS_RUNTIME_TYPE', type='literal', value=config.LITHOPS_RUNTIME_TYPE), EnvVarPrototype(name='LITHOPS_VERSION', type='literal', value=__version__), ] if payload_config_map_name: @@ -237,11 +251,10 @@ def _build_job_env_variables(self, action, payload_config_map_name=None): ) return env_variables - # ------------------------------------------------------------------ # - # CE resource helpers - # ------------------------------------------------------------------ # - def _list_jobs(self): + """ + Lists all job definitions in the current project + """ try: return self.ce_client.list_jobs(self.project_id).get_result().get('jobs', []) except ApiException as e: @@ -249,11 +262,17 @@ def _list_jobs(self): return [] def _is_lithops_job(self, job): + """ + Returns whether a job definition was created by Lithops + """ fn_name = job.get('name') or '' return fn_name.startswith(f'lithops-worker-{self.user_key}') @staticmethod def _read_job_env(job): + """ + Reads Lithops runtime metadata from a job definition env vars + """ runtime_type = None version = __version__ for env_var in job.get('run_env_variables', []): @@ -264,12 +283,16 @@ def _read_job_env(job): return runtime_type, version def _iter_lithops_runtimes(self, docker_image_name='all'): + """ + Yields deployed Lithops runtimes as + (image_name, memory, version, jobdef_name) tuples + """ for job in self._list_jobs(): if not self._is_lithops_job(job): continue runtime_type, version = self._read_job_env(job) - if runtime_type != LITHOPS_RUNTIME_TYPE: + if runtime_type != config.LITHOPS_RUNTIME_TYPE: continue image_name = job.get('image_reference') or '' @@ -280,6 +303,9 @@ def _iter_lithops_runtimes(self, docker_image_name='all'): yield image_name, memory, version, job['name'] def _delete_job_run(self, jobrun_name): + """ + Deletes a job run, ignoring 404 errors + """ try: self.ce_client.delete_job_run(self.project_id, jobrun_name) except ApiException as e: @@ -287,6 +313,9 @@ def _delete_job_run(self, jobrun_name): logger.debug(f"Deleting job run {jobrun_name} failed with {e.status_code} {e.message}") def _delete_job_definition(self, jobdef_name): + """ + Deletes a job definition, ignoring 404 errors + """ try: self.ce_client.delete_job(self.project_id, jobdef_name) except ApiException as e: @@ -294,6 +323,9 @@ def _delete_job_definition(self, jobdef_name): logger.debug(f"Deleting job {jobdef_name} failed with {e.status_code} {e.message}") def _delete_config_map(self, config_map_name): + """ + Deletes a configmap + """ try: logger.debug(f"Deleting ConfigMap {config_map_name}") self.ce_client.delete_config_map(self.project_id, config_map_name) @@ -301,6 +333,9 @@ def _delete_config_map(self, config_map_name): logger.debug(f"Deleting config map {config_map_name} failed with {e.status_code} {e.message}") def _delete_lithops_config_maps(self): + """ + Deletes leftover lithops config maps from the project + """ try: configmaps = self.ce_client.list_config_maps(self.project_id).get_result() except ApiException as e: @@ -321,6 +356,9 @@ def _delete_lithops_config_maps(self): @staticmethod def _job_run_failure_message(status_details): + """ + Extracts the failure reason from a job run status + """ for details in (status_details.get('indices_details') or {}).values(): if not isinstance(details, dict): continue @@ -335,6 +373,9 @@ def _job_run_failure_message(status_details): @staticmethod def _job_run_finished(job_run): + """ + Returns the job run state and its status details + """ status = job_run.get('status') status_details = job_run.get('status_details') or {} succeeded = status_details.get('succeeded') or 0 @@ -348,13 +389,16 @@ def _job_run_finished(job_run): return 'running', status_details def _wait_for_job_run(self, jobrun_name): + """ + Waits until a job run completes or raises if it fails + """ logger.debug(f"Waiting for job run {jobrun_name}") while True: try: job_run = self.ce_client.get_job_run(self.project_id, jobrun_name).get_result() except ApiException as e: logger.debug(f"Polling job run {jobrun_name} failed with {e.status_code} {e.message}") - time.sleep(JOB_RUN_POLL_INTERVAL) + time.sleep(config.JOB_RUN_POLL_INTERVAL) continue state, status_details = self._job_run_finished(job_run) @@ -367,10 +411,13 @@ def _wait_for_job_run(self, jobrun_name): f"Job run {jobrun_name} failed" + (f": {reason}" if reason else '') ) - time.sleep(JOB_RUN_POLL_INTERVAL) + time.sleep(config.JOB_RUN_POLL_INTERVAL) @retry_on_except def _create_config_map(self, config_map_name, payload): + """ + Creates a configmap + """ data = {'lithops.payload': utils.dict_to_b64str(payload)} logger.debug(f"Creating ConfigMap {config_map_name}") @@ -394,6 +441,9 @@ def _create_config_map(self, config_map_name, payload): @retry_on_except def _job_def_exists(self, jobdef_name): + """ + Checks whether a job definition already exists + """ logger.debug(f"Checking if job definition {jobdef_name} already exists") try: self.ce_client.get_job(self.project_id, jobdef_name) @@ -406,11 +456,10 @@ def _job_def_exists(self, jobdef_name): logger.debug(f"Job definition {jobdef_name} found") return True - # ------------------------------------------------------------------ # - # Runtime build / deploy - # ------------------------------------------------------------------ # - def build_runtime(self, docker_image_name, dockerfile, extra_args=[]): + """ + Builds a new runtime from a Docker file and pushes it to the Docker hub + """ logger.info(f'Building runtime {docker_image_name} from {dockerfile}') docker_path = utils.get_docker_path() @@ -448,6 +497,9 @@ def build_runtime(self, docker_image_name, dockerfile, extra_args=[]): logger.debug('Building done!') def _build_default_runtime(self, default_runtime_img_name): + """ + Builds the default runtime + """ dockerfile = "Dockefile.default-ce-runtime" python_version = utils.CURRENT_PY_VERSION base_image = "slim-bookworm" @@ -460,6 +512,9 @@ def _build_default_runtime(self, default_runtime_img_name): os.remove(dockerfile) def deploy_runtime(self, docker_image_name, memory, timeout): + """ + Deploys a new runtime from an already built Docker image + """ self._get_or_create_namespace() try: @@ -475,6 +530,9 @@ def deploy_runtime(self, docker_image_name, memory, timeout): return self._generate_runtime_meta(docker_image_name, memory) def _generate_runtime_meta(self, docker_image_name, memory): + """ + Extracts runtime metadata by running a metadata job in Code Engine + """ logger.info(f"Extracting metadata from: {docker_image_name}") jobdef_name = self._format_jobdef_name(docker_image_name, memory) @@ -485,22 +543,22 @@ def _generate_runtime_meta(self, docker_image_name, memory): config_map_name = self._create_config_map(f'lithops-{jobdef_name}-metadata', job_payload) try: - self._delete_job_run(METADATA_JOBRUN_NAME) + self._delete_job_run(config.METADATA_JOBRUN_NAME) self._run_job( jobdef_name=jobdef_name, - jobrun_name=METADATA_JOBRUN_NAME, + jobrun_name=config.METADATA_JOBRUN_NAME, total_workers=1, runtime_memory=memory, action='metadata', payload_config_map_name=config_map_name, ) - self._wait_for_job_run(METADATA_JOBRUN_NAME) + self._wait_for_job_run(config.METADATA_JOBRUN_NAME) except Exception as e: raise Exception( f"Unable to extract Python preinstalled modules from the runtime: {e}" ) from e finally: - self._delete_job_run(METADATA_JOBRUN_NAME) + self._delete_job_run(config.METADATA_JOBRUN_NAME) self._delete_config_map(config_map_name) data_key = '/'.join([JOBS_PREFIX, jobdef_name + '.meta']) @@ -510,6 +568,10 @@ def _generate_runtime_meta(self, docker_image_name, memory): return runtime_meta def _create_container_registry_secret(self): + """ + Create the container registry secret in the project + (only if credentials are present in config) + """ if not self._has_registry_credentials(): return @@ -521,7 +583,7 @@ def _create_container_registry_secret(self): ) try: - self.ce_client.delete_secret(self.project_id, REGISTRY_SECRET_NAME) + self.ce_client.delete_secret(self.project_id, config.REGISTRY_SECRET_NAME) except ApiException as e: if e.status_code != 404: raise e @@ -530,7 +592,7 @@ def _create_container_registry_secret(self): self.ce_client.create_secret( self.project_id, format='registry', - name=REGISTRY_SECRET_NAME, + name=config.REGISTRY_SECRET_NAME, data=secret_data, ) except ApiException as e: @@ -539,6 +601,9 @@ def _create_container_registry_secret(self): @retry_on_except def _create_job_definition(self, docker_image_name, runtime_memory, timeout=None): + """ + Creates a Job definition + """ self._create_container_registry_secret() jobdef_name = self._format_jobdef_name(docker_image_name, runtime_memory) @@ -554,8 +619,8 @@ def _create_job_definition(self, docker_image_name, runtime_memory, timeout=None time.sleep(1) kwargs = { - 'run_commands': [PYTHON_BIN], - 'run_arguments': [ENTRYPOINT_SCRIPT], + 'run_commands': [config.PYTHON_BIN], + 'run_arguments': [config.ENTRYPOINT_SCRIPT], 'run_env_variables': self._build_job_env_variables('run'), 'run_mode': 'task', 'scale_array_spec': '0', @@ -565,7 +630,7 @@ def _create_job_definition(self, docker_image_name, runtime_memory, timeout=None 'scale_retry_limit': 3, } if self._has_registry_credentials(): - kwargs['image_secret'] = REGISTRY_SECRET_NAME + kwargs['image_secret'] = config.REGISTRY_SECRET_NAME self.ce_client.create_job( self.project_id, @@ -577,11 +642,10 @@ def _create_job_definition(self, docker_image_name, runtime_memory, timeout=None logger.debug(f'Job Definition {jobdef_name} created') return jobdef_name - # ------------------------------------------------------------------ # - # Lifecycle - # ------------------------------------------------------------------ # - def _resolve_jobdef_name(self, runtime_name, memory, version=__version__): + """ + Resolves the job definition name for a runtime image + """ jobdef_name = self._format_jobdef_name(runtime_name, memory, version) if self._job_def_exists(jobdef_name): return jobdef_name @@ -599,6 +663,10 @@ def _resolve_jobdef_name(self, runtime_name, memory, version=__version__): return jobdef_name def delete_runtime(self, runtime_name, memory, version=__version__, jobdef_name=None): + """ + Deletes a runtime. + We need to delete the job definition. + """ if not self._get_or_create_namespace(create=False): logger.info(f"Project {self.project_name} does not exist") return @@ -610,6 +678,9 @@ def delete_runtime(self, runtime_name, memory, version=__version__, jobdef_name= self._delete_job_definition(jobdef_name) def clean(self, all=False): + """ + Deletes all runtimes from all packages + """ logger.info(f'Cleaning project {self.project_name}') if not self._get_or_create_namespace(create=False): logger.info(f"Project {self.project_name} does not exist") @@ -629,6 +700,10 @@ def clean(self, all=False): os.remove(self.cache_file) def list_runtimes(self, docker_image_name='all'): + """ + List all the runtimes + return: list of tuples (docker_image_name, memory, version, jobdef_name) + """ if not self._get_or_create_namespace(create=False): logger.info(f"Project {self.project_name} does not exist") return [] @@ -636,6 +711,9 @@ def list_runtimes(self, docker_image_name='all'): return list(self._iter_lithops_runtimes(docker_image_name)) def clear(self, job_keys=None): + """ + Clean all completed jobruns in the current executor + """ if not self._get_or_create_namespace(create=False): logger.info(f"Project {self.project_name} does not exist") return @@ -650,6 +728,10 @@ def clear(self, job_keys=None): pass def invoke(self, docker_image_name, runtime_memory, job_payload): + """ + Invoke -- return information about this invocation + For array jobs only remote_invocator is allowed + """ self._get_or_create_namespace() executor_id = job_payload['executor_id'] @@ -692,6 +774,9 @@ def invoke(self, docker_image_name, runtime_memory, job_payload): @retry_on_except def _run_job(self, jobdef_name, jobrun_name, total_workers, runtime_memory, action, payload_config_map_name): + """ + Creates and starts a Code Engine job run + """ self.ce_client.create_job_run( self.project_id, job_name=jobdef_name, @@ -704,6 +789,11 @@ def _run_job(self, jobdef_name, jobrun_name, total_workers, runtime_memory, ) def get_runtime_key(self, docker_image_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. + """ self._get_or_create_namespace() jobdef_name = self._format_jobdef_name(docker_image_name, 256, version) runtime_key = os.path.join( @@ -712,6 +802,10 @@ def get_runtime_key(self, docker_image_name, runtime_memory, version=__version__ return runtime_key def get_runtime_info(self): + """ + Method that returns all the relevant information about the runtime set + in config + """ if 'runtime' not in self.config or self.config['runtime'] == 'default': self.config['runtime'] = self._get_default_runtime_image_name() diff --git a/lithops/serverless/backends/code_engine/config.py b/lithops/serverless/backends/code_engine/config.py index 04da239ee..42b5b9b68 100644 --- a/lithops/serverless/backends/code_engine/config.py +++ b/lithops/serverless/backends/code_engine/config.py @@ -28,6 +28,13 @@ FH_ZIP_LOCATION = os.path.join(os.getcwd(), 'lithops_codeengine.zip') +REGISTRY_SECRET_NAME = 'lithops-regcred' +LITHOPS_RUNTIME_TYPE = 'lithops-runtime' +ENTRYPOINT_SCRIPT = '/lithops/lithopsentry.py' +PYTHON_BIN = '/usr/local/bin/python' +METADATA_JOBRUN_NAME = 'lithops-runtime-metadata' +JOB_RUN_POLL_INTERVAL = 2 + # https://cloud.ibm.com/docs/codeengine?topic=codeengine-mem-cpu-combo VALID_CPU_VALUES = [0.125, 0.25, 0.5, 1, 2, 4, 6, 8] VALID_MEMORY_VALUES = [256, 512, 1024, 2048, 4096, 8192, 12288, 16384, 24576, 32768] @@ -63,10 +70,14 @@ RUN pip install --upgrade --ignore-installed setuptools six pip \ && pip install --upgrade --no-cache-dir --ignore-installed \ gunicorn \ + pika \ flask \ gevent \ ibm-cos-sdk \ ibm-cloud-sdk-core \ + ibm-vpc \ + ibm-code-engine-sdk \ + kubernetes \ redis \ requests \ PyYAML \ @@ -92,6 +103,9 @@ def _validate_cpu_memory(runtime_cpu, runtime_memory): + """ + Validates that the CPU and memory pair is supported by Code Engine + """ min_memory, max_memory = VALID_CPU_MEMORY[runtime_cpu] if not min_memory <= runtime_memory <= max_memory: raise Exception( @@ -102,6 +116,9 @@ def _validate_cpu_memory(runtime_cpu, runtime_memory): def load_config(config_data): + """ + Loads and validates the Code Engine backend configuration + """ if 'ibm' not in config_data or config_data['ibm'] is None: raise Exception("'ibm' section is mandatory in the configuration") diff --git a/runtime/code_engine/Dockerfile b/runtime/code_engine/Dockerfile index fd2f9295f..82e54fb10 100644 --- a/runtime/code_engine/Dockerfile +++ b/runtime/code_engine/Dockerfile @@ -1,31 +1,29 @@ -# Python 3.10 -FROM python:3.10-slim-bookworm +# Python 3.12 +FROM python:3.12-slim-bookworm # Python 3.11 #FROM python:3.11-slim-bookworm RUN apt-get update \ - # add the required packages && apt-get install -y unzip \ - # cleanup package lists, they are not used anymore in this image - && rm -rf /var/lib/apt/lists/* \ - && apt-cache search linux-headers-generic + && rm -rf /var/lib/apt/lists/* COPY requirements.txt requirements.txt -RUN pip install --upgrade pip setuptools six gevent && pip install --no-cache-dir -r requirements.txt +RUN pip install --upgrade pip setuptools six gevent \ + && pip install --no-cache-dir -r requirements.txt ENV PORT=8080 ENV CONCURRENCY=1 ENV TIMEOUT=600 ENV PYTHONUNBUFFERED=TRUE -# Copy Lithops proxy and lib to the container image. ENV APP_HOME=/lithops WORKDIR $APP_HOME COPY lithops_codeengine.zip . RUN unzip lithops_codeengine.zip && rm lithops_codeengine.zip -CMD exec gunicorn --bind :$PORT --workers $CONCURRENCY --timeout $TIMEOUT lithopsentry:proxy +# CE jobs override the container command. Gunicorn is kept for legacy Knative-style starts. +CMD ["sh", "-c", "exec gunicorn --bind :$PORT --workers $CONCURRENCY --timeout $TIMEOUT lithopsentry:proxy"] diff --git a/runtime/code_engine/requirements.txt b/runtime/code_engine/requirements.txt index a587ec205..a254aeca0 100644 --- a/runtime/code_engine/requirements.txt +++ b/runtime/code_engine/requirements.txt @@ -4,12 +4,11 @@ gunicorn flask pika -boto3 ibm-cloud-sdk-core ibm-cos-sdk +ibm-vpc ibm-code-engine-sdk -scikit-learn -diskcache +kubernetes redis gevent requests @@ -19,8 +18,8 @@ cloudpickle ps-mem tblib psutil -kubernetes -# Optional packages (Add your packages below) -ibm-vpc +# Optional packages (add your packages below) +#scikit-learn +#diskcache #pandas