| """Helpers for validating startup runtime expectations from `.env`.""" |
|
|
| from __future__ import annotations |
|
|
| import os |
| from dataclasses import dataclass |
| from pathlib import Path |
|
|
| from dotenv import dotenv_values |
|
|
| _CONTAINER_RUNTIME_TARGETS = {"compose", "docker"} |
|
|
|
|
| @dataclass(frozen=True) |
| class RuntimeEnvironment: |
| """Describes whether the current process is running in a container runtime.""" |
|
|
| in_container: bool |
| in_docker: bool |
| in_kubernetes: bool |
|
|
| @property |
| def label(self) -> str: |
| if self.in_kubernetes: |
| return "Kubernetes" |
| if self.in_docker: |
| return "Docker" |
| return "host" |
|
|
|
|
| def _read_cgroup_content() -> str: |
| """Best-effort read of cgroup metadata for container detection.""" |
|
|
| for candidate in ("/proc/1/cgroup", "/proc/self/cgroup"): |
| try: |
| return Path(candidate).read_text(encoding="utf-8") |
| except OSError: |
| continue |
| return "" |
|
|
|
|
| def detect_runtime_environment( |
| environ: dict[str, str] | None = None, |
| ) -> RuntimeEnvironment: |
| """Detect whether the current process is running on host, Docker, or Kubernetes.""" |
|
|
| environ = environ or os.environ |
| cgroup_content = _read_cgroup_content().lower() |
|
|
| in_kubernetes = bool( |
| environ.get("KUBERNETES_SERVICE_HOST") |
| or Path("/var/run/secrets/kubernetes.io/serviceaccount").exists() |
| or "kubepods" in cgroup_content |
| or "kubernetes" in cgroup_content |
| ) |
| in_docker = bool( |
| Path("/.dockerenv").exists() |
| or Path("/run/.containerenv").exists() |
| or any( |
| marker in cgroup_content |
| for marker in ("docker", "containerd", "libpod", "podman") |
| ) |
| ) |
|
|
| return RuntimeEnvironment( |
| in_container=in_kubernetes or in_docker, |
| in_docker=in_docker, |
| in_kubernetes=in_kubernetes, |
| ) |
|
|
|
|
| def load_runtime_target_from_env_file(env_path: str | Path = ".env") -> str | None: |
| """Return the raw LIGHTRAG_RUNTIME_TARGET value from the `.env` file, if present.""" |
|
|
| env_values = dotenv_values(str(env_path)) |
| runtime_target = env_values.get("LIGHTRAG_RUNTIME_TARGET") |
| if runtime_target is None: |
| return None |
| return runtime_target.strip() |
|
|
|
|
| def validate_runtime_target( |
| runtime_target: str | None, |
| runtime_environment: RuntimeEnvironment | None = None, |
| ) -> tuple[bool, str | None]: |
| """Validate `.env` runtime target against the current runtime environment.""" |
|
|
| if runtime_target is None: |
| return True, None |
|
|
| normalized_target = runtime_target.strip().lower() |
| runtime_environment = runtime_environment or detect_runtime_environment() |
|
|
| if normalized_target == "host": |
| if runtime_environment.in_container: |
| return ( |
| False, |
| "Configuration error in .env: LIGHTRAG_RUNTIME_TARGET=host.\n" |
| "This value from .env requires the server process to run on the host, " |
| f"but the current process is running inside {runtime_environment.label}.", |
| ) |
| return True, None |
|
|
| if normalized_target in _CONTAINER_RUNTIME_TARGETS: |
| if runtime_environment.in_container: |
| return True, None |
| return ( |
| False, |
| f"Configuration error in .env: LIGHTRAG_RUNTIME_TARGET={runtime_target}.\n" |
| "This value from .env requires the server process to run inside Docker or " |
| "Kubernetes, but the current process is running on the host.", |
| ) |
|
|
| return ( |
| False, |
| f"Configuration error in .env: LIGHTRAG_RUNTIME_TARGET={runtime_target!r}.\n" |
| "This value from .env must be 'host' or 'compose' (alias: 'docker').", |
| ) |
|
|
|
|
| def validate_runtime_target_from_env_file( |
| env_path: str | Path = ".env", |
| runtime_environment: RuntimeEnvironment | None = None, |
| ) -> tuple[bool, str | None]: |
| """Load LIGHTRAG_RUNTIME_TARGET from `.env` and validate it if declared.""" |
|
|
| runtime_target = load_runtime_target_from_env_file(env_path) |
| return validate_runtime_target(runtime_target, runtime_environment) |
|
|
