// SPEC: PJ2026-01060307 控制面模块化 draft-2026-06-25-p0. search-script module for scripts/src/platform-infra-observability.ts.

// Moved mechanically from scripts/src/platform-infra-observability.ts:1814-2136 for #903.

// SPEC: PJ2026-01060501 OTel追踪 draft-2026-06-19-p0.
// Responsibility: YAML-first platform-infra OpenTelemetry tracing control commands.
import { Buffer } from "node:buffer";
import { readFileSync } from "node:fs";
import type { UniDeskConfig } from "../config";
import { rootPath } from "../config";
import type { RenderedCliResult } from "../output";
import {
  compactCapture,
  createYamlFieldReader,
  numberField,
  parseJsonOutput,
  redactSensitiveUnknown,
  shQuote,
  capture,
} from "../platform-infra-ops-library";

import type { ObservabilityConfig, ObservabilityTarget } from "./types";
import { fieldManager } from "./types";

export function tempoService(observability: ObservabilityConfig, target: ObservabilityTarget): string {
  return `apiVersion: v1
kind: Service
metadata:
  name: ${observability.traceBackend.serviceName}
  namespace: ${target.namespace}
  labels:
    app.kubernetes.io/name: ${observability.traceBackend.deploymentName}
    app.kubernetes.io/component: trace-backend
    app.kubernetes.io/part-of: platform-infra
    app.kubernetes.io/managed-by: unidesk
spec:
  type: ClusterIP
  selector:
    app.kubernetes.io/name: ${observability.traceBackend.deploymentName}
    app.kubernetes.io/component: trace-backend
  ports:
    - name: http
      port: ${observability.traceBackend.httpPort}
      targetPort: http
    - name: otlp-grpc
      port: ${observability.traceBackend.otlp.grpcPort}
      targetPort: otlp-grpc
    - name: otlp-http
      port: ${observability.traceBackend.otlp.httpPort}
      targetPort: otlp-http
`;
}

export function applyScript(params: {
  yaml: string;
  target: ObservabilityTarget;
  dryRun: boolean;
  wait: boolean;
  collectorDeploymentName: string;
  backendDeploymentName: string;
}): string {
  const encoded = Buffer.from(params.yaml, "utf8").toString("base64");
  const dryRunArg = params.dryRun ? "--dry-run=server" : "";
  const waitDisposition = params.wait ? "skipped-short-connection-use-status" : "skipped";
  return `
set -u
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
manifest="$tmp/platform-infra-observability.yaml"
printf '%s' '${encoded}' | base64 -d > "$manifest"
kubectl apply --server-side --field-manager=${fieldManager} ${dryRunArg} -f "$manifest" >"$tmp/apply.out" 2>"$tmp/apply.err"
apply_rc=$?
collector_rollout_rc=0
backend_rollout_rc=0
wait_disposition=${waitDisposition}
python3 - "$apply_rc" "$collector_rollout_rc" "$backend_rollout_rc" "$wait_disposition" "$tmp/apply.out" "$tmp/apply.err" "$tmp/collector-rollout.out" "$tmp/collector-rollout.err" "$tmp/backend-rollout.out" "$tmp/backend-rollout.err" <<'PY'
import json, os, sys
def text(path, limit=6000):
    try:
        return open(path, encoding="utf-8", errors="replace").read()[-limit:]
    except FileNotFoundError:
        return ""
apply_rc = int(sys.argv[1])
collector_rc = int(sys.argv[2])
backend_rc = int(sys.argv[3])
payload = {
  "ok": apply_rc == 0 and collector_rc == 0 and backend_rc == 0,
  "target": "${params.target.id}",
  "namespace": "${params.target.namespace}",
  "dryRun": ${params.dryRun ? "True" : "False"},
  "apply": {"exitCode": apply_rc, "stdout": text(sys.argv[5]), "stderr": text(sys.argv[6])},
  "rollout": {
    "disposition": sys.argv[4],
    "collector": {"exitCode": collector_rc, "stdout": text(sys.argv[7]), "stderr": text(sys.argv[8])},
    "backend": {"exitCode": backend_rc, "stdout": text(sys.argv[9]), "stderr": text(sys.argv[10])},
  },
}
print(json.dumps(payload, ensure_ascii=False, indent=2))
sys.exit(0 if payload["ok"] else 1)
PY
`;
}

