feat(env): auto-detect gke_ kubecontext prefix as prod; status.py context helpers

knoe/knoe_conf.py: normalize_environment() now recognises the gke_ prefix
(e.g. gke_plenary-truck-485623-p7_us-west3_knoe-dev-0) as the 'prod'
environment, matching real GKE kubecontext naming.

status.py: add _current_kubecontext() and _cfg_path_from_kubecontext()
so status.py auto-selects the correct cfg file based on the active
kubectl context without manual KNOE_CONF overrides.

Co-authored-by: Junie <junie@jetbrains.com>
This commit is contained in:
chrisfu 2026-05-23 21:31:07 -07:00
parent ef20c8a598
commit da0fd2c545
2 changed files with 102 additions and 11 deletions

View File

@ -95,7 +95,7 @@ def normalize_environment(env: str | None) -> str:
return "dev"
if s in ("service", "services", "k3s", "knoe-service-cluster") or s.startswith("knoe-service-"):
return "service"
if s in ("prod", "production", "k8s", "knoe-prod-cluster") or s.startswith("knoe-prod-"):
if s in ("prod", "production", "k8s", "knoe-prod-cluster") or s.startswith(("knoe-prod-", "gke_")):
return "prod"
if s in ("test", "testing"):
return "test"

111
status.py
View File

@ -36,14 +36,60 @@ PROJECT_ROOT = _SELF.parent # status.py lives at the project root
# ---------------------------------------------------------------------------
def _current_kubecontext() -> str:
"""Return the current kubectl context name, or empty string on failure."""
try:
res = subprocess.run(
["kubectl", "config", "current-context"],
capture_output=True,
text=True,
timeout=5,
)
if res.returncode == 0:
return res.stdout.strip()
except Exception:
pass
return ""
def _cfg_path_from_kubecontext(kubecontext: str) -> Path | None:
"""Derive the config file path from a kubecontext name.
Maps the kubecontext to an environment via ``normalize_environment`` and
returns the corresponding cfg file path if it exists.
"""
if not kubecontext:
return None
try:
from knoe import knoe_conf as knoe_conf_mgr
env = knoe_conf_mgr.normalize_environment(kubecontext)
cfg_name = knoe_conf_mgr.cfg_file_for_env(env)
if cfg_name:
conf_dir = knoe_conf_mgr.resolve_knoe_conf_dir(PROJECT_ROOT)
candidate = conf_dir / cfg_name
if candidate.exists():
return candidate
except Exception:
pass
return None
def _load_knoe_cfg(cfg_path: Path | None = None) -> configparser.ConfigParser:
"""Load knoe.cfg (with per-environment override layering).
Resolution order for the config file:
1. Explicit *cfg_path* argument (from ``-c`` CLI flag).
2. ``$KNOE_CONF/{k3d|k3s|gke}.cfg`` environment variable directory.
3. ``<PROJECT_ROOT>/conf/{k3d|k3s|gke}.cfg`` fallback.
2. Current kubecontext (``kubectl config current-context``) mapped to
the corresponding ``{k3d|k3s|gke}.cfg`` via environment normalisation.
3. ``$KNOE_CONF/{k3d|k3s|gke}.cfg`` environment variable directory.
4. ``<PROJECT_ROOT>/conf/{k3d|k3s|gke}.cfg`` fallback.
"""
if cfg_path is None:
# Try to derive config from the current kubecontext first.
kubectx = _current_kubecontext()
if kubectx:
cfg_path = _cfg_path_from_kubecontext(kubectx)
if cfg_path is None:
try:
from knoe import knoe_conf as knoe_conf_mgr
@ -84,14 +130,46 @@ def _cfg_get(
return fallback
def _get_namespace(cp: configparser.ConfigParser) -> str:
ns = _cfg_get(cp, "User", "NAMESPACE", "default")
return ns or "default"
def _get_namespace(cp: configparser.ConfigParser, env: str = "") -> str:
ns = _cfg_get(cp, "User", "NAMESPACE", "")
if not ns:
ns = _cfg_get(cp, "Global", "DATABASE_NAMESPACE", "")
if not ns:
ns = _cfg_get(cp, "Global", "NAMESPACE", "")
if ns and not (ns.startswith("${") and ns.endswith("}")):
return ns
# Environment-aware defaults (mirrors KnoeInstaller._secret_namespace)
if not env:
try:
from knoe import knoe_conf as knoe_conf_mgr
env = knoe_conf_mgr.normalize_environment(
_cfg_get(cp, "Global", "CLUSTER_ENV", "")
or _cfg_get(cp, "Global", "DEPLOYMENT_MODE", "")
)
except Exception:
pass
if env == "prod":
return "knoe-db-0"
return "default"
def _get_service_namespace(cp: configparser.ConfigParser) -> str:
ns = _cfg_get(cp, "User", "SERVICE_NAMESPACE", "default")
return ns or "default"
def _get_service_namespace(cp: configparser.ConfigParser, env: str = "") -> str:
ns = _cfg_get(cp, "User", "SERVICE_NAMESPACE", "")
if not ns:
ns = _cfg_get(cp, "Global", "SERVICE_NAMESPACE", "")
if ns and not (ns.startswith("${") and ns.endswith("}")) and ns != "default":
return ns
# Environment-aware defaults (mirrors KnoeInstaller._service_namespace)
if not env:
try:
from knoe import knoe_conf as knoe_conf_mgr
env = knoe_conf_mgr.normalize_environment(
_cfg_get(cp, "Global", "CLUSTER_ENV", "")
or _cfg_get(cp, "Global", "DEPLOYMENT_MODE", "")
)
except Exception:
pass
return "knoe-system"
def _get_etc_dir(cp: configparser.ConfigParser) -> Path:
@ -739,9 +817,22 @@ def main():
args = parser.parse_args()
cfg_path = Path(args.config) if args.config else None
# Detect env from kubecontext when no explicit config is given, so that
# namespace defaults are correct for the active cluster.
detected_env = ""
if cfg_path is None:
kubectx = _current_kubecontext()
if kubectx:
try:
from knoe import knoe_conf as knoe_conf_mgr
detected_env = knoe_conf_mgr.normalize_environment(kubectx)
except Exception:
pass
cp = _load_knoe_cfg(cfg_path)
namespace = _get_namespace(cp)
svc_namespace = _get_service_namespace(cp)
namespace = _get_namespace(cp, detected_env)
svc_namespace = _get_service_namespace(cp, detected_env)
etc_dir = _get_etc_dir(cp)
milestones = _build_milestones(namespace, svc_namespace, etc_dir, cp)