Newer
Older
"""Everything to do with Apps"""
import os
from sqlalchemy import ForeignKey, Integer, String, Boolean
from sqlalchemy.orm import relationship
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 __init__(self, slug, name, external=False, url=None):
self.slug = slug
self.name = name
self.external = external
self.url = url
def __repr__(self):
return f"{self.id} <{self.name}>"
"""
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}"
"""Returns an AppStatus object that describes the current cluster state"""
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)
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(),
@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")
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
if app.external:
self.installed = True
self.ready = True
self.message = "App is external"
return
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"
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,
}
class OAuthClientApp(db.Model): # pylint: disable=too-few-public-methods
"""
The OAuthClientApp object maps an OAuth client to the corresponding App.
This mapping exists so that
* you can have a different name for the OAuth client than for the app, and
* you can have multiple OAuth clients that belong to the same app.
"""
__tablename__ = "oauthclient_app"
oauthclient_id = db.Column(String(length=64), primary_key=True)
app_id = db.Column(Integer, ForeignKey("app.id"))
app = relationship("App")
def __repr__(self):
return (f"oauthclient_id: {self.oauthclient_id}, app_id: {self.app_id},"
f" app: {self.app}")