298 lines
10 KiB
Python
298 lines
10 KiB
Python
"""Everything to do with Apps"""
|
|
|
|
import os
|
|
import base64
|
|
|
|
from sqlalchemy import ForeignKey, Integer, String, Boolean
|
|
from sqlalchemy.orm import relationship
|
|
|
|
from database import db
|
|
import helpers.kubernetes as k8s
|
|
|
|
|
|
DEFAULT_APP_SUBDOMAINS = {
|
|
"nextcloud": "files",
|
|
"wordpress": "www",
|
|
"monitoring": "grafana",
|
|
}
|
|
|
|
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, server_default='0')
|
|
# The URL is only stored in the DB for external applications; otherwise the
|
|
# URL is stored in a configmap (see get_url)
|
|
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
|
|
|
|
For external applications: the URL is stored in the database
|
|
|
|
For internal applications: the URL is stored in a configmap named
|
|
`stackspin-{self.slug}-kustomization-variables under
|
|
`{self.slug_domain}`. This function reads that configmap. If the
|
|
configmap does not contain a URL for the application (which is
|
|
possible, if the app is not installed yet, for example), we return a
|
|
default URL.
|
|
"""
|
|
|
|
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 config map found with this domain name for this service, return
|
|
# that URL
|
|
if ks_config_map and domain_key in ks_config_map.keys():
|
|
return f"https://{ks_config_map[domain_key]}"
|
|
|
|
domain_secret = k8s.get_kubernetes_secret_data(
|
|
"stackspin-cluster-variables",
|
|
"flux-system")
|
|
domain = base64.b64decode(domain_secret['domain']).decode()
|
|
|
|
# See if there is another default subdomain for this app than just
|
|
# "slug.{domain}"
|
|
if self.slug in DEFAULT_APP_SUBDOMAINS:
|
|
return f"https://{DEFAULT_APP_SUBDOMAINS[self.slug]}.{domain}"
|
|
|
|
# No default known
|
|
return f"https://{self.slug}.{domain}"
|
|
|
|
def get_status(self):
|
|
"""Returns an AppStatus object that describes the current cluster state"""
|
|
return AppStatus(self)
|
|
|
|
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.
|
|
|
|
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)
|
|
|
|
def to_dict(self):
|
|
"""
|
|
represent this object as a dict, compatible for JSON output
|
|
"""
|
|
|
|
return {"id": self.id,
|
|
"name": self.name,
|
|
"slug": self.slug,
|
|
"external": self.external,
|
|
"status": self.get_status().to_dict(),
|
|
"url": self.get_url()}
|
|
|
|
@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 app: An app of which the kustomization and helmreleases
|
|
property will be used.
|
|
:type app: App
|
|
"""
|
|
def __init__(self, app):
|
|
if app.external:
|
|
self.installed = True
|
|
self.ready = True
|
|
self.message = "App is external"
|
|
return
|
|
|
|
kustomization = app.kustomization
|
|
if kustomization is not None and "status" in kustomization:
|
|
ks_ready, ks_message = AppStatus.check_condition(kustomization['status'])
|
|
self.installed = True
|
|
if ks_ready:
|
|
self.ready = ks_ready
|
|
self.message = "Installed"
|
|
return
|
|
else:
|
|
ks_ready = None
|
|
ks_message = "Kustomization does not exist"
|
|
self.installed = False
|
|
self.ready = False
|
|
self.message = "Not installed"
|
|
return
|
|
|
|
helmreleases = app.helmreleases
|
|
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, but the kustomization is not
|
|
self.ready = ks_ready
|
|
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"
|
|
|
|
def to_dict(self):
|
|
"""Represents this app status as a dict"""
|
|
return {
|
|
"installed": self.installed,
|
|
"ready": self.ready,
|
|
"message": self.message,
|
|
}
|