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/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/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..cf99c4a23 100644 --- a/lithops/serverless/backends/code_engine/code_engine.py +++ b/lithops/serverless/backends/code_engine/code_engine.py @@ -17,28 +17,26 @@ 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__) @@ -49,26 +47,28 @@ def decorated_func(*args, **kwargs): 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 @@ -83,114 +83,73 @@ 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 + Creates the Code Engine SDK client """ - 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 + Gets or creates 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,33 +158,63 @@ def get_k8s_namespace(project_id): return self.namespace - def _create_k8s_iam_client(self): + def _sync_project_config(self): """ - Creates the k8s client with the IAM token + Syncs project and namespace metadata into the backend config """ - if self.is_lithops_worker: - return - - if not self.token_manager: - self.token_manager = IAMTokenManager(self.iam_api_key) + 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() - token, expiry_time = self.token_manager.get_token() + @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' - if expiry_time != self.config.get('token_expiry_time'): - self.config['token_expiry_time'] = expiry_time + @staticmethod + def _format_cpu(cpu): + """ + Formats CPU value for the Code Engine API + """ + text = f'{cpu:.3f}'.rstrip('0').rstrip('.') + return text - configuration = client.Configuration.get_default_copy() - configuration.host = self.cluster - configuration.api_key = {"authorization": "Bearer " + token} - client.Configuration.set_default(configuration) + @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'): + 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): + """ + 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): @@ -236,6 +225,237 @@ def _get_default_runtime_image_name(self): 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=config.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 _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: + logger.debug(f"List all jobs failed with {e.status_code} {e.message}") + 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', []): + 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'): + """ + 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 != config.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): + """ + Deletes a job run, ignoring 404 errors + """ + 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): + """ + Deletes a job definition, ignoring 404 errors + """ + 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): + """ + Deletes a configmap + """ + 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): + """ + 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: + 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): + """ + 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 + 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): + """ + 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 + 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): + """ + 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(config.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(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}") + + 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): + """ + 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) + 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 build_runtime(self, docker_image_name, dockerfile, extra_args=[]): """ Builds a new runtime from a Docker file and pushes it to the Docker hub @@ -280,7 +500,6 @@ 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" @@ -307,35 +526,156 @@ 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 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) + + 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(config.METADATA_JOBRUN_NAME) + self._run_job( + jobdef_name=jobdef_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(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(config.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 delete_runtime(self, runtime_name, memory, version=__version__): + def _create_container_registry_secret(self): """ - Deletes a runtime - We need to delete job definition + Create the container registry secret in the project + (only if credentials are present in config) """ - if not self._get_or_create_namespace(create=False): - logger.info(f"Project {self.project_name} does not exist") + if not self._has_registry_credentials(): return - logger.info(f'Deleting runtime: {runtime_name} - {memory}MB') - self._create_k8s_iam_client() + 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: + self.ce_client.delete_secret(self.project_id, config.REGISTRY_SECRET_NAME) + except ApiException as e: + if e.status_code != 404: + raise e + 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.create_secret( + self.project_id, + format='registry', + name=config.REGISTRY_SECRET_NAME, + data=secret_data, ) except ApiException as e: - logger.debug(f"Deleting a jobdef failed with {e.status} {e.reason}") + 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}") + + 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': [config.PYTHON_BIN], + 'run_arguments': [config.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'] = config.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 _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 + + 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 + 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 + + 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 + + if not jobdef_name: + jobdef_name = self._resolve_jobdef_name(runtime_name, memory, version) + + logger.info(f'Deleting runtime: {runtime_name} - {memory}MB') + self._delete_job_definition(jobdef_name) def clean(self, all=False): """ @@ -348,71 +688,27 @@ 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 self.list_runtimes(): + self.delete_runtime(image_name, memory, version, jobdef_name=jobdef_name) + + self._delete_lithops_config_maps() 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) + return: list of tuples (docker_image_name, memory, version, jobdef_name) """ - 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" - ) - except ApiException as e: - logger.debug(f"List all jobdefinitions failed with {e.status} {e.reason}") - return runtimes - - for jobdef in jobdefs['items']: - try: - if not jobdef['metadata']['name'].startswith(f'lithops-worker-{self.user_key}'): - continue - if not jobdef['metadata']['labels']['type'] == 'lithops-runtime': - 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) - 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): """ @@ -422,22 +718,10 @@ def clear(self, job_keys=None): 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: - 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._delete_config_map(jobrun_name) - except ApiException as e: - logger.debug(f"Deleting a jobrun failed with {e.status} {e.reason}") + for job_key in job_keys or self.jobs: + jobrun_name = f'lithops-{job_key.lower()}' + self._delete_job_run(jobrun_name) + self._delete_config_map(jobrun_name) try: self.jobs.remove(job_key) except ValueError: @@ -449,11 +733,9 @@ def invoke(self, docker_image_name, runtime_memory, job_payload): 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 +743,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,155 +754,40 @@ 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): + """ + Creates and starts a Code Engine job run + """ + 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"]): - 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'] - - 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", - ) - - try: - self.core_api.delete_namespaced_secret("lithops-regcred", self.namespace) - except ApiException: - pass - - try: - self.core_api.create_namespaced_secret(self.namespace, secret) - except ApiException as e: - if e.status != 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, - ) - except ApiException as e: - if e.status == 404: - pass - else: - 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 - - 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. @@ -630,8 +796,9 @@ 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) - 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): @@ -642,169 +809,10 @@ def get_runtime_info(self): 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'], 'runtime_timeout': self.config['runtime_timeout'], '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 - ) - except ApiException as e: - if e.status == 404: - logger.debug(f"Job definition {jobdef_name} not found (404)") - return False - else: - 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' - - job_payload = copy.deepcopy(self.internal_storage.storage.config) - 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: - 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 - - 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() - except Exception: - pass - - 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: - 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): - """ - 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) - - logger.debug(f"Creating ConfigMap {config_map_name}") - - self.core_api.create_namespaced_config_map( - namespace=self.namespace, - body=cmap, - field_manager='lithops' - ) - - 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 - ) - except ApiException as e: - logger.debug(f"Deleting a configmap failed with {e.status} {e.reason}") diff --git a/lithops/serverless/backends/code_engine/config.py b/lithops/serverless/backends/code_engine/config.py index 4d65abefa..42b5b9b68 100644 --- a/lithops/serverless/backends/code_engine/config.py +++ b/lithops/serverless/backends/code_engine/config.py @@ -26,21 +26,42 @@ 'docker_server': 'docker.io' } -DEFAULT_GROUP = "codeengine.cloud.ibm.com" -DEFAULT_VERSION = "v1beta1" - 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] -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', +] -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',) +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 \ @@ -53,12 +74,13 @@ flask \ gevent \ ibm-cos-sdk \ + ibm-cloud-sdk-core \ ibm-vpc \ ibm-code-engine-sdk \ + kubernetes \ redis \ requests \ PyYAML \ - kubernetes \ numpy \ cloudpickle \ ps-mem \ @@ -70,85 +92,33 @@ 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 +CMD ["sh", "-c", "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 -""" +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( + 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): + """ + 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") @@ -158,7 +128,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']) @@ -166,13 +136,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']: @@ -180,13 +154,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'] 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() 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