| # 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, os, logging, json |
| import openserverless.kustomize as kus |
| import openserverless.kube as kube |
| import openserverless.couchdb_util as cu |
| import openserverless.config as cfg |
| import openserverless.couchdb_util |
| import openserverless.util as util |
| |
| from openserverless.user_config import UserConfig |
| from openserverless.user_metadata import UserMetadata |
| |
| from jinja2 import Environment, FileSystemLoader |
| loader = FileSystemLoader(["./openserverless/templates", "./openserverless/files"]) |
| env = Environment(loader=loader) |
| |
| def update_templated_doc(db, database, template, data): |
| tpl = env.get_template(template) |
| doc = json.loads(tpl.render(data)) |
| return db.update_doc(database, doc) |
| |
| def create(owner=None): |
| logging.info("create couchdb") |
| runtime = cfg.get('openserverless.kube') |
| u = cfg.get('couchdb.admin.user', "COUCHDB_ADMIN_USER", "whisk_admin") |
| p = cfg.get('couchdb.admin.password', "COUCHDB_ADMIN_PASSWORD", "some_passw0rd") |
| user = f"db_username={u}" |
| pasw = f"db_password={p}" |
| |
| img = cfg.get('operator.image', 'OPERATOR_IMAGE', "missing-operator-image") |
| tag = cfg.get('operator.tag', 'OPERATOR_TAG', "missing-operator-tag") |
| image = f"{img}:{tag}" |
| container_image = runtime in ['openshift'] and "ghcr.io/nuvolaris/couchdb:2.3.1-nuvolaris.23101915" or "apache/couchdb:2.3" |
| |
| config = json.dumps(cfg.getall()) |
| data = { |
| "runtime":runtime, |
| "container_image":container_image, |
| "image": image, |
| "config": config, |
| "name": "couchdb", |
| "container": "couchdb", |
| "size": cfg.get("couchdb.volume-size", "COUCHDB_VOLUME_SIZE", 10), |
| "dir": "/opt/couchdb/data", |
| "storageClass": cfg.get("openserverless.storageclass"), |
| "container_cpu_req": cfg.get('configs.couchdb.resources.cpu-req') or "500m", |
| "container_cpu_lim": cfg.get('configs.couchdb.resources.cpu-lim') or "1", |
| "container_mem_req": cfg.get('configs.couchdb.resources.mem-req') or "1G", |
| "container_mem_lim": cfg.get('configs.couchdb.resources.mem-lim') or "2G", |
| "container_manage_resources": cfg.exists('configs.couchdb.resources.cpu-req'), |
| "index": "1", |
| "replicationRole":"primary", |
| "appName":"openserverless-couchdb" |
| } |
| |
| tplp = ["set-attach.yaml"] |
| util.couch_affinity_tolerations_data(data) |
| if(data['affinity'] or data['tolerations']): |
| tplp.append("affinity-tolerance-sts-core-attach.yaml") |
| |
| kus.processTemplate("couchdb","couchdb-set-tpl.yaml",data,"couchdb-set_generated.yaml") |
| kust = kus.secretLiteral("couchdb-auth", user, pasw) |
| kust += kus.patchTemplates("couchdb",tplp,data) |
| spec = kus.restricted_kustom_list("couchdb", kust, templates=["couchdb-init.yaml"],templates_filter=["couchdb-set_generated.yaml","couchdb-svc.yaml"],data=data) |
| |
| if owner: |
| kopf.append_owner_reference(spec['items'], owner) |
| else: |
| cfg.put("state.couchdb.spec", spec) |
| res = kube.apply(spec) |
| |
| if res: |
| # dynamically detect couchdb pod and wait for readiness |
| util.wait_for_pod_ready("{.items[?(@.metadata.labels.name == 'couchdb')].metadata.name}") |
| |
| return res |
| |
| def delete(): |
| spec = cfg.get("state.couchdb.spec") |
| res = False |
| if spec: |
| res = kube.delete(spec) |
| logging.info(f"delete couchdb: {res}") |
| return res |
| |
| def check(f, what, res): |
| if f: |
| logging.info(f"OK: {what}") |
| return res and True |
| else: |
| logging.warn(f"ERR: {what}") |
| return False |
| |
| def init_system(db): |
| res = check(db.wait_db_ready(60), "wait_db_ready", True) |
| res = check(db.configure_single_node(), "configure_single_node", res) |
| res = check(db.configure_no_reduce_limit(), "configure_no_reduce_limit", res) |
| cuser = cfg.get('couchdb.controller.user', "COUCHDB_CONTROLLER_USER", "controller_admin") |
| cpasw = cfg.get('couchdb.controller.password', "COUCHDB_CONTROLLER_PASSWORD", "s0meP@ass1") |
| iuser = cfg.get('couchdb.invoker.user', "COUCHDB_INVOKER_USER", "invoker_admin") |
| ipasw = cfg.get('couchdb.invoker.password', "COUCHDB_INVOKER_PASSWORD", "s0meP@ass2") |
| res = check(db.add_user(cuser, cpasw), "add_user: controller", res) |
| return check(db.add_user(iuser, ipasw), "add_user: invoker", res) |
| |
| def init_subjects(db): |
| subjects_design_docs = [ |
| "auth_design_document_for_subjects_db_v2.0.0.json", |
| "filter_design_document.json", |
| "namespace_throttlings_design_document_for_subjects_db.json"] |
| dbn = "subjects" |
| res = check(db.wait_db_ready(60), "wait_db_ready", True) |
| res = check(db.create_db(dbn), "create_db: subjects", res) |
| members = [cfg.get('couchdb.controller.user'), cfg.get('couchdb.invoker.user')] |
| res = check(db.add_role(dbn, members), "add_role: subjects", res) |
| for i in subjects_design_docs: |
| res = check(update_templated_doc(db, dbn, i, {}), f"add {i}", res) |
| return res |
| |
| def init_activations(db): |
| activations_design_docs = [ |
| "whisks_design_document_for_activations_db_v2.1.0.json", |
| "whisks_design_document_for_activations_db_filters_v2.1.1.json", |
| "filter_design_document.json", |
| "activations_design_document_for_activations_db.json", |
| "logCleanup_design_document_for_activations_db.json" |
| ] |
| dbn = "activations" |
| res = check(db.wait_db_ready(60), "wait_db_ready", True) |
| res = check(db.create_db(dbn), "create_db: activations", res) |
| members = [cfg.get('couchdb.controller.user'), cfg.get('couchdb.invoker.user')] |
| res = check(db.add_role(dbn, members), "add_role: activations", res) |
| for i in activations_design_docs: |
| res = check(update_templated_doc(db, dbn, i, {}), f"add {i}", res) |
| return res |
| |
| def init_actions(db): |
| whisks_design_docs = [ |
| "whisks_design_document_for_entities_db_v2.1.0.json", |
| "filter_design_document.json" |
| ] |
| dbn = "whisks" |
| res = check(db.wait_db_ready(60), "wait_db_ready", True) |
| res = check(db.create_db(dbn), "create_db: whisks", res) |
| members = [cfg.get('couchdb.controller.user'), cfg.get('couchdb.invoker.user')] |
| res = check(db.add_role(dbn, members), "add_role: actions", res) |
| for i in whisks_design_docs: |
| res = check(update_templated_doc(db, dbn, i, {}), f"add {i}", res) |
| return res |
| |
| def add_initial_subjects(db): |
| res = check(db.wait_db_ready(60), "wait_db_ready", True) |
| dbn = "subjects" |
| for _, (name, value) in enumerate(cfg.getall("openwhisk.namespaces").items()): |
| [uuid, key] = value.split(":") |
| basename = name.split(".")[-1] |
| data = { "name": basename, "key": key, "uuid": uuid} |
| res = check(update_templated_doc(db, dbn, "subject.json", data), f"add {name}", res) |
| return res |
| |
| def init(): |
| # load openserverless config from the named crd |
| config = os.environ.get("OPENSERVERLESS_CONFIG") |
| if config: |
| import logging |
| logging.basicConfig(level=logging.INFO) |
| spec = json.loads(config) |
| cfg.configure(spec) |
| for k in cfg.getall(): logging.info(f"{k} = {cfg.get(k)}") |
| |
| # dynamically detect couchdb pod and wait for readiness |
| util.wait_for_pod_ready("{.items[?(@.metadata.labels.name == 'couchdb')].metadata.name}") |
| |
| db = openserverless.couchdb_util.CouchDB() |
| res = check(init_system(db), "init_system", True) |
| res = check(init_subjects(db), "init_subjects", res) |
| res = check(init_activations(db), "init_activations", res) |
| res = check(init_actions(db), "init_actions", res) |
| res = check(add_initial_subjects(db), "add_subjects", res) |
| res = check(init_users_metadata(db), "init_users_metadata", res) |
| res = check(init_compactions_config(db), "init_compactions_config", res) |
| |
| # job process status code should be negated if the job is successfull |
| return not res |
| |
| def add_subject(db, namespace, auth): |
| """ |
| Add a new Openwhisk Couchdb subject that will be authorized to interact with the specified Openwhisk namespace. |
| the auth parameters represents the subject authentication in the form of a uuid.uuid4():randomstr(64) |
| """ |
| res = check(db.wait_db_ready(60), "wait_db_ready", True) |
| dbn = "subjects" |
| [uuid, key] = auth.split(":") |
| data = { "name": namespace, "key": key, "uuid": uuid} |
| return check(update_templated_doc(db, dbn, "subject.json", data), f"add {namespace}", res) |
| |
| def init_users_metadata(db): |
| """ |
| Add a new Openwhisk Couchdb database to host openserverless user relevant informations |
| """ |
| dbn = "users_metadata" |
| res = check(db.wait_db_ready(60), "wait_db_ready", True) |
| res = check(db.create_db(dbn), "create_db: user_metadata", res) |
| return res |
| |
| def init_compactions_config(db): |
| """ |
| Activate the compactions config for the openserverless related databases |
| """ |
| res = check(db.wait_db_ready(60), "wait_db_ready", True) |
| res = check(db.enable_db_compaction("users_metadata"), "enable_db_compaction: user_metadata", res) |
| res = check(db.enable_db_compaction("users_subjects"), "enable_db_compaction: subjects", res) |
| res = check(db.enable_db_compaction("whisks"), "enable_db_compaction: whisks", res) |
| return res |
| |
| def create_ow_user(ucfg: UserConfig, user_metadata: UserMetadata): |
| subject = ucfg.get("namespace") |
| auth = ucfg.get("auth") |
| |
| logging.info(f"authorizing OpenWhisk namespace {subject}") |
| |
| try: |
| db = openserverless.couchdb_util.CouchDB() |
| |
| if(util.validate_ow_auth(auth)): |
| logging.info(f"{subject} authorization is valid, adding subject") |
| res = add_subject(db, subject, auth) |
| |
| if(res): |
| user_metadata.add_metadata("AUTH",auth) |
| |
| return res |
| else: |
| return None |
| except Exception as e: |
| logging.error(f"failed to authorize Openwhisk namespace {subject} authorization id and key: {e}") |
| return None |
| |
| def delete_ow_user(subject): |
| logging.info(f"removing auhorization for OpenWhisk namespace {subject}") |
| |
| try: |
| db = openserverless.couchdb_util.CouchDB() |
| selector = {"selector":{"subject": {"$eq": subject }}} |
| response = db.find_doc("subjects", json.dumps(selector)) |
| |
| if(response['docs']): |
| docs = list(response['docs']) |
| if(len(docs) > 0): |
| doc = docs[0] |
| logging.info(f"removing subjects documents {doc['_id']}") |
| return db.delete_doc("subjects",doc['_id']) |
| |
| logging.warn(f"auhorization for OpenWhisk namespace {subject} not found!") |
| return None |
| except Exception as e: |
| logging.error(f"failed to remove Openwhisk namespace {subject} authorization id and key: {e}") |
| return None |
| |
| |
| |
| |
