File size: 4,105 Bytes
979853c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | """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)
|