osm_nbi/k8s_topics.py

# -*- coding: utf-8 -*-

# Licensed 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 logging
import yaml
import shutil
import os
from http import HTTPStatus

from time import time
from osm_nbi.base_topic import BaseTopic, EngineException
from osm_nbi.acm_topic import ACMTopic, ACMOperationTopic, ProfileTopic

from osm_nbi.descriptor_topics import DescriptorTopic
from osm_nbi.validation import (
    ValidationError,
    validate_input,
    cluster_creation_new_schema,
    cluster_edit_schema,
    cluster_update_schema,
    infra_controller_profile_create_new_schema,
    infra_config_profile_create_new_schema,
    app_profile_create_new_schema,
    resource_profile_create_new_schema,
    infra_controller_profile_create_edit_schema,
    infra_config_profile_create_edit_schema,
    app_profile_create_edit_schema,
    resource_profile_create_edit_schema,
    cluster_registration_new_schema,
    attach_dettach_profile_schema,
    ksu_schema,
    app_instance_schema,
    app_instance_edit_schema,
    app_instance_update_schema,
    oka_schema,
    node_create_new_schema,
    node_edit_schema,
)
from osm_common.dbbase import deep_update_rfc7396, DbException
from osm_common.msgbase import MsgException
from osm_common.fsbase import FsException

__author__ = (
    "Shrinithi R <shrinithi.r@tataelxsi.co.in>",
    "Shahithya Y <shahithya.y@tataelxsi.co.in>",
)


class InfraContTopic(ProfileTopic):
    topic = "k8sinfra_controller"
    topic_msg = "k8s_infra_controller"
    schema_new = infra_controller_profile_create_new_schema
    schema_edit = infra_controller_profile_create_edit_schema

    def __init__(self, db, fs, msg, auth):
        super().__init__(db, fs, msg, auth)

    def new(self, rollback, session, indata=None, kwargs=None, headers=None):
        # To create the new infra controller profile
        return self.new_profile(rollback, session, indata, kwargs, headers)

    def default(self, rollback, session, indata=None, kwargs=None, headers=None):
        # To create the default infra controller profile while creating the cluster
        return self.default_profile(rollback, session, indata, kwargs, headers)

    def delete(self, session, _id, dry_run=False, not_send_msg=None):
        check = {"infra_controller_profiles": _id}
        self.check_dependency(check, operation_type="delete")
        self.delete_profile(session, _id, dry_run, not_send_msg)
        return _id


class InfraConfTopic(ProfileTopic):
    topic = "k8sinfra_config"
    topic_msg = "k8s_infra_config"
    schema_new = infra_config_profile_create_new_schema
    schema_edit = infra_config_profile_create_edit_schema

    def __init__(self, db, fs, msg, auth):
        super().__init__(db, fs, msg, auth)

    def new(self, rollback, session, indata=None, kwargs=None, headers=None):
        # To create the new infra config profile
        return self.new_profile(rollback, session, indata, kwargs, headers)

    def default(self, rollback, session, indata=None, kwargs=None, headers=None):
        # To create the default infra config profile while creating the cluster
        return self.default_profile(rollback, session, indata, kwargs, headers)

    def delete(self, session, _id, dry_run=False, not_send_msg=None):
        check = {"infra_config_profiles": _id}
        self.check_dependency(check, operation_type="delete")
        self.delete_profile(session, _id, dry_run, not_send_msg)
        return _id


class AppProfileTopic(ProfileTopic):
    topic = "k8sapp"
    topic_msg = "k8s_app"
    schema_new = app_profile_create_new_schema
    schema_edit = app_profile_create_edit_schema

    def __init__(self, db, fs, msg, auth):
        super().__init__(db, fs, msg, auth)

    def new(self, rollback, session, indata=None, kwargs=None, headers=None):
        # To create the new app profile
        return self.new_profile(rollback, session, indata, kwargs, headers)

    def default(self, rollback, session, indata=None, kwargs=None, headers=None):
        # To create the default app profile while creating the cluster
        return self.default_profile(rollback, session, indata, kwargs, headers)

    def delete(self, session, _id, dry_run=False, not_send_msg=None):
        check = {"app_profiles": _id}
        self.check_dependency(check, operation_type="delete")
        self.delete_profile(session, _id, dry_run, not_send_msg)
        return _id


class ResourceTopic(ProfileTopic):
    topic = "k8sresource"
    topic_msg = "k8s_resource"
    schema_new = resource_profile_create_new_schema
    schema_edit = resource_profile_create_edit_schema

    def __init__(self, db, fs, msg, auth):
        super().__init__(db, fs, msg, auth)

    def new(self, rollback, session, indata=None, kwargs=None, headers=None):
        # To create the new resource profile
        return self.new_profile(rollback, session, indata, kwargs, headers)

    def default(self, rollback, session, indata=None, kwargs=None, headers=None):
        # To create the default resource profile while creating the cluster
        return self.default_profile(rollback, session, indata, kwargs, headers)

    def delete(self, session, _id, dry_run=False, not_send_msg=None):
        check = {"resource_profiles": _id}
        self.check_dependency(check, operation_type="delete")
        self.delete_profile(session, _id, dry_run, not_send_msg)
        return _id


