"""Put the Langfuse project API keys into the live Application's inline values. `values-production.yaml` (tracked) carries only `aiService.config.langfuseBaseUrl`, which is not a secret. The two keys are, so they live here the same way `secret.jwtSecret` and `secret.grafanaAdminPassword` do: inline on the ArgoCD Application, injected from GitHub Secrets by this script, never in Git. Edits go through the API, not the ArgoCD UI. Editing that text box saved the block as a FOLDED scalar once and took production down for ~16 hours by swallowing a key into a comment -- see the header of infra/argocd/applications/medical-chatbot-app.yaml. Reading the object, editing the string and PUTting it back keeps the structure intact. Verification drives POST /api/chat, not pod health: ai-service adds the Langfuse exporter at startup, so a bad value would surface as a crashed or silently degraded service, and health checks stayed green through that same outage. Chat outranks tracing -- if chat stops answering, this reverts itself. Required env: ARGOCD_PRACTICE_URL, ARGOCD_PRACTICE_PASSWORD, LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY. """ from __future__ import annotations import json import os import sys import time import urllib.error import urllib.request APP_NAME = "medical-chatbot-app" SITE = "https://realvuxbaro.me" ROLLOUT_WAIT = 20 ROLLOUT_ROUNDS = 12 PUBLIC_FIELD = "langfusePublicKey" SECRET_FIELD = "langfuseSecretKey" def call(base, method, path, token=None, body=None): req = urllib.request.Request( f"{base}{path}", data=json.dumps(body).encode() if body is not None else None, method=method, headers={"Content-Type": "application/json"}, ) if token: req.add_header("Authorization", f"Bearer {token}") with urllib.request.urlopen(req, timeout=30) as resp: raw = resp.read() return json.loads(raw) if raw else {} def set_secret_keys(values: str, public_key: str, secret_key: str) -> str: """Set both keys under the top-level `secret:` block, and nowhere else. Replaces them if already present (so re-running after a key rotation is safe), otherwise appends them to the existing `secret:` block. Refuses to guess if there is no `secret:` block at all -- that would mean the inline values are not the shape this expects, which is exactly when blindly appending causes an outage. """ wanted = {PUBLIC_FIELD: public_key, SECRET_FIELD: secret_key} lines = values.splitlines() out: list[str] = [] in_secret = False seen = set() secret_end = -1 indent = " " for line in lines: stripped = line.strip() is_top_level = bool(line) and not line[0].isspace() if is_top_level: if in_secret: secret_end = len(out) # first line after the block in_secret = stripped == "secret:" elif in_secret and stripped: indent = line[: len(line) - len(line.lstrip())] field = stripped.split(":", 1)[0] if field in wanted: seen.add(field) out.append(f"{indent}{field}: {wanted[field]}") continue out.append(line) if in_secret: secret_end = len(out) if secret_end < 0: raise SystemExit( "No top-level `secret:` block in the inline values -- refusing to guess " "where the keys belong. Inspect with inspect-argocd-app.yml first." ) missing = [f"{indent}{name}: {value}" for name, value in wanted.items() if name not in seen] if missing: out[secret_end:secret_end] = missing return "\n".join(out) + "\n" def chat_works() -> bool: """Drive the real user path. True only on a genuine grounded answer.""" req = urllib.request.Request( f"{SITE}/api/chat", data=json.dumps({"content": "Chống chỉ định của Metformin là gì?"}).encode(), method="POST", headers={"Content-Type": "application/json"}, ) try: with urllib.request.urlopen(req, timeout=120) as resp: body = resp.read().decode(errors="replace") ok = resp.status == 200 and "disclaimer" in body and "gặp sự cố" not in body print(f" chat: HTTP {resp.status}, grounded={ok}") return ok except Exception as exc: # noqa: BLE001 - any failure is a failed check print(f" chat: FAILED -- {exc}") return False def put_values(base, token, values: str) -> None: app = call(base, "GET", f"/api/v1/applications/{APP_NAME}", token=token) app["spec"]["source"]["helm"]["values"] = values call(base, "PUT", f"/api/v1/applications/{APP_NAME}", token=token, body=app) try: call(base, "POST", f"/api/v1/applications/{APP_NAME}/sync", token=token, body={}) except urllib.error.HTTPError as exc: if exc.code != 400: # 400 = already syncing raise def main() -> int: base = os.environ["ARGOCD_PRACTICE_URL"].rstrip("/") password = os.environ["ARGOCD_PRACTICE_PASSWORD"] public_key = os.environ["LANGFUSE_PUBLIC_KEY"] secret_key = os.environ["LANGFUSE_SECRET_KEY"] if not (public_key and secret_key): print("Both LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY must be set.", file=sys.stderr) return 1 token = call(base, "POST", "/api/v1/session", body={"username": "admin", "password": password})["token"] print("=== baseline ===") if not chat_works(): print("Chat is already broken before any change -- refusing to touch anything.", file=sys.stderr) return 1 app = call(base, "GET", f"/api/v1/applications/{APP_NAME}", token=token) before = app["spec"]["source"]["helm"].get("values", "") after = set_secret_keys(before, public_key, secret_key) if after == before: print("Inline values already carry these exact keys -- nothing to do.") return 0 print("=== setting Langfuse keys ===") put_values(base, token, after) for i in range(ROLLOUT_ROUNDS): time.sleep(ROLLOUT_WAIT) status = call(base, "GET", f"/api/v1/applications/{APP_NAME}", token=token).get("status", {}) health = status.get("health", {}).get("status") print(f"poll {i + 1}/{ROLLOUT_ROUNDS}: sync={status.get('sync', {}).get('status')} health={health}") if health == "Healthy" and i >= 2: break print("=== verifying chat still answers ===") if chat_works(): print("Langfuse keys set; chat verified working.") return 0 print("Chat broke -- reverting.", file=sys.stderr) put_values(base, token, before) time.sleep(ROLLOUT_WAIT * 2) print(f"reverted; chat_ok_after_revert={chat_works()}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())