"""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 from .apps import APP_NOT_INSTALLED_STATUS 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 a string that describes the app state in the cluster""" kustomization = self.kustomization if kustomization is not None and "status" in kustomization: ks_ready, ks_message = App.check_condition(kustomization['status']) else: ks_ready = None for helmrelease in self.helmreleases['items']: hr_status = helmrelease['status'] hr_ready, hr_message = App.check_condition(hr_status) # For now, only show the message of the first HR that isn't ready if not hr_ready: break if ks_ready is None: return APP_NOT_INSTALLED_STATUS # *Should* not happen, but just in case: if (ks_ready is None and hr_ready is not None) or \ (hr_ready is None and ks_ready is not None): return ("This app is in a strange state. Contact a Stackspin" " administrator if this status stays for longer than 5 minutes") if ks_ready and hr_ready: return "App installed and running" if not hr_ready: return f"App HelmRelease status: {hr_message}" if not ks_ready: return f"App Kustomization status: {ks_message}" return "App is installing..." 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. 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 """ 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: role.delete() 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) 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.list_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") @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" 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}")