class ClusterTopic(ACMTopic):
    topic = "clusters"
    topic_msg = "cluster"
    schema_new = cluster_creation_new_schema
    schema_edit = attach_dettach_profile_schema

    def __init__(self, db, fs, msg, auth):
        super().__init__(db, fs, msg, auth)
        self.infra_contr_topic = InfraContTopic(db, fs, msg, auth)
        self.infra_conf_topic = InfraConfTopic(db, fs, msg, auth)
        self.resource_topic = ResourceTopic(db, fs, msg, auth)
        self.app_topic = AppProfileTopic(db, fs, msg, auth)

    @staticmethod
    def format_on_new(content, project_id=None, make_public=False):
        ACMTopic.format_on_new(content, project_id=project_id, make_public=make_public)
        content["current_operation"] = None

    def new(self, rollback, session, indata=None, kwargs=None, headers=None):
        """
        Creates a new k8scluster into database.
        :param rollback: list to append the created items at database in case a rollback must be done
        :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
        :param indata: params to be used for the k8cluster
        :param kwargs: used to override the indata
        :param headers: http request headers
        :return: the _id of k8scluster created at database. Or an exception of type
            EngineException, ValidationError, DbException, FsException, MsgException.
            Note: Exceptions are not captured on purpose. They should be captured at called
        """

        step = "checking quotas"  # first step must be defined outside try
        try:
            if self.multiproject:
                self.check_quota(session)

            content = self._remove_envelop(indata)

            step = "name unique check"
            self.cluster_unique_name_check(session, indata["name"])

            step = "validating input parameters"
            self._update_input_with_kwargs(content, kwargs)

            content = self._validate_input_new(content, session, force=session["force"])

            operation_params = indata.copy()

            self.check_conflict_on_new(session, content)
            self.format_on_new(
                content, project_id=session["project_id"], make_public=session["public"]
            )

            step = "filling cluster details from input data"
            content = self._create_cluster(
                content, rollback, session, indata, kwargs, headers
            )

            step = "creating cluster at database"
            _id = self.db.create(self.topic, content)

            op_id = self.format_on_operation(
                content,
                "create",
                operation_params,
            )

            pubkey, privkey = self._generate_age_key()
            content["age_pubkey"] = self.db.encrypt(
                pubkey, schema_version="1.11", salt=_id
            )
            content["age_privkey"] = self.db.encrypt(
                privkey, schema_version="1.11", salt=_id
            )

            # TODO: set age_pubkey and age_privkey in the default profiles
            rollback.append({"topic": self.topic, "_id": _id})
            self.db.set_one("clusters", {"_id": _id}, content)
            self._send_msg("create", {"cluster_id": _id, "operation_id": op_id})

            # To add the content in old collection "k8sclusters"
            self.add_to_old_collection(content, session)

            return _id, None
        except (
            ValidationError,
            EngineException,
            DbException,
            MsgException,
            FsException,
        ) as e:
            raise type(e)("{} while '{}'".format(e, step), http_code=e.http_code)

    def _validate_input_new(self, content, session, force=False):
        # validating vim and checking the mandatory parameters
        vim_type = self.check_vim(session, content["vim_account"])

        # for aws
        if vim_type == "aws":
            self._aws_check(content)

        # for azure and gcp
        elif vim_type in ["azure", "gcp"]:
            self._params_check(content)

        return super()._validate_input_new(content, force=session["force"])

    def _aws_check(self, indata):
        if "node_count" in indata or "node_size" in indata:
            raise ValueError("node_count and node_size are not allowed for AWS")
        return

    def _params_check(self, indata):
        if "node_count" not in indata and "node_size" not in indata:
            raise ValueError("node_count and node_size are mandatory parameter")
        return

    def _create_cluster(self, content, rollback, session, indata, kwargs, headers):
        private_subnet = indata.get("private_subnet")
        public_subnet = indata.get("public_subnet")

        # Enforce: if private_subnet is provided, public_subnet must also be provided
        if (private_subnet and not public_subnet) or (
            public_subnet and not private_subnet
        ):
            raise ValueError(
                "'public_subnet' must be provided if 'private_subnet' is given and viceversa."
            )

        # private Subnet validation
        if private_subnet:
            count = len(private_subnet)
            if count != 2:
                raise ValueError(
                    f"private_subnet must contain exactly 2 items, got {count}"
                )

        # public Subnet validation
        public_subnet = indata.get("public_subnet")
        if public_subnet:
            count = len(public_subnet)
            if count != 1:
                raise ValueError(
                    f"public_subnet must contain exactly 1 items, got {count}"
                )

        content["infra_controller_profiles"] = [
            self._create_default_profiles(
                rollback, session, indata, kwargs, headers, self.infra_contr_topic
            )
        ]
        content["infra_config_profiles"] = [
            self._create_default_profiles(
                rollback, session, indata, kwargs, headers, self.infra_conf_topic
            )
        ]
        content["resource_profiles"] = [
            self._create_default_profiles(
                rollback, session, indata, kwargs, headers, self.resource_topic
            )
        ]
        content["app_profiles"] = [
            self._create_default_profiles(
                rollback, session, indata, kwargs, headers, self.app_topic
            )
        ]
        content["created"] = "true"
        content["state"] = "IN_CREATION"
        content["operatingState"] = "PROCESSING"
        content["git_name"] = self.create_gitname(content, session)
        content["resourceState"] = "IN_PROGRESS.REQUEST_RECEIVED"

        # Get the vim_account details
        vim_account_details = self.db.get_one(
            "vim_accounts", {"name": content["vim_account"]}
        )

        # Add optional fields if they don't exist in the request
        if "region_name" not in indata:
            region_name = vim_account_details.get("config", {}).get("region_name")
            if region_name:
                content["region_name"] = region_name

        if "resource_group" not in indata:
            resource_group = vim_account_details.get("config", {}).get("resource_group")
            if resource_group:
                content["resource_group"] = resource_group

        version = "k8s_version" in content
        if not version:
            content["k8s_version"] = "1.32"
        # Additional cluster information, specific for each cluster type
        content["config"] = indata.get("config", {})
        content["node_count"] = indata.get("node_count", 0)
        content["ksu_count"] = 0
        self.logger.info(f"content is : {content}")
        return content

    def check_vim(self, session, name):
        try:
            vim_account_details = self.db.get_one("vim_accounts", {"name": name})
            if vim_account_details is not None:
                return vim_account_details["vim_type"]
        except ValidationError as e:
            raise EngineException(
                e,
                HTTPStatus.UNPROCESSABLE_ENTITY,
            )

    def _create_default_profiles(
        self, rollback, session, indata, kwargs, headers, topic
    ):
        topic = self.to_select_topic(topic)
        default_profiles = topic.default(rollback, session, indata, kwargs, headers)
        return default_profiles

    def to_select_topic(self, topic):
        if topic == "infra_controller_profiles":
            topic = self.infra_contr_topic
        elif topic == "infra_config_profiles":
            topic = self.infra_conf_topic
        elif topic == "resource_profiles":
            topic = self.resource_topic
        elif topic == "app_profiles":
            topic = self.app_topic
        return topic

    def show_one(self, session, _id, profile, filter_q=None, api_req=False):
        try:
            filter_q = self._get_project_filter(session)
            filter_q[self.id_field(self.topic, _id)] = _id
            content = self.db.get_one(self.topic, filter_q)
            existing_profiles = []
            topic = None
            topic = self.to_select_topic(profile)
            for profile_id in content[profile]:
                data = topic.show(session, profile_id, filter_q, api_req)
                existing_profiles.append(data)
            return existing_profiles
        except ValidationError as e:
            raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)

    def state_check(self, profile_id, session, topic):
        topic = self.to_select_topic(topic)
        content = topic.show(session, profile_id, filter_q=None, api_req=False)
        state = content["state"]
        if state == "CREATED":
            return
        else:
            raise EngineException(
                f" {profile_id}  is not in created state",
                HTTPStatus.UNPROCESSABLE_ENTITY,
            )

    def edit(self, session, _id, item, indata=None, kwargs=None):
        if item not in (
            "infra_controller_profiles",
            "infra_config_profiles",
            "app_profiles",
            "resource_profiles",
        ):
            self.schema_edit = cluster_edit_schema
            super().edit(session, _id, indata=item, kwargs=kwargs, content=None)
        else:
            indata = self._remove_envelop(indata)
            indata = self._validate_input_edit(
                indata, content=None, force=session["force"]
            )
            if indata.get("add_profile"):
                self.add_profile(session, _id, item, indata)
            elif indata.get("remove_profile"):
                self.remove_profile(session, _id, item, indata)
            else:
                error_msg = "Add / remove operation is only applicable"
                raise EngineException(error_msg, HTTPStatus.UNPROCESSABLE_ENTITY)

    def edit_extra_before(self, session, _id, indata=None, kwargs=None, content=None):
        check_dict = self.db.get_one(self.topic, {"_id": _id})
        if "name" in indata and check_dict["name"] != indata["name"]:
            self.check_unique_name(session, indata["name"])
            _filter = {"name": indata["name"]}
            topic_list = [
                "k8sclusters",
                "k8sinfra_controller",
                "k8sinfra_config",
                "k8sapp",
                "k8sresource",
            ]
            # Check unique name for k8scluster and profiles
            for topic in topic_list:
                if self.db.get_one(
                    topic, _filter, fail_on_empty=False, fail_on_more=False
                ):
                    raise EngineException(
                        "name '{}' already exists for {}".format(indata["name"], topic),
                        HTTPStatus.CONFLICT,
                    )
            # Replace name in k8scluster and profiles
            for topic in topic_list:
                data = self.db.get_one(topic, {"name": check_dict["name"]})
                data["name"] = indata["name"]
                self.db.replace(topic, data["_id"], data)
        return True

    def add_profile(self, session, _id, item, indata=None):
        check = {"cluster": _id, item: indata["add_profile"][0]["id"]}
        self.check_dependency(check)
        indata = self._remove_envelop(indata)
        operation_params = indata
        profile_id = indata["add_profile"][0]["id"]
        # check state
        self.state_check(profile_id, session, item)
        filter_q = self._get_project_filter(session)
        filter_q[self.id_field(self.topic, _id)] = _id
        content = self.db.get_one(self.topic, filter_q)
        profile_list = content[item]

        if profile_id not in profile_list:
            content["operatingState"] = "PROCESSING"
            op_id = self.format_on_operation(
                content,
                "add",
                operation_params,
            )
            self.db.set_one("clusters", {"_id": content["_id"]}, content)
            self._send_msg(
                "add",
                {
                    "cluster_id": _id,
                    "profile_id": profile_id,
                    "profile_type": item,
                    "operation_id": op_id,
                },
            )
        else:
            raise EngineException(
                f"{item} {profile_id} already exists", HTTPStatus.UNPROCESSABLE_ENTITY
            )

    def _get_default_profiles(self, session, topic):
        topic = self.to_select_topic(topic)
        existing_profiles = topic.list(session, filter_q=None, api_req=False)
        default_profiles = [
            profile["_id"]
            for profile in existing_profiles
            if profile.get("default", False)
        ]
        return default_profiles

    def remove_profile(self, session, _id, item, indata):
        check = {"cluster": _id, item: indata["remove_profile"][0]["id"]}
        self.check_dependency(check)
        indata = self._remove_envelop(indata)
        operation_params = indata
        profile_id = indata["remove_profile"][0]["id"]
        filter_q = self._get_project_filter(session)
        filter_q[self.id_field(self.topic, _id)] = _id
        content = self.db.get_one(self.topic, filter_q)
        profile_list = content[item]

        default_profiles = self._get_default_profiles(session, item)

        if profile_id in default_profiles:
            raise EngineException(
                "Cannot remove default profile", HTTPStatus.UNPROCESSABLE_ENTITY
            )
        if profile_id in profile_list:
            op_id = self.format_on_operation(
                content,
                "remove",
                operation_params,
            )
            self.db.set_one("clusters", {"_id": content["_id"]}, content)
            self._send_msg(
                "remove",
                {
                    "cluster_id": _id,
                    "profile_id": profile_id,
                    "profile_type": item,
                    "operation_id": op_id,
                },
            )
        else:
            raise EngineException(
                f"{item} {profile_id} does'nt exists", HTTPStatus.UNPROCESSABLE_ENTITY
            )

    def get_cluster_creds(self, session, _id, item):
        if not self.multiproject:
            filter_db = {}
        else:
            filter_db = self._get_project_filter(session)
        filter_db[BaseTopic.id_field(self.topic, _id)] = _id
        operation_params = None
        data = self.db.get_one(self.topic, filter_db)
        op_id = self.format_on_operation(data, item, operation_params)
        self.db.set_one(self.topic, {"_id": data["_id"]}, data)
        self._send_msg("get_creds", {"cluster_id": _id, "operation_id": op_id})
        return op_id

    def get_cluster_creds_file(self, session, _id, item, op_id):
        if not self.multiproject:
            filter_db = {}
        else:
            filter_db = self._get_project_filter(session)
        filter_db[BaseTopic.id_field(self.topic, _id)] = _id

        data = self.db.get_one(self.topic, filter_db)
        creds_flag = None
        for operations in data["operationHistory"]:
            if operations["op_id"] == op_id:
                creds_flag = operations["result"]
        self.logger.info("Creds Flag: {}".format(creds_flag))

        if creds_flag is True:
            credentials = data["credentials"]

            file_pkg = None
            current_path = _id

            self.fs.file_delete(current_path, ignore_non_exist=True)
            self.fs.mkdir(current_path)
            filename = "credentials.yaml"
            file_path = (current_path, filename)
            self.logger.info("File path: {}".format(file_path))
            file_pkg = self.fs.file_open(file_path, "a+b")

            credentials_yaml = yaml.safe_dump(
                credentials, indent=4, default_flow_style=False
            )
            file_pkg.write(credentials_yaml.encode(encoding="utf-8"))

            if file_pkg:
                file_pkg.close()
            file_pkg = None
            self.fs.sync(from_path=current_path)

            return (
                self.fs.file_open((current_path, filename), "rb"),
                "text/plain",
            )
        else:
            raise EngineException(
                "Not possible to get the credentials of the cluster",
                HTTPStatus.UNPROCESSABLE_ENTITY,
            )

    def update_item(self, session, _id, item, indata):
        if not self.multiproject:
            filter_db = {}
        else:
            filter_db = self._get_project_filter(session)
        # To allow project&user addressing by name AS WELL AS _id
        filter_db[BaseTopic.id_field(self.topic, _id)] = _id
        validate_input(indata, cluster_update_schema)
        data = self.db.get_one(self.topic, filter_db)
        operation_params = {}
        data["operatingState"] = "PROCESSING"
        data["resourceState"] = "IN_PROGRESS"
        operation_params = indata
        op_id = self.format_on_operation(
            data,
            item,
            operation_params,
        )
        self.db.set_one(self.topic, {"_id": _id}, data)
        data = {"cluster_id": _id, "operation_id": op_id}
        self._send_msg(item, data)
        return op_id

    def delete_extra_before(self, session, _id, db_content, not_send_msg=None):
        op_id = self.common_delete(_id, db_content)
        return {"cluster_id": _id, "operation_id": op_id, "force": session["force"]}

    def delete(self, session, _id, dry_run=False, not_send_msg=None):
        check_dict = {"cluster": _id}
        self.check_dependency(check_dict, operation_type="delete")
        filter_q = self._get_project_filter(session)
        filter_q[self.id_field(self.topic, _id)] = _id
        check_dict = self.db.get_one(self.topic, filter_q)
        op_id = check_dict["current_operation"]
        if check_dict["created"] == "false":
            raise EngineException(
                "Cannot delete registered cluster. Please deregister.",
                HTTPStatus.UNPROCESSABLE_ENTITY,
            )
        super().delete(session, _id, dry_run, not_send_msg)
        return op_id