export function statusScript(observability: ObservabilityConfig, target: ObservabilityTarget, full: boolean, generateValidationTrace = false): string {
  const endpointsJson = JSON.stringify(observability.probes.statusEndpoints);
  return `
set -u
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
capture_json() {
  name="$1"
  shift
  "$@" >"$tmp/$name.json" 2>"$tmp/$name.err"
  echo $? >"$tmp/$name.rc"
}
capture_raw() {
  name="$1"
  shift
  "$@" >"$tmp/$name.out" 2>"$tmp/$name.err"
  echo $? >"$tmp/$name.rc"
}
capture_json namespace kubectl get namespace ${shQuote(target.namespace)} -o json
capture_json deployments kubectl -n ${shQuote(target.namespace)} get deployment ${shQuote(observability.collector.deploymentName)} ${shQuote(observability.traceBackend.deploymentName)} -o json
capture_json services kubectl -n ${shQuote(target.namespace)} get service ${shQuote(observability.collector.serviceName)} ${shQuote(observability.traceBackend.serviceName)} -o json
capture_json pods kubectl -n ${shQuote(target.namespace)} get pods -l ${shQuote(`app.kubernetes.io/name in (${observability.collector.deploymentName},${observability.traceBackend.deploymentName})`)} -o json
capture_json events kubectl -n ${shQuote(target.namespace)} get events -o json
python3 - "$tmp" '${endpointsJson.replaceAll("'", "'\"'\"'")}' <<'PY'
import json, random, socket, subprocess, sys, time, urllib.error, urllib.request
tmp = sys.argv[1]
endpoints = json.loads(sys.argv[2])
namespace = "${target.namespace}"
def read(path, binary=False, limit=8000):
    try:
        mode = "rb" if binary else "r"
        with open(path, mode, encoding=None if binary else "utf-8", errors=None if binary else "replace") as fh:
            data = fh.read()
        if binary:
            data = data.decode("utf-8", errors="replace")
        return data[-limit:]
    except FileNotFoundError:
        return ""
def rc(name):
    try:
        return int(read(f"{tmp}/{name}.rc").strip() or "1")
    except ValueError:
        return 1
def parsed_json(name):
    raw = read(f"{tmp}/{name}.json", limit=1000000)
    try:
        return json.loads(raw) if raw else None
    except Exception:
        return None
def compact_section(name):
    return {
        "exitCode": rc(name),
        "stderrTail": read(f"{tmp}/{name}.err", limit=2000),
        "json": parsed_json(name),
    }
probe_results = []
for ep in endpoints:
    path = ep.get("path") or "/"
    if not path.startswith("/"):
        path = "/" + path
    proxy_path = f"/api/v1/namespaces/{namespace}/services/http:{ep['service']}:{ep['portName']}/proxy{path}"
    proc = subprocess.run(["kubectl", "get", "--raw", proxy_path], text=True, capture_output=True, timeout=20)
    probe_results.append({
        "name": ep["name"],
        "service": ep["service"],
        "portName": ep["portName"],
        "path": path,
        "exitCode": proc.returncode,
        "stdoutTail": proc.stdout[-1000:],
        "stderrTail": proc.stderr[-1000:],
        "ok": proc.returncode == 0,
    })
def free_port():
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.bind(("127.0.0.1", 0))
    port = sock.getsockname()[1]
    sock.close()
    return port
def parse_body(raw):
    try:
        return json.loads(raw) if raw else None
    except Exception:
        return raw[-2000:]
def http_request(method, url, body=None, timeout=10):
    data = None
    headers = {}
    if body is not None:
        data = json.dumps(body).encode("utf-8")
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            raw = resp.read(200000).decode("utf-8", errors="replace")
            status = int(getattr(resp, "status", 0) or 0)
            return {"ok": 200 <= status < 300, "status": status, "body": parse_body(raw)}
    except urllib.error.HTTPError as exc:
        raw = exc.read(200000).decode("utf-8", errors="replace")
        return {"ok": False, "status": exc.code, "body": parse_body(raw)}
    except Exception as exc:
        return {"ok": False, "error": type(exc).__name__ + ": " + str(exc)}
def wait_proxy(base_url):
    last = None
    for _ in range(30):
        last = http_request("GET", base_url + "/version", timeout=1)
        if last.get("ok") is True:
            return last
        time.sleep(0.2)
    return last or {"ok": False, "error": "kubectl proxy did not answer"}
def compact_http(result):
    if not isinstance(result, dict):
        return {"ok": False, "error": "missing-result"}
    compact = {"ok": result.get("ok") is True}
    if "status" in result:
        compact["status"] = result.get("status")
    if "error" in result:
        compact["error"] = result.get("error")
    body = result.get("body")
    if isinstance(body, dict):
        compact["bodyKeys"] = sorted(list(body.keys()))[:12]
    elif isinstance(body, str) and body:
        compact["bodyTail"] = body[-1000:]
    return compact
def validation_trace_payload(trace_id, span_id):
    now = time.time_ns()
    return {
        "resourceSpans": [{
            "resource": {"attributes": [
                {"key": "service.name", "value": {"stringValue": "unidesk-observability-validate"}},
                {"key": "deployment.environment", "value": {"stringValue": "${target.id}"}},
                {"key": "unidesk.node", "value": {"stringValue": "${target.id}"}},
                {"key": "k8s.namespace.name", "value": {"stringValue": namespace}},
            ]},
            "scopeSpans": [{
                "scope": {"name": "unidesk-cli-platform-infra-observability"},
                "spans": [{
                    "traceId": trace_id,
                    "spanId": span_id,
                    "name": "unidesk.observability.validate",
                    "kind": 1,
                    "startTimeUnixNano": str(now),
                    "endTimeUnixNano": str(now + 1000000),
                    "attributes": [
                        {"key": "unidesk.issue", "value": {"stringValue": "489"}},
                        {"key": "unidesk.spec", "value": {"stringValue": "${observability.metadata.spec}"}},
                        {"key": "unidesk.validation", "value": {"stringValue": "platform-infra-observability"}},
                    ],
                    "status": {"code": 1},
                }],
            }],
        }],
    }
def generate_test_trace():
    trace_id = ("%032x" % random.getrandbits(128))
    span_id = ("%016x" % random.getrandbits(64))
    port = free_port()
    base_url = "http://127.0.0.1:%s" % port
    stdout_path = f"{tmp}/kubectl-proxy.out"
    stderr_path = f"{tmp}/kubectl-proxy.err"
    with open(stdout_path, "w", encoding="utf-8") as stdout, open(stderr_path, "w", encoding="utf-8") as stderr:
        proc = subprocess.Popen(
            ["kubectl", "proxy", "--address=127.0.0.1", "--port", str(port), "--accept-hosts=^127\\\\.0\\\\.0\\\\.1$"],
            stdout=stdout,
            stderr=stderr,
            text=True,
        )
    try:
        proxy_ready = wait_proxy(base_url)
        if proxy_ready.get("ok") is not True:
            return {
                "ok": False,
                "stage": "proxy-start",
                "traceId": trace_id,
                "proxy": compact_http(proxy_ready),
                "proxyStderrTail": read(stderr_path, limit=2000),
            }
        collector_url = base_url + "/api/v1/namespaces/%s/services/http:${observability.collector.serviceName}:otlp-http/proxy/v1/traces" % namespace
        tempo_url = base_url + "/api/v1/namespaces/%s/services/http:${observability.traceBackend.serviceName}:http/proxy/api/traces/%s" % (namespace, trace_id)
        ingest = http_request("POST", collector_url, validation_trace_payload(trace_id, span_id), timeout=15)
        query = None
        if ingest.get("ok") is True:
            for _ in range(20):
                query = http_request("GET", tempo_url, timeout=10)
                if query.get("ok") is True:
                    break
                time.sleep(1)
        return {
            "ok": ingest.get("ok") is True and isinstance(query, dict) and query.get("ok") is True,
            "stage": "query" if ingest.get("ok") is True else "ingest",
            "traceId": trace_id,
            "spanId": span_id,
            "collectorEndpoint": "${observability.collector.serviceName}:otlp-http",
            "backendEndpoint": "${observability.traceBackend.serviceName}:http",
            "ingest": compact_http(ingest),
            "query": compact_http(query),
            "valuesPrinted": False,
        }
    finally:
        proc.terminate()
        try:
            proc.wait(timeout=3)
        except subprocess.TimeoutExpired:
            proc.kill()
            proc.wait(timeout=3)
runtime_ready = rc("namespace") == 0 and rc("deployments") == 0 and rc("services") == 0 and all(item["ok"] for item in probe_results)
validation_trace = generate_test_trace() if (${generateValidationTrace ? "True" : "False"} and runtime_ready) else None
payload = {
    "ok": runtime_ready and (validation_trace is None or validation_trace.get("ok") is True),
    "target": "${target.id}",
    "namespace": namespace,
    "sections": {
        "namespace": compact_section("namespace"),
        "deployments": compact_section("deployments"),
        "services": compact_section("services"),
        "pods": compact_section("pods"),
        "events": compact_section("events"),
    },
    "probes": probe_results,
    "validationTrace": validation_trace,
}
print(json.dumps(payload, ensure_ascii=False, indent=2 if ${full ? "True" : "False"} else None))
sys.exit(0 if payload["ok"] else 1)
PY
`;
}