mirror of
https://github.com/open-webui/terminals.git
synced 2026-07-22 04:05:25 -04:00
Timothy Jaeryang Baek
719 lines
22 KiB
Python
719 lines
22 KiB
Python
"""Open Terminal Operator — Kopf handlers for the Terminal CRD.
|
|
|
|
Watches ``Terminal`` custom resources (``terminals.openwebui.com/v1alpha1``)
|
|
and reconciles the underlying Kubernetes resources:
|
|
|
|
- **Secret** holding a generated API key
|
|
- **Pod** running the open-terminal container
|
|
- **Service** (ClusterIP) exposing port 8000
|
|
- **PVC** (optional) for persistent ``/home/user`` storage
|
|
|
|
The orchestrator creates/deletes Terminal CRs; this operator does the rest.
|
|
|
|
Ported from the ``kubernetes-controller`` branch with the ABC-compatible
|
|
``openwebui.com`` API group retained for extensibility.
|
|
"""
|
|
|
|
import base64
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import secrets
|
|
import string
|
|
from datetime import datetime, timezone
|
|
|
|
import kopf
|
|
import kubernetes
|
|
from kubernetes import client as k8s
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
_LOG_LEVELS = {
|
|
"DEBUG": logging.DEBUG,
|
|
"INFO": logging.INFO,
|
|
"WARNING": logging.WARNING,
|
|
"ERROR": logging.ERROR,
|
|
"CRITICAL": logging.CRITICAL,
|
|
}
|
|
|
|
GROUP = "openwebui.com"
|
|
VERSION = "v1alpha1"
|
|
PLURAL = "terminals"
|
|
|
|
RESTRICTED_POD_SECURITY_CONTEXT = {
|
|
"runAsNonRoot": True,
|
|
"seccompProfile": {"type": "RuntimeDefault"},
|
|
}
|
|
RESTRICTED_CONTAINER_SECURITY_CONTEXT = {
|
|
"allowPrivilegeEscalation": False,
|
|
"capabilities": {"drop": ["ALL"]},
|
|
"runAsNonRoot": True,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Startup
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _configured_log_level() -> int:
|
|
raw = os.environ.get("TERMINALS_LOG_LEVEL", "INFO")
|
|
level = raw.strip().upper()
|
|
if level in _LOG_LEVELS:
|
|
return _LOG_LEVELS[level]
|
|
|
|
log.warning(
|
|
"Invalid TERMINALS_LOG_LEVEL=%r; using INFO. Expected one of: %s",
|
|
raw,
|
|
", ".join(_LOG_LEVELS),
|
|
)
|
|
return logging.INFO
|
|
|
|
|
|
@kopf.on.startup()
|
|
def configure(settings: kopf.OperatorSettings, **_):
|
|
"""Load K8s config and configure kopf settings."""
|
|
try:
|
|
kubernetes.config.load_incluster_config()
|
|
except kubernetes.config.ConfigException:
|
|
kubernetes.config.load_kube_config()
|
|
log_level = _configured_log_level()
|
|
logging.getLogger().setLevel(log_level)
|
|
settings.posting.level = max(log_level, logging.WARNING)
|
|
settings.persistence.finalizer = "terminals.openwebui.com/finalizer"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _generate_api_key(length: int = 48) -> str:
|
|
alphabet = string.ascii_letters + string.digits
|
|
return "sk-" + "".join(secrets.choice(alphabet) for _ in range(length))
|
|
|
|
|
|
def _resource_name(name: str, suffix: str) -> str:
|
|
"""Derive child resource names from the Terminal CR name."""
|
|
return f"{name}-{suffix}"
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
|
|
_SIZE_RE = re.compile(r"^(\d+(?:\.\d+)?)\s*(Ki|Mi|Gi|Ti)?$")
|
|
_CPU_RE = re.compile(r"^(\d+(?:\.\d+)?)\s*(m)?$")
|
|
_SIZE_MULT = {"": 1, "Ki": 1024, "Mi": 1024**2, "Gi": 1024**3, "Ti": 1024**4}
|
|
|
|
|
|
def _parse_size(value: str) -> int:
|
|
m = _SIZE_RE.match(str(value).strip())
|
|
if not m:
|
|
return int(value)
|
|
num, suffix = float(m.group(1)), m.group(2) or ""
|
|
return int(num * _SIZE_MULT[suffix])
|
|
|
|
|
|
def _parse_cpu_nanos(value: str) -> int:
|
|
m = _CPU_RE.match(str(value).strip())
|
|
if not m:
|
|
return int(float(value) * 1_000_000_000)
|
|
num, suffix = float(m.group(1)), m.group(2) or ""
|
|
if suffix == "m":
|
|
return int(num * 1_000_000)
|
|
return int(num * 1_000_000_000)
|
|
|
|
|
|
def _format_cpu_count(value: str) -> str:
|
|
cores = _parse_cpu_nanos(value) / 1_000_000_000
|
|
if cores.is_integer():
|
|
return str(int(cores))
|
|
return f"{cores:.3f}".rstrip("0").rstrip(".")
|
|
|
|
|
|
def _deep_merge(*items: dict | None) -> dict:
|
|
result = {}
|
|
for item in items:
|
|
if not item:
|
|
continue
|
|
for key, value in item.items():
|
|
if isinstance(value, dict) and isinstance(result.get(key), dict):
|
|
result[key] = _deep_merge(result[key], value)
|
|
else:
|
|
result[key] = value
|
|
return result
|
|
|
|
|
|
def _resource_limit_env(resources_spec: dict) -> list[dict]:
|
|
limits = resources_spec.get("limits", {})
|
|
env = []
|
|
|
|
cpu_limit = limits.get("cpu")
|
|
if cpu_limit:
|
|
cpu_limit = str(cpu_limit)
|
|
env.append({"name": "OPEN_TERMINAL_CPU_LIMIT", "value": cpu_limit})
|
|
env.append(
|
|
{"name": "OPEN_TERMINAL_CPU_COUNT", "value": _format_cpu_count(cpu_limit)}
|
|
)
|
|
|
|
memory_limit = limits.get("memory")
|
|
if memory_limit:
|
|
memory_limit = str(memory_limit)
|
|
env.append({"name": "OPEN_TERMINAL_MEMORY_LIMIT", "value": memory_limit})
|
|
env.append(
|
|
{"name": "OPEN_TERMINAL_MEMORY_BYTES", "value": str(_parse_size(memory_limit))}
|
|
)
|
|
|
|
return env
|
|
|
|
|
|
def _set_env_var(env: list[dict], name: str, value: str) -> None:
|
|
for item in env:
|
|
if item.get("name") == name:
|
|
item["value"] = value
|
|
return
|
|
env.append({"name": name, "value": value})
|
|
|
|
|
|
def _owner_ref(body: dict) -> dict:
|
|
"""Build a single ownerReference dict for garbage collection."""
|
|
return {
|
|
"apiVersion": f"{GROUP}/{VERSION}",
|
|
"kind": "Terminal",
|
|
"name": body["metadata"]["name"],
|
|
"uid": body["metadata"]["uid"],
|
|
"controller": True,
|
|
"blockOwnerDeletion": True,
|
|
}
|
|
|
|
|
|
def _labels(name: str, user_id: str = "") -> dict[str, str]:
|
|
labels = {
|
|
"app.kubernetes.io/name": "open-terminal",
|
|
"app.kubernetes.io/instance": name,
|
|
"app.kubernetes.io/managed-by": "terminals",
|
|
"app.kubernetes.io/part-of": "open-terminal",
|
|
"openwebui.com/terminal": name,
|
|
}
|
|
if user_id:
|
|
labels["openwebui.com/user-id"] = user_id
|
|
return labels
|
|
|
|
|
|
def _set_condition(
|
|
status: dict,
|
|
cond_type: str,
|
|
cond_status: str,
|
|
reason: str = "",
|
|
message: str = "",
|
|
) -> list:
|
|
"""Create or update a condition in the conditions list."""
|
|
conditions = list(status.get("conditions") or [])
|
|
for c in conditions:
|
|
if c["type"] == cond_type:
|
|
c["status"] = cond_status
|
|
c["lastTransitionTime"] = _now_iso()
|
|
c["reason"] = reason
|
|
c["message"] = message
|
|
return conditions
|
|
conditions.append(
|
|
{
|
|
"type": cond_type,
|
|
"status": cond_status,
|
|
"lastTransitionTime": _now_iso(),
|
|
"reason": reason,
|
|
"message": message,
|
|
}
|
|
)
|
|
return conditions
|
|
|
|
|
|
def _parse_node_selector() -> dict[str, str] | None:
|
|
raw = os.environ.get("TERMINALS_KUBERNETES_NODE_SELECTOR", "").strip()
|
|
if not raw:
|
|
return None
|
|
if raw.startswith("{"):
|
|
data = json.loads(raw)
|
|
if not isinstance(data, dict):
|
|
raise ValueError("TERMINALS_KUBERNETES_NODE_SELECTOR must be an object")
|
|
return {str(key): str(value) for key, value in data.items()}
|
|
|
|
selector = {}
|
|
for pair in raw.split(","):
|
|
if "=" not in pair:
|
|
raise ValueError(
|
|
"TERMINALS_KUBERNETES_NODE_SELECTOR must be JSON or k=v pairs"
|
|
)
|
|
key, value = pair.split("=", 1)
|
|
selector[key.strip()] = value.strip()
|
|
return selector or None
|
|
|
|
|
|
def _parse_tolerations() -> list[dict] | None:
|
|
raw = os.environ.get("TERMINALS_KUBERNETES_TOLERATIONS", "").strip()
|
|
if not raw:
|
|
return None
|
|
data = json.loads(raw)
|
|
if not isinstance(data, list) or not all(isinstance(item, dict) for item in data):
|
|
raise ValueError("TERMINALS_KUBERNETES_TOLERATIONS must be a JSON array")
|
|
return data
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Manifest builders
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _build_pod_manifest(
|
|
name: str,
|
|
namespace: str,
|
|
spec: dict,
|
|
api_key: str,
|
|
owner_ref: dict,
|
|
pvc_name: str | None,
|
|
user_id: str = "",
|
|
) -> dict:
|
|
"""Build the Pod manifest for an Open Terminal instance."""
|
|
image = spec.get("image", "ghcr.io/open-webui/open-terminal:latest")
|
|
resources_spec = spec.get("resources", {})
|
|
packages = spec.get("packages", [])
|
|
pip_packages = spec.get("pipPackages", [])
|
|
|
|
env = [
|
|
{"name": "OPEN_TERMINAL_API_KEY", "value": api_key},
|
|
{"name": "OPEN_TERMINAL_HOST", "value": "0.0.0.0"},
|
|
{"name": "OPEN_TERMINAL_PORT", "value": "8000"},
|
|
]
|
|
if packages:
|
|
env.append({"name": "OPEN_TERMINAL_PACKAGES", "value": " ".join(packages)})
|
|
if pip_packages:
|
|
env.append(
|
|
{"name": "OPEN_TERMINAL_PIP_PACKAGES", "value": " ".join(pip_packages)}
|
|
)
|
|
for key, value in spec.get("env", {}).items():
|
|
key = str(key)
|
|
if key != "OPEN_TERMINAL_API_KEY" and value is not None:
|
|
_set_env_var(env, key, str(value))
|
|
for item in _resource_limit_env(resources_spec):
|
|
_set_env_var(env, item["name"], item["value"])
|
|
|
|
volume_mounts = []
|
|
volumes = []
|
|
if pvc_name:
|
|
volume_mounts.append({"name": "home", "mountPath": "/home/user"})
|
|
volumes.append(
|
|
{"name": "home", "persistentVolumeClaim": {"claimName": pvc_name}}
|
|
)
|
|
|
|
container = {
|
|
"name": "open-terminal",
|
|
"image": image,
|
|
"ports": [{"containerPort": 8000, "name": "http", "protocol": "TCP"}],
|
|
"env": env,
|
|
"volumeMounts": volume_mounts,
|
|
"readinessProbe": {
|
|
"httpGet": {"path": "/health", "port": 8000},
|
|
"initialDelaySeconds": 3,
|
|
"periodSeconds": 5,
|
|
},
|
|
"livenessProbe": {
|
|
"httpGet": {"path": "/health", "port": 8000},
|
|
"initialDelaySeconds": 10,
|
|
"periodSeconds": 15,
|
|
},
|
|
}
|
|
|
|
requests = resources_spec.get("requests", {})
|
|
limits = resources_spec.get("limits", {})
|
|
if requests or limits:
|
|
container["resources"] = {}
|
|
if requests:
|
|
container["resources"]["requests"] = requests
|
|
if limits:
|
|
container["resources"]["limits"] = limits
|
|
|
|
restricted = bool(spec.get("restricted"))
|
|
container_security_context = _deep_merge(
|
|
RESTRICTED_CONTAINER_SECURITY_CONTEXT if restricted else {},
|
|
spec.get("containerSecurityContext"),
|
|
)
|
|
if container_security_context:
|
|
container["securityContext"] = container_security_context
|
|
|
|
pod_labels = _labels(name, user_id)
|
|
pod_spec = {
|
|
"containers": [container],
|
|
"volumes": volumes,
|
|
"restartPolicy": "Always",
|
|
"enableServiceLinks": False,
|
|
"automountServiceAccountToken": False,
|
|
}
|
|
node_selector = _parse_node_selector()
|
|
if node_selector:
|
|
pod_spec["nodeSelector"] = node_selector
|
|
tolerations = _parse_tolerations()
|
|
if tolerations:
|
|
pod_spec["tolerations"] = tolerations
|
|
pod_security_context = _deep_merge(
|
|
RESTRICTED_POD_SECURITY_CONTEXT if restricted else {},
|
|
spec.get("podSecurityContext"),
|
|
)
|
|
if pod_security_context:
|
|
pod_spec["securityContext"] = pod_security_context
|
|
|
|
return {
|
|
"apiVersion": "v1",
|
|
"kind": "Pod",
|
|
"metadata": {
|
|
"name": _resource_name(name, "pod"),
|
|
"namespace": namespace,
|
|
"labels": pod_labels,
|
|
"ownerReferences": [owner_ref],
|
|
},
|
|
"spec": pod_spec,
|
|
}
|
|
|
|
|
|
def _build_service_manifest(
|
|
name: str, namespace: str, owner_ref: dict, user_id: str = ""
|
|
) -> dict:
|
|
return {
|
|
"apiVersion": "v1",
|
|
"kind": "Service",
|
|
"metadata": {
|
|
"name": _resource_name(name, "svc"),
|
|
"namespace": namespace,
|
|
"labels": _labels(name, user_id),
|
|
"ownerReferences": [owner_ref],
|
|
},
|
|
"spec": {
|
|
"type": "ClusterIP",
|
|
"selector": {"openwebui.com/terminal": name},
|
|
"ports": [
|
|
{"name": "http", "port": 8000, "targetPort": 8000, "protocol": "TCP"}
|
|
],
|
|
},
|
|
}
|
|
|
|
|
|
def _build_secret_manifest(
|
|
name: str, namespace: str, api_key: str, owner_ref: dict,
|
|
user_id: str = "",
|
|
) -> dict:
|
|
return {
|
|
"apiVersion": "v1",
|
|
"kind": "Secret",
|
|
"metadata": {
|
|
"name": _resource_name(name, "apikey"),
|
|
"namespace": namespace,
|
|
"labels": _labels(name, user_id),
|
|
"ownerReferences": [owner_ref],
|
|
},
|
|
"type": "Opaque",
|
|
"data": {
|
|
"api-key": base64.b64encode(api_key.encode()).decode(),
|
|
},
|
|
}
|
|
|
|
|
|
def _build_pvc_manifest(
|
|
name: str, namespace: str, spec: dict, owner_ref: dict,
|
|
user_id: str = "",
|
|
) -> dict:
|
|
persistence = spec.get("persistence", {})
|
|
size = persistence.get("size", "1Gi")
|
|
storage_class = persistence.get("storageClass", "")
|
|
|
|
# NOTE: PVCs intentionally have NO ownerReference so they survive
|
|
# Terminal CR deletion. User workspace data must persist across
|
|
# idle cleanup and refresh cycles.
|
|
pvc = {
|
|
"apiVersion": "v1",
|
|
"kind": "PersistentVolumeClaim",
|
|
"metadata": {
|
|
"name": _resource_name(name, "pvc"),
|
|
"namespace": namespace,
|
|
"labels": _labels(name, user_id),
|
|
},
|
|
"spec": {
|
|
"accessModes": ["ReadWriteOnce"],
|
|
"resources": {"requests": {"storage": size}},
|
|
},
|
|
}
|
|
if storage_class:
|
|
pvc["spec"]["storageClassName"] = storage_class
|
|
return pvc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Create handler
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@kopf.on.create(GROUP, VERSION, PLURAL)
|
|
async def on_create(body, spec, name, namespace, patch, **_):
|
|
"""Create all child resources for a new Terminal CR."""
|
|
log.info("Creating terminal %s/%s for user %s", namespace, name, spec.get("userId"))
|
|
|
|
user_id = spec.get("userId", "")
|
|
owner_ref = _owner_ref(body)
|
|
api_key = _generate_api_key()
|
|
core_v1 = k8s.CoreV1Api()
|
|
|
|
# -- Status: Provisioning
|
|
patch.status["phase"] = "Provisioning"
|
|
patch.status["lastActivityAt"] = _now_iso()
|
|
patch.status["conditions"] = _set_condition(
|
|
{}, "Ready", "False", "Provisioning", "Creating child resources"
|
|
)
|
|
|
|
# -- PVC (if persistence enabled)
|
|
persistence = spec.get("persistence", {})
|
|
pvc_name = None
|
|
if persistence.get("enabled", True):
|
|
pvc_name = _resource_name(name, "pvc")
|
|
pvc_manifest = _build_pvc_manifest(name, namespace, spec, owner_ref, user_id=user_id)
|
|
try:
|
|
core_v1.create_namespaced_persistent_volume_claim(
|
|
namespace=namespace, body=pvc_manifest
|
|
)
|
|
log.info("Created PVC %s/%s", namespace, pvc_name)
|
|
except k8s.exceptions.ApiException as e:
|
|
if e.status == 409:
|
|
log.info("PVC %s/%s already exists", namespace, pvc_name)
|
|
else:
|
|
raise
|
|
|
|
# -- Secret (API key)
|
|
secret_name = _resource_name(name, "apikey")
|
|
secret_manifest = _build_secret_manifest(name, namespace, api_key, owner_ref, user_id=user_id)
|
|
try:
|
|
core_v1.create_namespaced_secret(namespace=namespace, body=secret_manifest)
|
|
log.info("Created Secret %s/%s", namespace, secret_name)
|
|
except k8s.exceptions.ApiException as e:
|
|
if e.status == 409:
|
|
log.info("Secret %s/%s already exists, reading existing key", namespace, secret_name)
|
|
existing = core_v1.read_namespaced_secret(secret_name, namespace)
|
|
api_key = base64.b64decode(existing.data["api-key"]).decode()
|
|
else:
|
|
raise
|
|
|
|
# -- Service
|
|
svc_name = _resource_name(name, "svc")
|
|
svc_manifest = _build_service_manifest(name, namespace, owner_ref, user_id=user_id)
|
|
try:
|
|
core_v1.create_namespaced_service(namespace=namespace, body=svc_manifest)
|
|
log.info("Created Service %s/%s", namespace, svc_name)
|
|
except k8s.exceptions.ApiException as e:
|
|
if e.status == 409:
|
|
log.info("Service %s/%s already exists", namespace, svc_name)
|
|
else:
|
|
raise
|
|
|
|
# -- Pod
|
|
pod_name = _resource_name(name, "pod")
|
|
pod_manifest = _build_pod_manifest(
|
|
name, namespace, spec, api_key, owner_ref, pvc_name, user_id=user_id
|
|
)
|
|
try:
|
|
core_v1.create_namespaced_pod(namespace=namespace, body=pod_manifest)
|
|
log.info("Created Pod %s/%s", namespace, pod_name)
|
|
except k8s.exceptions.ApiException as e:
|
|
if e.status == 409:
|
|
log.info("Pod %s/%s already exists", namespace, pod_name)
|
|
else:
|
|
raise
|
|
|
|
# -- Update status
|
|
service_url = f"http://{svc_name}.{namespace}.svc:8000"
|
|
patch.status["podName"] = pod_name
|
|
patch.status["serviceName"] = svc_name
|
|
patch.status["serviceUrl"] = service_url
|
|
patch.status["apiKeySecret"] = secret_name
|
|
patch.status["phase"] = "Pending"
|
|
patch.status["conditions"] = _set_condition(
|
|
{"conditions": patch.status.get("conditions", [])},
|
|
"Ready",
|
|
"False",
|
|
"PodNotReady",
|
|
"Waiting for pod to become ready",
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Delete handler (cleanup is automatic via ownerReferences, but log it)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@kopf.on.delete(GROUP, VERSION, PLURAL)
|
|
async def on_delete(name, namespace, **_):
|
|
"""Log deletion — child resources are cleaned up via ownerReferences."""
|
|
log.info(
|
|
"Terminal %s/%s deleted. Child resources will be garbage-collected.",
|
|
namespace,
|
|
name,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pod watcher — update Terminal status when pod phase changes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@kopf.on.event("v1", "pods", labels={"app.kubernetes.io/managed-by": "terminals"})
|
|
async def on_pod_event(event, body, **_):
|
|
"""Watch terminal pods and reflect readiness back into the Terminal CR status."""
|
|
pod = body
|
|
labels = pod.get("metadata", {}).get("labels", {})
|
|
terminal_name = labels.get("openwebui.com/terminal")
|
|
if not terminal_name:
|
|
return
|
|
|
|
namespace = pod["metadata"]["namespace"]
|
|
pod_phase = (pod.get("status") or {}).get("phase", "Unknown")
|
|
|
|
# Check container readiness
|
|
container_statuses = (pod.get("status") or {}).get("containerStatuses", [])
|
|
is_ready = any(cs.get("ready", False) for cs in container_statuses)
|
|
|
|
custom_api = k8s.CustomObjectsApi()
|
|
try:
|
|
terminal = custom_api.get_namespaced_custom_object(
|
|
group=GROUP,
|
|
version=VERSION,
|
|
namespace=namespace,
|
|
plural=PLURAL,
|
|
name=terminal_name,
|
|
)
|
|
except k8s.exceptions.ApiException as e:
|
|
if e.status == 404:
|
|
return
|
|
raise
|
|
|
|
current_status = terminal.get("status", {})
|
|
current_phase = current_status.get("phase")
|
|
|
|
# Don't update if terminal is being torn down
|
|
if current_phase in ("Idle",):
|
|
return
|
|
|
|
new_phase = current_phase
|
|
if is_ready and pod_phase == "Running":
|
|
new_phase = "Running"
|
|
elif pod_phase in ("Pending",):
|
|
new_phase = "Pending"
|
|
elif pod_phase in ("Failed", "Unknown"):
|
|
new_phase = "Error"
|
|
|
|
if new_phase == current_phase and current_phase == "Running" and is_ready:
|
|
return # No change needed
|
|
|
|
conditions = _set_condition(
|
|
current_status,
|
|
"Ready",
|
|
"True" if is_ready else "False",
|
|
"PodReady" if is_ready else "PodNotReady",
|
|
f"Pod phase: {pod_phase}",
|
|
)
|
|
|
|
status_patch = {
|
|
"status": {
|
|
"phase": new_phase,
|
|
"conditions": conditions,
|
|
}
|
|
}
|
|
|
|
if is_ready and new_phase == "Running":
|
|
status_patch["status"]["lastActivityAt"] = _now_iso()
|
|
|
|
try:
|
|
custom_api.patch_namespaced_custom_object_status(
|
|
group=GROUP,
|
|
version=VERSION,
|
|
namespace=namespace,
|
|
plural=PLURAL,
|
|
name=terminal_name,
|
|
body=status_patch,
|
|
)
|
|
except k8s.exceptions.ApiException as e:
|
|
if e.status == 404:
|
|
return
|
|
log.warning("Failed to patch Terminal %s/%s status: %s", namespace, terminal_name, e)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Idle timeout timer
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@kopf.timer(GROUP, VERSION, PLURAL, interval=60, idle=60)
|
|
async def idle_check(spec, status, name, namespace, **_):
|
|
"""Periodically check if a terminal has exceeded its idle timeout."""
|
|
phase = (status or {}).get("phase")
|
|
if phase not in ("Running", "Idle"):
|
|
return
|
|
|
|
last_activity = (status or {}).get("lastActivityAt")
|
|
if not last_activity:
|
|
return
|
|
|
|
timeout_minutes = spec.get("idleTimeoutMinutes", 30)
|
|
try:
|
|
last_dt = datetime.fromisoformat(last_activity.replace("Z", "+00:00"))
|
|
except (ValueError, TypeError):
|
|
return
|
|
|
|
elapsed = (datetime.now(timezone.utc) - last_dt).total_seconds() / 60
|
|
|
|
if elapsed < timeout_minutes:
|
|
return
|
|
|
|
log.info(
|
|
"Terminal %s/%s idle for %.1f min (timeout=%d). Deleting pod.",
|
|
namespace,
|
|
name,
|
|
elapsed,
|
|
timeout_minutes,
|
|
)
|
|
|
|
pod_name = (status or {}).get("podName")
|
|
if not pod_name:
|
|
return
|
|
|
|
# Delete the pod to free resources; the PVC, Secret, and CRD remain
|
|
core_v1 = k8s.CoreV1Api()
|
|
try:
|
|
core_v1.delete_namespaced_pod(name=pod_name, namespace=namespace)
|
|
except k8s.exceptions.ApiException as e:
|
|
if e.status == 404:
|
|
log.info("Pod %s/%s already gone", namespace, pod_name)
|
|
else:
|
|
raise
|
|
|
|
# Update status to Idle
|
|
custom_api = k8s.CustomObjectsApi()
|
|
try:
|
|
custom_api.patch_namespaced_custom_object_status(
|
|
group=GROUP,
|
|
version=VERSION,
|
|
namespace=namespace,
|
|
plural=PLURAL,
|
|
name=name,
|
|
body={
|
|
"status": {
|
|
"phase": "Idle",
|
|
"conditions": _set_condition(
|
|
status,
|
|
"Ready",
|
|
"False",
|
|
"IdleTimeout",
|
|
f"Pod deleted after {elapsed:.0f} min of inactivity",
|
|
),
|
|
}
|
|
},
|
|
)
|
|
except k8s.exceptions.ApiException:
|
|
pass
|