class NodeGroupTopic(ACMTopic):
    topic = "nodegroups"
    topic_msg = "nodegroup"
    schema_new = node_create_new_schema
    schema_edit = node_edit_schema

    def __init__(self, db, fs, msg, auth):
        BaseTopic.__init__(self, db, fs, msg, auth)

    @staticmethod
    def format_on_new(content, project_id=None, make_public=False):
        BaseTopic.format_on_new(content, project_id=project_id, make_public=make_public)
        content["current_operation"] = None
        content["state"] = "IN_CREATION"
        content["operatingState"] = "PROCESSING"
        content["resourceState"] = "IN_PROGRESS"

    def new(self, rollback, session, indata=None, kwargs=None, headers=None):
        self.logger.info(f"Indata: {indata}")
        self.check_unique_name(session, indata["name"])

        indata = self._remove_envelop(indata)
        self._update_input_with_kwargs(indata, kwargs)
        if not indata.get("private_subnet") and not indata.get("public_subnet"):
            raise EngineException(
                "Please provide atleast one subnet",
                HTTPStatus.UNPROCESSABLE_ENTITY,
            )
        content = self._validate_input_new(indata, session["force"])

        self.logger.info(f"Indata: {indata}")
        self.logger.info(f"Content: {content}")
        cluster_id = content["cluster_id"]
        db_cluster = self.db.get_one("clusters", {"_id": cluster_id})
        private_subnet = db_cluster.get("private_subnet")
        public_subnet = db_cluster.get("public_subnet")
        if content.get("private_subnet"):
            for subnet in content["private_subnet"]:
                if subnet not in private_subnet:
                    raise EngineException(
                        "No External subnet is used to add nodegroup",
                        HTTPStatus.UNPROCESSABLE_ENTITY,
                    )
        if content.get("public_subnet"):
            for subnet in content["public_subnet"]:
                if subnet not in public_subnet:
                    raise EngineException(
                        "No External subnet is used to add nodegroup",
                        HTTPStatus.UNPROCESSABLE_ENTITY,
                    )

        operation_params = {}
        for content_key, content_value in content.items():
            operation_params[content_key] = content_value
        self.format_on_new(
            content, session["project_id"], make_public=session["public"]
        )
        content["git_name"] = self.create_gitname(content, session)
        self.logger.info(f"Operation Params: {operation_params}")
        op_id = self.format_on_operation(
            content,
            "create",
            operation_params,
        )
        node_count = db_cluster.get("node_count")
        new_node_count = node_count + 1
        self.logger.info(f"New Node count: {new_node_count}")
        db_cluster["node_count"] = new_node_count
        self.db.set_one("clusters", {"_id": cluster_id}, db_cluster)
        _id = self.db.create(self.topic, content)
        self._send_msg("add_nodegroup", {"nodegroup_id": _id, "operation_id": op_id})
        return _id, op_id

    def list(self, session, filter_q=None, api_req=False):
        db_filter = {}
        if filter_q.get("cluster_id"):
            db_filter["cluster_id"] = filter_q.get("cluster_id")
        data_list = self.db.get_list(self.topic, db_filter)
        cluster_data = self.db.get_one("clusters", {"_id": db_filter["cluster_id"]})
        self.logger.info(f"Cluster Data: {cluster_data}")
        self.logger.info(f"Data: {data_list}")
        if filter_q.get("cluster_id"):
            outdata = {}
            outdata["count"] = cluster_data["node_count"]
            outdata["data"] = data_list
            self.logger.info(f"Outdata: {outdata}")
            return outdata
        if api_req:
            data_list = [self.sol005_projection(inst) for inst in data_list]
        return data_list

    def delete(self, session, _id, dry_run=False, not_send_msg=None):
        if not self.multiproject:
            filter_q = {}
        else:
            filter_q = self._get_project_filter(session)
        filter_q[self.id_field(self.topic, _id)] = _id
        item_content = self.db.get_one(self.topic, filter_q)
        item_content["state"] = "IN_DELETION"
        item_content["operatingState"] = "PROCESSING"
        item_content["resourceState"] = "IN_PROGRESS"
        self.check_conflict_on_del(session, _id, item_content)
        op_id = self.format_on_operation(
            item_content,
            "delete",
            None,
        )
        self.db.set_one(self.topic, {"_id": item_content["_id"]}, item_content)
        self._send_msg(
            "delete_nodegroup",
            {"nodegroup_id": _id, "operation_id": op_id},
            not_send_msg=not_send_msg,
        )
        return op_id

    def update_item(self, session, _id, item, indata):
        content = None
        try:
            if not content:
                content = self.db.get_one(self.topic, {"_id": _id})
            indata = self._validate_input_edit(indata, content, force=session["force"])
            _id = content.get("_id") or _id

            content = self.check_conflict_on_edit(session, content, indata, _id=_id)
            op_id = self.format_on_edit(content, indata)
            op_id = ACMTopic.format_on_operation(
                content,
                "scale",
                indata,
            )
            self.logger.info(f"op_id: {op_id}")
            content["operatingState"] = "PROCESSING"
            content["resourceState"] = "IN_PROGRESS"
            self.db.replace(self.topic, _id, content)
            self._send_msg(
                "scale_nodegroup", {"nodegroup_id": _id, "operation_id": op_id}
            )
            return op_id
        except ValidationError as e:
            raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)

    def edit(self, session, _id, indata, kwargs):
        content = None

        # Override descriptor with query string kwargs
        if kwargs:
            self._update_input_with_kwargs(indata, kwargs)
        try:
            if indata and session.get("set_project"):
                raise EngineException(
                    "Cannot edit content and set to project (query string SET_PROJECT) at same time",
                    HTTPStatus.UNPROCESSABLE_ENTITY,
                )
            # TODO self._check_edition(session, indata, _id, force)
            if not content:
                content = self.db.get_one(self.topic, {"_id": _id})

            indata = self._validate_input_edit(indata, content, force=session["force"])
            self.logger.info(f"Indata: {indata}")

            # To allow project addressing by name AS WELL AS _id. Get the _id, just in case the provided one is a name
            _id = content.get("_id") or _id

            content = self.check_conflict_on_edit(session, content, indata, _id=_id)
            if "name" in indata and "description" in indata:
                content["name"] = indata["name"]
                content["description"] = indata["description"]
            elif "name" in indata:
                content["name"] = indata["name"]
            elif "description" in indata:
                content["description"] = indata["description"]
            op_id = self.format_on_edit(content, indata)
            self.db.set_one(self.topic, {"_id": _id}, content)
            return op_id
        except ValidationError as e:
            raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)


