"""Everything to do with Apps""" import os from sqlalchemy import ForeignKey, Integer, String from sqlalchemy.orm import relationship from database import db import helpers.kubernetes as k8s class App(db.Model): """ The App object, interact with the App database object. Data is stored in the local database. """ id = db.Column(Integer, primary_key=True) name = db.Column(String(length=64)) slug = db.Column(String(length=64), unique=True) def __repr__(self): return f"{self.id} <{self.name}>" def get_status(self): """Returns an AppStatus object that describes the current cluster state""" return AppStatus(self.kustomization, self.helmreleases) def install(self): """Creates a Kustomization in the Kubernetes cluster that installs this application""" # Generate the necessary passwords, etc. from a template self.__generate_secrets() # Create add- kustomization self.__create_kustomization() def uninstall(self): """ Delete the app kustomization. In our case, this triggers a deletion of the app's PVCs (so deletes all data), as well as any other Kustomizations and HelmReleases related to the app. It also triggers a deletion of the OAuth2Client object, but does not delete the secrets generated by the `install` command. It also does not remove the TLS secret generated by cert-manager. """ self.__delete_kustomization() def delete(self): """ Fully deletes an application This includes user roles, all kubernetes objects and also PVCs, so your data will be *gone* """ # Delete all roles first for role in self.roles: db.session.delete(role) db.session.commit() db.session.delete(self) return db.session.commit() def __generate_secrets(self): """Generates passwords for app installation""" # Create app variables secret if self.variables_template_filepath: k8s.create_variables_secret(self.slug, self.variables_template_filepath) k8s.create_variables_secret( self.slug, os.path.join( self.__get_templates_dir(), "stackspin-oauth-variables.yaml.jinja" ) ) def __create_kustomization(self): """Creates the `add-{app_slug}` kustomization in the Kubernetes cluster""" kustomization_template_filepath = \ os.path.join(self.__get_templates_dir(), "add-app-kustomization.yaml.jinja") k8s.store_kustomization(kustomization_template_filepath, self.slug) def __delete_kustomization(self): """Deletes kustomization for this app""" k8s.delete_kustomization(f"add-{self.slug}") @property def variables_template_filepath(self): """Path to the variables template used to generate secrets the app needs""" variables_template_filepath = os.path.join(self.__get_templates_dir(), f"stackspin-{self.slug}-variables.yaml.jinja") if os.path.exists(variables_template_filepath): return variables_template_filepath return None @property def namespace(self): """ Returns the Kubernetes namespace of this app FIXME: This should probably become a database field. """ if self.slug in ['nextcloud', 'wordpress', 'wekan', 'zulip']: return 'stackspin-apps' return 'stackspin' @property def roles(self): """ All roles that are linked to this app """ return AppRole.query.filter_by( app_id=self.id ).all() @property def kustomization(self): """Returns the kustomization object for this app""" return k8s.get_kustomization(self.slug) @property def helmreleases(self): """Returns the helmreleases associated with the kustomization for this app""" return k8s.get_all_helmreleases(self.namespace, f"kustomize.toolkit.fluxcd.io/name={self.slug}") @staticmethod def __get_templates_dir(): """Returns directory that contains the Jinja templates used to create app secrets.""" return os.path.join(os.path.dirname(os.path.realpath(__file__)), "templates") class AppRole(db.Model): # pylint: disable=too-few-public-methods """ The AppRole object, stores the roles Users have on Apps """ user_id = db.Column(String(length=64), primary_key=True) app_id = db.Column(Integer, ForeignKey("app.id"), primary_key=True) role_id = db.Column(Integer, ForeignKey("role.id")) role = relationship("Role") def __repr__(self): return (f"role_id: {self.role_id}, user_id: {self.user_id}," f" app_id: {self.app_id}, role: {self.role}") class AppStatus(): # pylint: disable=too-few-public-methods """ Represents the status of an app in the Kubernetes cluster. This class can answer a few questions, like "is the app installed?", but can also return raw status messages from Kustomizations and HelmReleases This constructor sets three variables: self.installed (bool): Whether the app should be installed self.ready (bool): Whether the app is installed correctly self.message (str): Information about the status :param kustomization_status: The status of the Kustomization of this app: :type kustomization_status: str :param helmrelease_status: The status of the helmreleases of this app :type helmrelease_status: str[] """ def __init__(self, kustomization, helmreleases): self.helmreleases = {} if kustomization is not None and "status" in kustomization: ks_ready, ks_message = AppStatus.check_condition(kustomization['status']) self.installed = True else: ks_ready = None ks_message = "Kustomization does not exist" self.installed = False self.ready = False self.message = "Not installed" return for helmrelease in helmreleases: hr_status = helmrelease['status'] hr_ready, hr_message = AppStatus.check_condition(hr_status) # For now, only show the message of the first HR that isn't ready if not hr_ready: self.ready = False self.message = f"HelmRelease {helmrelease['metadata']['name']} status: {hr_message}" return # If we end up here, all HRs are ready if ks_ready: self.ready = True self.message = "Installed" else: self.ready = False self.message = f"App Kustomization status: {ks_message}" def __repr__(self): return f"Installed: {self.installed}\tReady: {self.ready}\tMessage: {self.message}" @staticmethod def check_condition(status): """ Returns a tuple that has true/false for readiness and a message Ready, in this case means that the condition's type == "Ready" and its status == "True". If the condition type "Ready" does not occur, the status is interpreted as not ready. The message that is returned is the message that comes with the condition with type "Ready" :param status: Kubernetes resource's "status" object. :type status: dict """ for condition in status["conditions"]: if condition["type"] == "Ready": return condition["status"] == "True", condition["message"] return False, "Condition with type 'Ready' not found"