""" App model with Mixin pattern for better organization. All code remains in a single file, but functionality is organized into logical mixin classes: - LifecycleMixin: create, delete, restart, save - DeployMixin: deploy, scale, mount, pipeline, clean - KubernetesMixin: pod/deployment/event operations - HealthMixin: HTTP/TCP health verification - ConfigMixin: configuration and environment management - AutoscaleMixin: HPA autoscaling - UsageMixin: usage reporting - StateMixin: state synchronization """ import base64 import functools import json import logging import random import re import requests import string import socket import time from collections import namedtuple, OrderedDict from contextlib import closing from urllib.parse import urljoin from datetime import datetime, timezone from docker import auth as docker_auth from django.conf import settings from django.db import models from django.template import Template, Context from django.db.models import F, Func, Value, JSONField from django.contrib.auth import get_user_model from rest_framework.exceptions import ValidationError from api.tasks import send_app_log from api.utils import get_httpclient, dict_diff, get_next_uid from api.exceptions import AlreadyExists, DryccException, ServiceUnavailable from api.utils import ( CacheLock, DeployLock, generate_app_name, apply_tasks, validate_reserved_names, ) from scheduler import KubeHTTPException, KubeException from .gateway import Gateway, Route from .limit import LimitPlan from .config import Config from .service import Service from .release import Release from .tls import TLS from .appsettings import AppSettings from .volume import Volume from .base import UuidAuditedModel, PTYPE_WEB, PTYPE_RUN, DEFAULT_HTTP_PORT User = get_user_model() logger = logging.getLogger(__name__) # http://kubernetes.io/v1.1/docs/design/identifiers.html def validate_app_id(value): """Check that the value follows the kubernetes name constraints.""" match = re.match(r"^[a-z]([a-z0-9-]{3,}[a-z0-9])$", value) if not match: raise ValidationError( "App name must start with an alphabetic character, " "cannot end with a hyphen and can only contain " "a-z (lowercase), 0-9 and hyphens." ) validate_reserved_names(value) def validate_app_structure(value): """Error if the dict values aren't ints >= 0.""" try: for k, v in value.items(): if int(v) < 0: raise ValueError("Must be greater than or equal to zero") validate_reserved_names(k) except ValueError as err: raise ValidationError(str(err)) class AppBaseMixin: """Base mixin providing common utilities for all App mixins. All other App mixins should inherit from this class to access: - lock: distributed app lock - log: application logging - ptypes: process type property - check_ptypes: validate process types - get_deploy_name: K8s deployment name generation - get_system_actor: system user for automated operations """ @property def ptypes(self): """List of process types from app structure.""" return list(self.structure.keys()) def log(self, message, level=logging.INFO): """Log a message in the context of this application.""" send_app_log.delay(self.id, message, level) logger.log(level, "[{}]: {}".format(self.id, message)) def lock(self): """Acquire distributed lock for this app.""" return CacheLock(f"app:lock:{self.id}") def check_ptypes(self, ptypes: set): """Check available procfile types.""" if not ptypes: ptypes = self.ptypes else: invalid_ptypes = ptypes.difference(self.ptypes) if len(invalid_ptypes) != 0: raise DryccException( "process type {} is not included in procfile".format(",".join(invalid_ptypes)) ) return ptypes def get_deploy_name(self, ptype): """Get Kubernetes deployment name for a process type.""" return f"{self.id}-{ptype}" def get_system_actor(self): """Get system actor for automated operations (not a real user).""" return namedtuple("SystemActor", ["username"])("system") class LifecycleMixin(AppBaseMixin): """Mixin for App lifecycle operations: create, restart.""" def create(self, *args, **kwargs): """ Create application with initial config, settings, release, domain and k8s resources. """ cfg = self._set_default_config() try: self.release_set.latest() except Release.DoesNotExist: Release.objects.create(version=1, app=self, config=cfg, build=None) namespace = self.id self.log("creating Namespace {} and services".format(namespace), level=logging.DEBUG) try: self.scheduler.ns.get(namespace) except KubeException: try: self.scheduler.ns.create(namespace) except KubeException as e: raise ServiceUnavailable("Could not create the Namespace in Kubernetes") from e self._create_network_policy() try: self.appsettings_set.latest() except AppSettings.DoesNotExist: AppSettings.objects.create(app=self, routable=True, autodeploy=True, autorollback=True) try: self.tls_set.latest() except TLS.DoesNotExist: TLS.objects.create(app=self) def restart(self, **kwargs): """Restart deployments with the kubectl rollout api.""" deployments = [] if self.structure[kwargs["type"]] > 0: deployments.append(self.get_deploy_name(kwargs["type"])) try: tasks = [ ( functools.partial(self.scheduler.deployment.restart, self.id, deployment), lambda future: self.log( f'restart {kwargs["type"]} callback: {future.result()}' ), ) for deployment in deployments ] apply_tasks(tasks) except Exception as e: err = "warning, some pods failed to restart:\n{}".format(str(e)) self.log(err, logging.WARNING) def _create_network_policy(self): """Create network policy for the app namespace.""" if not settings.DRYCC_NETWORK_POLICY_TEMPLATE: return name = namespace = self.id json_str = ( Template(settings.DRYCC_NETWORK_POLICY_TEMPLATE) .render(Context({"workspace": str(self.workspace.id)})) .strip() ) kwargs = json.loads(json_str) if json_str else {} try: self.scheduler.networkpolicy.get(namespace, name) self.scheduler.networkpolicy.patch(namespace, name, **kwargs) except KubeHTTPException as e: if e.response.status_code == 404: try: self.scheduler.networkpolicy.create(namespace, name, **kwargs) except KubeException as ex: raise ServiceUnavailable( "Could not create the NetworkPolicy in Kubernetes" ) from ex else: raise ServiceUnavailable("Could not get the NetworkPolicy in Kubernetes") from e def _set_default_config(self): """Set default configuration for the app.""" plan = LimitPlan.get_default() limits = {PTYPE_WEB: plan.id, PTYPE_RUN: plan.id} try: config = self.config_set.latest() limits[PTYPE_WEB] = config.limits.get(PTYPE_WEB, plan.id) limits[PTYPE_RUN] = config.limits.get(PTYPE_RUN, plan.id) except Config.DoesNotExist: config = Config.objects.create(app=self, limits=limits) for ptype in self.ptypes: limits[ptype] = config.limits.get(ptype, plan.id) if limits != config.limits: config.limits = limits config.save(update_fields=["limits"]) return config class DeployMixin(AppBaseMixin): """Mixin for deploy operations: deploy, scale, mount, pipeline, clean.""" def deploy(self, release, ptypes=None, force_deploy=False, rollback_on_failure=True): """Deploy a new release to this application.""" if release is None or release.build is None: raise DryccException("No build associated with this release") self.create() prev_release = release.previous() self._merge_structure(release, prev_release) app_settings = self.appsettings_set.latest() volumes = self.volume_set.all() deploys = {} for scale_type, replicas in self.structure.items(): if ptypes is not None and scale_type not in ptypes: continue scale_type_volumes = [_ for _ in volumes if scale_type in _.path.keys()] deploys[scale_type] = self.gather_app_settings( release, app_settings, scale_type, replicas, volumes=scale_type_volumes ) self._deploy(deploys, ptypes, prev_release, release, force_deploy, rollback_on_failure) if app_settings.autodeploy: self.clean(release=release) release.clean(ptypes) def scale(self, user, structure): """Scale containers up or down.""" err_msg = None release = Release.latest(self) if PTYPE_RUN in structure or release is None or release.build is None: if PTYPE_RUN in structure: err_msg = "Cannot set scale for reserved types, procfile type is: run" else: err_msg = "No build associated with this release" self.log(err_msg, logging.WARNING) raise DryccException(err_msg) app_settings = self.appsettings_set.latest() self._scale(user, structure, release, app_settings) def mount(self, user, volume, structure=None): """Mount volumes to app.""" release = Release.latest(self) if release is None or release.build is None: raise DryccException("No build associated with this release") app_settings = self.appsettings_set.latest() self._mount(user, volume, app_settings, structure=structure) def pipeline(self, release, ptypes, force_deploy=False): """Execute pipeline for a release.""" prefix = f"[pipeline] release {release.version_name}" try: if release is not None and release.build is not None: if release.build.dryccfile: for run in release.get_runners(ptypes): self.log(f"{prefix} starts running pipeline.run {run['image']}") job_name = self.run( self.get_system_actor(), run["image"], command=run["command"], args=run["args"], timeout=run["timeout"], expires=run["timeout"], envs=self.build_env_vars(release, run["ptype"]), ) state, labels = "initializing", {"job-name": job_name} for count, state in enumerate( self.scheduler.pod.watch( self.id, labels, settings.DRYCC_PILELINE_RUN_TIMEOUT ) ): self.log(f"{prefix} waiting for pipeline.run: {state} * {count}") if state != "down": raise DryccException(f"pipeline run state error: {state}") self.log(f"{prefix} starts running...") rollback_on_failure = self.appsettings_set.latest().autorollback if not rollback_on_failure: self.log(f"{prefix} deploy do not rollback on failure") self.deploy(release, ptypes, force_deploy, rollback_on_failure) release.state, release.failed = "succeed", False ptypes = list(ptypes) if ptypes is not None else ptypes release.add_condition(state="succeed", action="pipeline", ptypes=ptypes) except Exception as e: release.failed, release.state, msg = True, "crashed", f"{type(e).__name__}: {e}" ptypes = list(ptypes) if ptypes is not None else ptypes release.add_condition(state="crashed", action="pipeline", ptypes=ptypes, exception=msg) self.log(f"{prefix} pipeline runtime error, {msg}", logging.ERROR) finally: DeployLock(self.pk).release(ptypes) release.save(update_fields=["state", "failed"]) self.log(f"{prefix} run completed...") def clean(self, release=None, ptypes=None): """Clean old k8s resources.""" release = release if release else self.release_set.latest() pre_release = release.previous(None) if pre_release is not None: removed = {} for ptype in pre_release.ptypes: if ptype not in release.ptypes and self.structure.get(ptype, 0) > 0: if ptypes is None or ptype in ptypes: removed[ptype] = 0 self.scale(self.get_system_actor(), removed) self._merge_structure(release, pre_release) labels = {"heritage": "drycc"} if ptypes: labels["type__in"] = ptypes resource_apis = [self.scheduler.deployments, self.scheduler.secret] for api in resource_apis: resources = api.get(self.id, labels=labels).json()["items"] if resources is not None: for resource in resources: name = resource["metadata"]["name"] ptype = resource["metadata"].get("labels", {}).get("type") if ptype and ptype not in self.structure: api.delete(self.id, name, ignore_exception=True) self.log(f"cleanup old kubernetes resources for {self.id}") def _deploy(self, deploys, ptypes, prev_release, release, force_deploy, rollback_on_failure): """Internal deploy implementation.""" deploys = OrderedDict(sorted(deploys.items(), key=lambda d: d[1].get("routable"))) self._check_deployment_in_progress(deploys, force_deploy) try: tasks = [] lock = DeployLock(self.pk) for scale_type, kwargs in deploys.items(): self.set_application_config(release, scale_type) tasks.append( ( functools.partial( self.scheduler.deploy, namespace=self.id, name=self.get_deploy_name(scale_type), image=release.get_deploy_image(scale_type), command=release.get_deploy_command(scale_type), args=release.get_deploy_args(scale_type), **kwargs, ), lambda future: self.log( "deploy and unlock callback: " f"{[future.result(), lock.release([scale_type])]}" ), ) ) try: apply_tasks(tasks) except KubeException as e: if ( rollback_on_failure and prev_release is not None and prev_release.build is not None ): err = "There was a problem deploying {}. Rolling back to release {}.".format( release.version_name, prev_release.version_name ) self.log(err, logging.ERROR) self.deploy(prev_release, ptypes, True, False) raise DryccException("{}\n{}".format(err, str(e))) from e raise except Exception as e: err = "(app::deploy): {}".format(e) self.log(err, logging.ERROR) raise ServiceUnavailable(err) from e for ptype in deploys.keys(): if ptype == PTYPE_WEB: target_port = release.get_port(ptype) self._create_default_ingress(target_port) service = self.service_set.filter(ptype=ptype).first() if not service: continue if prev_release and prev_release.build: continue if ptype == PTYPE_WEB: self.verify_http_health(service, **deploys[ptype]) else: self.verify_tcp_health(service, **deploys[ptype]) def _scale(self, user, structure, release, app_settings): """Internal scale implementation.""" self.create() try: for target, count in structure.copy().items(): structure[target] = int(count) validate_app_structure(structure) except (TypeError, ValueError, ValidationError) as e: err_msg = "Invalid scaling format: {}".format(e) self.log(err_msg) raise DryccException(err_msg) new_scale = dict_diff(structure, self.structure).get("changed", {}) old_scale = dict_diff(self.structure, structure).get("changed", {}) if new_scale: try: self._scale_pods(new_scale, release, app_settings) except ServiceUnavailable: self._scale_pods(old_scale, release, app_settings) raise App.objects.filter(id=self.id).update( structure=Func( F("structure"), Value(new_scale, JSONField()), function="jsonb_concat" ) ) msg = "{} scaled pods ".format(user.username) + " ".join( "{}={}".format(k, v) for k, v in list(structure.items()) ) self.log(msg) return True return False def _scale_pods(self, scale_types, release, app_settings): """Scale pods for given process types.""" volumes = Volume.objects.filter(app=self).exclude(path={}) tasks = [] for scale_type, replicas in scale_types.items(): scale_type_volumes = [volume for volume in volumes if scale_type in volume.path.keys()] data = self.gather_app_settings( release, app_settings, scale_type, replicas, volumes=scale_type_volumes ) self.set_application_config(release, scale_type) tasks.append( ( functools.partial( self.scheduler.scale, namespace=self.id, name=self.get_deploy_name(scale_type), image=release.get_deploy_image(scale_type), command=release.get_deploy_command(scale_type), args=release.get_deploy_args(scale_type), **data, ), lambda future: self.log(f"scale {scale_type} callback: {future.result()}"), ) ) try: apply_tasks(tasks) except Exception as e: err = "(scale): {}".format(e) self.log(err, logging.ERROR) raise ServiceUnavailable(err) from e def _mount(self, user, volume, app_settings, structure=None): """Internal mount implementation.""" volumes = Volume.objects.filter(app=self) tasks = [] for scale_type, replicas in structure.items() if structure else self.structure.items(): if scale_type != PTYPE_RUN: release = self.release_set.filter( deployed_ptypes__contains=scale_type, failed=False ).latest() replicas = self.structure.get(scale_type, 0) scale_type_volumes = [ volume for volume in volumes if scale_type in volume.path.keys() ] data = self.gather_app_settings( release, app_settings, scale_type, replicas, volumes=scale_type_volumes ) deployment = self.scheduler.deployment.get( self.id, self.get_deploy_name(scale_type) ).json() spec_annotations = deployment["spec"]["template"]["metadata"].get( "annotations", {} ) self.set_application_config(release, scale_type) tasks.append( ( functools.partial( self.scheduler.deployment.patch, namespace=self.id, name=self.get_deploy_name(scale_type), image=release.get_deploy_image(scale_type), command=release.get_deploy_command(scale_type), args=release.get_deploy_args(scale_type), spec_annotations=spec_annotations, resource_version=deployment["metadata"]["resourceVersion"], **data, ), lambda future: self.log( f"mount {volume} for {scale_type} callback: {future.result()}" ), ) ) try: apply_tasks(tasks) except Exception as e: err = f"(changed volume mount for {volume}: {e}" self.log(err, logging.ERROR) raise ServiceUnavailable(err) from e self.log(f"{user.username} changed volume mount for {volume}") def _merge_structure(self, release, prev_release): """Scale to default structure based on release type.""" lock = self.lock() if not lock.acquire(): raise ServiceUnavailable(f"could not acquire app lock for {self.id}") try: self.refresh_from_db() default_structure = {} for ptype in release.ptypes: default_structure[ptype] = 1 if ptype == PTYPE_WEB else 0 if (self.structure != default_structure) or ( prev_release and prev_release.build and prev_release.build.type != release.build.type ): for ptype, scale in self.structure.items(): if ptype not in release.ptypes and scale == 0: continue default_structure[ptype] = scale self.structure = default_structure # Use QuerySet.update() instead of self.save() to avoid triggering # App.save() side effects (self.create() which provisions k8s resources). # We only need to persist the structure field to the database here. App.objects.filter(id=self.id).update(structure=default_structure) finally: lock.release() return self.structure def _check_deployment_in_progress(self, deploys, force_deploy=False): """Check if any deployment is in progress.""" if force_deploy: return for scale_type, kwargs in deploys.items(): name = self.get_deploy_name(scale_type) in_progress, deploy_okay = self.scheduler.deployment.in_progress( self.id, name, kwargs.get("deploy_timeout"), kwargs.get("deploy_batches"), kwargs.get("replicas"), kwargs.get("tags"), ) if in_progress and not deploy_okay: raise AlreadyExists("Deployment for {} is already in progress".format(name)) def _create_default_ingress(self, target_port): """Create default ingress resources (service, gateway, route).""" try: service = self.service_set.filter(ptype=PTYPE_WEB).latest() except Service.DoesNotExist: service = Service(app=self, ptype=PTYPE_WEB) service.add_port(DEFAULT_HTTP_PORT, "TCP", target_port) service.save() else: if service.update_port(DEFAULT_HTTP_PORT, "TCP", target_port): service.save() try: gateway = self.gateway_set.filter(name=self.id).latest() if gateway.change_default_tls(): gateway.save() except Gateway.DoesNotExist: gateway = Gateway(app=self, name=self.id) added, msg = gateway.add(DEFAULT_HTTP_PORT, "HTTP") if not added: raise DryccException(msg) gateway.save() try: route = self.route_set.filter(name=self.id).latest() if route.change_default_tls(): route.save() except Route.DoesNotExist: route = Route( app=self, kind="HTTPRoute", name=self.id, rules=[ { "backendRefs": [ { "kind": "Service", "name": service.name, "port": DEFAULT_HTTP_PORT, "weight": 100, } ] } ], ) attached, msg = route.attach(gateway.name, DEFAULT_HTTP_PORT) if not attached: raise DryccException(msg) route.save() class RunMixin(AppBaseMixin): """Mixin for one-off command execution.""" def run( self, user, image=None, command=None, args=None, volumes=None, timeout=3600, expires=3600, **kwargs, ): """Run a one-off command in an ephemeral app container.""" def pod_name(size=5, chars=string.ascii_lowercase + string.digits): return "".join(random.choice(chars) for _ in range(size)) release = Release.latest(self) if release is None or release.build is None: raise DryccException("No build associated with this release to run this command") app_settings = self.appsettings_set.latest() volume_list = [] if volumes: for volume in Volume.objects.filter(app=self, name__in=volumes.keys()): volume.path[PTYPE_RUN] = volumes.get(volume.name, None) volume_list.append(volume) else: for volume in Volume.objects.filter(app=self): if PTYPE_RUN in volume.path.keys(): volume_list.append(volume) data = self.gather_app_settings( release, app_settings, ptype=PTYPE_RUN, replicas=1, volumes=volume_list ) data["restart_policy"] = "Never" data["active_deadline_seconds"] = timeout data["ttl_seconds_after_finished"] = expires name = self.get_deploy_name(PTYPE_RUN) + "-" + pod_name() self.log("{} on {} runs '{}'".format(user.username, name, command)) kwargs.update(data) try: self.set_application_config(release, PTYPE_RUN) self.scheduler.job.create( self.id, name, image=image, command=command, args=args, **kwargs ) except Exception as e: err = "{} ({}): {}".format(name, PTYPE_RUN, e) raise ServiceUnavailable(err) from e return name class KubernetesMixin(AppBaseMixin): """Mixin for K8s resource operations: pods, deployments, events.""" def describe_pod(self, pod_name): """Describe pod details.""" def get_command_and_args(pod, container_name): command, args = [], [] for container in pod["spec"]["containers"]: if container["name"] == container_name: args = container.get("args", []) command = container.get("command", []) break return command, args result = [] try: pod = self.scheduler.pod.get(self.id, pod_name).json() if pod["status"]["phase"] == "Pending": statuses = pod["spec"]["containers"] else: statuses = pod["status"]["containerStatuses"] for status in statuses: command, args = get_command_and_args(pod, status["name"]) result.append( { "container": status["name"], "image": status["image"], "command": command, "args": args, "state": status.get("state", {}), "lastState": status.get("lastState", {}), "ready": status.get("ready", False), "restartCount": status.get("restartCount", 0), "status": pod["status"].get("phase", ""), "reason": pod["status"].get("reason", ""), "message": pod["status"].get("message", ""), } ) except KubeHTTPException as e: if e.response.status_code != 404: raise e return result def list_pods(self, *args, **kwargs): """List pods for this app.""" try: labels = self._scheduler_filter(**kwargs) if "name" in kwargs: pods = [self.scheduler.pod.get(self.id, kwargs["name"]).json()] else: pods = self.scheduler.pod.get(self.id, labels=labels).json()["items"] or [] data = [] for p in pods: labels = p["metadata"]["labels"] if "startTime" in p["status"]: started = p["status"]["startTime"] else: started = str( datetime.now(timezone.utc).strftime(settings.DRYCC_DATETIME_FORMAT) ) state = str(self.scheduler.pod.state(p)) if p["status"]["phase"] != "Pending": ready = len([1 for s in p["status"]["containerStatuses"] if s["ready"]]) restarts = sum([s["restartCount"] for s in p["status"]["containerStatuses"]]) else: restarts, ready = 0, 0 item = { "name": p["metadata"]["name"], "state": state, "release": labels["version"], "type": labels["type"], "started": started, "ready": "%s/%s" % (ready, len(p["spec"]["containers"])), "restarts": restarts, } data.append(item) data.sort(key=lambda x: x["started"], reverse=True) return data except KubeHTTPException: pass except Exception as e: err = "(list pods): {}".format(e) self.log(err, logging.ERROR) raise ServiceUnavailable(err) from e def delete_pod(self, **kwargs): """Delete a specific pod.""" pod_name = kwargs.get("pod_name") try: pod = self.scheduler.pod.get(self.id, pod_name).json() if pod["metadata"]["labels"].get("heritage") == "drycc": self.scheduler.pod.delete(self.id, pod_name) except KubeHTTPException as e: if e.response.status_code != 404: raise e def describe_deployment(self, deployment_name): """Describe deployment details.""" result = [] try: deployment = self.scheduler.deployment.get(self.id, deployment_name).json() for container in deployment["spec"]["template"]["spec"]["containers"]: limits = container.get("resources", {}).get("limits", {}) result.append( { "container": container["name"], "image": container["image"], "command": container.get("command", []), "args": container.get("args", []), "liveness_probe": container.get("livenessProbe", {}), "readiness_probe": container.get("readinessProbe", {}), "limits": limits, "volume_mounts": container.get("volumeMounts", []), "node_selector": deployment["spec"]["template"]["spec"].get( "nodeSelector", {} ), } ) except KubeHTTPException as e: if e.response.status_code != 404: raise e return result def list_deployments(self, *args, **kwargs): """List deployments for this app.""" ptypes = self.release_set.latest().ptypes try: labels = self._scheduler_filter(**kwargs) if "name" in kwargs: deployments = [self.scheduler.deployment.get(self.id, kwargs["name"]).json()] else: deployments = ( self.scheduler.deployment.get(self.id, labels=labels).json()["items"] or [] ) data = [] for p in deployments: labels = p["spec"]["template"]["metadata"]["labels"] if p["metadata"]["creationTimestamp"]: started = p["metadata"]["creationTimestamp"] else: started = str( datetime.now(timezone.utc).strftime(settings.DRYCC_DATETIME_FORMAT) ) item = { "name": labels["type"], "release": labels["version"], "ready": "%s/%s" % (p["status"].get("readyReplicas", 0), p["spec"].get("replicas", 0)), "garbage": False if labels["type"] in ptypes else True, "up_to_date": p["status"].get("updatedReplicas", 0), "available_replicas": p["status"].get("availableReplicas", 0), "started": started, } data.append(item) data.sort(key=lambda x: x["started"], reverse=True) return data except KubeHTTPException: pass except Exception as e: err = "(list deployments): {}".format(e) self.log(err, logging.ERROR) raise ServiceUnavailable(err) from e def list_events(self, ref_kind, ref_name, *args, **kwargs): """List events for this app.""" try: fields = {"regarding.kind": ref_kind, "regarding.name": ref_name} kwargs["fields"] = fields events = self.scheduler.events.get(self.id, **kwargs).json()["items"] data = [] for e in events: item = { "reason": e["reason"], "message": e["note"], "created": e["metadata"]["creationTimestamp"], } data.append(item) data.sort(key=lambda x: x["created"], reverse=False) return data except KubeHTTPException: pass except Exception as e: err = "(list event): {}".format(e) self.log(err, logging.ERROR) raise ServiceUnavailable(err) from e def _scheduler_filter(self, **kwargs): """Build scheduler labels filter.""" labels = {"app": self.id, "heritage": "drycc"} if "type" in kwargs: labels.update({"type": kwargs["type"]}) if "version" in kwargs: if isinstance(kwargs["version"], int): version = "v{}".format(kwargs["version"]) else: version = kwargs["version"] labels.update({"version": version}) return labels class HealthMixin(AppBaseMixin): """Mixin for health check operations.""" def verify_http_health(self, service, **kwargs): """Verify HTTP service health.""" app_type = kwargs.get("app_type") self.log( "Waiting for service to be ready to serve traffic to process type {}".format(app_type), level=logging.DEBUG, ) url = "http://{}:{}".format(service.domain, service.ports[0]["port"]) if ( "livenessProbe" in kwargs.get("healthcheck", {}) and "httpGet" in kwargs["healthcheck"]["livenessProbe"] ): allowed = [200] handler = kwargs["healthcheck"]["livenessProbe"]["httpGet"] url = urljoin(url, handler.get("path", "/")) req_timeout = handler.get("timeoutSeconds", 1) else: allowed = set(range(200, 599)) allowed.remove(404) req_timeout = 3 start = time.time() failed = False response = None for _ in range(10): try: with get_httpclient() as session: response = session.get(url, timeout=req_timeout) failed = False except requests.exceptions.RequestException: failed = True time.sleep(1) continue if (time.time() - start) > (req_timeout * 10): break if response.status_code in allowed: break time.sleep(1) if (response and response.status_code == 404) or failed: delta = round(time.time() - start) self.log( "Router was not ready to serve traffic " "to process type {} in time, waited {} seconds".format( app_type, delta ), level=logging.WARNING, ) return self.log( "Router is ready to serve traffic to process type {}".format(app_type), level=logging.DEBUG, ) def verify_tcp_health(self, service, **kwargs): """Verify TCP service health.""" for _ in range(10): with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock: sock.settimeout(3) if sock.connect_ex((service.domain, service.ports[0]["port"])) == 0: break else: time.sleep(3) class ConfigMixin(AppBaseMixin): """Mixin for configuration and environment management.""" def set_application_config(self, release, ptype): """Create/update application config as a Kubernetes secret.""" labels = {"version": release.version_name, "type": ptype, "class": "env"} secrets_env = {} for key, value in self.build_env_vars(release, ptype).items(): secrets_env[key.lower().replace("_", "-")] = str(value) secrets_env = OrderedDict(sorted(secrets_env.items(), key=lambda t: t[0])) secret_name = "{}-{}-{}-env".format(self.id, ptype, release.version_name) try: self.scheduler.secret.get(self.id, secret_name) except KubeHTTPException: self.scheduler.secret.create(self.id, secret_name, secrets_env, labels=labels) else: self.scheduler.secret.update(self.id, secret_name, secrets_env, labels=labels) def image_pull_secret(self, namespace, ptype, registry, image): """Set imagePullSecret for registry.""" docker_config, name, create = self._get_private_registry_config(ptype, image, registry) if create is None: return elif create: data = {".dockerconfigjson": docker_config} try: self.scheduler.secret.get(namespace, name) except KubeHTTPException: self.scheduler.secret.create( namespace, name, data, secret_type="kubernetes.io/dockerconfigjson" ) else: self.scheduler.secret.update( namespace, name, data, secret_type="kubernetes.io/dockerconfigjson" ) return name def build_env_vars(self, release, ptype): """Build environment variables dict.""" if release is None or release.build is None: raise DryccException("No build associated with this release to run this command") default_env = { "DRYCC_APP": self.id, "DRYCC_WORKSPACE": self.workspace_id, "WORKFLOW_RELEASE": release.version_name, "WORKFLOW_RELEASE_SUMMARY": release.summary, "WORKFLOW_RELEASE_CREATED_AT": str( release.created.strftime(settings.DRYCC_DATETIME_FORMAT) ), "SOURCE_VERSION": release.build.sha, } default_env.update(release.config.envs(ptype)) port = release.get_port(ptype) if port: default_env["PORT"] = port return default_env def _get_private_registry_config(self, ptype, image, registry=None): """Get private registry configuration.""" name = settings.REGISTRY_SECRET_PREFIX + "-" + ptype if registry: hostname = registry.get("hostname", None) if not hostname: hostname, _ = docker_auth.split_repo_name(image) if hostname == docker_auth.INDEX_NAME: hostname = docker_auth.INDEX_URL username = registry.get("username") password = registry.get("password") elif settings.REGISTRY_LOCATION == "off-cluster": secret = self.scheduler.secret.get( settings.WORKFLOW_NAMESPACE, "controller-creds" ).json() hostname = secret["data"]["registry-host"] if hostname == "": hostname = docker_auth.INDEX_URL username = secret["data"]["registry-username"] password = secret["data"]["registry-password"] name = name + "-" + settings.REGISTRY_LOCATION else: return None, None, None auth = bytes("{}:{}".format(username, password), "UTF-8") docker_config = json.dumps( {"auths": {hostname: {"auth": base64.b64encode(auth).decode(encoding="UTF-8")}}} ) return docker_config, name, True def _set_default_limit(self, config, ptype): """Set default limit for a process type.""" if ptype not in config.limits: plan = LimitPlan.get_default() config.limits[ptype] = plan.id config.save(update_fields=["limits"]) return config def _get_volumes_and_mounts(self, ptype, volumes): """Get Kubernetes volume configurations.""" k8s_volumes, k8s_volume_mounts = [], [] if volumes: for volume in volumes: k8s_volumes.append( {"name": volume.name, "persistentVolumeClaim": {"claimName": volume.name}} ) k8s_volume_mounts.append( {"name": volume.name, "mountPath": volume.path.get(ptype)} ) return k8s_volumes, k8s_volume_mounts def gather_app_settings(self, release, app_settings, ptype, replicas, volumes=None): """Gather all required information for deploying an application.""" envs = self.build_env_vars(release, ptype) config = self._set_default_limit(release.config, ptype) limit_plan = LimitPlan.objects.get(id=config.limits.get(ptype)) batches = int(envs.get("DRYCC_DEPLOY_BATCHES", settings.DRYCC_DEPLOY_BATCHES)) deploy_timeout = int(envs.get("DRYCC_DEPLOY_TIMEOUT", settings.DRYCC_DEPLOY_TIMEOUT)) deployment_history = envs.get( "KUBERNETES_DEPLOYMENTS_REVISION_HISTORY_LIMIT", settings.KUBERNETES_DEPLOYMENTS_REVISION_HISTORY_LIMIT, ) termination_grace_period_seconds = int( config.termination_grace_period.get( ptype, envs.get( "KUBERNETES_POD_TERMINATION_GRACE_PERIOD_SECONDS", settings.KUBERNETES_POD_TERMINATION_GRACE_PERIOD_SECONDS, ), ) ) image_pull_policy = envs.get("IMAGE_PULL_POLICY", settings.IMAGE_PULL_POLICY) registry = config.registry.get(ptype, {}) image_pull_secret_name = self.image_pull_secret( self.id, ptype, registry, release.get_deploy_image(ptype) ) routable = True if (ptype == PTYPE_WEB and app_settings.routable) else False volumes, volume_mounts = self._get_volumes_and_mounts(ptype, volumes) volumes.extend(limit_plan.pod_volumes) volume_mounts.extend(limit_plan.container_volume_mounts) return { "tags": config.tags.get(ptype, {}), "envs": envs, "registry": registry, "replicas": replicas, "version": release.version_name, "app_type": ptype, "resources": {"limits": limit_plan.limits, "requests": limit_plan.requests}, "build_type": release.build.type, "annotations": limit_plan.annotations, "lifecycle": config.lifecycle.get(ptype, {}), "healthcheck": config.healthcheck.get(ptype, {}), "runtime_class_name": limit_plan.runtime_class_name, "dns_policy": settings.DRYCC_APP_DNS_POLICY, "routable": routable, "deploy_batches": batches, "restart_policy": "Always", "deploy_timeout": deploy_timeout, "deployment_revision_history_limit": deployment_history, "release_summary": release.summary, "termination_grace_period_seconds": termination_grace_period_seconds, "image_pull_secret_name": image_pull_secret_name, "image_pull_policy": image_pull_policy, "volumes": volumes, "volume_mounts": volume_mounts, "node_selector": limit_plan.node_selector, "pod_security_context": limit_plan.pod_security_context, "container_security_context": limit_plan.container_security_context, } class AutoscaleMixin(AppBaseMixin): """Mixin for autoscaling operations.""" def autoscale(self, proc_type, autoscale): """Set autoscale rules for the application.""" if proc_type == PTYPE_RUN: raise DryccException("Cannot set autoscale for reserved types, procfile type is: run") name = "{}-{}".format(self.id, proc_type) target = {"apiVersion": "apps/v1", "kind": "Deployment", "metadata": {"name": name}} try: self.scheduler.hpa.get(self.id, name) if autoscale is None: self.scheduler.hpa.delete(self.id, name) else: self.scheduler.hpa.update( self.id, name, app_type=proc_type, target=target, **autoscale ) except KubeHTTPException as e: if e.response.status_code == 404: self.scheduler.hpa.create( self.id, name, app_type=proc_type, target=target, **autoscale ) else: raise ServiceUnavailable(str(e)) from e class UsageMixin(AppBaseMixin): """Mixin for usage reporting.""" def to_usage(self, timestamp: float): """Generate usage data for billing.""" usage = [] config = self.config_set.latest() for ptype, scale in self.structure.items(): plan = config.limits.get(ptype) usage.append( { "app_id": str(self.uuid), "workspace": self.workspace_id, "name": plan, "type": "limits", "unit": "number", "usage": scale, "kwargs": {"ptype": ptype}, "timestamp": int(timestamp), "identifier": self.uuid.hex, } ) return usage class StateMixin(AppBaseMixin): """Mixin for state synchronization operations.""" def state_to_k8s(self): """Sync app state to Kubernetes.""" release = Release.latest(self) if release is None or release.build is None: self.log("the last release does not have a build, skipping deployment...") return ptypes = set() for ptype, scale in self.structure.items(): response = self.scheduler.deployment.get( self.id, self.get_deploy_name(ptype), ignore_exception=True ) if response.status_code == 404 and scale > 0: ptypes.add(ptype) elif response.status_code != 200: data = response.json() self.log( "get deployment status_code {}, message: {}".format( response.status_code, data.get("message", "") ), logging.ERROR, ) if len(ptypes) == 0: self.log("the cluster status is the latest, skipping deployment...") return self.deploy(release, ptypes, False, False) class App( LifecycleMixin, DeployMixin, RunMixin, KubernetesMixin, HealthMixin, ConfigMixin, AutoscaleMixin, UsageMixin, StateMixin, UuidAuditedModel, ): """ Application used to service requests on behalf of end-users. Mixin organization: - LifecycleMixin: create, restart - DeployMixin: deploy, scale, mount, pipeline, clean - RunMixin: one-off command execution - KubernetesMixin: pod/deployment/event operations - HealthMixin: HTTP/TCP health verification - ConfigMixin: configuration and environment management - AutoscaleMixin: HPA autoscaling - UsageMixin: usage reporting - StateMixin: state synchronization """ id = models.SlugField(max_length=63, unique=True, null=True, validators=[validate_app_id]) uid = models.PositiveIntegerField(unique=True, editable=False) workspace = models.ForeignKey("Workspace", on_delete=models.CASCADE) structure = models.JSONField(default=dict, blank=True, validators=[validate_app_structure]) class Meta: verbose_name = "Application" ordering = ["id"] def save(self, *args, **kwargs): if not self.id: self.id = generate_app_name() while App.objects.filter(id=self.id).exists(): self.id = generate_app_name() try: self.release_set.latest() except Release.DoesNotExist: try: if self.scheduler.ns.get(self.id).status_code == 200: err = "{} already exists as a namespace in this kuberenetes setup".format( self.id ) self.log(err, logging.INFO) raise AlreadyExists(err) except KubeHTTPException: pass if not self.uid: self.uid = get_next_uid(App) application = super().save(**kwargs) self.create(*args, **kwargs) return application def delete(self, *args, **kwargs): self.log("deleting environment") try: self.scheduler.ns.get(self.id) try: self.scheduler.ns.delete(self.id) for _ in range(30): try: self.scheduler.ns.get(self.id) except KubeHTTPException as e: if e.response.status_code == 404: break except KubeException as e: raise ServiceUnavailable( "Could not delete Kubernetes Namespace {} " "within 30 seconds".format(self.id) ) from e except KubeHTTPException: pass return super().delete(*args, **kwargs) def __str__(self): return self.id