dashboard/areas/apps/models.py

"""Everything to do with Apps"""

import os

from sqlalchemy import ForeignKey, Integer, String, Boolean
from sqlalchemy.orm import relationship
from database import db
import helpers.kubernetes as k8s

# Circular import, need fixing
#from .apps import APP_NOT_INSTALLED_STATUS

APP_NOT_INSTALLED_STATUS = "Not installed"

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)
    external = db.Column(Boolean, unique=False, nullable=False, default=True, server_default='0')
    url = db.Column(String(length=128), unique=False)

    def __repr__(self):
        return f"{self.id} <{self.name}>"

    def get_url(self):
        """Returns the URL where this application is running"""

        if self.external:
            return self.url

        # Get domain name from configmap
        ks_config_map = k8s.get_kubernetes_config_map_data(
                f"stackspin-{self.slug}-kustomization-variables",
                "flux-system")
        domain_key = f"{self.slug}_domain"
        if ks_config_map is None or domain_key not in ks_config_map.keys():
            return None

        return ks_config_map[domain_key]


    def get_status(self):
        """Returns a string that describes the app state in the cluster"""

        if self.external:
            return "External app"


        # TODO: Get some kind of caching for those values, as this is called
        #       on every app list, causing significant delays in the interface

        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-<app> 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:
            db.session.delete(role)
            #role.delete()
        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)

    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}")