class ClusterOpsTopic(ACMTopic):
    topic = "clusters"
    topic_msg = "cluster"
    schema_new = cluster_registration_new_schema

    def __init__(self, db, fs, msg, auth):
        super().__init__(db, fs, msg, auth)
        self.infra_contr_topic = InfraContTopic(db, fs, msg, auth)
        self.infra_conf_topic = InfraConfTopic(db, fs, msg, auth)
        self.resource_topic = ResourceTopic(db, fs, msg, auth)
        self.app_topic = AppProfileTopic(db, fs, msg, auth)

    @staticmethod
    def format_on_new(content, project_id=None, make_public=False):
        ACMTopic.format_on_new(content, project_id=project_id, make_public=make_public)
        content["current_operation"] = None

    def add(self, rollback, session, indata, kwargs=None, headers=None):
        step = "checking quotas"
        try:
            self.check_quota(session)
            step = "name unique check"
            self.cluster_unique_name_check(session, indata["name"])
            # self.check_unique_name(session, indata["name"])
            step = "validating input parameters"
            cls_add_request = self._remove_envelop(indata)
            self._update_input_with_kwargs(cls_add_request, kwargs)
            cls_add_request = self._validate_input_new(
                cls_add_request, session["force"]
            )
            operation_params = cls_add_request

            step = "filling cluster details from input data"
            cls_add_request = self._add_cluster(
                cls_add_request, rollback, session, indata, kwargs, headers
            )

            step = "registering the cluster at database"
            self.format_on_new(
                cls_add_request, session["project_id"], make_public=session["public"]
            )
            op_id = self.format_on_operation(
                cls_add_request,
                "register",
                operation_params,
            )
            _id = self.db.create(self.topic, cls_add_request)
            pubkey, privkey = self._generate_age_key()
            cls_add_request["age_pubkey"] = self.db.encrypt(
                pubkey, schema_version="1.11", salt=_id
            )
            cls_add_request["age_privkey"] = self.db.encrypt(
                privkey, schema_version="1.11", salt=_id
            )
            # TODO: set age_pubkey and age_privkey in the default profiles
            self.db.set_one(self.topic, {"_id": _id}, cls_add_request)
            rollback.append({"topic": self.topic, "_id": _id})
            self._send_msg("register", {"cluster_id": _id, "operation_id": op_id})

            # To add the content in old collection "k8sclusters"
            self.add_to_old_collection(cls_add_request, session)

            return _id, None
        except (
            ValidationError,
            EngineException,
            DbException,
            MsgException,
            FsException,
        ) as e:
            raise type(e)("{} while '{}'".format(e, step), http_code=e.http_code)

    def _add_cluster(self, cls_add_request, rollback, session, indata, kwargs, headers):
        cls_add = {
            "name": cls_add_request["name"],
            "credentials": cls_add_request["credentials"],
            "vim_account": cls_add_request["vim_account"],
            "bootstrap": cls_add_request["bootstrap"],
            "openshift": cls_add_request.get("openshift", False),
            "infra_controller_profiles": [
                self._create_default_profiles(
                    rollback, session, indata, kwargs, headers, self.infra_contr_topic
                )
            ],
            "infra_config_profiles": [
                self._create_default_profiles(
                    rollback, session, indata, kwargs, headers, self.infra_conf_topic
                )
            ],
            "resource_profiles": [
                self._create_default_profiles(
                    rollback, session, indata, kwargs, headers, self.resource_topic
                )
            ],
            "app_profiles": [
                self._create_default_profiles(
                    rollback, session, indata, kwargs, headers, self.app_topic
                )
            ],
            "created": "false",
            "state": "IN_CREATION",
            "operatingState": "PROCESSING",
            "git_name": self.create_gitname(cls_add_request, session),
            "resourceState": "IN_PROGRESS.REQUEST_RECEIVED",
        }
        # Add optional fields if they exist in the request
        if "description" in cls_add_request:
            cls_add["description"] = cls_add_request["description"]
        return cls_add

    def check_vim(self, session, name):
        try:
            vim_account_details = self.db.get_one("vim_accounts", {"name": name})
            if vim_account_details is not None:
                return name
        except ValidationError as e:
            raise EngineException(
                e,
                HTTPStatus.UNPROCESSABLE_ENTITY,
            )

    def _create_default_profiles(
        self, rollback, session, indata, kwargs, headers, topic
    ):
        topic = self.to_select_topic(topic)
        default_profiles = topic.default(rollback, session, indata, kwargs, headers)
        return default_profiles

    def to_select_topic(self, topic):
        if topic == "infra_controller_profiles":
            topic = self.infra_contr_topic
        elif topic == "infra_config_profiles":
            topic = self.infra_conf_topic
        elif topic == "resource_profiles":
            topic = self.resource_topic
        elif topic == "app_profiles":
            topic = self.app_topic
        return topic

    def remove(self, session, _id, dry_run=False, not_send_msg=None):
        """
        Delete item by its internal _id
        :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
        :param _id: server internal id
        :param dry_run: make checking but do not delete
        :param not_send_msg: To not send message (False) or store content (list) instead
        :return: operation id (None if there is not operation), raise exception if error or not found, conflict, ...
        """

        # To allow addressing projects and users by name AS WELL AS by _id
        if not self.multiproject:
            filter_q = {}
        else:
            filter_q = self._get_project_filter(session)
        filter_q[self.id_field(self.topic, _id)] = _id
        item_content = self.db.get_one(self.topic, filter_q)

        op_id = self.format_on_operation(
            item_content,
            "deregister",
            None,
        )
        self.db.set_one(self.topic, {"_id": _id}, item_content)

        self.check_conflict_on_del(session, _id, item_content)
        if dry_run:
            return None

        if self.multiproject and session["project_id"]:
            # remove reference from project_read if there are more projects referencing it. If it last one,
            # do not remove reference, but delete
            other_projects_referencing = next(
                (
                    p
                    for p in item_content["_admin"]["projects_read"]
                    if p not in session["project_id"] and p != "ANY"
                ),
                None,
            )

            # check if there are projects referencing it (apart from ANY, that means, public)....
            if other_projects_referencing:
                # remove references but not delete
                update_dict_pull = {
                    "_admin.projects_read": session["project_id"],
                    "_admin.projects_write": session["project_id"],
                }
                self.db.set_one(
                    self.topic, filter_q, update_dict=None, pull_list=update_dict_pull
                )
                return None
            else:
                can_write = next(
                    (
                        p
                        for p in item_content["_admin"]["projects_write"]
                        if p == "ANY" or p in session["project_id"]
                    ),
                    None,
                )
                if not can_write:
                    raise EngineException(
                        "You have not write permission to delete it",
                        http_code=HTTPStatus.UNAUTHORIZED,
                    )

        # delete
        self._send_msg(
            "deregister",
            {"cluster_id": _id, "operation_id": op_id},
            not_send_msg=not_send_msg,
        )
        return _id


