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