app.py

"""
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