class KsusTopic(ACMTopic):
    topic = "ksus"
    okapkg_topic = "okas"
    topic_msg = "ksu"
    schema_new = ksu_schema
    schema_edit = ksu_schema
    MAP_PROFILE = {
        "infra_controller_profiles": "infra-controllers",
        "infra_config_profiles": "infra-configs",
        "resource_profiles": "managed_resources",
        "app_profiles": "apps",
    }

    def __init__(self, db, fs, msg, auth):
        super().__init__(db, fs, msg, auth)
        self.logger = logging.getLogger("nbi.ksus")

    @staticmethod
    def format_on_new(content, project_id=None, make_public=False):
        BaseTopic.format_on_new(content, project_id=project_id, make_public=make_public)
        content["current_operation"] = None
        content["state"] = "IN_CREATION"
        content["operatingState"] = "PROCESSING"
        content["resourceState"] = "IN_PROGRESS"

    def new(self, rollback, session, indata=None, kwargs=None, headers=None):
        _id_list = []
        for content in indata["ksus"]:
            check_dict = {content["profile"]["profile_type"]: content["profile"]["_id"]}
            check_dict["okas"] = []

            for okas in content["oka"]:
                if "_id" in okas and "sw_catalog_path" in okas:
                    raise EngineException(
                        "Cannot create ksu with both OKA and SW catalog path",
                        HTTPStatus.UNPROCESSABLE_ENTITY,
                    )
                elif "_id" not in okas and "sw_catalog_path" not in okas:
                    raise EngineException(
                        "Cannot create ksu. Either oka id or SW catalog path is required for all OKA in a KSU",
                        HTTPStatus.UNPROCESSABLE_ENTITY,
                    )
                elif "_id" in okas:
                    check_dict["okas"].append(okas["_id"])
            self.check_dependency(check_dict)

            # Override descriptor with query string kwargs
            content = self._remove_envelop(content)
            self._update_input_with_kwargs(content, kwargs)
            content = self._validate_input_new(input=content, force=session["force"])

            # Check for unique name
            self.check_unique_name(session, content["name"])

            self.check_conflict_on_new(session, content)

            operation_params = {}
            for content_key, content_value in content.items():
                operation_params[content_key] = content_value
            self.format_on_new(
                content, project_id=session["project_id"], make_public=session["public"]
            )
            op_id = self.format_on_operation(
                content,
                operation_type="create",
                operation_params=operation_params,
            )
            content["git_name"] = self.create_gitname(content, session)

            # Update Oka_package usage state
            for okas in content["oka"]:
                if "_id" in okas.keys():
                    self.update_usage_state(session, okas)

            profile_id = content["profile"].get("_id")
            profile_type = content["profile"].get("profile_type")
            db_cluster_list = self.db.get_list("clusters")
            for db_cluster in db_cluster_list:
                if db_cluster.get("created") == "true":
                    profile_list = db_cluster[profile_type]
                    if profile_id in profile_list:
                        ksu_count = db_cluster.get("ksu_count")
                        new_ksu_count = ksu_count + 1
                        self.logger.info(f"New KSU count: {new_ksu_count}")
                        db_cluster["ksu_count"] = new_ksu_count
                        self.db.set_one(
                            "clusters", {"_id": db_cluster["_id"]}, db_cluster
                        )

            _id = self.db.create(self.topic, content)
            rollback.append({"topic": self.topic, "_id": _id})
            _id_list.append(_id)
        data = {"ksus_list": _id_list, "operation_id": op_id}
        self._send_msg("create", data)
        return _id_list, op_id

    def clone(self, rollback, session, _id, indata, kwargs, headers):
        check_dict = {
            "ksu": _id,
            indata["profile"]["profile_type"]: indata["profile"]["_id"],
        }
        self.check_dependency(check_dict)
        filter_db = self._get_project_filter(session)
        filter_db[BaseTopic.id_field(self.topic, _id)] = _id
        data = self.db.get_one(self.topic, filter_db)

        op_id = self.format_on_operation(
            data,
            "clone",
            indata,
        )
        self.db.set_one(self.topic, {"_id": data["_id"]}, data)
        self._send_msg("clone", {"ksus_list": [data["_id"]], "operation_id": op_id})
        return op_id

    def update_usage_state(self, session, oka_content):
        _id = oka_content["_id"]
        filter_db = self._get_project_filter(session)
        filter_db[BaseTopic.id_field(self.topic, _id)] = _id

        data = self.db.get_one(self.okapkg_topic, filter_db)
        if data["_admin"]["usageState"] == "NOT_IN_USE":
            usage_state_update = {
                "_admin.usageState": "IN_USE",
            }
            self.db.set_one(
                self.okapkg_topic, {"_id": _id}, update_dict=usage_state_update
            )

    def move_ksu(self, session, _id, indata=None, kwargs=None, content=None):
        check_dict = {
            "ksu": _id,
            indata["profile"]["profile_type"]: indata["profile"]["_id"],
        }
        self.check_dependency(check_dict)
        indata = self._remove_envelop(indata)

        # Override descriptor with query string kwargs
        if kwargs:
            self._update_input_with_kwargs(indata, kwargs)
        try:
            if indata and session.get("set_project"):
                raise EngineException(
                    "Cannot edit content and set to project (query string SET_PROJECT) at same time",
                    HTTPStatus.UNPROCESSABLE_ENTITY,
                )
            # TODO self._check_edition(session, indata, _id, force)
            if not content:
                content = self.show(session, _id)
            indata = self._validate_input_edit(
                input=indata, content=content, force=session["force"]
            )
            operation_params = indata
            deep_update_rfc7396(content, indata)

            # To allow project addressing by name AS WELL AS _id. Get the _id, just in case the provided one is a name
            _id = content.get("_id") or _id
            op_id = self.format_on_operation(
                content,
                "move",
                operation_params,
            )
            if content.get("_admin"):
                now = time()
                content["_admin"]["modified"] = now
            content["operatingState"] = "PROCESSING"
            content["resourceState"] = "IN_PROGRESS"

            self.db.replace(self.topic, _id, content)
            data = {"ksus_list": [content["_id"]], "operation_id": op_id}
            self._send_msg("move", data)
            return op_id
        except ValidationError as e:
            raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)

    def check_conflict_on_edit(self, session, final_content, edit_content, _id):
        if final_content["name"] != edit_content["name"]:
            self.check_unique_name(session, edit_content["name"])
        return final_content

    @staticmethod
    def format_on_edit(final_content, edit_content):
        op_id = ACMTopic.format_on_operation(
            final_content,
            "update",
            edit_content,
        )
        final_content["operatingState"] = "PROCESSING"
        final_content["resourceState"] = "IN_PROGRESS"
        if final_content.get("_admin"):
            now = time()
            final_content["_admin"]["modified"] = now
        return op_id

    def edit(self, session, _id, indata, kwargs):
        _id_list = []
        if _id == "update":
            for ksus in indata["ksus"]:
                content = ksus
                _id = content["_id"]
                _id_list.append(_id)
                content.pop("_id")
                op_id = self.edit_ksu(session, _id, content, kwargs)
        else:
            content = indata
            _id_list.append(_id)
            op_id = self.edit_ksu(session, _id, content, kwargs)

        data = {"ksus_list": _id_list, "operation_id": op_id}
        self._send_msg("edit", data)

    def cluster_list_ksu(self, session, filter_q=None, api_req=None):
        db_filter = {}
        if filter_q.get("cluster_id"):
            db_filter["_id"] = filter_q.get("cluster_id")
        ksu_data_list = []

        cluster_data = self.db.get_one("clusters", db_filter)
        profiles_list = [
            "infra_controller_profiles",
            "infra_config_profiles",
            "app_profiles",
            "resource_profiles",
        ]
        for profile in profiles_list:
            data_list = []
            for profile_id in cluster_data[profile]:
                filter_q = {"profile": {"_id": profile_id, "profile_type": profile}}
                data_list = self.db.get_list(self.topic, filter_q)
            for ksu_data in data_list:
                ksu_data["package_name"] = []
                ksu_data["package_path"] = []
                for okas in ksu_data["operationHistory"][0]["operationParams"]["oka"]:
                    sw_catalog_path = okas.get("sw_catalog_path")
                    if sw_catalog_path:
                        parts = sw_catalog_path.rsplit("/", 2)
                        self.logger.info(f"Parts: {parts}")
                        ksu_data["package_name"].append(parts[-2])
                        ksu_data["package_path"].append("/".join(parts[:-1]))
                    else:
                        oka_id = okas["_id"]
                        db_oka = self.db.get_one("okas", {"_id": oka_id})
                        oka_type = self.MAP_PROFILE[
                            db_oka.get("profile_type", "infra_controller_profiles")
                        ]
                        ksu_data["package_name"].append(db_oka["git_name"].lower())
                        ksu_data["package_path"].append(
                            f"{oka_type}/{db_oka['git_name'].lower()}"
                        )
                ksu_data_list.append(ksu_data)

        outdata = {}
        outdata["count"] = cluster_data["ksu_count"]
        outdata["data"] = ksu_data_list
        self.logger.info(f"Outdata: {outdata}")
        return outdata

    def edit_ksu(self, session, _id, indata, kwargs):
        check_dict = {
            "ksu": _id,
        }
        if indata.get("profile"):
            check_dict[indata["profile"]["profile_type"]] = indata["profile"]["_id"]
        if indata.get("oka"):
            check_dict["okas"] = []
            for oka in indata["oka"]:
                if oka.get("_id") is not None:
                    check_dict["okas"].append(oka["_id"])

        self.check_dependency(check_dict)
        content = None
        indata = self._remove_envelop(indata)

        # Override descriptor with query string kwargs
        if kwargs:
            self._update_input_with_kwargs(indata, kwargs)
        try:
            if indata and session.get("set_project"):
                raise EngineException(
                    "Cannot edit content and set to project (query string SET_PROJECT) at same time",
                    HTTPStatus.UNPROCESSABLE_ENTITY,
                )
            # TODO self._check_edition(session, indata, _id, force)
            if not content:
                content = self.show(session, _id)

            for okas in indata["oka"]:
                if not okas["_id"]:
                    okas.pop("_id")
                if not okas["sw_catalog_path"]:
                    okas.pop("sw_catalog_path")

            indata = self._validate_input_edit(indata, content, force=session["force"])

            # To allow project addressing by name AS WELL AS _id. Get the _id, just in case the provided one is a name
            _id = content.get("_id") or _id

            content = self.check_conflict_on_edit(session, content, indata, _id=_id)
            op_id = self.format_on_edit(content, indata)
            self.db.replace(self.topic, _id, content)
            return op_id
        except ValidationError as e:
            raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)

    def delete_ksu(self, session, _id, indata, dry_run=False, not_send_msg=None):
        _id_list = []
        if _id == "delete":
            for ksus in indata["ksus"]:
                content = ksus
                _id = content["_id"]
                content.pop("_id")
                op_id, not_send_msg_ksu = self.delete(session, _id)
                if not not_send_msg_ksu:
                    _id_list.append(_id)
        else:
            op_id, not_send_msg_ksu = self.delete(session, _id)
            if not not_send_msg_ksu:
                _id_list.append(_id)

        if _id_list:
            data = {
                "ksus_list": _id_list,
                "operation_id": op_id,
                "force": session["force"],
            }
            self._send_msg("delete", data, not_send_msg)
        return op_id

    def delete(self, session, _id):
        if not self.multiproject:
            filter_q = {}
        else:
            filter_q = self._get_project_filter(session)
        filter_q[self.id_field(self.topic, _id)] = _id
        item_content = self.db.get_one(self.topic, filter_q)

        check = {
            "ksu": _id,
            item_content["profile"]["profile_type"]: item_content["profile"]["_id"],
        }
        self.check_dependency(check, operation_type="delete")

        item_content["state"] = "IN_DELETION"
        item_content["operatingState"] = "PROCESSING"
        item_content["resourceState"] = "IN_PROGRESS"
        op_id = self.format_on_operation(
            item_content,
            "delete",
            None,
        )
        self.db.set_one(self.topic, {"_id": item_content["_id"]}, item_content)

        # Check if the profile exists. If it doesn't, no message should be sent to Kafka
        not_send_msg = None
        profile_id = item_content["profile"]["_id"]
        profile_type = item_content["profile"]["profile_type"]
        profile_collection_map = {
            "app_profiles": "k8sapp",
            "resource_profiles": "k8sresource",
            "infra_controller_profiles": "k8sinfra_controller",
            "infra_config_profiles": "k8sinfra_config",
        }
        profile_collection = profile_collection_map[profile_type]
        profile_content = self.db.get_one(
            profile_collection, {"_id": profile_id}, fail_on_empty=False
        )
        if not profile_content:
            self.db.del_one(self.topic, filter_q)
            not_send_msg = True
        return op_id, not_send_msg


