| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| # |
| import kopf, logging, time, os |
| import openserverless.kube as kube |
| import openserverless.kustomize as kus |
| import openserverless.config as cfg |
| import openserverless.util as util |
| import openserverless.minio_util as mutil |
| import openserverless.openwhisk as openwhisk |
| import openserverless.minio_ingress as minio_ingress |
| import openserverless.operator_util as operator_util |
| |
| from openserverless.user_config import UserConfig |
| from openserverless.user_metadata import UserMetadata |
| from openserverless.minio_util import MinioClient |
| |
| def _add_miniouser_metadata(ucfg: UserConfig, user_metadata:UserMetadata): |
| """ |
| adds entries for minio connectivity S3_HOST, S3_PORT, S3_ACCESS_KEY, S3_SECRET_KEY |
| this is becasue MINIO |
| """ |
| |
| try: |
| minio_service = util.get_service("{.items[?(@.spec.selector.app == 'minio')]}") |
| if(minio_service): |
| minio_host = f"{minio_service['metadata']['name']}.{minio_service['metadata']['namespace']}.svc.cluster.local" |
| access_key = ucfg.get('namespace') |
| secret_key = ucfg.get("object-storage.password") |
| user_metadata.add_metadata("S3_PROVIDER","minio") |
| user_metadata.add_metadata("S3_HOST",minio_host) |
| user_metadata.add_metadata("S3_ACCESS_KEY",access_key) |
| user_metadata.add_metadata("S3_SECRET_KEY",secret_key) |
| |
| user_metadata.add_safely_from_cm("S3_API_URL", '{.metadata.annotations.s3_api_url}') |
| user_metadata.add_safely_from_cm("S3_CONSOLE_URL", '{.metadata.annotations.s3_console_url}') |
| |
| ports = list(minio_service['spec']['ports']) |
| for port in ports: |
| if(port['name']=='minio-api'): |
| user_metadata.add_metadata("S3_PORT",port['port']) |
| |
| |
| return None |
| except Exception as e: |
| logging.error(f"failed to build MINIO metadata for {ucfg.get('namespace')}: {e}") |
| return None |
| |
| def find_content_path(filename): |
| absolute_path = os.path.dirname(__file__) |
| relative_path = "../deploy/content" |
| return os.path.join(absolute_path, relative_path, filename) |
| |
| def create(owner=None): |
| logging.info(f"*** configuring minio standalone") |
| |
| data = util.get_minio_config_data() |
| |
| tplp = ["00-minio-pvc.yaml","01-minio-dep.yaml","02-minio-svc.yaml"] |
| |
| if(data['affinity'] or data['tolerations']): |
| tplp.append("affinity-tolerance-dep-core-attach.yaml") |
| |
| kust = kus.patchTemplates("minio", tplp, data) |
| spec = kus.kustom_list("minio", kust, templates=[], data=data) |
| |
| if owner: |
| kopf.append_owner_reference(spec['items'], owner) |
| else: |
| cfg.put("state.minio.spec", spec) |
| |
| res = kube.apply(spec) |
| |
| # dynamically detect minio pod and wait for readiness |
| util.wait_for_pod_ready("{.items[?(@.metadata.labels.app == 'minio')].metadata.name}") |
| create_nuv_storage(data) |
| minio_ingress.create_minio_ingresses(data, owner) |
| logging.info("*** configured minio standalone") |
| return res |
| |
| def _annotate_nuv_metadata(data): |
| """ |
| annotate openserverless configmap with entries for minio connectivity S3_ENDPOINT, S3_PORT, S3_ACCESS_KEY, S3_SECRET_KEY |
| this is becasue MINIO |
| """ |
| try: |
| minio_service = util.get_service("{.items[?(@.spec.selector.app == 'minio')]}") |
| if(minio_service): |
| minio_host = f"{minio_service['metadata']['name']}.{minio_service['metadata']['namespace']}.svc.cluster.local" |
| access_key = data["minio_nuv_user"] |
| secret_key = data["minio_nuv_password"] |
| openwhisk.annotate(f"s3_host={minio_host}") |
| openwhisk.annotate(f"s3_access_key={access_key}") |
| openwhisk.annotate(f"s3_secret_key={secret_key}") |
| openwhisk.annotate("s3_provider=minio") |
| |
| ports = list(minio_service['spec']['ports']) |
| for port in ports: |
| if(port['name']=='minio-api'): |
| openwhisk.annotate(f"s3_port={port['port']}") |
| return None |
| except Exception as e: |
| logging.error(f"failed to build minio_host for openserverless: {e}") |
| return None |
| |
| def create_nuv_storage(data): |
| """ |
| Creates openserverless MINIO custom resources |
| """ |
| logging.info(f"*** configuring MINIO storage for openserverless") |
| # introduce a 10 seconds delay to be sure that MINIO server is up and running completely as pod readines not to be enough |
| time.sleep(10) |
| minioClient = mutil.MinioClient() |
| res = minioClient.add_user(data["minio_nuv_user"], data["minio_nuv_password"]) |
| |
| if(res): |
| _annotate_nuv_metadata(data) |
| bucket_policy_names = [] |
| |
| logging.info("*** adding openserverless MINIO data bucket") |
| res = minioClient.make_bucket("openserverless-data") |
| bucket_policy_names.append("openserverless-data/*") |
| |
| if(res): |
| openwhisk.annotate("s3_bucket_data=openserverless-data") |
| |
| logging.info("*** adding openserverless MINIO static public bucket") |
| res = minioClient.make_public_bucket("openserverless-web") |
| bucket_policy_names.append("openserverless-web/*") |
| |
| if(res): |
| openwhisk.annotate("s3_bucket_static=openserverless-web") |
| content_path = find_content_path("index.html") |
| |
| if(content_path): |
| logging.info(f"uploading example content to openserverless-web from {content_path}") |
| res = minioClient.upload_folder_content(content_path,"openserverless-web") |
| else: |
| logging.warn("could not find example static content to upload") |
| |
| if(len(bucket_policy_names)>0): |
| logging.info(f"granting rw access to created policies under username {data['minio_nuv_user']}") |
| minioClient.assign_rw_bucket_policy_to_user(data["minio_nuv_user"],bucket_policy_names) |
| |
| logging.info("*** configured MINIO storage for openserverless") |
| |
| def assign_bucket_quota(bucket_name, quota, minioClient:MinioClient): |
| if not quota.lower() in ['auto'] and quota.isnumeric(): |
| logging.info(f"*** setting quota on bucket {bucket_name} with hardlimit to {quota}m") |
| res = minioClient.assign_quota_to_bucket(bucket_name,quota) |
| |
| if res: |
| logging.info(f"*** quota on bucket {bucket_name} set successfully") |
| else: |
| logging.warn(f"*** skipping quota set on bucket {bucket_name}. Requested quota values is {quota}") |
| |
| |
| def create_ow_storage(state, ucfg: UserConfig, user_metadata: UserMetadata, owner=None): |
| minioClient = mutil.MinioClient() |
| namespace = ucfg.get("namespace") |
| secretkey = ucfg.get("object-storage.password") |
| |
| logging.info(f"*** configuring storage for namespace {namespace}") |
| |
| res = minioClient.add_user(namespace, secretkey) |
| state['storage_user']=res |
| |
| if(res): |
| _add_miniouser_metadata(ucfg, user_metadata) |
| |
| bucket_policy_names = [] |
| |
| if(ucfg.get('object-storage.data.enabled')): |
| bucket_name = ucfg.get('object-storage.data.bucket') |
| logging.info(f"*** adding private bucket {bucket_name} for {namespace}") |
| res = minioClient.make_bucket(bucket_name) |
| bucket_policy_names.append(f"{bucket_name}/*") |
| state['storage_data']=res |
| |
| if(res): |
| user_metadata.add_metadata("S3_BUCKET_DATA",bucket_name) |
| ucfg.put("S3_BUCKET_DATA",bucket_name) |
| |
| if ucfg.exists('object-storage.quota'): |
| assign_bucket_quota(bucket_name,ucfg.get('object-storage.quota'), minioClient) |
| |
| if(ucfg.get('object-storage.route.enabled')): |
| bucket_name = ucfg.get("object-storage.route.bucket") |
| logging.info(f"*** adding public bucket {bucket_name} for {namespace}") |
| res = minioClient.make_public_bucket(bucket_name) |
| bucket_policy_names.append(f"{bucket_name}/*") |
| |
| if(res): |
| user_metadata.add_metadata("S3_BUCKET_STATIC",bucket_name) |
| ucfg.put("S3_BUCKET_STATIC",bucket_name) |
| |
| if ucfg.exists('object-storage.quota'): |
| assign_bucket_quota(bucket_name,ucfg.get('object-storage.quota'), minioClient) |
| |
| content_path = find_content_path("index.html") |
| |
| if(content_path): |
| logging.info(f"uploading example content to {bucket_name} from {content_path}") |
| res = minioClient.upload_folder_content(content_path,bucket_name) |
| else: |
| logging.warn("could not find example static content to upload") |
| |
| state['storage_route']=res |
| |
| if(len(bucket_policy_names)>0): |
| logging.info(f"granting rw access to created policies under namespace {namespace}") |
| minioClient.assign_rw_bucket_policy_to_user(namespace,bucket_policy_names) |
| |
| return state |
| |
| def delete_ow_storage(ucfg): |
| minioClient = mutil.MinioClient() |
| namespace = ucfg.get("namespace") |
| |
| if(ucfg.get('object-storage.data.enabled')): |
| bucket_name = ucfg.get('object-storage.data.bucket') |
| logging.info(f"*** removing private bucket {bucket_name} for {namespace}") |
| res = minioClient.force_bucket_remove(bucket_name) |
| |
| if(ucfg.get('object-storage.route.enabled')): |
| bucket_name = ucfg.get("object-storage.route.bucket") |
| logging.info(f"*** removing public bucket {bucket_name} for {namespace}") |
| res = minioClient.force_bucket_remove(bucket_name) |
| |
| return minioClient.delete_user(namespace) |
| |
| def delete_by_owner(): |
| spec = kus.build("minio") |
| res = kube.delete(spec) |
| logging.info(f"delete minio: {res}") |
| return res |
| |
| def delete_by_spec(): |
| spec = cfg.get("state.minio.spec") |
| res = False |
| if spec: |
| res = kube.delete(spec) |
| logging.info(f"delete minio: {res}") |
| return res |
| |
| def delete(owner=None): |
| data = util.get_minio_config_data() |
| minio_ingress.delete_minio_ingresses(data, owner) |
| |
| if owner: |
| return delete_by_owner() |
| else: |
| return delete_by_spec() |
| |
| def patch(status, action, owner=None): |
| """ |
| Called by the operator patcher to create/delete minio component |
| """ |
| try: |
| logging.info(f"*** handling request to {action} minio") |
| if action == 'create': |
| msg = create(owner) |
| operator_util.patch_operator_status(status,'minio','on') |
| else: |
| msg = delete(owner) |
| operator_util.patch_operator_status(status,'minio','off') |
| |
| logging.info(msg) |
| logging.info(f"*** hanlded request to {action} minio") |
| except Exception as e: |
| logging.error('*** failed to update minio: %s' % e) |
| operator_util.patch_operator_status(status,'minio','error') |
| |
| def patch_ingresses(status, action, owner=None): |
| """ |
| Called by the operator patcher to create/delete minio component |
| """ |
| try: |
| logging.info(f"*** handling request to {action} minio ingresses") |
| data = util.get_minio_config_data() |
| if action == 'update': |
| msg = minio_ingress.create_minio_ingresses(data, owner) |
| operator_util.patch_operator_status(status,'minio-ingresses','on') |
| |
| logging.info(msg) |
| logging.info(f"*** hanlded request to {action} minio ingresses") |
| except Exception as e: |
| logging.error('*** failed to update minio ingresses: %s' % e) |
| operator_util.patch_operator_status(status,'minio-ingresses','error') |
