256 lines
8.9 KiB
Python
256 lines
8.9 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
|
|
|
|
from flask import current_app
|
|
|
|
from constants import APP_NOT_INSTALLED_STATUS
|
|
|
|
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, default=True, 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"
|
|
# No config map found, or configmap not configured to contain the
|
|
# domain (yet). Return the default for this app
|
|
if ks_config_map is None or domain_key not in ks_config_map.keys():
|
|
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}"
|
|
return f"https://{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)
|
|
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"
|
|
|
|
def to_json(self):
|
|
"""
|
|
represent this object as a json object. Return JSON object
|
|
"""
|
|
|
|
return {"id": self.id,
|
|
"name": self.name,
|
|
"slug": self.slug,
|
|
"external": self.external,
|
|
"url": self.get_url(),
|
|
"status": self.get_status()}
|
|
|
|
|
|
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}")
|