class AppInstanceTopic(ACMTopic):
    topic = "appinstances"
    okapkg_topic = "okas"
    topic_msg = "appinstance"
    schema_new = app_instance_schema
    schema_edit = app_instance_edit_schema

    def __init__(self, db, fs, msg, auth):
        super().__init__(db, fs, msg, auth)
        self.logger = logging.getLogger("nbi.appinstances")

    @staticmethod
    def format_on_new(content, project_id=None, make_public=False):
        BaseTopic.format_on_new(content, project_id=project_id, make_public=make_public)
        content["current_operation"] = None
        content["state"] = "IN_CREATION"
        content["operatingState"] = "PROCESSING"
        content["resourceState"] = "IN_PROGRESS"

    def new(self, rollback, session, indata=None, kwargs=None, headers=None):
        if indata.get("oka") and indata.get("sw_catalog_path"):
            raise EngineException(
                "Cannot create app instance with both OKA and SW catalog path",
                HTTPStatus.UNPROCESSABLE_ENTITY,
            )

        # Override descriptor with query string kwargs
        content = self._remove_envelop(indata)
        self._update_input_with_kwargs(content, kwargs)
        content = self._validate_input_new(input=content, force=session["force"])

        # Check for unique name
        self.check_unique_name(session, content["name"])

        self.check_conflict_on_new(session, content)

        operation_params = {}
        for content_key, content_value in content.items():
            operation_params[content_key] = content_value
        self.format_on_new(
            content, project_id=session["project_id"], make_public=session["public"]
        )
        op_id = self.format_on_operation(
            content,
            operation_type="create",
            operation_params=operation_params,
        )
        content["git_name"] = self.create_gitname(content, session)

        oka_id = content.get("oka")
        if oka_id:
            self.update_oka_usage_state(session, oka_id)

        _id = self.db.create(self.topic, content)
        rollback.append({"topic": self.topic, "_id": _id})
        self._send_msg("create", {"appinstance": _id, "operation_id": op_id})
        return _id, op_id

    def update_oka_usage_state(self, session, oka_id):
        filter_db = self._get_project_filter(session)
        filter_db[BaseTopic.id_field(self.topic, oka_id)] = oka_id

        data = self.db.get_one(self.okapkg_topic, filter_db)
        if data["_admin"]["usageState"] == "NOT_IN_USE":
            usage_state_update = {
                "_admin.usageState": "IN_USE",
            }
            self.db.set_one(
                self.okapkg_topic, {"_id": oka_id}, update_dict=usage_state_update
            )

    def check_conflict_on_edit(self, session, final_content, edit_content, _id):
        if final_content["name"] != edit_content["name"]:
            self.check_unique_name(session, edit_content["name"])
        return final_content

    @staticmethod
    def format_on_edit(final_content, edit_content):
        op_id = ACMTopic.format_on_operation(
            final_content,
            "update",
            edit_content,
        )
        final_content["operatingState"] = "PROCESSING"
        final_content["resourceState"] = "IN_PROGRESS"
        if final_content.get("_admin"):
            now = time()
            final_content["_admin"]["modified"] = now
        return op_id

    def edit(self, session, _id, indata, kwargs):
        content = None
        indata = self._remove_envelop(indata)

        # Override descriptor with query string kwargs
        if kwargs:
            self._update_input_with_kwargs(indata, kwargs)
        try:
            if indata and session.get("set_project"):
                raise EngineException(
                    "Cannot edit content and set to project (query string SET_PROJECT) at same time",
                    HTTPStatus.UNPROCESSABLE_ENTITY,
                )
            # TODO self._check_edition(session, indata, _id, force)
            if not content:
                content = self.show(session, _id)

            indata = self._validate_input_edit(indata, content, force=session["force"])

            # To allow project addressing by name AS WELL AS _id. Get the _id, just in case the provided one is a name
            _id = content.get("_id") or _id

            content = self.check_conflict_on_edit(session, content, indata, _id=_id)
            op_id = self.format_on_edit(content, indata)
            self.db.replace(self.topic, _id, content)
            return op_id
        except ValidationError as e:
            raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)

    def update_item(self, session, _id, item, indata):
        if not self.multiproject:
            filter_db = {}
        else:
            filter_db = self._get_project_filter(session)
        # To allow project&user addressing by name AS WELL AS _id
        filter_db[BaseTopic.id_field(self.topic, _id)] = _id
        self.logger.info(f"Item: {item}")
        self.logger.info(f"Indata before validation: {indata}")
        validate_input(indata, app_instance_update_schema)
        self.logger.info(f"Indata after validation: {indata}")
        data = self.db.get_one(self.topic, filter_db)
        operation_params = {}
        data["operatingState"] = "PROCESSING"
        data["resourceState"] = "IN_PROGRESS"
        operation_params = indata
        self.logger.info(f"Operation params: {operation_params}")
        op_id = self.format_on_operation(
            data,
            "update",
            operation_params,
        )
        self.db.set_one(self.topic, {"_id": _id}, data)
        self._send_msg("update", {"appinstance": _id, "operation_id": op_id})
        return op_id

    def delete(self, session, _id, not_send_msg=None):
        if not self.multiproject:
            filter_q = {}
        else:
            filter_q = self._get_project_filter(session)
        filter_q[self.id_field(self.topic, _id)] = _id
        item_content = self.db.get_one(self.topic, filter_q)
        item_content["state"] = "IN_DELETION"
        item_content["operatingState"] = "PROCESSING"
        item_content["resourceState"] = "IN_PROGRESS"
        op_id = self.format_on_operation(
            item_content,
            "delete",
            None,
        )
        self.db.set_one(self.topic, {"_id": item_content["_id"]}, item_content)

        # Check if the profile exists. If it doesn't, no message should be sent to Kafka
        not_send_msg2 = not_send_msg
        profile_id = item_content["profile"]
        profile_type = item_content["profile_type"]
        profile_collection_map = {
            "app_profiles": "k8sapp",
            "resource_profiles": "k8sresource",
            "infra_controller_profiles": "k8sinfra_controller",
            "infra_config_profiles": "k8sinfra_config",
        }
        profile_collection = profile_collection_map[profile_type]
        profile_content = self.db.get_one(
            profile_collection, {"_id": profile_id}, fail_on_empty=False
        )
        if not profile_content:
            self.db.del_one(self.topic, filter_q)
            not_send_msg2 = True
        self._send_msg(
            "delete",
            {"appinstance": _id, "operation_id": op_id, "force": session["force"]},
            not_send_msg=not_send_msg2,
        )
        return op_id


