diff --git a/client/python/babylon_async/LICENSE b/client/python/babylon_async/LICENSE new file mode 100644 index 000000000..c9235f8a6 --- /dev/null +++ b/client/python/babylon_async/LICENSE @@ -0,0 +1,7 @@ +Copyright © 2026 Red Hat + +Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/client/python/babylon_async/README.md b/client/python/babylon_async/README.md new file mode 100644 index 000000000..592acbad6 --- /dev/null +++ b/client/python/babylon_async/README.md @@ -0,0 +1 @@ +# Babylon Asynchronous Client Library diff --git a/client/python/babylon_async/pyproject.toml b/client/python/babylon_async/pyproject.toml new file mode 100644 index 000000000..0d2fe5c64 --- /dev/null +++ b/client/python/babylon_async/pyproject.toml @@ -0,0 +1,23 @@ +[build-system] +requires = ["hatchling >= 1.26"] +build-backend = "hatchling.build" + +[project] +name = "babylon_async" +version = "0.0.1" +authors = [ + { name="Johnathan Kupferer", email="jkupfere@redhat.com" }, +] +description = "Babylon asynchronous library package" +readme = "README.md" +requires-python = ">=3.9" +classifiers = [ + "Programming Language :: Python :: 3", + "Operating System :: OS Independent", +] +license = "MIT" +license-files = ["LICENSE"] + +[project.urls] +Homepage = "https://github.com/redhat-cop/babylon" +Issues = "https://github.com/redhat-cop/babylon/issues" diff --git a/client/python/babylon_async/src/babylon_async/__init__.py b/client/python/babylon_async/src/babylon_async/__init__.py new file mode 100644 index 000000000..28ebd973f --- /dev/null +++ b/client/python/babylon_async/src/babylon_async/__init__.py @@ -0,0 +1,9 @@ +from .client import BabylonClient +from .exceptions import BabylonApiException + +from .anarchyrun import AnarchyRun +from .anarchysubject import AnarchySubject +from .catalogitem import CatalogItem +from .tenantclusterpool import TenantClusterPool +from .resourceclaim import ResourceClaim +from .resourcepool import ResourcePool diff --git a/client/python/babylon_async/src/babylon_async/anarchygovernor.py b/client/python/babylon_async/src/babylon_async/anarchygovernor.py new file mode 100644 index 000000000..440cf5a6e --- /dev/null +++ b/client/python/babylon_async/src/babylon_async/anarchygovernor.py @@ -0,0 +1,73 @@ +from __future__ import annotations +from typing import List, Mapping + +from .k8s_object import K8sObject + +class AnarchyGovernor(K8sObject): + api_group = "anarchy.gpte.redhat.com" + api_version = "v1" + kind = "AnarchyGovernor" + plural = "anarchygovernors" + api_group_version = f"{api_group}/{api_version}" + + def get_sandboxes(self) -> List[Mapping]: + return [ + SandboxSpec(item) + for item in self.spec.job_vars['__meta__'].get('sandboxes', []) + ] + + @property + def job_vars(self) -> Mapping: + return self.spec.job_vars + + @property + def spec(self) -> AnarchyGovernorSpec: + return AnarchyGovernorSpec(self._definition) + + @property + def vars(self) -> Mapping: + return self.spec.vars + +class AnarchyGovernorSpec: + def __init__(self, definition): + self._definition = definition + + @property + def job_vars(self) -> Mapping: + return self._definition.get('vars', {}).get('job_vars', {}) + + @property + def vars(self) -> Mapping: + return self._definition.get('vars', {}) + +class SandboxSpec: + def __init__(self, definition): + self._definition = definition + + @property + def alias(self) -> str|None: + return self._definition.get('alias') + + @property + def annotations(self) -> Mapping[str, str]|None: + return self._definition.get('annotations') + + @property + def cloud_selector(self) -> Mapping[str, str]|None: + return self._definition.get('cloud_selector') + + @property + def kind(self) -> str: + return self._definition['kind'] + + @property + def namespace_suffix(self) -> str|None: + return self._definition.get('namespace_suffix') + + @property + def quota(self) -> Mapping[str, str]|None: + return self._definition.get('quota') + + @property + def var(self) -> str|None: + return self._definition.get('var') diff --git a/client/python/babylon_async/src/babylon_async/anarchyrun.py b/client/python/babylon_async/src/babylon_async/anarchyrun.py new file mode 100644 index 000000000..eeee682f1 --- /dev/null +++ b/client/python/babylon_async/src/babylon_async/anarchyrun.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +from datetime import datetime + +from kubernetes_asyncio.client import ( + ApiException as KubernetesApiException +) + +from .k8s_object import K8sObject + +class AnarchyRun(K8sObject): + api_group = "anarchy.gpte.redhat.com" + api_version = "v1" + kind = "AnarchyRun" + plural = "anarchyruns" + api_group_version = f"{api_group}/{api_version}" + + @property + def is_active(self) -> bool: + value = self.metadata.labels.get('anarchy.gpte.redhat.com/runner') + if value.startswith('anarchy-runner-'): + return True + return self.state in {'failed', 'pending'} + + @property + def is_finished(self) -> bool: + return self.state in {'canceled', 'successful'} + + @property + def is_queued(self) -> bool: + return self.state == 'queued' + + @property + def runner_pod_name(self) -> str|None: + value = self.metadata.labels.get('anarchy.gpte.redhat.com/runner') + if value.startswith('anarchy-runner-'): + return value + return None + + @property + def spec(self) -> AnarchyRunSpec: + return AnarchyRunSpec(self._definition['spec']) + + @property + def state(self) -> str|None: + value = self.metadata.labels.get('anarchy.gpte.redhat.com/runner') + if value in {'canceled', 'failed', 'pending', 'queued', 'successful'}: + return value + if value.startswith('anarchy-runner-'): + return 'running' + return None + + @property + def status(self) -> AnarchyRunStatus|None: + if 'status' not in self._definition: + return None + return AnarchyRunStatus(self._definition['status']) + + async def check_runner_pod_exists(self) -> bool: + if self.runner_pod_name is None: + return True + try: + await self.client.core_v1_api.read_namespaced_pod( + name=self.runner_pod_name, + namespace=self.metadata.namespace, + ) + return True + except KubernetesApiException as exception: + if exception.status == 404: + return False + raise + + async def set_runner_state(self, value) -> None: + await self.patch({ + "metadata": { + "labels": { + "anarchy.gpte.redhat.com/runner": value, + } + } + }) + +class AnarchyRunSpec: + def __init__(self, definition): + self._definition = definition + +class AnarchyRunStatus: + def __init__(self, definition): + self._definition = definition + + @property + def retry_after_datetime(self) -> datetime|None: + ts = self.retry_after_timestamp + if ts is None: + return None + return datetime.strptime(ts, '%Y-%m-%dT%H:%M:%S%z') + + @property + def retry_after_timestamp(self) -> str|None: + return self._definition.get('retryAfter') + + @property + def run_post_datetime(self) -> datetime|None: + ts = self.run_post_timestamp + if ts is None: + return None + return datetime.strptime(ts, '%Y-%m-%dT%H:%M:%S%z') + + @property + def run_post_timestamp(self) -> str|None: + return self._definition.get('runPostTimestamp') diff --git a/client/python/babylon_async/src/babylon_async/anarchysubject.py b/client/python/babylon_async/src/babylon_async/anarchysubject.py new file mode 100644 index 000000000..fe91c7d4c --- /dev/null +++ b/client/python/babylon_async/src/babylon_async/anarchysubject.py @@ -0,0 +1,56 @@ +from __future__ import annotations +from typing import Generator + +from .k8s_object import K8sObject +from .resourcereference import ResourceReference + +from .anarchyrun import AnarchyRun + +class AnarchySubject(K8sObject): + api_group = "anarchy.gpte.redhat.com" + api_version = "v1" + kind = "AnarchySubject" + plural = "anarchysubjects" + api_group_version = f"{api_group}/{api_version}" + + @property + def spec(self) -> AnarchySubjectSpec: + return AnarchySubjectSpec(self._definition['spec']) + + @property + def status(self) -> AnarchySubjectStatus|None: + if 'status' not in self._definition: + return None + return AnarchySubjectStatus(self._definition['status']) + + async def get_anarchy_run(self, name:str) -> AnarchyRun: + return await self.client.get_anarchy_run( + name=name, + namespace=self.namespace, + ) + + async def list_anarchy_runs(self) -> Generator[AnarchyRun, None, None]: + async for anarchy_run in self.client.list_anarchy_runs_for_anarchy_subject(self): + yield anarchy_run + +class AnarchySubjectSpec: + def __init__(self, definition): + self._definition = definition + +class AnarchySubjectStatus: + def __init__(self, definition): + self._definition = definition + + @property + def runs(self) -> AnarchySubjectStatusRuns: + return AnarchySubjectStatusRuns(self._definition.get('runs', {})) + +class AnarchySubjectStatusRuns: + def __init__(self, definition): + self._definition = definition + + @property + def active(self) -> list[ResourceReference]: + return [ + ResourceReference(item) for item in self._definition.get('active', []) + ] diff --git a/client/python/babylon_async/src/babylon_async/catalogitem.py b/client/python/babylon_async/src/babylon_async/catalogitem.py new file mode 100644 index 000000000..4dd4f45a7 --- /dev/null +++ b/client/python/babylon_async/src/babylon_async/catalogitem.py @@ -0,0 +1,352 @@ +from __future__ import annotations +from copy import deepcopy +from typing import Any, List, Mapping + +from datetime import datetime, timedelta + +import deepmerge +import jinja2 +import openapi_schema_validator +import pytimeparse + +from .k8s_object import K8sObject +from .resourcepool import ResourcePool +from .resourceprovider import ResourceProvider + +template_merger = deepmerge.Merger( + [(dict, ["merge"])], + ["override"], + ["override"], +) + +class CatalogItem(K8sObject): + api_group = "babylon.gpte.redhat.com" + api_version = "v1" + kind = "CatalogItem" + plural = "catalogitems" + api_group_version = f"{api_group}/{api_version}" + + @property + def asset_uuid(self) -> str: + return self.metadata.labels.get('gpte.redhat.com/asset-uuid') + + @property + def category(self) -> str: + return self.spec.category + + @property + def external_url(self) -> str|None: + return self.spec.external_url + + @property + def display_name(self) -> str: + return self.spec.display_name + + @property + def parameters(self) -> list[CatalogItemSpecParameter]: + return self.spec.parameters + + @property + def provider(self) -> str: + return self.metadata.labels.get('babylon.gpte.redhat.com/Provider') + + @property + def spec(self) -> CatalogItemSpec: + return CatalogItemSpec(self._definition['spec']) + + async def check_resource_pool_match(self, + resource_pool: ResourcePool, + cache: bool=False, + ) -> bool: + """Evaluate whether ResourcePool may match this catalog item.""" + # External catalog items will never match a pool + if self.external_url is not None: + return False + + # Gather job vars to extract parameter values + all_job_vars = {} + for resource in resource_pool.spec.resources: + all_job_vars.update( + resource.template.get('spec', {}).get('vars', {}).get('job_vars', {}) + ) + + parameter_values = {} + for parameter in self.parameters: + if parameter.name in all_job_vars: + if parameter.validate(all_job_vars[parameter.name]): + parameter_values[parameter.name] = all_job_vars[parameter.name] + elif parameter.default is not None: + parameter_values[parameter.name] = parameter.default + + resource_provider = await self.get_resource_provider(cache=True) + try: + resource_provider_resources = await resource_provider.get_resources( + parameter_values=parameter_values, + cache=True, + ) + except (TypeError, jinja2.exceptions.UndefinedError): + # Template evaluation failed, so must not match + return False + + # Match is impossible if resource pool has more resources than the provider specifies + if len(resource_pool.spec.resources) > len(resource_provider_resources): + return False + + # Discard fields that don't need to match + for resource in resource_provider_resources: + resource.pop('name') + resource['template']['spec']['vars'].pop('action_schedule', None) + + for idx, pool_resource in enumerate(resource_pool.spec.resources): + resource_provider_resource = resource_provider_resources[idx] + # All vars from the resource_provider must match, but pool may have other vars + template_cmp = template_merger.merge( + deepcopy(pool_resource.template), + resource_provider_resource['template'], + ) + if pool_resource.provider.name != resource_provider_resource['provider']['name']: + return False + if pool_resource.template != template_cmp: + return False + return True + + async def get_resource_provider(self, cache:bool=False) -> ResourceProvider: + return await self.client.get_resource_provider( + cache=cache, + name=self.name, + ) + +class CatalogItemSpec: + def __init__(self, definition): + self._definition = definition + + @property + def agnosticv_repo(self) -> CatalogItemSpecAgnosticvRepo|None: + if 'agnosticvRepo' not in self._definition: + return None + return CatalogItemSpecAgnosticvRepo(self._definition['agnosticvRepo']) + + @property + def category(self) -> str: + return self._definition['category'] + + @property + def description(self) -> CatalogItemSpecDescription|None: + if 'description' not in self._definition: + return None + return CatalogItemSpecDescription(self._definition['description']) + + @property + def display_name(self) -> str: + return self._definition['displayName'] + + @property + def external_url(self) -> str|None: + return self._definition.get('externalUrl') + + @property + def last_update(self) -> CatalogItemSpecLastUpdate|None: + if 'last_update' not in self._definition: + return None + return CatalogItemSpecLastUpdate(self._definition['last_update']) + + @property + def lifespan(self) -> CatalogItemSpecLifespan|None: + if 'lifespan' not in self._definition: + return None + return CatalogItemSpecLifespan(self._definition['lifespan']) + + @property + def parameters(self) -> List[CatalogItemSpecParameter]: + return [ + CatalogItemSpecParameter(item) + for item in self._definition.get('parameters', []) + ] + + @property + def provision_time_estimate(self) -> timedelta|None: + if 'provisionTimeEstimate' not in self._definition: + return None + return timedelta(seconds=pytimeparse.parse(self._definition['provisionTimeEstimate'])) + + @property + def runtime(self) -> CatalogItemSpecRuntime|None: + if 'runtime' not in self._definition: + return None + return CatalogItemSpecRuntime(self._definition['runtime']) + + @property + def terms_of_service(self) -> str|None: + return self._definition.get('termsOfService') + + @property + def workshop_lab_ui_redirect(self) -> bool|None: + return self._definition.get('workshopLabUiRedirect') + + @property + def workshop_ui_max_instances(self) -> bool|None: + return self._definition.get('workshopUiMaxInstances') + + @property + def workshop_user_mode(self) -> str: + return self._definition.get('workshopUserMode') + +class CatalogItemSpecAgnosticvRepo: + def __init__(self, definition): + self._definition = definition + + @property + def git(self) -> CatalogItemSpecAgnosticvRepoGit: + return CatalogItemSpecAgnosticvRepoGit(self._definition['git']) + + @property + def name(self) -> str: + return self._definition['name'] + +class CatalogItemSpecAgnosticvRepoGit: + def __init__(self, definition): + self._definition = definition + + @property + def ref(self) -> str: + return self._definition['ref'] + + @property + def url(self) -> str: + return self._definition['url'] + +class CatalogItemSpecDescription: + def __init__(self, definition): + self._definition = definition + + @property + def content(self) -> str: + return self._definition['content'] + + @property + def format(self) -> str: + return self._definition['format'] + +class CatalogItemSpecLastUpdate: + def __init__(self, definition): + self._definition = definition + + @property + def git(self) -> CatalogItemSpecLastUpdateGit: + return CatalogItemSpecLastUpdateGit(self._definition['git']) + +class CatalogItemSpecLastUpdateGit: + def __init__(self, definition): + self._definition = definition + + @property + def author(self) -> str: + return self._definition['author'] + + @property + def committer(self) -> str: + return self._definition['committer'] + + @property + def hash(self) -> str: + return self._definition['hash'] + + @property + def message(self) -> str: + return self._definition['message'] + + @property + def when_author(self) -> datetime: + return datetime.strptime(self._definition['when_author'], '%Y-%m-%dT%H:%M:%S%z') + + @property + def when_committer(self) -> datetime: + return datetime.strptime(self._definition['when_committer'], '%Y-%m-%dT%H:%M:%S%z') + +class CatalogItemSpecLifespan: + def __init__(self, definition): + self._definition = definition + + @property + def default(self) -> timedelta: + return timedelta(seconds=pytimeparse.parse(self._definition['default'])) + + @property + def maximum(self) -> timedelta: + return timedelta(seconds=pytimeparse.parse(self._definition['maximum'])) + + @property + def relative_maximum(self) -> timedelta: + return timedelta(seconds=pytimeparse.parse(self._definition['relativeMaximum'])) + +class CatalogItemSpecParameter: + def __init__(self, definition): + self._definition = definition + + @property + def annotation(self) -> str|None: + return self._definition.get('annotation') + + @property + def description(self) -> str|None: + return self._definition.get('description') + + @property + def default(self) -> Any: + if self.openapi_v3_schema is None: + return None + return self.openapi_v3_schema.get('default') + + @property + def form_label(self) -> str|None: + return self._definition.get('formLabel') + + @property + def form_require_condition(self) -> str|None: + return self._definition.get('formRequireCondition') + + @property + def name(self) -> str: + return self._definition['name'] + + @property + def openapi_v3_schema(self) -> Mapping|None: + return self._definition.get('openAPIV3Schema') + + @property + def openapi_v3_schema_validator(self) -> openapi_schema_validator.OAS30Validator|None: + schema = self._definition.get('openAPIV3Schema') + if schema is None: + return None + return openapi_schema_validator.OAS30Validator(schema) + + @property + def required(self) -> bool: + return self._definition.get('required', False) + + @property + def validation(self) -> str|None: + return self._definition.get('validation') + + def validate(self, value:Any) -> bool: + # This ideally would also check `validation`, but these checks are + # currently only possible to check on the client side. + if self.openapi_v3_schema is None: + return True + try: + self.openapi_v3_schema_validator.validate(value) + return True + except openapi_schema_validator.validators.ValidationError: + return False + +class CatalogItemSpecRuntime: + def __init__(self, definition): + self._definition = definition + + @property + def default(self) -> timedelta: + return timedelta(seconds=pytimeparse.parse(self._definition['default'])) + + @property + def maximum(self) -> timedelta: + return timedelta(seconds=pytimeparse.parse(self._definition['maximum'])) diff --git a/client/python/babylon_async/src/babylon_async/client.py b/client/python/babylon_async/src/babylon_async/client.py new file mode 100644 index 000000000..6feb6f8e4 --- /dev/null +++ b/client/python/babylon_async/src/babylon_async/client.py @@ -0,0 +1,557 @@ +from __future__ import annotations + +from typing import Any, Generator, List, Mapping + +import os + +import kubernetes_asyncio +from inflection import singularize +from kubernetes_asyncio.client import ( + ApiClient, CoreV1Api, CustomObjectsApi, + ApiException as KubernetesApiException +) + +from .anarchygovernor import AnarchyGovernor +from .anarchyrun import AnarchyRun +from .anarchysubject import AnarchySubject +from .catalogitem import CatalogItem +from .exceptions import BabylonApiException +from .k8s_object import K8sObject +from .namespace import Namespace +from .resourceclaim import ResourceClaim +from .resourcehandle import ResourceHandle +from .resourcepool import ResourcePool +from .resourceprovider import ResourceProvider +from .user import User +from .workshop import Workshop +from .workshopprovision import WorkshopProvision + +class BabylonClient: + @classmethod + async def create(cls, api_client:ApiClient=None) -> BabylonClient: + client = cls(api_client=api_client) + await client.init() + return client + + def __init__(self, api_client:ApiClient=None): + self.api_client = api_client + self.core_v1_api:CoreV1Api|None = None + self.custom_objects_api:CustomObjectsApi|None = None + self.is_admin:bool = False + + async def __aenter__(self) -> None: + await self.init() + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: + await self.close() + + async def close(self): + await self.api_client.close() + + async def init(self) -> None: + if self.api_client is None: + if os.path.exists('/run/secrets/kubernetes.io/serviceaccount'): + # If running in a container then use incluster config + kubernetes_asyncio.config.load_incluster_config() + else: + # Default to using user's kube config + await kubernetes_asyncio.config.load_kube_config() + + self.api_client = ApiClient() + + self.core_v1_api = CoreV1Api(self.api_client) + self.custom_objects_api = CustomObjectsApi(self.api_client) + await self.__init_is_admin() + + async def __init_is_admin(self) -> None: + """Check if user has admin rights in babylon and set is_admin flag.""" + self.is_admin = await self.check_access( + group="anarchy.gpte.redhat.com", + plural="anarchysubjects", + verb="update", + ) + + async def check_access(self, + group:str, + plural:str, + verb:str, + namespace:str|None=None, + ) -> bool: + resource_attributes = { + "group": group, + "resource": plural, + "verb": verb, + } + if namespace is not None: + resource_attributes['namespace'] = namespace + (data, status, _) = await self.api_client.call_api( + '/apis/authorization.k8s.io/v1/selfsubjectaccessreviews', + 'POST', + auth_settings=['BearerToken'], + body={ + "apiVersion": "authorization.k8s.io/v1", + "kind": "SelfSubjectAccessReview", + "spec": { + "resourceAttributes": resource_attributes, + }, + }, + response_types_map={ + 201: "object", + 401: None, + }, + ) + if status == 201: + return data.get('status', {}).get('allowed', False) + + reason = f"Failed to check access to {verb} {plural}.{group}" + if namespace is not None: + reason += f" in {namespace}" + raise BabylonApiException(status, reason) + + #async def order_service(self, + # catalog_item: CatalogItem, + # user: User, + # parameters:Mapping={}, + #) -> ResourceClaim: + # pass + + # Generic methods + async def create_object(self, + definition:Mapping[str,Any], + plural:str, + version:str, + group:str|None=None, + namespace:str|None=None, + ) -> Mapping: + try: + if group is None: + singular = singularize(plural) + if namespace is None: + return self.api_client.sanitize_for_serialization( + await getattr(self.core_v1_api, f"create_{singular}")(definition) + ) + return self.api_client.sanitize_for_serialization( + await getattr( + self.core_v1_api, f"create_namespaced_{singular}" + )(namespace, definition) + ) + if namespace is None: + return await self.custom_objects_api.create_cluster_custom_object( + body=definition, + group=group, + plural=plural, + version=version, + ) + return await self.custom_objects_api.create_namespaced_custom_object( + body=definition, + group=group, + namespace=namespace, + plural=plural, + version=version, + ) + except KubernetesApiException as exception: + raise BabylonApiException(kubernetes_api_exception=exception) from exception + + async def delete_object(self, + name:str, + plural:str, + version:str, + group:str|None=None, + namespace:str|None=None, + ) -> Mapping: + try: + if group is None: + singular = singularize(plural) + if namespace is None: + return self.api_client.sanitize_for_serialization( + await getattr(self.core_v1_api, f"delete_{singular}")(name) + ) + return self.api_client.sanitize_for_serialization( + await getattr( + self.core_v1_api, f"delete_namespaced_{singular}" + )(name, namespace) + ) + if namespace is None: + return await self.custom_objects_api.delete_cluster_custom_object( + group=group, + name=name, + plural=plural, + version=version, + ) + return await self.custom_objects_api.delete_namespaced_custom_object( + group=group, + name=name, + namespace=namespace, + plural=plural, + version=version, + ) + except KubernetesApiException as exception: + raise BabylonApiException(kubernetes_api_exception=exception) from exception + + async def get_object(self, + plural:str, + name:str, + version:str, + group:str|None=None, + namespace:str|None=None, + ) -> Mapping: + try: + if group is None: + singular = singularize(plural) + if namespace is None: + return self.api_client.sanitize_for_serialization( + await getattr(self.core_v1_api, f"read_{singular}")(name) + ) + return self.api_client.sanitize_for_serialization( + getattr(self.core_v1_api, f"read_namespaced_{singular}")(name, namespace) + ) + if namespace is None: + return await self.custom_objects_api.get_cluster_custom_object( + group=group, + name=name, + plural=plural, + version=version, + ) + return await self.custom_objects_api.get_namespaced_custom_object( + group=group, + name=name, + namespace=namespace, + plural=plural, + version=version, + ) + except KubernetesApiException as exception: + raise BabylonApiException(kubernetes_api_exception=exception) from exception + + async def list_object(self, + plural:str, + version:str, + batch_size:int=50, + label_selector:str|None=None, + group:str|None=None, + namespace:str|None=None, + ) -> Generator[Mapping, None, None]: + # FIXME - wrap with a try/except + _continue = None + method = None + if group is None: + singular = singularize(plural) + if namespace is None: + try: + method = getattr(self.core_v1_api, f"list_{singular}") + except AttributeError: + method = getattr(self.core_v1_api, f"list_{singular}_for_all_namespaces") + else: + method = getattr(self.core_v1_api, f"list_namespaced_{singular}")(namespace) + + while True: + if method is None: + if namespace is None: + obj_list = await self.custom_objects_api.list_cluster_custom_object( + _continue=_continue, + group=group, + label_selector=label_selector, + limit=batch_size, + plural=plural, + version=version, + ) + else: + obj_list = await self.custom_objects_api.list_namespaced_custom_object( + _continue=_continue, + group=group, + label_selector=label_selector, + limit=batch_size, + namespace=namespace, + plural=plural, + version=version, + ) + else: + if namespace is None: + obj_list = self.api_client.sanitize_for_serialization( + await method(_continue=_continue) + ) + else: + obj_list = self.api_client.sanitize_for_serialization( + await method(namespace=namespace, _continue=_continue) + ) + for definition in obj_list.get('items', []): + yield definition + _continue = obj_list['metadata'].get('continue') + if not _continue: + return + + async def patch_object(self, + group:str, + plural:str, + name:str, + version:str, + patch:Mapping|List[Mapping], + namespace:str|None=None, + ) -> Mapping: + # FIXME - wrap with a try/except + content_type = ( + 'application/json-patch+json' + if isinstance(patch, list) else + 'application/merge-patch+json' + ) + if namespace is None: + return await self.custom_objects_api.patch_cluster_custom_object( + group=group, + name=name, + plural=plural, + version=version, + body=patch, + _content_type=content_type, + ) + return await self.custom_objects_api.patch_namespaced_custom_object( + group=group, + name=name, + namespace=namespace, + plural=plural, + version=version, + body=patch, + _content_type=content_type, + ) + + async def patch_object_status(self, + group:str, + plural:str, + name:str, + version:str, + patch:Mapping|List[Mapping], + namespace:str|None=None, + ) -> Mapping: + # FIXME - wrap with a try/except + content_type = ( + 'application/json-patch+json' + if isinstance(patch, list) else + 'application/merge-patch+json' + ) + if namespace is None: + return await self.custom_objects_api.patch_cluster_custom_object_status( + group=group, + name=name, + plural=plural, + version=version, + body=patch, + _content_type=content_type, + ) + return await self.custom_objects_api.patch_namespaced_custom_object_status( + group=group, + name=name, + namespace=namespace, + plural=plural, + version=version, + body=patch, + _content_type=content_type, + ) + + # AnarchyGovernor methods + async def get_anarchy_governor(self, + name:str, + namespace:str, + ) -> AnarchyGovernor: + return await AnarchyGovernor.get(client=self, name=name, namespace=namespace) + + + # AnarchyRun methods + async def get_anarchy_run(self, + name:str, + namespace:str, + ) -> AnarchyRun: + return await AnarchyRun.get( + client=self, + name=name, + namespace=namespace, + ) + + async def list_anarchy_runs(self, + label_selector:str=None, + namespace:str=None, + ) -> Generator[AnarchyRun, None, None]: + async for anarchy_run in AnarchyRun.list( + client=self, + label_selector=label_selector, + namespace=namespace + ): + yield anarchy_run + + async def list_anarchy_runs_for_anarchy_subject(self, + anarchy_subject:AnarchySubject, + ) -> Generator[AnarchyRun, None, None]: + async for anarchy_run in self.list_anarchy_runs( + label_selector=f"anarchy.gpte.redhat.com/subject={anarchy_subject.name}", + namespace=anarchy_subject.namespace, + ): + yield anarchy_run + + # AnarchySubject methods + async def list_anarchy_subjects(self, + label_selector:str=None, + namespace:str=None, + ) -> Generator[AnarchySubject, None, None]: + async for anarchy_subject in AnarchySubject.list( + client=self, + label_selector=label_selector, + namespace=namespace, + ): + yield anarchy_subject + + # CatalogItem methods + async def get_catalog_item(self, name:str, namespace:str) -> CatalogItem: + return await CatalogItem.get(client=self, name=name, namespace=namespace) + + async def list_catalog_items(self, + label_selector:str|None=None, + namespace:str|None=None, + ) -> Generator[CatalogItem, None, None]: + async for catalog_item in CatalogItem.list( + client=self, + label_selector=label_selector, + namespace=namespace, + ): + yield catalog_item + + # Namespace methods + async def get_current_user_service_namespace(self) -> Namespace: + # Getting the current user service namespace involves checking access + # and some guesswork. If the user is not an admin they will net be able + # to list either namespaces or projects. + n = 0 + user = await self.get_user('~') + base_name = "user-" + user.metadata.name.replace('.', '-').replace('@', '-') + while n < 10: + name = base_name if n==0 else f"{base_name}-{n}" + if self.is_admin: + try: + return await self.get_namespace(name) + except BabylonApiException as exception: + if exception.status != 404: + raise + else: + if await self.check_access( + group=ResourceClaim.api_group, + namespace=name, + plural=ResourceClaim.plural, + verb="update", + ): + return Namespace(self, {"metadata": {"name": name}}) + + n += 1 + raise BabylonApiException(404, "No service namespace found for current user.") + + async def get_user_service_namespace(self, user:User) -> Namespace: + pass + + # Namespace methods + async def get_namespace(self, + name:str, + ) -> Namespace: + return await Namespace.get(client=self, name=name) + + async def list_namespaces(self, + label_selector:str|None=None, + ) -> Generator[Namespace, None, None]: + async for namespace in Namespace.list( + client=self, + label_selector=label_selector, + ): + yield namespace + + # ResourceClaim methods + async def create_resource_claim(self, + namespace:str, + provider_name:str, + auto_detach:bool=False, + name:str|None=None, + owner:K8sObject|None=None, + parameter_values:Mapping[str,Any]|None=None, + ): + if parameter_values is None: + parameter_values = {} + return await ResourceClaim.create_with_provider( + auto_detach=auto_detach, + client=self, + name=name, + namespace=namespace, + owner=owner, + parameter_values=parameter_values, + provider_name=provider_name, + ) + + async def get_resource_claim(self, name:str, namespace:str) -> ResourceClaim: + return await ResourceClaim.get(client=self, name=name, namespace=namespace) + + async def list_resource_claims(self, + label_selector:str|None=None, + namespace:str|None=None, + ) -> Generator[ResourceClaim, None, None]: + async for resource_claim in ResourceClaim.list( + client=self, + label_selector=label_selector, + namespace=namespace, + ): + yield resource_claim + + # ResourceHandle methods + async def get_resource_handle(self, name:str) -> ResourceHandle: + return await ResourceHandle.get(client=self, name=name, namespace="poolboy") + + # ResourcePool methods + async def get_resource_pool(self, name:str) -> ResourcePool: + return await ResourcePool.get(client=self, name=name, namespace="poolboy") + + async def list_resource_pools(self, + label_selector:str|None=None, + ) -> Generator[ResourcePool, None, None]: + async for resource_pool in ResourcePool.list( + client=self, + label_selector=label_selector, + namespace="poolboy", + ): + yield resource_pool + + # ResourceProvider methods + async def get_resource_provider(self, name:str, cache:bool=False) -> ResourceProvider: + return await ResourceProvider.get(cache=cache, client=self, name=name, namespace="poolboy") + + async def list_resource_providers(self, + label_selector:str|None=None, + ) -> Generator[ResourceProvider, None, None]: + async for resource_provider in ResourceProvider.list( + client=self, + label_selector=label_selector, + namespace="poolboy", + ): + yield resource_provider + + # User methods + async def get_current_user(self) -> User: + return await User.get(client=self, name='~') + + async def get_user(self, name:str) -> User: + return await User.get(client=self, name=name) + + # Workshop methods + async def list_workshops(self, + label_selector:str=None, + namespace:str=None, + ) -> Generator[Workshop, None, None]: + async for workshop in Workshop.list( + client=self, + label_selector=label_selector, + namespace=namespace + ): + yield workshop + + # WorkshopProvision methods + async def list_workshop_provisions(self, + label_selector:str=None, + namespace:str=None, + ) -> Generator[WorkshopProvision, None, None]: + async for workshop_provision in WorkshopProvision.list( + client=self, + label_selector=label_selector, + namespace=namespace + ): + yield workshop_provision diff --git a/client/python/babylon_async/src/babylon_async/exceptions.py b/client/python/babylon_async/src/babylon_async/exceptions.py new file mode 100644 index 000000000..142225f00 --- /dev/null +++ b/client/python/babylon_async/src/babylon_async/exceptions.py @@ -0,0 +1,30 @@ +from __future__ import annotations +from kubernetes_asyncio.client import ApiException as KubernetesApiException + +class BabylonApiException(Exception): + def __init__(self, + status:int|None=None, + reason:str|None=None, + kubernetes_api_exception:KubernetesApiException|None=None, + ): + if kubernetes_api_exception is not None: + if reason is None: + reason = kubernetes_api_exception.reason + if status is None: + status = kubernetes_api_exception.status + self.kubernetes_api_exception = kubernetes_api_exception + self.reason = reason + self.status = status + + def __str__(self): + error_message = f"({self.status})\nReason: {self.reason}\n" + + if self.kubernetes_api_exception is not None: + if self.kubernetes_api_exception.headers: + error_message += "HTTP response headers: {0}\n".format( + self.kubernetes_api_exception.headers) + + if self.kubernetes_api_exception.body: + error_message += "HTTP response body: {0}\n".format(self.kubernetes_api_exception.body) + + return error_message diff --git a/client/python/babylon_async/src/babylon_async/k8s_object.py b/client/python/babylon_async/src/babylon_async/k8s_object.py new file mode 100644 index 000000000..b5f816675 --- /dev/null +++ b/client/python/babylon_async/src/babylon_async/k8s_object.py @@ -0,0 +1,268 @@ +from __future__ import annotations +from collections.abc import Mapping + +from datetime import datetime + +from copy import deepcopy +from typing import List + +class K8sObject: + @classmethod + async def create(cls, client, + definition:Mapping, + name:str|None=None, + namespace:str|None=None, + owner:K8sObject|None=None + ): + definition['apiVersion'] = cls.api_group_version + definition['kind'] = cls.kind + + if 'metadata' not in definition: + definition['metadata'] = {} + + if name is not None: + if name.endswith('*'): + definition['metadata']['generateName'] = name[:-1] + else: + definition['metadata']['name'] = name + + if owner is not None: + definition['metadata']['ownerReferences'] = [owner.as_owner_reference()] + + definition = await client.create_object( + definition=definition, + group=cls.api_group, + namespace=namespace, + plural=cls.plural, + version=cls.api_version, + ) + return cls(client=client, definition=definition) + + @classmethod + async def get(cls, client, name:str, namespace:str|None=None, cache:bool=False): + if cache and hasattr(cls, '__cache__') and (name, namespace) in cls.__cache__: + return cls.__cache__.get((name, namespace)) + + definition = await client.get_object( + group=cls.api_group, + name=name, + namespace=namespace, + plural=cls.plural, + version=cls.api_version, + ) + obj = cls(client=client, definition=definition) + if cache: + obj.__cache_put() + return obj + + @classmethod + async def list(cls, + client, + label_selector:str|None=None, + namespace:str|None=None, + ) -> Mapping: + async for definition in client.list_object( + group=cls.api_group, + label_selector=label_selector, + namespace=namespace, + plural=cls.plural, + version=cls.api_version, + ): + yield cls(client=client, definition=definition) + + def __init__(self, client, definition): + self.client = client + self._definition = definition + + def __str__(self) -> str: + if self.namespace is None: + return f"{self.kind} {self.name}" + return f"{self.kind} {self.name} in {self.namespace}" + + @property + def creation_datetime(self) -> datetime: + return self.metadata.creation_datetime + + @property + def creation_timestamp(self) -> str: + return self.metadata.creation_timestamp + + @property + def is_deleting(self) -> bool: + return self.metadata.deletion_timestamp is not None + + @property + def metadata(self) -> K8sObjectMetadata: + return K8sObjectMetadata( + self._definition['metadata'], + ) + + @property + def name(self) -> str: + return self.metadata.name + + @property + def namespace(self) -> str|None: + return self.metadata.namespace + + @property + def uid(self) -> str: + return self.metadata.uid + + def __cache_put(self) -> None: + cls = self.__class__ + if not hasattr(cls, '__cache__'): + setattr(cls, '__cache__', {}) + cls.__cache__[(self.name, self.namespace)] = self + + def as_owner_reference(self, + block_owner_deletion:bool|None=None, + controller:bool=True, + ) -> mapping: + """Return reference mapping to object for use as owner reference.""" + ret = self.as_reference() + if block_owner_deletion is not None: + ret['blockOwnerDeletion'] = block_owner_deletion + ret['controller'] = controller + ret['uid'] = self.uid + return ret + + def as_reference(self) -> Mapping: + """Return reference mapping to object.""" + ret = { + "apiVersion": self.api_group_version, + "kind": self.kind, + "name": self.name, + } + if self.namespace is not None: + ret['namespace'] = self.namespace + return ret + + def get_definition(self) -> Mapping: + return deepcopy(self._definition) + + def update_definition(self, definition: Mapping) -> None: + self._definition = definition + + async def delete(self) -> None: + await self.client.delete_object( + group=self.api_group, + name=self.metadata.name, + namespace=self.metadata.namespace, + plural=self.plural, + version=self.api_version, + ) + + async def patch(self, patch: Mapping|List[Mapping]) -> None: + """Apply patch to object and update definition.""" + definition = await self.client.patch_object( + group=self.api_group, + name=self.metadata.name, + namespace=self.metadata.namespace, + patch=patch, + plural=self.plural, + version=self.api_version, + ) + self.update_definition(definition) + + async def patch_status(self, patch: Mapping|List[Mapping]) -> None: + """Apply json patch to object and update definition.""" + definition = await self.client.patch_object_status( + group=self.api_group, + name=self.metadata.name, + namespace=self.metadata.namespace, + patch=patch, + plural=self.plural, + version=self.api_version, + ) + self.update_definition(definition) + + async def refresh(self) -> None: + """Refetch object to refresh definition""" + self._definition = self.client.get_object( + group=self.api_gorup, + name=self.metadata.name, + namespace=self.metadata.namespace, + plural=self.plural, + version=self.api_version, + ) + + +class K8sObjectMetadata: + def __init__(self, definition): + self._definition = definition + + @property + def annotations(self) -> Mapping[str, str]: + return self._definition.get('annotations', {}) + + @property + def creation_datetime(self) -> datetime: + return datetime.strptime(self._definition['creationTimestamp'], '%Y-%m-%dT%H:%M:%S%z') + + @property + def creation_timestamp(self) -> str: + return self._definition['creationTimestamp'] + + @property + def deletion_datetime(self) -> datetime|None: + if 'deletionTimestamp' not in self._definition: + return None + return datetime.strptime(self._definition['deletionTimestamp'], '%Y-%m-%dT%H:%M:%S%z') + + @property + def deletion_timestamp(self) -> str|None: + return self._definition.get('deletionTimestamp') + + @property + def finalizers(self) -> List[str]|None: + return self._definition.get('finalizers') + + @property + def labels(self) -> Mapping[str, str]: + return self._definition.get('labels', {}) + + @property + def name(self) -> str: + return self._definition['name'] + + @property + def namespace(self) -> str|None: + return self._definition.get('namespace') + + @property + def owner_references(self) -> List[K8sObjectMetadataOwnerReference]|None: + if 'ownerReferences' not in self._definition: + return None + return [ + K8sObjectMetadataOwnerReference(item) + for item in self._definition['ownerReferences'] + ] + + @property + def uid(self) -> str: + return self._definition.get('uid') + +class K8sObjectMetadataOwnerReference: + def __init__(self, definition): + self._definition = definition + + @property + def api_version(self) -> str: + return self._definition['apiVersion'] + + @property + def controller(self) -> bool: + return self._definition['controller'] + + @property + def kind(self) -> str: + return self._definition['kind'] + + @property + def name(self) -> str: + return self._definition['name'] + + @property + def uid(self) -> str: + return self._definition['uid'] diff --git a/client/python/babylon_async/src/babylon_async/namespace.py b/client/python/babylon_async/src/babylon_async/namespace.py new file mode 100644 index 000000000..46df6bfb0 --- /dev/null +++ b/client/python/babylon_async/src/babylon_async/namespace.py @@ -0,0 +1,10 @@ +from __future__ import annotations + +from .k8s_object import K8sObject + +class Namespace(K8sObject): + api_group = None + api_version = "v1" + kind = "Namespace" + plural = "namespaces" + api_group_version = "v1" diff --git a/client/python/babylon_async/src/babylon_async/poolboy_templating.py b/client/python/babylon_async/src/babylon_async/poolboy_templating.py new file mode 100644 index 000000000..7942b995b --- /dev/null +++ b/client/python/babylon_async/src/babylon_async/poolboy_templating.py @@ -0,0 +1,355 @@ +import copy +import functools +import json +import random +import re +from datetime import datetime, timedelta, timezone + +import jinja2 +import jmespath +import pytimeparse +from str2bool import str2bool +from strgen import StringGenerator + +MAX_RECURSION_DEPTH = 100 + +def timedelta_to_str(td:timedelta) -> str: + total_seconds = int(td.total_seconds()) + days = total_seconds // 86400 + hours = (total_seconds % 86400) // 3600 + minutes = (total_seconds % 3600) // 60 + seconds = total_seconds % 60 + + ret = "" + if days > 0: + ret += f"{days}d" + if hours > 0: + ret += f"{hours:d}h" + if minutes > 0: + ret += f"{minutes}m" + if seconds > 0: + ret += f"{seconds}s" + return ret + +class TimeStamp(object): + def __init__(self, set_datetime=None): + if not set_datetime: + self.datetime = datetime.now(timezone.utc) + elif isinstance(set_datetime, datetime): + self.datetime = set_datetime + elif isinstance(set_datetime, str): + self.datetime = datetime.strptime(set_datetime, "%Y-%m-%dT%H:%M:%S%z") + else: + self.datetime = set_datetime + + def __call__(self, arg): + return TimeStamp(arg) + + def __eq__(self, other): + return self.datetime == other.datetime + + def __ne__(self, other): + return self.datetime != other.datetime + + def __ge__(self, other): + return self.datetime >= other.datetime + + def __gt__(self, other): + return self.datetime > other.datetime + + def __le__(self, other): + return self.datetime <= other.datetime + + def __lt__(self, other): + return self.datetime < other.datetime + + def __str__(self): + return self.datetime.strftime('%FT%TZ') + + def __add__(self, interval): + return self.add(interval) + + def add(self, interval): + ret = TimeStamp(self.datetime) + if isinstance(interval, timedelta): + ret.datetime += interval + elif isinstance(interval, str): + ret.datetime += timedelta(seconds=pytimeparse.parse(interval)) + elif isinstance(interval, number.Number): + ret.datetime += timedelta(seconds=interval) + else: + raise Exception(f"Unable to add {interval} to Timestamp") + return ret + + @property + def utcnow(self): + return TimeStamp() + +def error_if_undefined(result): + if isinstance(result, jinja2.Undefined): + result._fail_with_undefined_error() + else: + return result + +def seconds_to_interval(seconds:int) -> str: + if seconds % 86400 == 0: + return f"{int(seconds / 86400)}d" + elif seconds % 3600 == 0: + return f"{int(seconds / 3600)}h" + elif seconds % 60 == 0: + return f"{int(seconds / 60)}m" + else: + return f"{int(seconds)}s" + +def timedelta_to_str(td:timedelta) -> str: + total_seconds = int(td.total_seconds()) + days = total_seconds // 86400 + hours = (total_seconds % 86400) // 3600 + minutes = (total_seconds % 3600) // 60 + seconds = total_seconds % 60 + + ret = "" + if days > 0: + ret += f"{days}d" + if hours > 0: + ret += f"{hours:d}h" + if minutes > 0: + ret += f"{minutes}m" + if seconds > 0: + ret += f"{seconds}s" + return ret + +jinja2env = jinja2.Environment( + finalize = error_if_undefined, + undefined = jinja2.ChainableUndefined, +) +jinja2env.filters['bool'] = lambda x: bool(str2bool(x)) if isinstance(x, str) else bool(x) +jinja2env.filters['json_query'] = lambda x, query: jmespath.search(query, x) +jinja2env.filters['merge_list_of_dicts'] = lambda a: functools.reduce(lambda d1, d2: {**(d1 or {}), **(d2 or {})}, a) if a else {} +jinja2env.filters['object'] = lambda x: json.dumps(x) +jinja2env.filters['parse_time_interval'] = lambda x: timedelta(seconds=pytimeparse.parse(str(x))) +jinja2env.filters['strgen'] = lambda x: StringGenerator(x).render() +jinja2env.filters['to_datetime'] = lambda s, f='%Y-%m-%d %H:%M:%S': datetime.strptime(str(s), f) +jinja2env.filters['to_json'] = lambda x: json.dumps(x) + +# Regex to detect if it looks like this value should be rendered as a raw type +# rather than a string. +# +# So given a template in YAML of: +# +# example: +# boolean_as_string: "{{ 1 == 1 }}" +# boolean_raw: "{{ (1 == 1) | bool }}" +# float_as_string: "{{ 1 / 3 }}" +# float_raw: "{{ (1 / 3) | float }}" +# number_as_string: "{{ 1 + 1 }}" +# number_raw: "{{ (1 + 1) | int }}" +# object_as_string: "{{ {'user': {'name': 'alice'}} }}" +# object_raw: "{{ {'user': {'name': 'alice'}} | object }}" +# +# Will render to: +# +# example: +# boolean_as_string: 'True' +# boolean_raw: true +# float_as_string: '0.3333333333333333' +# float_raw: 0.3333333333333333 +# number_as_string: '2' +# number_raw: 2 +# object_as_string: '{''user'': {''name'': ''alice''}}' +# object_raw: +# user: +# name: alice +type_filter_match_re = re.compile(r'^{{(?!.*{{).*\| *(bool|float|int|list|object) *}}$') + +def check_condition(condition, variables={}, template_variables={}): + return jinja2process( + template="{{ (" + condition + ") | bool}}", + variables=variables, + template_variables=template_variables, + ) + +def template_output_typing(template_out, template): + '''Convert template output to corresponding type as determined by final filter.''' + type_filter_match = type_filter_match_re.match(template) + if type_filter_match: + type_filter = type_filter_match.group(1) + try: + if type_filter == 'bool': + return bool(str2bool(template_out)) + elif type_filter == 'float': + return float(template_out) + elif type_filter == 'int': + return int(template_out) + elif type_filter == 'list': + return json.loads(template_out) + elif type_filter == 'object': + return json.loads(template_out) + except ValueError: + pass + return template_out + +def j2now(utc=False, fmt=None): + dt = datetime.now(timezone.utc if utc else None) + return dt.strftime(fmt) if fmt else dt + +j2template_cache = {} +def j2template_get(template: str): + if template in j2template_cache: + return j2template_cache[template] + j2template = jinja2env.from_string(template) + j2template_cache[template] = j2template + return j2template + + +def jinja2process(template, omit=None, variables={}, template_variables={}): + variables = copy.copy(variables) + variables['datetime'] = datetime + variables['now'] = j2now + variables['omit'] = omit + variables['timedelta'] = timedelta + variables['timezone'] = timezone + variables['timestamp'] = TimeStamp() + variables['__recursion_depth__'] = 0 + j2template = j2template_get(template) + + class TemplateVariable(object): + '''Variable may contain template string referencing other variables.''' + def __init__(self, value): + self.value = value + self.j2template = j2template_get(value) if '{{' in value or '{%' in value else None + + def get_typed_value(self): + if self.j2template: + template_out = self.__str__() + return template_output_typing(template_out, self.value) + return self.value + + def __bool__(self): + return str2bool(self.__str__()) + + def __contains__(self, item): + return item in self.get_typed_value() + + def __eq__(self, cmp): + if isinstance(TemplateVariable, cmp): + return self.get_typed_value() == cmp.get_typed_value() + else: + return self.get_typed_value() == cmp + + def __float__(self): + return float(self.__str__()) + + def __ge__(self, cmp): + if isinstance(TemplateVariable, cmp): + return self.get_typed_value() >= cmp.get_typed_value() + else: + return self.get_typed_value() >= cmp + + def __getattr__(self, key): + return self[key] + + def __getitem__(self, key): + return self.get_typed_value()[key] + + def __gt__(self, cmp): + if isinstance(TemplateVariable, cmp): + return self.get_typed_value() > cmp.get_typed_value() + else: + return self.get_typed_value() > cmp + + def __int__(self): + return int(self.__str__()) + + def __le__(self, cmp): + if isinstance(TemplateVariable, cmp): + return self.get_typed_value() <= cmp.get_typed_value() + else: + return self.get_typed_value() <= cmp + + def __len__(self): + return len(self.get_typed_value()); + + def __lt__(self, cmp): + if isinstance(TemplateVariable, cmp): + return self.get_typed_value() < cmp.get_typed_value() + else: + return self.get_typed_value() < cmp + + def __ne__(self, cmp): + return not self == cmp + + def __repr__(self): + return self.__str__() + + def __str__(self): + if self.j2template is None: + return str(self.value) + if variables['__recursion_depth__'] > MAX_RECURSION_DEPTH: + raise RuntimeError("Template variable exceeded max recursion") + variables['__recursion_depth__'] += 1 + ret = self.j2template.render(variables) + variables['__recursion_depth__'] -= 1 + return ret + + for name, value in template_variables.items(): + variables[name] = TemplateVariable(value) if isinstance(value, str) else value + + template_out = j2template.render(variables) + return template_output_typing(template_out, template) + +def recursive_process_template_strings(template, variables={}, template_variables={}): + """Take a template and recursively process template strings within it. + The template may be any type. + If it is a dictionary or list then all values will be recursively procesed. + Strings will be handled as templates and values replaced by output. + If a template string ends in a bool, float, int, or object filter then + output will be converted from string to the corresponding type. + + This method is used in various places including generating status summary, + validating parameters, checking heath/readiness, and generating resource definitions. + + Keyword arguments: + variables -- simple key/value pair variables + template_variables -- variables which may contain template strings and should only come from trusted sources + """ + omit = '__omit_place_holder__' + ''.join(random.choices('abcdef0123456789', k=40)) + return __recursive_strip_omit( + __recursive_process_template_strings( + omit = omit, + template = template, + variables = variables, + template_variables = template_variables, + ), + omit = omit, + ) + +def __recursive_process_template_strings(template, omit, variables, template_variables): + if isinstance(template, dict): + return { + key: __recursive_process_template_strings(val, omit=omit, variables=variables, template_variables=template_variables) + for key, val in template.items() + } + elif isinstance(template, list): + return [ + __recursive_process_template_strings(item, omit=omit, variables=variables, template_variables=template_variables) + for item in template + ] + elif isinstance(template, str): + return jinja2process(template, omit=omit, variables=variables, template_variables=template_variables) + else: + return template + +def __recursive_strip_omit(value, omit): + if isinstance(value, dict): + return { + key: __recursive_strip_omit(val, omit=omit) + for key, val in value.items() + if val != omit + } + elif isinstance(value, list): + return [ + __recursive_strip_omit(item, omit=omit) for item in value if item != omit + ] + elif value != omit: + return value diff --git a/client/python/babylon_async/src/babylon_async/resourceclaim.py b/client/python/babylon_async/src/babylon_async/resourceclaim.py new file mode 100644 index 000000000..1e920837f --- /dev/null +++ b/client/python/babylon_async/src/babylon_async/resourceclaim.py @@ -0,0 +1,547 @@ +from __future__ import annotations +from typing import Any, List, Mapping + +from datetime import datetime, timedelta, timezone + +import pytimeparse + +from .exceptions import BabylonApiException +from .k8s_object import K8sObject +from .resourcehandle import ResourceHandle +from .resourcereference import ResourceReference + +class ResourceClaim(K8sObject): + api_group = "poolboy.gpte.redhat.com" + api_version = "v1" + kind = "ResourceClaim" + plural = "resourceclaims" + api_group_version = f"{api_group}/{api_version}" + + @classmethod + async def create_with_provider(cls, client, + namespace:str, + provider_name:str, + auto_detach:bool=False, + name:str|None=None, + owner:K8sObject|None=None, + parameter_values:Mapping[str,Any]|None=None, + ): + if name is None: + name = f"{provider_name}-*" + if parameter_values is None: + parameter_values = {} + definition = { + "spec": { + "provider": { + "name": provider_name, + "parameterValues": parameter_values, + } + } + } + if auto_detach: + definition["spec"] = { + "when": "status.resources | json_query(\"[?state.spec.vars.current_state == 'provision-failed']\") | length != 0", + } + + return await ResourceClaim.create( + client=client, + definition=definition, + name=name, + namespace=namespace, + owner=owner, + ) + + @property + def asset_uuid(self) -> str|None: + # Fetching asset uuid from status is more reliable than annotations + if self.status is not None or self.status.summary is not None and 'agnosticv' in self.status.summary: + return self.status.summary['agnosticv']['asset_uuid'] + # Fallback to attempt to get from annotations + if self.metadata.annotations is not None: + return self.metadata.annotations.get('gpte.redhat.com/asset-uuid') + return None + + @property + def catalog_display_name(self) -> str|None: + # Try getting the catalog display name from annotations, but it is better + # get the display name by fetching the CatalogItem object itself. + if self.metadata.annotations is None: + return None + return self.metadata.annotations.get('babylon.gpte.redhat.com/catalogDisplayName') + + @property + def catalog_item_display_name(self) -> str|None: + # Try getting the catalog item display name from annotations, but it is better + # get the display name by fetching the CatalogItem object itself. + if self.metadata.annotations is None: + return None + return self.metadata.annotations.get('babylon.gpte.redhat.com/catalogItemDisplayName') + + @property + def catalog_item_name(self) -> str|None: + # Fetching catalog item name from status is more reliable than annotations + if self.status is not None or self.status.summary is not None and 'catalog_item_name' in self.status.summary: + return self.status.summary['catalog_item_name'] + # Fallback to attempt to fetch from labels + if self.metadata.labels is not None: + return self.metadata.labels.get('babylon.gpte.redhat.com/catalogItemName') + return None + + @property + def catalog_item_namespace(self) -> str|None: + # Fetching catalog item namespace from status is more reliable than annotations + if self.status is not None or self.status.summary is not None and 'catalog_item_namespace' in self.status.summary: + return self.status.summary['catalog_item_namespace'] + # Fallback to attempt to fetch from labels + if self.metadata.labels is not None: + return self.metadata.labels.get('babylon.gpte.redhat.com/catalogItemNamespace') + return None + + @property + def guid(self) -> str|None: + if self.status.resource_handle is None: + return None + return self.status.resource_handle.name[5:] + + @property + def lifespan_end_datetime(self) -> datetime|None: + """Return lifespan end if set""" + try: + return self.status.lifespan.end_datetime + except AttributeError: + return None + + @property + def lifespan_start_datetime(self) -> datetime|None: + """Return lifespan start if set""" + try: + return self.status.lifespan.start_datetime + except AttributeError: + return None + + @property + def provision_data(self) -> Mapping|None: + if self.status is None or self.status.summary is None: + return None + return self.status.summary.get('provision_data') + + @property + def requested_lifespan_end_datetime(self) -> datetime: + """Return requested lifespan end datetime.""" + lifespan = self.spec.lifespan + if lifespan is None: + return None + return lifespan.end_datetime + + @property + def requested_lifespan_end_timestamp(self) -> str|None: + """Return requested lifespan end as timestamp string""" + lifespan = self.spec.lifespan + if lifespan is None: + return None + return lifespan.end_timestamp + + @property + def requested_stop_datetime(self) -> datetime|None: + """Return requested stop datetime if set.""" + ts = self.requested_stop_timestamp + if ts is None: + return None + return datetime.strptime(ts, '%Y-%m-%dT%H:%M:%S%z') + + @property + def requested_stop_timestamp(self) -> str|None: + """Return requested stop timestamp if set.""" + try: + return self.spec.provider.parameter_values.get('stop_timestamp') + except AttributeError: + return None + + @property + def resource_handle(self) -> ResourceReference|None: + return self.status.resource_handle + + @property + def resource_handle_name(self) -> str|None: + """Return ResourceHandle name if defined.""" + try: + return self.status.resource_handle.name + except AttributeError: + return None + + @property + def runtime_default(self) -> timedelta|None: + if self.status is None or self.status.summary is None or 'runtime_default' not in self.status.summary: + return None + return timedelta(seconds=pytimeparse.parse(self.status.summary['runtime_default'])) + + @property + def runtime_maximum(self) -> timedelta|None: + if self.status is None or self.status.summary is None or 'runtime_maximum' not in self.status.summary: + return None + return timedelta(seconds=pytimeparse.parse(self.status.summary['runtime_maximum'])) + + @property + def spec(self) -> ResourceClaimSpec: + return ResourceClaimSpec(self._definition['spec']) + + @property + def state(self) -> str|None: + if self.status is None or self.status.summary is None: + return None + return self.status.summary.get('state') + + @property + def status(self) -> ResourceClaimStatus|None: + if 'status' not in self._definition: + return None + return ResourceClaimStatus(self._definition['status']) + + @property + def stop_datetime(self) -> datetime|None: + """Return effective stop datetime if set.""" + ts = self.stop_timestamp + if ts is None: + return None + return datetime.strptime(ts, '%Y-%m-%dT%H:%M:%S%z') + + @property + def stop_timestamp(self) -> str|None: + """Return effective stop timestamp if set.""" + provider = self.status.provider + if provider is None: + return None + return provider.parameter_values.get('stop_timestamp') + + @property + def white_glove(self) -> bool: + if self.metadata.labels is None: + return False + return self.metadata.labels.get('demo.redhat.com/white-glove') == 'true' + + @property + def workshop_id(self) -> str|None: + if self.metadata.labels is not None: + return self.metadata.labels.get('babylon.gpte.redhat.com/workshop-id') + return None + + @property + def workshop_provision_name(self) -> str|None: + if self.metadata.owner_references is not None: + for owner_reference in self.metadata.owner_references: + if owner_reference.kind == 'WorkshopProvision': + return owner_reference.name + if self.metadata.labels is not None: + return self.metadata.labels.get('babylon.gpte.redhat.com/workshop-provision') + return None + + @property + def workshop_uid(self) -> str|None: + if self.metadata.labels is not None: + return self.metadata.labels.get('babylon.gpte.redhat.com/workshop-uid') + return None + + async def disable_autodestroy(self) -> bool: + """Set lifespan end such that it is effectively disaled. + Returns boolean indicating if change was made.""" + + # Treat stop at 2100-12-31T00:00:00Z as effectively never + never = datetime(2100, 12, 31, tzinfo=timezone.utc) + + lifespan_end = self.lifespan_end_datetime + if lifespan_end is not None and lifespan_end >= never: + return False + + await self.set_lifespan_end(never, True) + + async def disable_autostop(self) -> bool: + """Set auto-stop schedule such that stop is effectively disabled. + Returns boolean indicating if change was made.""" + + # Treat stop at 2100-12-31T00:00:00Z as effectively never + never = datetime(2100, 12, 31, tzinfo=timezone.utc) + + stop_datetime = self.stop_datetime + if stop_datetime is not None and stop_datetime >= never: + return False + + await self.set_requested_stop_datetime(never) + return True + + async def get_resource_handle(self) -> ResourceHandle|None: + """Return ResourceHandle associated with this ResourceClaim""" + name = self.resource_handle_name + if name is None: + return None + return await self.client.get_resource_handle(name) + + async def set_lifespan_end(self, dt:datetime, override_limits:bool=False) -> None: + """Set lifespan end for ResourceClaim. + If limit_override is set then also update ResourceHandle to ensure + value will be applied.""" + if override_limits: + try: + resource_handle = await self.get_resource_handle() + if resource_handle: + req_maximum = dt - self.lifespan_start_datetime + req_relative_maximum = dt - datetime.now(timezone.utc) + + set_values = {"end": dt} + if resource_handle.lifespan_maximum_timedelta < req_maximum: + set_values['maximum'] = req_maximum + if resource_handle.lifespan_relative_maximum_timedelta < req_relative_maximum: + set_values['relative_maximum'] = req_relative_maximum + + await resource_handle.set_lifespan(**set_values) + except BabylonApiException as err: + if err.status != 404: + raise + + if dt == self.requested_lifespan_end_datetime: + return + + await self.patch({ + "spec": { + "lifespan": { + "end": dt.strftime('%FT%TZ') + } + } + }) + + + async def set_requested_stop_datetime(self, dt:datetime) -> None: + """Set auto-stop schedule to specified datetime.""" + await self.patch([{ + "op": "add", + "path": "/spec/provider/parameterValues/stop_timestamp", + "value": dt.strftime('%FT%TZ'), + }]) + + +class ResourceClaimSpec: + def __init__(self, definition): + self._definition = definition + + @property + def auto_detach(self) -> ResourceClaimSpecAutoDetach|None: + if 'autoDetach' not in self._definition: + return None + return ResourceClaimSpecAutoDetach(self._definition['autoDetach']) + + @property + def lifespan(self) -> ResourceClaimSpecLifespan|None: + if 'lifespan' not in self._definition: + return None + return ResourceClaimSpecLifespan(self._definition['lifespan']) + + @property + def provider(self) -> ResourceClaimSpecProvider|None: + if 'provider' not in self._definition: + return None + return ResourceClaimSpecProvider(self._definition['provider']) + + +class ResourceClaimSpecAutoDetach: + def __init__(self, definition): + self._definition = definition + + @property + def when(self) -> str: + return self._definition['when'] + +class ResourceClaimSpecLifespan: + def __init__(self, definition): + self._definition = definition + + @property + def end(self) -> datetime|None: + """Alias for end_datetime""" + return self.end_datetime + + @property + def end_datetime(self) -> datetime|None: + """Return requested lifespan end as datetime object""" + ts = self.end_timestamp + if ts is None: + return None + return datetime.strptime(ts, '%Y-%m-%dT%H:%M:%S%z') + + @property + def end_timestamp(self) -> str|None: + """Return requested lifespan end timestamp string if defined""" + return self._definition.get('end') + + @property + def start(self) -> datetime|None: + """Alias for start_datetime""" + return self.start_datetime + + @property + def start_datetime(self) -> datetime|None: + """Return requested lifespan start as datetime object""" + ts = self.start_timestamp + if ts is None: + return None + return datetime.strptime(ts, '%Y-%m-%dT%H:%M:%S%z') + + @property + def start_timestamp(self) -> str|None: + """Return requested lifespan start timestamp string if defined""" + return self._definition.get('start') + +class ResourceClaimSpecProvider: + def __init__(self, definition): + self._definition = definition + + @property + def name(self) -> str: + return self._definition['name'] + + @property + def parameter_values(self) -> Mapping[str, Any]: + return self._definition.get('parameterValues', {}) + +class ResourceClaimStatus: + def __init__(self, definition): + self._definition = definition + + @property + def healthy(self) -> bool|None: + return self._definition.get('healthy') + + @property + def lifespan(self) -> ResourceClaimStatusLifespan|None: + if 'lifespan' not in self._definition: + return None + return ResourceClaimStatusLifespan(self._definition['lifespan']) + + @property + def provider(self) -> ResourceClaimStatusProvider|None: + if 'provider' not in self._definition: + return None + return ResourceClaimStatusProvider(self._definition['provider']) + + @property + def ready(self) -> bool|None: + return self._definition.get('ready') + + @property + def resource_handle(self) -> ResourceReference|None: + if 'resourceHandle' not in self._definition: + return None + return ResourceReference(self._definition['resourceHandle']) + + @property + def resources(self) -> List[ResourceClaimStatusResource]|None: + if 'resources' not in self._definition: + return None + return [ + ResourceClaimStatusResource(item) + for item in self._definition['resources'] + ] + + @property + def summary(self) -> Mapping|None: + return self._definition.get('summary') + +class ResourceClaimStatusLifespan: + def __init__(self, definition): + self._definition = definition + + @property + def end(self) -> datetime|None: + """Alias for end_datetime""" + return self.end_datetime + + @property + def end_datetime(self) -> datetime|None: + """Return lifespan end as datetime object""" + ts = self.end_timestamp + if ts is None: + return None + return datetime.strptime(ts, '%Y-%m-%dT%H:%M:%S%z') + + @property + def end_timestamp(self) -> str|None: + """Return lifespan end timestamp string if defined""" + return self._definition.get('end') + + @property + def first_ready(self) -> datetime|None: + if 'firstReady' in self._definition: + return datetime.strptime(self._definition['firstReady'], '%Y-%m-%dT%H:%M:%S%z') + return None + + @property + def maximum(self) -> timedelta|None: + if 'maximum' in self._definition: + return timedelta(seconds=pytimeparse.parse(self._definition['maximum'])) + return None + + @property + def relative_maximum(self) -> timedelta|None: + if 'relativeMaximum' in self._definition: + return timedelta(seconds=pytimeparse.parse(self._definition['relativeMaximum'])) + return None + + @property + def start(self) -> datetime|None: + """Alias for start_datetime""" + return self.start_datetime + + @property + def start_datetime(self) -> datetime|None: + """Return lifespan start as datetime object""" + ts = self.start_timestamp + if ts is None: + return None + return datetime.strptime(ts, '%Y-%m-%dT%H:%M:%S%z') + + @property + def start_timestamp(self) -> str|None: + """Return lifespan start timestamp string if defined""" + return self._definition.get('start') + +class ResourceClaimStatusProvider: + def __init__(self, definition): + self._definition = definition + + @property + def name(self) -> str: + return self._definition['name'] + + @property + def parameter_values(self) -> Mapping[str, Any]: + return self._definition.get('parameterValues', {}) + +class ResourceClaimStatusResource: + def __init__(self, definition): + self._definition = definition + + @property + def healthy(self) -> bool|None: + return self._definition.get('healthy') + + @property + def name(self) -> str|None: + return self._definition.get('name') + + @property + def provider(self) -> ResourceReference|None: + if 'provider' not in self._definition: + return None + return ResourceReference(self._definition['provider']) + + @property + def ready(self) -> bool|None: + return self._definition.get('ready') + + @property + def reference(self) -> ResourceReference|None: + if 'reference' not in self._definition: + return None + return ResourceReference(self._definition['reference']) + + @property + def state(self) -> Mapping|None: + return self._definition.get('state') diff --git a/client/python/babylon_async/src/babylon_async/resourcehandle.py b/client/python/babylon_async/src/babylon_async/resourcehandle.py new file mode 100644 index 000000000..76fee4912 --- /dev/null +++ b/client/python/babylon_async/src/babylon_async/resourcehandle.py @@ -0,0 +1,249 @@ +from __future__ import annotations +from typing import Any, Mapping + +from datetime import datetime, timedelta + +import pytimeparse + +from .k8s_object import K8sObject +from .poolboy_templating import timedelta_to_str + +class ResourceHandle(K8sObject): + api_group = "poolboy.gpte.redhat.com" + api_version = "v1" + kind = "ResourceHandle" + plural = "resourcehandles" + api_group_version = f"{api_group}/{api_version}" + + @classmethod + async def create_with_provider(cls, client, + provider_name:str, + lifespan:Mapping[str,str|datetime|timedelta]=None, + owner:K8sObject|None=None, + parameter_values:Mapping[str,Any]=None, + ) -> ResourceHandle: + if lifespan is None: + lifespan = {} + if name is None: + name = f"{provider_name}-*" + if parameter_values is None: + parameter_values = {} + definition = { + "spec": { + "lifespan": { + key: ( + timedelta_to_str(value) if isinstance(value, timedelta) else + value.strftime('%FT%TZ') if isinstance(value, datetime) else + value + ) for key, value in lifespan.items() + }, + "provider": { + "name": provider_name, + "parameterValues": parameter_values, + } + } + } + + return await ResourceHandle.create( + client=client, + definition=definition, + name="guid-*", + namespace="poolboy", + owner=owner, + ) + + @property + def asset_uuid(self) -> str|None: + # Fetching asset uuid from status is more reliable than annotations + if self.status is not None or self.status.summary is not None and 'agnosticv' in self.status.summary: + return self.status.summary['agnosticv']['asset_uuid'] + # Fallback to attempt to get from annotations + if self.metadata.annotations is not None: + return self.metadata.annotations.get('gpte.redhat.com/asset-uuid') + return None + + @property + def catalog_display_name(self) -> str|None: + # Try getting the catalog display name from annotations, but it is better + # get the display name by fetching the CatalogItem object itself. + if self.metadata.annotations is None: + return None + return self.metadata.annotations.get('babylon.gpte.redhat.com/catalogDisplayName') + + @property + def catalog_item_display_name(self) -> str|None: + # Try getting the catalog item display name from annotations, but it is better + # get the display name by fetching the CatalogItem object itself. + if self.metadata.annotations is None: + return None + return self.metadata.annotations.get('babylon.gpte.redhat.com/catalogItemDisplayName') + + @property + def catalog_item_name(self) -> str|None: + # Fetching catalog item name from status is more reliable than annotations + if self.status is not None or self.status.summary is not None and 'catalog_item_name' in self.status.summary: + return self.status.summary['catalog_item_name'] + # Fallback to attempt to fetch from labels + if self.metadata.labels is not None: + return self.metadata.labels.get('babylon.gpte.redhat.com/catalogItemName') + return None + + @property + def catalog_item_namespace(self) -> str|None: + # Fetching catalog item namespace from status is more reliable than annotations + if self.status is not None or self.status.summary is not None and 'catalog_item_namespace' in self.status.summary: + return self.status.summary['catalog_item_namespace'] + # Fallback to attempt to fetch from labels + if self.metadata.labels is not None: + return self.metadata.labels.get('babylon.gpte.redhat.com/catalogItemNamespace') + return None + + @property + def guid(self) -> str|None: + return self.name[5:] + + @property + def lifespan_end_datetime(self) -> datetime|None: + """Return lifespan end as datetime if set""" + try: + return self.spec.lifespan.end_datetime + except AttributeError: + return None + + @property + def lifespan_maximum_timedelta(self) -> timedelta: + return self.spec.lifespan.maximum_timedelta + + @property + def lifespan_relative_maximum_timedelta(self) -> timedelta: + return self.spec.lifespan.relative_maximum_timedelta + + @property + def provision_data(self) -> Mapping|None: + if self.status is None or self.status.summary is None: + return None + return self.status.summary.get('provision_data') + + @property + def spec(self) -> ResourceHandleSpec: + return ResourceHandleSpec(self._definition['spec']) + + @property + def status(self) -> ResourceHandleStatus|None: + if 'status' not in self._definition: + return None + return ResourceHandleStatus(self._definition['status']) + + async def set_lifespan(self, + default:timedelta|str|None=None, + end:datetime|str|None=None, + maximum:timedelta|str|None=None, + relative_maximum:timedelta|str|None=None, + ): + if isinstance(default, datetime): + default = timedelta_to_str(default) + if isinstance(end, datetime): + end = end.strftime('%FT%TZ') + if isinstance(maximum, timedelta): + maximum = timedelta_to_str(maximum) + if isinstance(relative_maximum, timedelta): + relative_maximum = timedelta_to_str(relative_maximum) + + lifespan = {} + if default is not None: + lifespan['default'] = default + if end is not None: + lifespan['end'] = end + if maximum is not None: + lifespan['maximum'] = maximum + if relative_maximum is not None: + lifespan['relativeMaximum'] = relative_maximum + + await self.patch({"spec": {"lifespan": lifespan}}) + + +class ResourceHandleSpec: + def __init__(self, definition): + self._definition = definition + + @property + def lifespan(self) -> ResourceHandleSpecLifespan|None: + if 'lifespan' not in self._definition: + return None + return ResourceHandleSpecLifespan(self._definition['lifespan']) + + @property + def provider(self) -> ResourceHandleSpecProvider|None: + if 'provider' not in self._definition: + return None + return ResourceHandleSpecProvider(self._definition['provider']) + +class ResourceHandleSpecLifespan: + def __init__(self, definition): + self._definition = definition + + @property + def end(self) -> datetime|None: + """Alias for end_datetime""" + return self.end_datetime + + @property + def end_datetime(self) -> datetime|None: + """Return requested lifespan end as datetime object""" + ts = self.end_timestamp + if ts is None: + return None + return datetime.strptime(ts, '%Y-%m-%dT%H:%M:%S%z') + + @property + def end_timestamp(self) -> str|None: + """Return requested lifespan end timestamp string if defined""" + return self._definition.get('end') + + @property + def maximum(self) -> str: + """Return lifespan maximum""" + return self._definition['maximum'] + + @property + def maximum_timedelta(self) -> timedelta: + """Return lifespan maximum as timedelta""" + return timedelta(seconds=pytimeparse.parse(self.maximum)) + + @property + def relative_maximum(self) -> str: + """Return lifespan relative maximum""" + return self._definition['relativeMaximum'] + + @property + def relative_maximum_timedelta(self) -> timedelta: + """Return lifespan relative maximum as timedelta""" + return timedelta(seconds=pytimeparse.parse(self.relative_maximum)) + +class ResourceHandleSpecProvider: + def __init__(self, definition): + self._definition = definition + + @property + def name(self) -> str: + return self._definition['name'] + + @property + def parameter_values(self) -> Mapping[str, Any]: + return self._definition.get('parameterValues', {}) + +class ResourceHandleStatus: + def __init__(self, definition): + self._definition = definition + + @property + def healthy(self) -> bool|None: + return self._definition.get('healthy') + + @property + def ready(self) -> bool|None: + return self._definition.get('ready') + + @property + def summary(self) -> Mapping|None: + return self._definition.get('summary') diff --git a/client/python/babylon_async/src/babylon_async/resourcepool.py b/client/python/babylon_async/src/babylon_async/resourcepool.py new file mode 100644 index 000000000..4a20fc46a --- /dev/null +++ b/client/python/babylon_async/src/babylon_async/resourcepool.py @@ -0,0 +1,60 @@ +from __future__ import annotations +from typing import Any, List, Mapping + +from .k8s_object import K8sObject +from .resourcereference import ResourceReference + +class ResourcePool(K8sObject): + api_group = "poolboy.gpte.redhat.com" + api_version = "v1" + kind = "ResourcePool" + plural = "resourcepools" + api_group_version = f"{api_group}/{api_version}" + + @property + def spec(self) -> ResourcePoolSpec: + return ResourcePoolSpec(self._definition['spec']) + + +class ResourcePoolSpec: + def __init__(self, definition): + self._definition = definition + + @property + def resources(self) -> List[ResourcePoolSpecResource]: + return [ + ResourcePoolSpecResource(item) + for item in self._definition.get('resources', []) + ] + +class ResourcePoolSpecResource: + def __init__(self, definition): + self._definition = definition + + @property + def provider(self) -> ResourcePoolSpecResourceProvider: + return ResourcePoolSpecResourceProvider(self._definition['provider']) + + @property + def template(self) -> Mapping[str, Any]: + return self.definition.get('template', {}) + +class ResourcePoolSpecResourceProvider: + def __init__(self, definition): + self._definition = definition + + @property + def api_version(self) -> str: + return self._definition.get('apiVersion') + + @property + def kind(self) -> str: + return self._definition.get('kind') + + @property + def name(self) -> str: + return self._definition.get('name') + + @property + def namespace(self) -> str: + return self._definition.get('namespace') diff --git a/client/python/babylon_async/src/babylon_async/resourceprovider.py b/client/python/babylon_async/src/babylon_async/resourceprovider.py new file mode 100644 index 000000000..9434e9d47 --- /dev/null +++ b/client/python/babylon_async/src/babylon_async/resourceprovider.py @@ -0,0 +1,306 @@ +from __future__ import annotations +from typing import Any, List, Mapping + +from .k8s_object import K8sObject +from .poolboy_templating import check_condition, recursive_process_template_strings + +class ResourceProvider(K8sObject): + api_group = "poolboy.gpte.redhat.com" + api_version = "v1" + kind = "ResourceProvider" + plural = "resourceproviders" + api_group_version = f"{api_group}/{api_version}" + + @property + def has_template_definition(self) -> bool: + return self.spec.has_template_definition + + @property + def linked_resource_providers(self) -> list[ResourceProviderSpecLinkedResourceProvider]: + return self.spec.linked_resource_providers + + @property + def parameters(self) -> List[ResourceProviderSpecParameter]: + return self.spec.parameters + + @property + def resource_name(self) -> str: + return self._definition.get('resourceName') or self.name + + @property + def spec(self) -> ResourceProviderSpec: + return ResourceProviderSpec(self._definition['spec']) + + @property + def vars(self) -> Mapping[str, Any]: + return self.spec.vars + + def get_parameter_defaults(self, parameter_values:Mapping) -> Mapping: + ret = {} + for parameter in self.parameters: + value = parameter.get_default( + variables=parameter_values, + template_variables=self.vars, + ) + if value is not None: + ret[parameter.name] = value + return ret + + def processed_template(self, + parameter_values: Mapping, + ) -> Mapping: + return recursive_process_template_strings( + self.spec.template.definition, + variables=parameter_values, + template_variables=self.vars, + ) + + async def get_resources(self, + parameter_values: Mapping, + cache: bool=False, + resource_name: str|None=None, + ): + """Generate spec resources as used in ResourceHandle with given parameter_values""" + + parameter_values = { + **self.get_parameter_defaults(parameter_values=parameter_values), + **parameter_values, + } + + resources = [] + for linked_resource_provider in self.linked_resource_providers: + if not linked_resource_provider.check_when( + parameter_values=parameter_values, + resource_provider=self, + ): + continue + resource_provider = await self.client.get_resource_provider( + cache=cache, + name=linked_resource_provider.name, + ) + resources.extend( + await resource_provider.get_resources( + cache=cache, + parameter_values={ + key: recursive_process_template_strings( + value, + variables={ + **resource_provider.get_parameter_defaults( + parameter_values=parameter_values + ), + **parameter_values, + }, + template_variables={ + "resources": [], + "resource_claim": {}, + **self.vars, + } + ) + for key, value in linked_resource_provider.parameter_values.items() + } + ) + ) + + if self.has_template_definition: + resources.append({ + "name": resource_name or self.resource_name, + "provider": self.as_reference(), + "template": self.processed_template(parameter_values=parameter_values), + }) + + return resources + + async def get_linked_resource_providers(self, cache:bool=False): + linked_resource_providers = [] + for linked_resource_provider in self.linked_resource_providers: + resource_provider = await self.client.get_resource_provider( + cache=cache, + name=linked_resource_provider.name, + ) + linked_resource_providers.append(resource_provider) + return linked_resource_providers + +class ResourceProviderSpec: + def __init__(self, definition): + self._definition = definition + + @property + def default(self) -> Mapping: + return self._definition.get('default', {}) + + @property + def has_template_definition(self) -> bool: + return 'override' in self._definition or ( + self.template is not None and + self.template.definition is not None + ) + + @property + def linked_resource_providers(self) -> List[ResourceProviderSpecLinkedResourceProvider]: + return [ + ResourceProviderSpecLinkedResourceProvider(item) + for item in self._definition.get('linkedResourceProviders', []) + ] + + @property + def parameters(self) -> List[ResourceProviderSpecParameter]: + return [ + ResourceProviderSpecParameter(item) + for item in self._definition.get('parameters', []) + ] + + @property + def match_ignore(self) -> list[str]: + return self._definition.get('match_ignore', []) + + @property + def override(self) -> Mapping: + return self._definition.get('override', {}) + + @property + def resource_name(self) -> str|None: + return self._definition.get('resourceName') + + @property + def template(self) -> ResourceProviderSpecTemplate|None: + if 'template' not in self._definition: + return None + return ResourceProviderSpecTemplate(self._definition['template']) + + @property + def vars(self) -> Mapping[str, Any]: + return self._definition.get('vars', {}) + +class ResourceProviderSpecLinkedResourceProvider: + def __init__(self, definition): + self._definition = definition + + @property + def name(self) -> str: + return self._definition['name'] + + @property + def parameter_values(self) -> Mapping[str, Any]: + return self._definition.get('parameterValues', {}) + + @property + def resource_name(self) -> str: + return self._definition.get('resourceName', self.name) + + @property + def template_vars(self) -> List[ResourceProviderSpecLinkedResourceProviderTemplateVar]: + return [ + ResourceProviderSpecLinkedResourceProviderTemplateVar(item) + for item in self._definition.get('templateVars', []) + ] + + @property + def wait_for(self) -> str|None: + return self._definition.get('waitFor') + + @property + def when(self) -> str|None: + return self._definition.get('when') + + def check_when(self, + parameter_values: Mapping, + resource_provider: ResourceProvider, + ): + if not self.when: + return True + return check_condition( + self.when, + variables={ + **parameter_values, + **self.parameter_values, + }, + template_variables=resource_provider.spec.vars, + ) + +class ResourceProviderSpecLinkedResourceProviderTemplateVar: + def __init__(self, definition): + self._definition = definition + + @property + def _from(self) -> str: + return self._definition['from'] + + @property + def name(self) -> str: + return self._definition['name'] + +class ResourceProviderSpecParameter: + def __init__(self, definition): + self._definition = definition + + @property + def allow_update(self) -> bool: + return self._definition.get('allowUpdate', False) + + @property + def default(self) -> ResourceProviderSpecParameterDefault|None: + if 'default' not in self._definition: + return None + return ResourceProviderSpecParameterDefault(self._definition['default']) + + @property + def name(self) -> str: + return self._definition['name'] + + @property + def required(self) -> bool: + return self._definition.get('required', False) + + @property + def validation(self) -> ResourceProviderSpecParameterValidation|None: + if 'validation' not in self._definition: + return None + return ResourceProviderSpecParameterValidation(self._definition['validation']) + + def get_default(self, variables:Mapping, template_variables:Mapping) -> Any: + if self.default is not None: + if self.default.template is not None: + return recursive_process_template_strings( + self.default.template, + variables=variables, + template_variables=template_variables, + ) + if self.default.value is not None: + return self.default.value + + if self.validation is not None and self.validation.openapi_v3_schema is not None: + return self.validation.openapi_v3_schema.get('default') + + return None + +class ResourceProviderSpecParameterDefault: + def __init__(self, definition): + self._definition = definition + + @property + def template(self) -> Any: + return self._definition.get('template') + + @property + def value(self) -> Any: + return self._definition.get('value') + +class ResourceProviderSpecParameterValidation: + def __init__(self, definition): + self._definition = definition + + @property + def openapi_v3_schema(self) -> Mapping|None: + return self._definition.get('openAPIV3Schema') + +class ResourceProviderSpecTemplate: + def __init__(self, definition): + self._definition = definition + + @property + def definition(self) -> Mapping|None: + return self._definition.get('definition') + + @property + def enable(self) -> bool: + return self._definition.get('enable', False) diff --git a/client/python/babylon_async/src/babylon_async/resourcereference.py b/client/python/babylon_async/src/babylon_async/resourcereference.py new file mode 100644 index 000000000..469a39574 --- /dev/null +++ b/client/python/babylon_async/src/babylon_async/resourcereference.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +class ResourceReference: + def __init__(self, definition): + self.definition = definition + + @property + def api_version(self) -> str: + return self.definition['apiVersion'] + + @property + def kind(self) -> str: + return self.definition['kind'] + + @property + def name(self) -> str: + return self.definition['name'] + + @property + def namespace(self) -> str|None: + return self.definition.get('namespace') + + @property + def uid(self) -> str|None: + return self.definition.get('uid') diff --git a/client/python/babylon_async/src/babylon_async/tenantclusterpool.py b/client/python/babylon_async/src/babylon_async/tenantclusterpool.py new file mode 100644 index 000000000..f34f85ddc --- /dev/null +++ b/client/python/babylon_async/src/babylon_async/tenantclusterpool.py @@ -0,0 +1,400 @@ +from __future__ import annotations + +import json + +from hashlib import sha1 +from typing import Any, List, Mapping + +from .exceptions import BabylonApiException +from .k8s_object import K8sObject + +class TenantClusterPool(K8sObject): + api_group = "babylon.gpte.redhat.com" + api_version = "v1" + kind = "TenantClusterPool" + plural = "tenantclusterpools" + api_group_version = f"{api_group}/{api_version}" + + @property + def cluster_provisioning(self) -> TenantClusterPoolSpecClusterProvisioning: + """Configuration for provisioning clusters which provide capacity for tenants.""" + return self.spec.cluster_provisioning + + @property + def clusters(self) -> List[TenantClusterPoolStatusCluster]: + """Return clusters from status if defined else empty list.""" + status = self.status + if status is None: + return [] + return self.status.clusters or [] + + @property + def max_clusters(self) -> int|None: + """Maximum number of clusters to provision. If undefined then no + maximum is applied.""" + return self.spec.max_clusters + + @property + def min_available_sandbox_placements(self) -> int: + """Minimum number of unassigned sandbox placements to maintain + on shared clusters. A new shared cluster will be provisioned + if less than this number remain.""" + return self.spec.min_available_sandbox_placements + + @property + def min_clusters(self) -> int: + """Minimum number of clusters to provision. If undefined then no + minimum is applied.""" + return self.spec.min_clusters or 0 + + @property + def sandbox_host(self) -> TenantClusterPoolSpecSandboxHost: + """Configuration for registering clusters as tenant hosts with the Sandbox API.""" + return self.spec.sandbox_host + + @property + def spec(self) -> TenantClusterPoolSpec: + """Configuration for TenantClusterPool""" + return TenantClusterPoolSpec(self._definition['spec']) + + @property + def status(self) -> TenantClusterPoolStatus|None: + """Status of TenantClusterPool""" + if 'status' not in self._definition: + return None + return TenantClusterPoolStatus(self._definition['status']) + + async def add_cluster_to_status(self, + resource_claim_name:str, + retries:int=10, + sandbox_api_state:str="pending" + ) -> None: + """Add cluster named by resource claim to status.""" + name = resource_claim_name.split('.', 1)[1].replace('.', '-') + for attempt in range(retries+1): + # Loop until successfully added + patch = [] + if self.status is None: + patch.append({ + "op": "test", + "path": "/status", + "value": None, + }) + patch.append({ + "op": "add", + "path": "/status", + "value": { + "clusters": [], + } + }) + elif self.status.clusters is None: + patch.append({ + "op": "test", + "path": "/status/clusters", + "value": None, + }) + patch.append({ + "op": "add", + "path": "/status/clusters", + "value": [], + }) + + patch.append({ + "op": "add", + "path": "/status/clusters/-", + "value": { + "name": name, + "resourceClaimName": resource_claim_name, + "sandboxApiState": sandbox_api_state, + } + }) + + try: + await self.patch_status(patch) + return + except BabylonApiException as err: + if attempt == retries: + raise + if err.status != 422: + raise + await self.refresh() + + async def remove_cluster_from_status(self, + resource_claim_name:str, + retries:int=10, + ) -> None: + """Remove cluster named by resource claim from status.""" + for attempt in range(retries+1): + for idx, cluster in reversed(list(enumerate(self.status.clusters))): + if cluster.resource_claim_name != resource_claim_name: + continue + try: + await self.patch_status([{ + "op": "test", + "path": f"/status/clusters/{idx}/resourceClaimName", + "value": resource_claim_name, + }, { + "op": "remove", + "path": f"/status/clusters/{idx}", + }]) + except BabylonApiException as err: + if attempt == retries: + raise + if err.status != 422: + raise + await self.refresh() + continue + return + + async def set_cluster_sandbox_api_state(self, + resource_claim_name:str, + sandbox_api_state:str, + retries:int=10, + ) -> None: + """Set sandbox api state of cluster in status.""" + for attempt in range(retries+1): + for idx, cluster in reversed(list(enumerate(self.status.clusters))): + if cluster.resource_claim_name != resource_claim_name: + continue + try: + await self.patch_status([{ + "op": "test", + "path": f"/status/clusters/{idx}/resourceClaimName", + "value": resource_claim_name, + }, { + "op": "add", + "path": f"/status/clusters/{idx}/sandboxApiState", + "value": sandbox_api_state, + }]) + except BabylonApiException as err: + if attempt == retries: + raise + if err.status != 422: + raise + await self.refresh() + continue + return + + +class TenantClusterPoolSpec: + """Configuration for TenantClusterPool""" + def __init__(self, definition): + self._definition = definition + + @property + def cluster_provisioning(self) -> TenantClusterPoolSpecClusterProvisioning: + """Configuration for provisioning clusters which provide capacity for tenants.""" + return TenantClusterPoolSpecClusterProvisioning( + self._definition['clusterProvisioning'], + ) + + @property + def max_clusters(self) -> int|None: + """Maximum number of clusters to provision. If undefined then no + maximum is applied.""" + return self._definition.get('maxClusters') + + @property + def min_available_sandbox_placements(self) -> int: + """Minimum number of unassigned sandbox placements to maintain + on shared clusters. A new shared cluster will be provisioned + if less than this number remain.""" + return self._definition.get('minAvailableSandboxPlacements', 0) + + @property + def min_clusters(self) -> int: + """Maximum number of clusters to provision. If undefined then no + minimum is applied.""" + return self._definition.get('minClusters', 0) + + @property + def sandbox_host(self) -> TenantClusterPoolSpecSandboxHost: + """Configuration for registering clusters as tenant hosts with the Sandbax API.""" + return TenantClusterPoolSpecSandboxHost( + self._definition['sandboxHost'], + ) + + @property + def tenant_pools(self) -> List[TenantClusterPoolSpecTenantPool]: + """Configuration to maintain pools of provisioned tenant items.""" + return [ + TenantClusterPoolSpecTenantPool(item) + for item in self._definition.get('tenantPools', []) + ] + + +class TenantClusterPoolSpecClusterProvisioning: + """Configuration for provisioning clusters which provide capacity for tenants.""" + def __init__(self, definition): + self._definition = definition + + @property + def provider(self) -> TenantClusterPoolSpecClusterProvisioningProvider: + """Configuration for provisioning clusters which provide capacity for tenants.""" + return TenantClusterPoolSpecClusterProvisioningProvider( + self._definition['provider'], + ) + + +class TenantClusterPoolSpecClusterProvisioningProvider: + """Configuration for provisioning clusters which provide capacity for tenants.""" + def __init__(self, definition): + self._definition = definition + + @property + def name(self) -> str: + """ResourceProvider name""" + return self._definition['name'] + + @property + def parameter_values(self) -> Mapping[str, Any]: + """Parameter values used when creating the tenant ResourceHandle""" + return self._definition.get('parameterValues', {}) + + +class TenantClusterPoolSpecSandboxHost: + """Configuration for registering clusters as tenant hosts with the Sandbax API.""" + def __init__(self, definition): + self._definition = definition + + @property + def annotations(self) -> Mapping[str, str]: + """Annotations used to onboard provisioned clusters to the sandbox + API.""" + return self._definition['annotations'] + + @property + def deployer_admin_sa_token_refresh_interval(self) -> str: + """Admin service account token refresh. Defaults to `3h` if unset.""" + return self._definition.get('deployer_admin_sa_token_refresh_interval', '3h') + + @property + def deployer_admin_sa_token_target_var(self) -> str: + """Variable to set in tenant deployment. + Defaults to `openshift_cluster_admin_token` if unspecified.""" + return self._definition.get( + 'deployer_admin_sa_token_target_var', + 'openshift_cluster_admin_token', + ) + + @property + def deployer_admin_sa_token_ttl(self) -> str: + """Service Account token refresh interval. + Defaults to `6h` if unset.""" + return self._definition.get('deployer_admin_sa_token_ttl', '6h') + + @property + def max_cpu_usage_percentage(self) -> int|None: + """Maximum CPU utilization on host to allow tenant placement.""" + return self._definition.get('max_cpu_usage_percentage') + + @property + def max_memory_usage_percentage(self) -> int|None: + """Maximum memory utilization on host to allow tenant placement.""" + return self._definition.get('max_memory_usage_percentage') + + @property + def max_placements(self) -> int: + """Maximum number of tenant placements supported by this host.""" + return self._definition['max_placements'] + + @property + def quota_required(self) -> bool: + """Configure whether tenant sandbox request must include quota.""" + return self._definition.get('quota_required', False) + + +class TenantClusterPoolSpecTenantPool: + """Configuration to keep a pool of provisioned tenant items.""" + def __init__(self, definition): + self._definition = definition + + @property + def min_available(self) -> int: + """Minimum number of unbound ResourceHandles to maintain pooled and + available for ResourceClaims.""" + return self._definition.get('minAvailable', 0) + + @property + def hash_identifier(self) -> str: + """Short hash identifier for the tenant pool.""" + return sha1( + json.dumps({ + "name": self.provider.name, + "parameterValues": self.provider.parameter_values, + }).encode('utf-8') + ).hexdigest() + + @property + def provider(self) -> TenantClusterPoolSpecTenantPoolProvider: + """ResourceProvider configuration used to provision pooled + ResourceHandles for the tenant.""" + return TenantClusterPoolSpecTenantPoolProvider( + self._definition['provider'], + ) + +class TenantClusterPoolSpecTenantPoolProvider: + """ResourceProvider configuration used to provision pooled ResourceHandles + for the tenant.""" + def __init__(self, definition): + self._definition = definition + + @property + def name(self) -> str: + """ResourceProvider name""" + return self._definition['name'] + + @property + def parameter_values(self) -> Mapping[str, Any]: + """Parameter values used when creating the tenant ResourceHandle""" + return self._definition.get('parameterValues', {}) + +class TenantClusterPoolStatus: + """Status of TenantClusterPool""" + def __init__(self, definition): + self._definition = definition + + @property + def clusters(self) -> List[TenantClusterPoolStatusCluster]|None: + """Return clusters from status if defined.""" + if 'clusters' not in self._definition: + return None + return [ + TenantClusterPoolStatusCluster(item) + for item in self._definition['clusters'] + ] + +class TenantClusterPoolStatusCluster: + """Status of cluster in TenantClusterPool""" + def __init__(self, definition): + self._definition = definition + + @property + def name(self) -> str: + """ResourceClaim name used to request the cluster.""" + return self._definition['name'] + + @property + def resource_claim_name(self) -> str: + """ResourceClaim name used to request the cluster.""" + return self._definition['resourceClaimName'] + + @property + def sandbox_api_state(self) -> str: + """Sandbox API state may be: + + pending - Cluster is being provisioned and will be + registered with the SandboxAPI when ready. + + available - Cluster has been on-boarded to the + SandboxAPI and is available for placements. + + disabled - Cluster is currently disabled in the + SandboxAPI. + + removed - Cluster was found removed from the + SandboxAPI. As this indicates a manual override the + cluster will not be made available again by the + babylon-cluster-tenant-pool-manager.""" + + return self._definition['sandboxApiState'] diff --git a/client/python/babylon_async/src/babylon_async/user.py b/client/python/babylon_async/src/babylon_async/user.py new file mode 100644 index 000000000..0b9c9e7e4 --- /dev/null +++ b/client/python/babylon_async/src/babylon_async/user.py @@ -0,0 +1,20 @@ +from __future__ import annotations +from typing import List + +from .k8s_object import K8sObject + +class User(K8sObject): + api_group = "user.openshift.io" + api_version = "v1" + kind = "User" + plural = "users" + api_group_version = f"{api_group}/{api_version}" + + @property + def groups(self) -> List[str]|None: + """Groups list is populated by the OpenShift API only when fetching current user, `~`""" + return self._definition.get('groups') + + @property + def identities(self) -> List[str]: + return self._definition.get('identities', []) diff --git a/client/python/babylon_async/src/babylon_async/workshop.py b/client/python/babylon_async/src/babylon_async/workshop.py new file mode 100644 index 000000000..a28d80e27 --- /dev/null +++ b/client/python/babylon_async/src/babylon_async/workshop.py @@ -0,0 +1,37 @@ +from __future__ import annotations +from typing import Generator + +from .k8s_object import K8sObject +from .workshopprovision import WorkshopProvision + +class Workshop(K8sObject): + api_group = "babylon.gpte.redhat.com" + api_version = "v1" + kind = "Workshop" + plural = "workshops" + api_group_version = f"{api_group}/{api_version}" + + @property + def spec(self) -> WorkshopSpec: + return WorkshopSpec(self._definition['spec']) + + @property + def status(self) -> WorkshopStatus|None: + if 'status' not in self._definition: + return None + return WorkshopStatus(self._definition['status']) + + async def list_workshop_provisions(self) -> Generator[WorkshopProvision, None, None]: + async for workshop_provision in self.client.list_workshop_provisions( + label_selector=f"babylon.gpte.redhat.com/workshop={self.name}", + namespace=self.namespace, + ): + yield workshop_provision + +class WorkshopSpec: + def __init__(self, definition): + self._definition = definition + +class WorkshopStatus: + def __init__(self, definition): + self._definition = definition diff --git a/client/python/babylon_async/src/babylon_async/workshopprovision.py b/client/python/babylon_async/src/babylon_async/workshopprovision.py new file mode 100644 index 000000000..e3c6423b9 --- /dev/null +++ b/client/python/babylon_async/src/babylon_async/workshopprovision.py @@ -0,0 +1,61 @@ +from __future__ import annotations +from typing import Any, Mapping + +from .k8s_object import K8sObject +from .resourceprovider import ResourceProvider + +class WorkshopProvision(K8sObject): + api_group = "babylon.gpte.redhat.com" + api_version = "v1" + kind = "WorkshopProvision" + plural = "workshopprovisions" + api_group_version = f"{api_group}/{api_version}" + + @property + def parameter_values(self) -> Mapping: + return self.spec.parameters + + @property + def resource_provider_name(self) -> str: + return self.spec.catalog_item.name + + @property + def spec(self) -> WorkshopProvisionSpec: + return WorkshopProvisionSpec(self._definition['spec']) + + @property + def status(self) -> WorkshopProvisionStatus|None: + if 'status' not in self._definition: + return None + return WorkshopProvisionStatus(self._definition['status']) + + async def get_resource_provider(self) -> ResourceProvider: + return await self.client.get_resource_provider(name=self.resource_provider_name) + +class WorkshopProvisionSpec: + def __init__(self, definition): + self._definition = definition + + @property + def catalog_item(self) -> WorkshopProvisionSpecCatalogItem: + return WorkshopProvisionSpecCatalogItem(self._definition['catalogItem']) + + @property + def parameters(self) -> Mapping: + return self._definition.get('parameters') + +class WorkshopProvisionSpecCatalogItem: + def __init__(self, definition): + self._definition = definition + + @property + def name(self) -> str: + return self._definition['name'] + + @property + def namespace(self) -> str: + return self._definition['namespace'] + +class WorkshopProvisionStatus: + def __init__(self, definition): + self._definition = definition diff --git a/client/python/babylon_async/src/find-cnv-workshops.py b/client/python/babylon_async/src/find-cnv-workshops.py new file mode 100755 index 000000000..750fc546c --- /dev/null +++ b/client/python/babylon_async/src/find-cnv-workshops.py @@ -0,0 +1,51 @@ +#!/usr/bin/env python + +import asyncio + +from babylon_async import BabylonClient + +import deepmerge +template_merger = deepmerge.Merger( + [(dict, ["merge"])], + ["override"], + ["override"], +) + +async def find_cnv_workshops(babylon) -> None: + async for workshop in babylon.list_workshops(): + service_namespace = await babylon.get_namespace(workshop.namespace) + async for workshop_provision in workshop.list_workshop_provisions(): + resource_provider = await workshop_provision.get_resource_provider() + + resources = await resource_provider.get_resources(parameter_values=workshop_provision.parameter_values) + for item in resources: + item_resource_provider = await babylon.get_resource_provider(item['provider']['name']) + anarchy_governor_name = item_resource_provider.spec.override['spec']['governor'] + anarchy_namespace = item_resource_provider.spec.override['metadata']['namespace'] + #FIXME sandboxes are actually conditional... ideally it would check + #_vars = {} + #template_merger.merge(_vars, item['template']['spec']['vars']) + #template_merger.merge(_vars, item_resource_provider.spec.override['spec']['vars']) + #print(_vars['job_vars']) + anarchy_governor = await babylon.get_anarchy_governor( + name=anarchy_governor_name, + namespace=anarchy_namespace, + ) + for sandbox in anarchy_governor.get_sandboxes(): + if sandbox.cloud_selector.get('cloud') == 'cnv': + print(f"{workshop} <{service_namespace.metadata.annotations['openshift.io/requester']}>") + + +async def main() -> None: + from argparse import ArgumentParser + argparser = ArgumentParser( + prog="find-cnv-workshops.py", + description="Find workshops using CNV", + ) + args = argparser.parse_args() + + async with BabylonClient() as babylon: + await find_cnv_workshops(babylon) + +if __name__ == '__main__': + asyncio.run(main()) diff --git a/client/python/babylon_async/src/find-useless-pools.py b/client/python/babylon_async/src/find-useless-pools.py new file mode 100755 index 000000000..16d9f4cc9 --- /dev/null +++ b/client/python/babylon_async/src/find-useless-pools.py @@ -0,0 +1,95 @@ +#!/usr/bin/env python + +import asyncio + +from copy import deepcopy + +from babylon_async import BabylonClient, CatalogItem, ResourcePool + +import deepmerge +template_merger = deepmerge.Merger( + [(dict, ["merge"])], + ["override"], + ["override"], +) + +async def find_useless_pools(): + async with BabylonClient() as babylon: + all_catalog_items = [] + async for catalog_item in babylon.list_catalog_items(): + all_catalog_items.append(catalog_item) + + async for resource_pool in babylon.list_resource_pools(): + await report_resource_pool_if_no_catalog_item( + all_catalog_items=all_catalog_items, + resource_pool=resource_pool, + ) + +async def report_resource_pool_if_no_catalog_item( + all_catalog_items:list[CatalogItem], + resource_pool:ResourcePool, +): + all_job_vars = {} + for resource in resource_pool.spec.resources: + all_job_vars.update( + resource.template.get('spec', {}).get('vars', {}).get('job_vars', {}) + ) + + for catalog_item in all_catalog_items: + if catalog_item.external_url is not None: + continue + parameter_values = {} + for parameter in catalog_item.parameters: + if parameter.name in all_job_vars: + if parameter.validate(all_job_vars[parameter.name]): + parameter_values[parameter.name] = all_job_vars[parameter.name] + elif parameter.default is not None: + parameter_values[parameter.name] = parameter.default + resource_provider = await catalog_item.get_resource_provider(cache=True) + resource_provider_resources = await resource_provider.get_resources( + parameter_values=parameter_values, + cache=True, + ) + + # Match is impossible if resource pool has more resources than the provider specifies + if len(resource_pool.spec.resources) > len(resource_provider_resources): + continue + + # Discard fields that don't need to match + for resource in resource_provider_resources: + resource.pop('name') + resource['template']['spec']['vars'].pop('action_schedule', None) + + catalog_item_match = None + for idx, pool_resource in enumerate(resource_pool.spec.resources): + resource_provider_resource = resource_provider_resources[idx] + # All vars from the resource_provider must match, but pool may have other vars + template_cmp = template_merger.merge( + deepcopy(pool_resource.template), + resource_provider_resource['template'], + ) + if pool_resource.provider.name != resource_provider_resource['provider']['name']: + catalog_item_match = False + break + if pool_resource.template != template_cmp: + catalog_item_match = False + break + else: + catalog_item_match = True + if catalog_item_match: + break + else: + print(f"{resource_pool} is useless!") + +async def main(): + from argparse import ArgumentParser + argparser = ArgumentParser( + prog="find-useless-pools.py", + description="Find resource pools that cannot match any catalog items", + ) + args = argparser.parse_args() + + await find_useless_pools() + +if __name__ == '__main__': + asyncio.run(main()) diff --git a/client/python/babylon_async/src/fix-subjects.py b/client/python/babylon_async/src/fix-subjects.py new file mode 100755 index 000000000..1625359da --- /dev/null +++ b/client/python/babylon_async/src/fix-subjects.py @@ -0,0 +1,89 @@ +#!/usr/bin/env python + +import asyncio + +from babylon_async import BabylonClient +from babylon_async.exceptions import BabylonApiException + +from kubernetes_asyncio.client import ( + ApiException as KubernetesApiException +) + +async def fix_subjects(dry_run:bool=False, namespace:str|None=None): + async with BabylonClient() as babylon: + async for anarchy_subject in babylon.list_anarchy_subjects(namespace=namespace): + await fix_subject(anarchy_subject, dry_run=dry_run) + + +async def fix_subject(anarchy_subject, dry_run=False): + await fix_runs(anarchy_subject, dry_run=dry_run) + +async def fix_runs(anarchy_subject, dry_run=False): + oldest_queued_run = None + has_active_run = False + if anarchy_subject.status is not None: + status_patch = [] + for idx, anarchy_run_ref in enumerate(anarchy_subject.status.runs.active): + anarchy_run = None + try: + anarchy_run = await anarchy_subject.get_anarchy_run(anarchy_run_ref.name) + except BabylonApiException as exception: + if exception.status != 404: + raise + if anarchy_run is None: + if dry_run: + print(f"Would remove missing AnarchyRun {anarchy_run_ref.name} from {anarchy_subject}") + else: + print(f"Removing missing AnarchyRun {anarchy_run_ref.name} from {anarchy_subject}") + status_patch.insert(0, { + "op": "remove", + "path": f"/status/runs/active/{idx}", + }) + status_patch.insert(0, { + "op": "test", + "path": f"/status/runs/active/{idx}/name", + "value": anarchy_run_ref.name, + }) + if len(status_patch) > 0 and not dry_run: + print(status_patch) + try: + await anarchy_subject.patch_status(status_patch) + except KubernetesApiException as exception: + if exception.status not in (404, 422): + raise + async for anarchy_run in anarchy_subject.list_anarchy_runs(): + if anarchy_run.is_active: + has_active_run = True + if not await anarchy_run.check_runner_pod_exists(): + if dry_run: + print(f"{anarchy_run} references missing runner pod {anarchy_run.runner_pod_name}") + else: + print(f"Resetting {anarchy_run} to pending due to reference missing runner pod {anarchy_run.runner_pod_name}") + await anarchy_run.set_runner_state('pending') + elif anarchy_run.is_queued: + if ( + oldest_queued_run is None or + oldest_queued_run.creation_datetime > anarchy_run.creation_datetime + ): + oldest_queued_run = anarchy_run + if not has_active_run and oldest_queued_run is not None: + print(f"{oldest_queued_run} is queued but there are no active runs for {anarchy_subject}") + + +async def main(): + from argparse import ArgumentParser + argparser = ArgumentParser( + prog="fix-subjects.py", + description="Fix common known issues which may occur with AnarchySubjects", + ) + argparser.add_argument("-d", "--dry-run", action='store_true') + argparser.add_argument("-n", "--namespace") + args = argparser.parse_args() + + await fix_subjects( + dry_run=args.dry_run, + namespace=args.namespace, + ) + +if __name__ == '__main__': + asyncio.run(main()) diff --git a/client/python/babylon_async/src/label-pools.py b/client/python/babylon_async/src/label-pools.py new file mode 100755 index 000000000..21893c04b --- /dev/null +++ b/client/python/babylon_async/src/label-pools.py @@ -0,0 +1,132 @@ +#!/usr/bin/env python + +import asyncio + +from copy import deepcopy + +from babylon_async import BabylonClient, CatalogItem, ResourcePool + +def catalog_item_label(catalog_item): + return f"catalog-item.demo.redhat.com/{catalog_item.name}" + +def cloud_provider_label(value): + return f"cloud-provider.demo.redhat.com/{value}" + +def resource_provider_label(resource_provider): + return f"resource-provider.demo.redhat.com/{resource_provider.name}" + +all_catalog_items = None +async def get_all_catalog_items(babylon: BabylonClient) -> list[CatalogItem]: + """Return all catalog items either from cache or get and cache the first time.""" + global all_catalog_items + if all_catalog_items is not None: + return all_catalog_items + all_catalog_items = [] + async for catalog_item in babylon.list_catalog_items(): + all_catalog_items.append(catalog_item) + return all_catalog_items + +async def get_catalog_items_for_resource_pool( + babylon: BabylonClient, + resource_pool: ResourcePool, +) -> list[CatalogItem]: + catalog_items = [] + all_catalog_items = await get_all_catalog_items(babylon) + for catalog_item in all_catalog_items: + if await catalog_item.check_resource_pool_match(resource_pool, cache=True): + catalog_items.append(catalog_item) + return catalog_items + +async def label_resource_pools( + dry_run:bool=False, +): + """Label ResourcePools with cloud provider labels.""" + async with BabylonClient() as babylon: + async for resource_pool in babylon.list_resource_pools(): + await label_resource_pool( + babylon=babylon, + dry_run=dry_run, + resource_pool=resource_pool, + ) + +async def label_resource_pool( + babylon:BabylonClient, + resource_pool:ResourcePool, + dry_run:bool=False, +): + labels = {} + for resource in resource_pool.spec.resources: + resource_template_job_vars = resource.template.get('spec', {}).get('vars', {}).get('job_vars', {}) + if 'cloud_provider' in resource_template_job_vars: + labels[cloud_provider_label(resource_template_job_vars['cloud_provider'])] = "" + resource_provider = await babylon.get_resource_provider(resource.provider.name) + labels[resource_provider_label(resource_provider)] = "" + + anarchy_namespace = resource_provider.spec.override.get('metadata', {}).get('namespace') + anarchy_governor_name = resource_provider.spec.override.get('spec', {}).get('governor') + + anarchy_governor = await babylon.get_anarchy_governor( + name=anarchy_governor_name, + namespace=anarchy_namespace, + ) + if 'cloud_provider' in anarchy_governor.job_vars: + labels[cloud_provider_label(anarchy_governor.job_vars['cloud_provider'])] = "" + + catalog_items = await get_catalog_items_for_resource_pool( + babylon=babylon, + resource_pool=resource_pool, + ) + if len(catalog_items) > 0: + for catalog_item in catalog_items: + labels[catalog_item_label(catalog_item)] = catalog_item.uid + else: + labels["catalog-item.demo.redhat.com/NONE"] = "" + + patch = [] + for label, value in labels.items(): + if resource_pool.metadata.labels.get(label) == value: + continue + patch.append({ + "op": "add", + "path": f"/metadata/labels/{label.replace('/', '~1')}", + "value": value + }) + if dry_run: + print(f"Would add label {label} to {resource_pool}") + else: + print(f"Adding label {label} to {resource_pool}") + + for label, value in resource_pool.metadata.labels.items(): + if label in labels or not ( + label.startswith('catalog-item.demo.redhat.com/') or + label.startswith('cloud-provider.demo.redhat.com/') or + label.startswith('resource-provider.demo.redhat.com/') + ): + continue + patch.append({ + "op": "remove", + "path": f"/metadata/labels/{label.replace('/', '~1')}", + }) + if dry_run: + print(f"Would remove label {label} from {resource_pool}") + else: + print(f"Removing label {label} from {resource_pool}") + + if len(patch) == 0 or dry_run: + return + + await resource_pool.patch(patch) + +async def main(): + from argparse import ArgumentParser + argparser = ArgumentParser( + prog="label-pools.py", + description="Label ResourcePools by cloud provider", + ) + argparser.add_argument("-d", "--dry-run", action='store_true') + args = argparser.parse_args() + + await label_resource_pools(dry_run=args.dry_run) + +if __name__ == '__main__': + asyncio.run(main())