211 lines
7.1 KiB
Python
211 lines
7.1 KiB
Python
"""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-<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)
|
|
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.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"
|