class OkaTopic(DescriptorTopic, ACMOperationTopic):
    topic = "okas"
    topic_msg = "oka"
    schema_new = oka_schema
    schema_edit = oka_schema

    def __init__(self, db, fs, msg, auth):
        super().__init__(db, fs, msg, auth)
        self.logger = logging.getLogger("nbi.oka")

    @staticmethod
    def format_on_new(content, project_id=None, make_public=False):
        DescriptorTopic.format_on_new(
            content, project_id=project_id, make_public=make_public
        )
        content["current_operation"] = None
        content["state"] = "PENDING_CONTENT"
        content["operatingState"] = "PROCESSING"
        content["resourceState"] = "IN_PROGRESS"

    def check_conflict_on_del(self, session, _id, db_content):
        usage_state = db_content["_admin"]["usageState"]
        if usage_state == "IN_USE":
            raise EngineException(
                "There is a KSU using this package",
                http_code=HTTPStatus.CONFLICT,
            )

    def check_conflict_on_edit(self, session, final_content, edit_content, _id):
        if "name" in edit_content:
            if final_content["name"] == edit_content["name"]:
                name = edit_content["name"]
                raise EngineException(
                    f"No update, new name for the OKA is the same: {name}",
                    http_code=HTTPStatus.CONFLICT,
                )
            else:
                self.check_unique_name(session, edit_content["name"])
        elif (
            "description" in edit_content
            and final_content["description"] == edit_content["description"]
        ):
            description = edit_content["description"]
            raise EngineException(
                f"No update, new description for the OKA is the same: {description}",
                http_code=HTTPStatus.CONFLICT,
            )
        return final_content

    def edit(self, session, _id, indata=None, kwargs=None, content=None):
        indata = self._remove_envelop(indata)

        # Override descriptor with query string kwargs
        if kwargs:
            self._update_input_with_kwargs(indata, kwargs)
        try:
            if indata and session.get("set_project"):
                raise EngineException(
                    "Cannot edit content and set to project (query string SET_PROJECT) at same time",
                    HTTPStatus.UNPROCESSABLE_ENTITY,
                )
            # TODO self._check_edition(session, indata, _id, force)
            if not content:
                content = self.show(session, _id)

            indata = self._validate_input_edit(indata, content, force=session["force"])

            # To allow project addressing by name AS WELL AS _id. Get the _id, just in case the provided one is a name
            _id = content.get("_id") or _id

            content = self.check_conflict_on_edit(session, content, indata, _id=_id)
            op_id = self.format_on_edit(content, indata)
            deep_update_rfc7396(content, indata)

            self.db.replace(self.topic, _id, content)
            return op_id
        except ValidationError as e:
            raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)

    def delete(self, session, _id, dry_run=False, not_send_msg=None):
        check = {"oka": _id}
        self.check_dependency(check, operation_type="delete")
        if not self.multiproject:
            filter_q = {}
        else:
            filter_q = self._get_project_filter(session)
        filter_q[self.id_field(self.topic, _id)] = _id
        item_content = self.db.get_one(self.topic, filter_q)
        item_content["state"] = "IN_DELETION"
        item_content["operatingState"] = "PROCESSING"
        self.check_conflict_on_del(session, _id, item_content)
        op_id = self.format_on_operation(
            item_content,
            "delete",
            None,
        )
        self.db.set_one(self.topic, {"_id": item_content["_id"]}, item_content)
        self._send_msg(
            "delete",
            {"oka_id": _id, "operation_id": op_id, "force": session["force"]},
            not_send_msg=not_send_msg,
        )
        return op_id

    def new(self, rollback, session, indata=None, kwargs=None, headers=None):
        # _remove_envelop
        if indata:
            if "userDefinedData" in indata:
                indata = indata["userDefinedData"]

        content = {"_admin": {"userDefinedData": indata, "revision": 0}}

        self._update_input_with_kwargs(content, kwargs)
        content = BaseTopic._validate_input_new(
            self, input=kwargs, force=session["force"]
        )

        self.check_unique_name(session, content["name"])
        operation_params = {}
        for content_key, content_value in content.items():
            operation_params[content_key] = content_value
        self.format_on_new(
            content, session["project_id"], make_public=session["public"]
        )
        op_id = self.format_on_operation(
            content,
            operation_type="create",
            operation_params=operation_params,
        )
        content["git_name"] = self.create_gitname(content, session)
        _id = self.db.create(self.topic, content)
        rollback.append({"topic": self.topic, "_id": _id})
        return _id, op_id

    def upload_content(self, session, _id, indata, kwargs, headers):
        if headers["Method"] in ("PUT", "PATCH"):
            check = {"oka": _id}
            self.check_dependency(check)
        current_desc = self.show(session, _id)

        compressed = None
        content_type = headers.get("Content-Type")
        if (
            content_type
            and "application/gzip" in content_type
            or "application/x-gzip" in content_type
        ):
            compressed = "gzip"
        if content_type and "application/zip" in content_type:
            compressed = "zip"
        filename = headers.get("Content-Filename")
        if not filename and compressed:
            filename = "package.tar.gz" if compressed == "gzip" else "package.zip"
        elif not filename:
            filename = "package"

        revision = 1
        if "revision" in current_desc["_admin"]:
            revision = current_desc["_admin"]["revision"] + 1

        file_pkg = None
        fs_rollback = []

        try:
            start = 0
            # Rather than using a temp folder, we will store the package in a folder based on
            # the current revision.
            proposed_revision_path = _id + ":" + str(revision)
            # all the content is upload here and if ok, it is rename from id_ to is folder

            if start:
                if not self.fs.file_exists(proposed_revision_path, "dir"):
                    raise EngineException(
                        "invalid Transaction-Id header", HTTPStatus.NOT_FOUND
                    )
            else:
                self.fs.file_delete(proposed_revision_path, ignore_non_exist=True)
                self.fs.mkdir(proposed_revision_path)
                fs_rollback.append(proposed_revision_path)

            storage = self.fs.get_params()
            storage["folder"] = proposed_revision_path
            storage["zipfile"] = filename

            file_path = (proposed_revision_path, filename)
            file_pkg = self.fs.file_open(file_path, "a+b")

            if isinstance(indata, dict):
                indata_text = yaml.safe_dump(indata, indent=4, default_flow_style=False)
                file_pkg.write(indata_text.encode(encoding="utf-8"))
            else:
                indata_len = 0
                indata = indata.file
                while True:
                    indata_text = indata.read(4096)
                    indata_len += len(indata_text)
                    if not indata_text:
                        break
                    file_pkg.write(indata_text)

            # Need to close the file package here so it can be copied from the
            # revision to the current, unrevisioned record
            if file_pkg:
                file_pkg.close()
            file_pkg = None

            # Fetch both the incoming, proposed revision and the original revision so we
            # can call a validate method to compare them
            current_revision_path = _id + "/"
            self.fs.sync(from_path=current_revision_path)
            self.fs.sync(from_path=proposed_revision_path)

            # Is this required?
            if revision > 1:
                try:
                    self._validate_descriptor_changes(
                        _id,
                        filename,
                        current_revision_path,
                        proposed_revision_path,
                    )
                except Exception as e:
                    shutil.rmtree(
                        self.fs.path + current_revision_path, ignore_errors=True
                    )
                    shutil.rmtree(
                        self.fs.path + proposed_revision_path, ignore_errors=True
                    )
                    # Only delete the new revision.  We need to keep the original version in place
                    # as it has not been changed.
                    self.fs.file_delete(proposed_revision_path, ignore_non_exist=True)
                    raise e

            indata = self._remove_envelop(indata)

            # Override descriptor with query string kwargs
            if kwargs:
                self._update_input_with_kwargs(indata, kwargs)

            current_desc["_admin"]["storage"] = storage
            current_desc["_admin"]["onboardingState"] = "ONBOARDED"
            current_desc["_admin"]["operationalState"] = "ENABLED"
            current_desc["_admin"]["modified"] = time()
            current_desc["_admin"]["revision"] = revision

            deep_update_rfc7396(current_desc, indata)

            # Copy the revision to the active package name by its original id
            shutil.rmtree(self.fs.path + current_revision_path, ignore_errors=True)
            os.rename(
                self.fs.path + proposed_revision_path,
                self.fs.path + current_revision_path,
            )
            self.fs.file_delete(current_revision_path, ignore_non_exist=True)
            self.fs.mkdir(current_revision_path)
            self.fs.reverse_sync(from_path=current_revision_path)

            shutil.rmtree(self.fs.path + _id)
            kwargs = {}
            kwargs["package"] = filename
            if headers["Method"] == "POST":
                current_desc["state"] = "IN_CREATION"
                op_id = current_desc.get("operationHistory", [{"op_id": None}])[-1].get(
                    "op_id"
                )
            elif headers["Method"] in ("PUT", "PATCH"):
                op_id = self.format_on_operation(
                    current_desc,
                    "update",
                    kwargs,
                )
                current_desc["operatingState"] = "PROCESSING"
                current_desc["resourceState"] = "IN_PROGRESS"

            self.db.replace(self.topic, _id, current_desc)

            #  Store a copy of the package as a point in time revision
            revision_desc = dict(current_desc)
            revision_desc["_id"] = _id + ":" + str(revision_desc["_admin"]["revision"])
            self.db.create(self.topic + "_revisions", revision_desc)
            fs_rollback = []

            if headers["Method"] == "POST":
                self._send_msg("create", {"oka_id": _id, "operation_id": op_id})
            elif headers["Method"] == "PUT" or "PATCH":
                self._send_msg("edit", {"oka_id": _id, "operation_id": op_id})

            return True

        except EngineException:
            raise
        finally:
            if file_pkg:
                file_pkg.close()
            for file in fs_rollback:
                self.fs.file_delete(file, ignore_non_exist=True)