@@ -39,6 +39,11 @@ interface TestTargetSpec {
ttlSeconds : number ;
waitTimeoutSeconds : number ;
fieldManager : string ;
taskState : {
rootPath : string ;
statusRequestTimeoutSeconds : number ;
logTailLines : number ;
} ;
cleanup : {
mode : "delete-namespace" ;
requireOwnershipLabels : true ;
@@ -141,6 +146,13 @@ interface SecretMaterial {
employeePassword : string ;
}
type RemoteCapture = typeof capture ;
interface AsyncJobIdentity {
id : string ;
stateDir : string ;
}
const DEFAULT_CONFIG_PATH = rootPath ( "config" , "pikaoa.yaml" ) ;
const MANAGED_BY = "unidesk-pikaoa-test-target" ;
const TEST_RUNTIME_LABEL = "pikaoa.unidesk.io/test-runtime" ;
@@ -161,21 +173,22 @@ export function pikaoaTestTargetHelp(): Record<string, unknown> {
guarantees : [
"未声明专用 target 时 plan/status 返回 configured=false, start/stop 在远端调用前失败。" ,
"validationOnly target 只用于本地渲染,不连接 route。" ,
"start 只提交具名远程任务并立即返回;status 用短连接读取阶段、终态和有界事件尾部。" ,
"正式 delivery namespace、PaC、GitOps、公网暴露和生产数据库不属于该入口的写入范围。" ,
"Secret 只披露 sourceRef、targetKey、presence 和 fingerprint,不输出值。" ,
] ,
} ;
}
export async function runPikaoaCommand ( config : UniDeskConfig , args : string [ ] ) : Promise < RenderedCliResult > {
export async function runPikaoaCommand ( config : UniDeskConfig , args : string [ ] , remoteCapture : RemoteCapture = capture ) : Promise < RenderedCliResult > {
if ( args . length === 0 || args . some ( isHelpToken ) ) return renderMachine ( "pikaoa test-target" , pikaoaTestTargetHelp ( ) , "json" ) ;
if ( args [ 0 ] !== "test-target" ) throw inputError ( "pikaoa 只支持 test-target" , "unsupported-pikaoa-command" , "pikaoa" , [ "test-target" ] ) ;
const options = parseOptions ( args . slice ( 1 ) ) ;
const selection = readSelection ( options ) ;
if ( options . action === "plan" ) return renderResult ( planPayload ( options , selection ) , options ) ;
if ( options . action === "status" ) return await statusResult ( config , options , selection ) ;
if ( options . action === "start" ) return await startResult ( config , options , selection ) ;
return await stopResult ( config , options , selection ) ;
if ( options . action === "status" ) return await statusResult ( config , options , selection , remoteCapture ) ;
if ( options . action === "start" ) return await startResult ( config , options , selection , remoteCapture ) ;
return await stopResult ( config , options , selection , remoteCapture ) ;
}
function parseOptions ( args : string [ ] ) : TestTargetOptions {
@@ -269,6 +282,7 @@ function parseTarget(id: string, root: Record<string, unknown>, configLabel: str
const workerProbes = record ( probes . worker , ` ${ path } .probes.worker ` ) ;
const cleanup = record ( root . cleanup , ` ${ path } .cleanup ` ) ;
const migration = record ( root . migration , ` ${ path } .migration ` ) ;
const taskState = record ( root . taskState , ` ${ path } .taskState ` ) ;
if ( exporterFailure . blocking !== false ) throw new Error ( ` ${ path } .observability.exporterFailure.blocking 必须为 false ` ) ;
if ( exporterFailure . warning !== true ) throw new Error ( ` ${ path } .observability.exporterFailure.warning 必须为 true ` ) ;
if ( cleanup . requireOwnershipLabels !== true ) throw new Error ( ` ${ path } .cleanup.requireOwnershipLabels 必须为 true ` ) ;
@@ -283,6 +297,11 @@ function parseTarget(id: string, root: Record<string, unknown>, configLabel: str
ttlSeconds : positiveInteger ( root . ttlSeconds , ` ${ path } .ttlSeconds ` , 86 _400 ) ,
waitTimeoutSeconds : positiveInteger ( root . waitTimeoutSeconds , ` ${ path } .waitTimeoutSeconds ` , 900 ) ,
fieldManager : kubernetesName ( nonEmpty ( root . fieldManager , ` ${ path } .fieldManager ` ) , ` ${ path } .fieldManager ` , 63 ) ,
taskState : {
rootPath : taskStateRoot ( taskState . rootPath , ` ${ path } .taskState.rootPath ` ) ,
statusRequestTimeoutSeconds : positiveInteger ( taskState . statusRequestTimeoutSeconds , ` ${ path } .taskState.statusRequestTimeoutSeconds ` , 30 ) ,
logTailLines : positiveInteger ( taskState . logTailLines , ` ${ path } .taskState.logTailLines ` , 100 ) ,
} ,
cleanup : {
mode : exactString ( cleanup . mode , ` ${ path } .cleanup.mode ` , "delete-namespace" ) as "delete-namespace" ,
requireOwnershipLabels : true ,
@@ -451,16 +470,25 @@ function renderedPlan(selection: Selection, context: RenderContext): Record<stri
command : target.migration.command ,
applyPhase : "after-postgres-before-workloads" ,
} ,
taskState : {
jobId : asyncJobIdentity ( target , context ) . id ,
rootPath : target.taskState.rootPath ,
statusRequestTimeoutSeconds : target.taskState.statusRequestTimeoutSeconds ,
logTailLines : target.taskState.logTailLines ,
mutation : false ,
} ,
manifestFingerprint : structuralFingerprint ( structuralManifest ) ,
mutation : false ,
} ;
}
async function statusResult ( config : UniDeskConfig , options : TestTargetOptions , selection : Selection ) : Promise < RenderedCliResult > {
async function statusResult ( config : UniDeskConfig , options : TestTargetOptions , selection : Selection , remoteCapture : RemoteCapture ) : Promise < RenderedCliResult > {
const planned = planPayload ( options , selection ) ;
if ( selection . target === null ) return renderResult ( planned , options ) ;
if ( selection . target === null ) {
return renderResult ( { . . . planned , remoteQueried : false , status : { remoteQueried : false , reason : "test-target-not-configured" } } , options ) ;
}
if ( selection . target . validationOnly ) {
return renderResult ( { . . . planned , status : { remoteQueried : false , reason : "validation-only-target" } } , options ) ;
return renderResult ( { . . . planned , remoteQueried : false , status : { remoteQueried : false , reason : "validation-only-target" } } , options ) ;
}
const context = renderContext ( options , selection . target , false ) ;
if ( context === null ) return renderResult ( { . . . planned , ok : false , status : { remoteQueried : false , reason : "instance-required" } } , options ) ;
@@ -468,13 +496,13 @@ async function statusResult(config: UniDeskConfig, options: TestTargetOptions, s
if ( blockers . length > 0 ) {
return renderResult ( { . . . planned , ok : false , mutation : false , remoteQueried : false , blockers , status : { remoteQueried : false , reason : "preflight-blocked" } } , options ) ;
}
const remote = await capture ( config , selection . target . route , [ "sh" ] , statusScript ( selection . target , context ) ) ;
const remote = await remoteCapture ( config , selection . target . route , [ "sh" ] , statusScript ( selection . target , context ) ) ;
const parsed = parseJsonOutput ( remote . stdout ) ;
const status = parsed === null ? { ok : false , remote : compactCapture ( remote , { full : true } ) } : summarizeRemoteStatus ( parsed , selection . target , context ) ;
return renderResult ( { . . . planned , ok : remote.exitCode === 0 && status . ok !== false , status : { remoteQueried : true , . . . status } } , options ) ;
return renderResult ( { . . . planned , ok : remote.exitCode === 0 && status . ok !== false , remoteQueried : true , status : { remoteQueried : true , . . . status } } , options ) ;
}
async function startResult ( config : UniDeskConfig , options : TestTargetOptions , selection : Selection ) : Promise < RenderedCliResult > {
async function startResult ( config : UniDeskConfig , options : TestTargetOptions , selection : Selection , remoteCapture : RemoteCapture ) : Promise < RenderedCliResult > {
const planned = planPayload ( options , selection ) ;
const context = selection . target === null ? null : renderContext ( options , selection . target ) ;
const blockers = safetyBlockers ( options , selection , context ) ;
@@ -485,24 +513,28 @@ async function startResult(config: UniDeskConfig, options: TestTargetOptions, se
}
const material = readSecretMaterial ( selection , selection . target ) ;
if ( ! material . ok ) {
return renderResult ( { . . . planned , ok : false , mutation : false , blockers : material.blockers , secret : { sources : material.sources , valuesPrinted : false } , failure : "secret-preflight-blocked" } , options ) ;
return renderResult ( { . . . planned , ok : false , mutation : false , remoteQueried : false , blockers : material.blockers , secret : { sources : material.sources , valuesPrinted : false } , failure : "secret-preflight-blocked" } , options ) ;
}
const manifest = buildManifest ( selection . target , context , material . values ) ;
const remote = await capture ( config , selection . target . route , [ "sh" ] , startScript ( selection . target , context , manifest ) ) ;
const remote = await remoteCapture ( config , selection . target . route , [ "sh" ] , startScript ( selection . target , context , manifest ) ) ;
const parsed = parseJsonOutput ( remote . stdout ) ;
const mutation = parsed ? . mutation === true ;
return renderResult ( {
. . . planned ,
ok : remote.exitCode === 0 && parsed ? . ok === true ,
mutation ,
mode : mutation ? "confirmed" : "preflight-blocked" ,
remoteQueried : true ,
mode : mutation ? "submitted" : parsed === null ? "submit-failed" : "existing-task" ,
code : parsed?.code ? ? "start-submit-unreadable" ,
jobId : parsed?.jobId ? ? asyncJobIdentity ( selection . target , context ) . id ,
task : parsed?.task ? ? null ,
secret : { sources : material.sources , valuesPrinted : false } ,
remote : parsed ? ? compactCapture ( remote , { full : true } ) ,
manifestFingerprint : structuralFingerprint ( manifest ) ,
} , options ) ;
}
async function stopResult ( config : UniDeskConfig , options : TestTargetOptions , selection : Selection ) : Promise < RenderedCliResult > {
async function stopResult ( config : UniDeskConfig , options : TestTargetOptions , selection : Selection , remoteCapture : RemoteCapture ) : Promise < RenderedCliResult > {
const planned = planPayload ( options , selection ) ;
const context = selection . target === null ? null : renderContext ( options , selection . target , false ) ;
const blockers = safetyBlockers ( options , selection , context , false ) ;
@@ -511,13 +543,17 @@ async function stopResult(config: UniDeskConfig, options: TestTargetOptions, sel
if ( selection . target === null || context === null || blockers . length > 0 ) {
return renderResult ( { . . . planned , ok : false , mutation : false , remoteQueried : false , blockers : uniqueBlockers ( blockers ) , failure : "preflight-blocked" } , options ) ;
}
const remote = await capture ( config , selection . target . route , [ "sh" ] , stopScript ( selection . target , context ) ) ;
const remote = await remoteCapture ( config , selection . target . route , [ "sh" ] , stopScript ( selection . target , context ) ) ;
const parsed = parseJsonOutput ( remote . stdout ) ;
return renderResult ( {
. . . planned ,
ok : remote.exitCode === 0 && parsed ? . ok === true ,
mutation : parsed?.mutation === true ,
remoteQueried : true ,
mode : parsed?.mutation === true ? "confirmed" : "no-change" ,
code : parsed?.code ? ? "stop-result-unreadable" ,
jobId : parsed?.jobId ? ? asyncJobIdentity ( selection . target , context ) . id ,
startTaskDisposition : parsed?.startTaskDisposition ? ? "unknown" ,
remote : parsed ? ? compactCapture ( remote , { full : true } ) ,
} , options ) ;
}
@@ -538,6 +574,12 @@ function renderContext(options: TestTargetOptions, target: TestTargetSpec, requi
} ;
}
function asyncJobIdentity ( target : TestTargetSpec , context : RenderContext ) : AsyncJobIdentity {
const digest = createHash ( "sha256" ) . update ( target . id ) . update ( "\0" ) . update ( context . instanceId ) . digest ( "hex" ) . slice ( 0 , 20 ) ;
const id = ` pikaoa- ${ digest } ` ;
return { id , stateDir : ` ${ target . taskState . rootPath } / ${ id } ` } ;
}
function safetyBlockers ( options : TestTargetOptions , selection : Selection , context : RenderContext | null , requireCommit = true ) : Blocker [ ] {
const blockers : Blocker [ ] = [ ] ;
const target = selection . target ;
@@ -771,6 +813,117 @@ function readSecretMaterial(selection: Selection, target: TestTargetSpec): {
}
function startScript ( target : TestTargetSpec , context : RenderContext , manifest : Record < string , unknown > [ ] ) : string {
const job = asyncJobIdentity ( target , context ) ;
const requestId = createHash ( "sha256" )
. update ( target . id )
. update ( "\0" )
. update ( context . instanceId )
. update ( "\0" )
. update ( context . commit )
. digest ( "hex" ) ;
const runnerEncoded = Buffer . from ( startRunnerScript ( target , context , manifest , job ) , "utf8" ) . toString ( "base64" ) ;
return `
set -u
umask 077
state_root= ${ shQuote ( target . taskState . rootPath ) }
state_dir= ${ shQuote ( job . stateDir ) }
status_file=" $ state_dir/status.json"
request_file=" $ state_dir/request-id"
pid_file=" $ state_dir/pid"
job_id= ${ shQuote ( job . id ) }
request_id= ${ shQuote ( requestId ) }
emit_existing() {
python3 - " $ request_file" " $ status_file" " $ pid_file" " $ job_id" " $ request_id" <<'PY'
import json, os, pathlib, sys
request_path, status_path, pid_path = map(pathlib.Path, sys.argv[1:4])
job_id, expected_request = sys.argv[4:6]
actual_request = request_path.read_text(errors="replace").strip() if request_path.exists() else ""
if actual_request != expected_request:
print(json.dumps({"ok": False, "mutation": False, "code": "start-instance-conflict", "jobId": job_id, "submitted": False, "task": None, "valuesPrinted": False}))
raise SystemExit(0)
try:
task = json.loads(status_path.read_text())
except Exception:
print(json.dumps({"ok": False, "mutation": False, "code": "start-state-unreadable", "jobId": job_id, "submitted": False, "task": None, "valuesPrinted": False}))
raise SystemExit(0)
state = str(task.get("state") or "unknown")
runner_alive = False
pid = task.get("pid")
if isinstance(pid, int) and pid > 0:
try:
os.kill(pid, 0)
runner_alive = True
except OSError:
pass
code_by_state = {
"queued": "start-already-running",
"running": "start-already-running",
"succeeded": "start-already-succeeded",
"failed": "start-already-failed",
"canceled": "start-already-canceled",
}
code = "start-worker-missing" if state in ("queued", "running") and not runner_alive else code_by_state.get(state, "start-state-unknown")
ok = state not in ("failed", "canceled", "unknown") and code != "start-worker-missing"
print(json.dumps({"ok": ok, "mutation": False, "code": code, "jobId": job_id, "submitted": False, "task": task, "valuesPrinted": False}))
PY
}
if ! mkdir -p " $ state_root"; then
printf '%s \ n' '{"ok":false,"mutation":false,"code":"task-state-root-create-failed","valuesPrinted":false}'
exit 41
fi
if ! mkdir " $ state_dir" 2>/dev/null; then
if [ -d " $ state_dir" ]; then emit_existing; exit 0; fi
printf '%s \ n' '{"ok":false,"mutation":false,"code":"task-state-create-failed","valuesPrinted":false}'
exit 42
fi
started_at=" $ (date -u +%Y-%m-%dT%H:%M:%SZ)"
printf '%s \ n' " $ request_id" >" $ request_file"
printf '%s \ n' " $ started_at" >" $ state_dir/started-at"
if ! printf '%s' ${ shQuote ( runnerEncoded ) } | base64 -d >" $ state_dir/job.sh"; then
rm -rf " $ state_dir"
printf '%s \ n' '{"ok":false,"mutation":false,"code":"task-runner-write-failed","valuesPrinted":false}'
exit 43
fi
chmod 0700 " $ state_dir/job.sh"
python3 - " $ status_file" " $ job_id" ${ shQuote ( target . id ) } ${ shQuote ( context . instanceId ) } ${ shQuote ( context . namespace ) } ${ shQuote ( context . commit ) } " $ started_at" <<'PY'
import json, os, pathlib, sys
path = pathlib.Path(sys.argv[1])
payload = {
"ok": True, "jobId": sys.argv[2], "targetId": sys.argv[3], "instanceId": sys.argv[4],
"namespace": sys.argv[5], "sourceCommit": sys.argv[6], "state": "running", "phase": "submitted",
"code": "start-running", "startedAt": sys.argv[7], "updatedAt": sys.argv[7], "finishedAt": None,
"exitCode": None, "pid": None, "valuesPrinted": False,
}
tmp = path.with_name(path.name + ".tmp")
tmp.write_text(json.dumps(payload, separators=(",", ":")) + " \\ n")
os.replace(tmp, path)
PY
if [ " $ ?" -ne 0 ]; then
rm -rf " $ state_dir"
printf '%s \ n' '{"ok":false,"mutation":false,"code":"task-state-initialize-failed","valuesPrinted":false}'
exit 44
fi
nohup sh " $ state_dir/job.sh" >/dev/null 2>&1 </dev/null &
pid= $ !
printf '%s \ n' " $ pid" >" $ pid_file"
python3 - " $ status_file" " $ pid" <<'PY'
import json, os, pathlib, sys
path = pathlib.Path(sys.argv[1])
try:
payload = json.loads(path.read_text())
except Exception:
raise SystemExit(0)
if payload.get("pid") is None and payload.get("phase") == "submitted":
payload["pid"] = int(sys.argv[2])
tmp = path.with_name(path.name + ".tmp.submit")
tmp.write_text(json.dumps(payload, separators=(",", ":")) + " \\ n")
os.replace(tmp, path)
PY
printf '{"ok":true,"mutation":true,"code":"start-submitted","submitted":true,"jobId":"%s","pid":%s,"task":{"state":"running","phase":"submitted","code":"start-running"},"valuesPrinted":false} \ n' " $ job_id" " $ pid"
` ;
}
function startRunnerScript ( target : TestTargetSpec , context : RenderContext , manifest : Record < string , unknown > [ ] , job : AsyncJobIdentity ) : string {
const migration = manifest . filter ( ( object ) = > object . kind === "Job" ) ;
const workloads = manifest . filter ( ( object ) = > object . kind === "Deployment" ) ;
const foundation = manifest . filter ( ( object ) = > object . kind !== "Job" && object . kind !== "Deployment" ) ;
@@ -778,132 +931,317 @@ function startScript(target: TestTargetSpec, context: RenderContext, manifest: R
const foundationEncoded = encode ( foundation ) ;
const migrationEncoded = encode ( migration ) ;
const workloadsEncoded = encode ( workloads ) ;
return `
set -u
return ` #!/bin/sh
set -e u
umask 077
state_dir= ${ shQuote ( job . stateDir ) }
status_file=" $ state_dir/status.json"
events_file=" $ state_dir/events.ndjson"
started_at=" $ (cat " $ state_dir/started-at")"
job_id= ${ shQuote ( job . id ) }
namespace= ${ shQuote ( context . namespace ) }
managed_by= ${ shQuote ( MANAGED_BY ) }
target_id= ${ shQuote ( target . id ) }
instance_id= ${ shQuote ( context . instanceId ) }
source_commit= ${ shQuote ( context . commit ) }
current_phase=runner-start
failure_code=start-runner-failed
tmp=""
write_status() {
state=" $ 1"; phase=" $ 2"; code=" $ 3"; exit_code=" $ 4"
now=" $ (date -u +%Y-%m-%dT%H:%M:%SZ)"
python3 - " $ status_file" " $ events_file" " $ job_id" " $ target_id" " $ instance_id" " $ namespace" " $ source_commit" " $ state" " $ phase" " $ code" " $ started_at" " $ now" " $ exit_code" " $ $ " <<'PY'
import json, os, pathlib, sys
status_path = pathlib.Path(sys.argv[1])
events_path = pathlib.Path(sys.argv[2])
state, phase, code = sys.argv[8:11]
exit_code = None if sys.argv[13] == "null" else int(sys.argv[13])
terminal = state in ("succeeded", "failed", "canceled")
payload = {
"ok": state in ("running", "succeeded"), "jobId": sys.argv[3], "targetId": sys.argv[4],
"instanceId": sys.argv[5], "namespace": sys.argv[6], "sourceCommit": sys.argv[7],
"state": state, "phase": phase, "code": code, "startedAt": sys.argv[11], "updatedAt": sys.argv[12],
"finishedAt": sys.argv[12] if terminal else None, "exitCode": exit_code, "pid": int(sys.argv[14]),
"valuesPrinted": False,
}
tmp = status_path.with_name(status_path.name + ".tmp." + sys.argv[14])
tmp.write_text(json.dumps(payload, separators=(",", ":")) + " \\ n")
os.replace(tmp, status_path)
event = {"at": sys.argv[12], "state": state, "phase": phase, "code": code, "valuesPrinted": False}
with events_path.open("a", encoding="utf-8") as handle:
handle.write(json.dumps(event, separators=(",", ":")) + " \\ n")
PY
}
check_canceled() {
if [ -f " $ state_dir/cancel-requested" ]; then failure_code=start-canceled; exit 130; fi
}
finish() {
rc= $ ?
trap - EXIT HUP INT TERM
if [ -f " $ state_dir/cancel-requested" ]; then
write_status canceled canceled start-canceled " $ rc" || true
elif [ " $ rc" -eq 0 ]; then
write_status succeeded completed started 0 || true
else
write_status failed " $ current_phase" " $ failure_code" " $ rc" || true
fi
[ -z " $ tmp" ] || rm -rf " $ tmp"
exit " $ rc"
}
trap finish EXIT
trap 'failure_code=start-runner-hangup; exit 129' HUP
trap 'failure_code=start-runner-interrupted; exit 130' INT
trap 'failure_code=start-runner-terminated; exit 143' TERM
rm -f " $ state_dir/job.sh"
tmp=" $ (mktemp -d)"
trap 'rm -rf " $ tmp"' EXIT
if kubectl get namespace " $ namespace" >/dev/null 2>&1; then
actual_managed=" $ (kubectl get namespace " $ namespace" -o jsonpath='{.metadata.labels.app \\ .kubernetes \\ .io/managed-by}' 2>/dev/null || true)"
actual_target=" $ ( kubectl get namespace "$ namespace" -o jsonpath='{.metadata.labels.pikaoa \\ .unidesk \\ .io/target}' 2 >/dev/null || true)"
actual_instance =" $ (kubectl get namespace " $ namespace" -o jsonpath='{.metadata.labels.pikaoa \\ .unidesk \\ .io/instance }' 2>/dev/null || true)"
current_phase=ownership-check
write_status running " $ current_phase" start-running null
check_canceled
if kubectl get namespace "$ namespace" --request-timeout= ${ target . taskState . statusRequestTimeoutSeconds } s >/dev/null 2>&1; then
actual_managed =" $ (kubectl get namespace " $ namespace" --request-timeout= ${ target . taskState . statusRequestTimeoutSeconds } s -o jsonpath='{.metadata.labels.app \\ .kubernetes \\ .io/managed-by }' 2>/dev/null || true)"
actual_target=" $ (kubectl get namespace " $ namespace" --request-timeout= ${ target . taskState . statusRequestTimeoutSeconds } s -o jsonpath='{.metadata.labels.pikaoa \\ .unidesk \\ .io/target}' 2>/dev/null || true)"
actual_instance=" $ (kubectl get namespace " $ namespace" --request-timeout= ${ target . taskState . statusRequestTimeoutSeconds } s -o jsonpath='{.metadata.labels.pikaoa \\ .unidesk \\ .io/instance}' 2>/dev/null || true)"
if [ " $ actual_managed" != " $ managed_by" ] || [ " $ actual_target" != " $ target_id" ] || [ " $ actual_instance" != " $ instance_id" ]; then
printf '%s \ n' '{"ok":false,"mutation":false," code":" namespace-ownership-mismatch","valuesPrinted":false}'
failure_ code= namespace-ownership-mismatch
exit 42
fi
fi
printf '%s' ${ shQuote ( foundationEncoded ) } | base64 -d >" $ tmp/foundation.yaml"
printf '%s' ${ shQuote ( migrationEncoded ) } | base64 -d >" $ tmp/migration.yaml"
printf '%s' ${ shQuote ( workloadsEncoded ) } | base64 -d >" $ tmp/workloads.yaml"
current_phase=foundation-apply
write_status running " $ current_phase" start-running null
check_canceled
if ! kubectl apply --server-side --force-conflicts --field-manager= ${ shQuote ( target . fieldManager ) } -f " $ tmp/foundation.yaml" >" $ tmp/apply.out" 2>" $ tmp/apply.err"; then
cat " $ tmp/apply.err" >&2
printf '%s \ n' '{"ok":false,"mutation":true,"code":"apply-failed","valuesPrinted":false}'
failure_code=apply-failed
exit 43
fi
wait_rc=0
kubectl -n " $ namespace" rollout status statefulset/pikaoa-postgres --timeout= ${ target . waitTimeoutSeconds } s >/dev/null 2>" $ tmp/wait.err" || wait_rc= $ ?
if [ " $ wait_rc" -ne 0 ]; then
cat "$ tmp/wait.err" >&2
printf '%s \ n' '{"ok":false,"mutation":true," code":" postgres-wait-failed","valuesPrinted":false}'
current_phase=postgres-rollout
write_status running " $ current_phase" start-running null
check_canceled
if ! kubectl -n " $ namespace" rollout status statefulset/pikaoa-postgres --timeout= ${ target . waitTimeoutSeconds } s >/dev/null 2> "$ tmp/wait.err"; then
failure_ code= postgres-wait-failed
exit 44
fi
current_phase=migration-apply
write_status running " $ current_phase" start-running null
check_canceled
if ! kubectl apply --server-side --force-conflicts --field-manager= ${ shQuote ( target . fieldManager ) } -f " $ tmp/migration.yaml" >" $ tmp/migration-apply.out" 2>" $ tmp/migration-apply.err"; then
cat " $ tmp/ migration-apply.err" >&2
printf '%s \ n' '{"ok":false,"mutation":true,"code":"migration-apply-failed","valuesPrinted":false}'
failure_code= migration-apply-failed
exit 45
fi
current_phase=migration-wait
write_status running " $ current_phase" start-running null
check_canceled
if ! kubectl -n " $ namespace" wait --for=condition=complete job/ ${ target . migration . jobName } --timeout= ${ target . waitTimeoutSeconds } s >/dev/null 2>" $ tmp/migration-wait.err"; then
cat " $ tmp/ migration-wait.err" >&2
printf '%s \ n' '{"ok":false,"mutation":true,"code":"migration-wait-failed","valuesPrinted":false}'
failure_code= migration-wait-failed
exit 46
fi
current_phase=workloads-apply
write_status running " $ current_phase" start-running null
check_canceled
if ! kubectl apply --server-side --force-conflicts --field-manager= ${ shQuote ( target . fieldManager ) } -f " $ tmp/workloads.yaml" >" $ tmp/workloads-apply.out" 2>" $ tmp/workloads-apply.err"; then
cat " $ tmp/ workloads-apply.err" >&2
printf '%s \ n' '{"ok":false,"mutation":true,"code":"workloads-apply-failed","valuesPrinted":false}'
failure_code= workloads-apply-failed
exit 47
fi
kubectl -n " $ namespace" rollout status deployment/pikaoa-api --timeout= ${ target . waitTimeoutSeconds } s >/dev/null 2>" $ tmp/wait.err" || wait_rc= $ ?
if [ " $ wa it_rc" -eq 0 ]; then kubectl -n " $ namespace" rollout status deployment/pikaoa-worker --timeout= ${ target . waitTimeoutSeconds } s >/dev/null 2>>" $ tmp/wait.err" || wait_rc= $ ?; fi
if [ " $ wait_rc" -ne 0 ]; then
cat "$ tmp/wait.err" >&2
printf '%s \ n' '{"ok":false,"mutation":true,"code":"rollout-wait-failed","valuesPrinted":false}'
current_phase=api-rollout
wr ite_status running " $ current_phase" start-running null
check_canceled
if ! kubectl -n " $ namespace" rollout status deployment/pikaoa-api --timeout= ${ target . waitTimeoutSeconds } s >/dev/null 2> "$ tmp/wait.err"; then
failure_code=api-rollout-wait-failed
exit 48
fi
printf '%s \ n' '{"ok":true,"mutation":true,"code":"started","valuesPrinted":false}'
current_phase=worker-rollout
write_status running " $ current_phase" start-running null
check_canceled
if ! kubectl -n " $ namespace" rollout status deployment/pikaoa-worker --timeout= ${ target . waitTimeoutSeconds } s >/dev/null 2>>" $ tmp/wait.err"; then
failure_code=worker-rollout-wait-failed
exit 49
fi
` ;
}
function statusScript ( target : TestTargetSpec , context : RenderContext ) : string {
const job = asyncJobIdentity ( target , context ) ;
return `
set -u
umask 077
state_dir= ${ shQuote ( job . stateDir ) }
job_id= ${ shQuote ( job . id ) }
namespace= ${ shQuote ( context . namespace ) }
if ! kubectl get namespace " $ namespace" >/dev/null 2>&1; then
printf '%s \ n' '{"ok":true,"exists":false,"mutation":false,"valuesPrinted":false}'
exit 0
tmp=" $ (mktemp -d)"
trap 'rm -rf " $ tmp"' EXIT
runtime_code=runtime-not-found
if kubectl get namespace " $ namespace" --request-timeout= ${ target . taskState . statusRequestTimeoutSeconds } s -o json >" $ tmp/namespace.json" 2>" $ tmp/namespace.err"; then
runtime_code=runtime-present
if ! kubectl -n " $ namespace" get deployment,statefulset,job,service,persistentvolumeclaim,pod --request-timeout= ${ target . taskState . statusRequestTimeoutSeconds } s -o json >" $ tmp/resources.json" 2>" $ tmp/resources.err"; then
runtime_code=runtime-resource-query-failed
fi
elif ! grep -Eiq 'not[[:space:]-]*found' " $ tmp/namespace.err"; then
runtime_code=runtime-query-failed
fi
printf '%s' '{"ok":true,"exists":true,"mutation":false,"namespace": '
kubectl get namespace " $ namespace" -o json
printf '%s' ',"resources":'
kubectl -n " $ namespace" get deployment,statefulset,job,service,persistentvolumeclaim,pod -o json
printf '%s \ n' ',"valuesPrinted":false}'
python3 - " $ state_dir" " $ job_id" " $ runtime_code" " $ tmp/namespace.json" " $ tmp/resources.json" ${ shQuote ( String ( target . taskState . logTailLines ) ) } <<'PY '
import json, os, pathlib, sys
state_dir = pathlib.Path(sys.argv[1])
job_id, runtime_code = sys.argv[2:4]
namespace_path, resources_path = map(pathlib.Path, sys.argv[4:6])
tail_lines = int(sys.argv[6])
status_path = state_dir / "status.json"
task = None
task_error = None
if status_path.exists():
try:
task = json.loads(status_path.read_text())
except Exception:
task_error = "start-state-unreadable"
pid = task.get("pid") if isinstance(task, dict) else None
runner_alive = False
if isinstance(pid, int) and pid > 0:
try:
os.kill(pid, 0)
runner_alive = True
except OSError:
pass
events = []
events_path = state_dir / "events.ndjson"
if events_path.exists():
for line in events_path.read_text(errors="replace").splitlines()[-tail_lines:]:
try:
item = json.loads(line)
except Exception:
continue
if isinstance(item, dict):
events.append({key: item.get(key) for key in ("at", "state", "phase", "code", "valuesPrinted")})
if task_error is not None:
code, ok, exists = task_error, False, True
elif task is None:
code, ok, exists = "start-not-found", True, False
else:
state = str(task.get("state") or "unknown")
if state in ("queued", "running") and not runner_alive:
code, ok = "start-worker-missing", False
else:
code = {"queued": "start-running", "running": "start-running", "succeeded": "start-succeeded", "failed": "start-failed", "canceled": "start-canceled"}.get(state, "start-state-unknown")
ok = state not in ("failed", "canceled", "unknown")
exists = True
runtime = {"queried": True, "exists": runtime_code in ("runtime-present", "runtime-resource-query-failed"), "code": runtime_code}
for key, path in (("namespace", namespace_path), ("resources", resources_path)):
if path.exists():
try:
runtime[key] = json.loads(path.read_text())
except Exception:
runtime["code"] = "runtime-response-unreadable"
payload = {"ok": ok, "mutation": False, "code": code, "jobId": job_id, "taskExists": exists, "task": task, "runnerAlive": runner_alive, "logTail": events, "runtime": runtime, "valuesPrinted": False}
print(json.dumps(payload, separators=(",", ":")))
PY
` ;
}
function stopScript ( target : TestTargetSpec , context : RenderContext ) : string {
const job = asyncJobIdentity ( target , context ) ;
return `
set -u
umask 077
state_dir= ${ shQuote ( job . stateDir ) }
job_id= ${ shQuote ( job . id ) }
namespace= ${ shQuote ( context . namespace ) }
managed_by= ${ shQuote ( MANAGED_BY ) }
target_id= ${ shQuote ( target . id ) }
instance_id= ${ shQuote ( context . instanceId ) }
if ! kubectl get namespace " $ namespace" >/dev/null 2>&1; then
printf '%s \ n' '{"ok":true,"mutation":false,"code":"already-absent","valuesPrinted":false} '
mark_cancel() {
python3 - " $ state_dir" <<'PY '
import json, pathlib, sys
state_dir = pathlib.Path(sys.argv[1])
status_path = state_dir / "status.json"
if not status_path.exists():
print("not-found|false")
raise SystemExit(0)
try:
state = str(json.loads(status_path.read_text()).get("state") or "unknown")
except Exception:
print("state-unreadable|false")
raise SystemExit(0)
if state in ("queued", "running"):
cancel_path = state_dir / "cancel-requested"
if cancel_path.exists():
print("cancel-already-requested|false")
else:
cancel_path.touch(mode=0o600, exist_ok=False)
print("cancel-requested|true")
else:
print("already-terminal|false")
PY
}
tmp=" $ (mktemp -d)"
trap 'rm -rf " $ tmp"' EXIT
if ! kubectl get namespace " $ namespace" --request-timeout= ${ target . taskState . statusRequestTimeoutSeconds } s -o json >" $ tmp/namespace.json" 2>" $ tmp/namespace.err"; then
if ! grep -Eiq 'not[[:space:]-]*found' " $ tmp/namespace.err"; then
printf '{"ok":false,"mutation":false,"code":"runtime-query-failed","jobId":"%s","startTaskDisposition":"unknown","valuesPrinted":false} \ n' " $ job_id"
exit 45
fi
cancel_result=" $ (mark_cancel)"
cancel_disposition=" \ ${ cancel_result % % | * } "
cancel_mutation=" \ ${ cancel_result # # * | } "
printf '{"ok":true,"mutation":%s,"code":"already-absent","jobId":"%s","startTaskDisposition":"%s","valuesPrinted":false} \ n' " $ cancel_mutation" " $ job_id" " $ cancel_disposition"
exit 0
fi
actual_managed=" $ (kubectl get namespace " $ namespace" -o jsonpath='{.metadata.labels.app \\ .kubernetes \\ .io/managed-by}' 2>/dev/null || true)"
actual_target=" $ (kubectl get namespace " $ namespace" -o jsonpath='{.metadata.labels.pikaoa \\ .unidesk \\ .io/target}' 2>/dev/null || true)"
actual_instance=" $ (kubectl get namespace " $ namespace" -o jsonpath='{.metadata.labels.pikaoa \\ .unidesk \\ .io/instance}' 2>/dev/null || true)"
actual_managed=" $ (kubectl get namespace " $ namespace" --request-timeout= ${ target . taskState . statusRequestTimeoutSeconds } s -o jsonpath='{.metadata.labels.app\\ .kubernetes \\ .io/managed-by}' 2>/dev/null || true)"
actual_target=" $ (kubectl get namespace " $ namespace" --request-timeout= ${ target . taskState . statusRequestTimeoutSeconds } s -o jsonpath='{.metadata.labels.pikaoa\\ .unidesk \\ .io/target}' 2>/dev/null || true)"
actual_instance=" $ (kubectl get namespace " $ namespace" --request-timeout= ${ target . taskState . statusRequestTimeoutSeconds } s -o jsonpath='{.metadata.labels.pikaoa\\ .unidesk \\ .io/instance}' 2>/dev/null || true)"
if [ " $ actual_managed" != " $ managed_by" ] || [ " $ actual_target" != " $ target_id" ] || [ " $ actual_instance" != " $ instance_id" ]; then
printf '%s \ n' ' {"ok":false,"mutation":false,"code":"namespace-ownership-mismatch","valuesPrinted":false}'
printf '{"ok":false,"mutation":false,"code":"namespace-ownership-mismatch","jobId":"%s","startTaskDisposition":"not-touched","valuesPrinted":false} \ n' " $ job_id"
exit 42
fi
kubectl delete namespace " $ namespace" --wait=false >/dev/null
printf '%s \ n' '{"ok":true,"mutation":true,"code":"stop-requested","valuesPrinted":false}'
cancel_result=" $ (mark_cancel)"
cancel_disposition=" \ ${ cancel_result % % | * } "
cancel_mutation=" \ ${ cancel_result # # * | } "
if ! kubectl delete namespace " $ namespace" --wait=false --request-timeout= ${ target . taskState . statusRequestTimeoutSeconds } s >/dev/null 2>" $ tmp/delete.err"; then
printf '{"ok":false,"mutation":%s,"code":"namespace-delete-failed","jobId":"%s","startTaskDisposition":"%s","valuesPrinted":false} \ n' " $ cancel_mutation" " $ job_id" " $ cancel_disposition"
exit 43
fi
printf '{"ok":true,"mutation":true,"code":"stop-requested","jobId":"%s","startTaskDisposition":"%s","valuesPrinted":false} \ n' " $ job_id" " $ cancel_disposition"
` ;
}
function summarizeRemoteStatus ( parsed : Record < string , unknown > , target : TestTargetSpec , context : RenderContext ) : Record < string , unknown > {
if ( parsed . exists !== true ) return { ok : parsed.ok === true , exists : false , mutation : false , valuesPrinted : false } ;
const namespace = optionalRecord ( parsed . namespace , "status.namespace" ) ? ? { } ;
const runtime = optionalRecord ( parsed . runtime , "status.runtime" ) ? ? { } ;
const namespace = optionalRecord ( runtime . namespace , "status.runtime. namespace" ) ? ? { } ;
const metadata = optionalRecord ( namespace . metadata , "status.namespace.metadata" ) ? ? { } ;
const labels = optionalRecord ( metadata . labels , "status.namespace.metadata.labels" ) ? ? { } ;
const resources = optionalRecord ( parsed . resources , "status.resources" ) ? ? { } ;
const resources = optionalRecord ( runtime . resources , "status.runtime. resources" ) ? ? { } ;
const items = Array . isArray ( resources . items ) ? resources . items . filter ( isRecord ) : [ ] ;
return {
ok : parsed.ok === true ,
exists : true ,
code : parsed.code ? ? "start-status-unknown" ,
jobId : parsed.jobId ? ? asyncJobIdentity ( target , context ) . id ,
taskExists : parsed.taskExists === true ,
task : optionalRecord ( parsed . task , "status.task" ) ,
runnerAlive : parsed.runnerAlive === true ,
logTail : Array.isArray ( parsed . logTail ) ? parsed . logTail . filter ( isRecord ) : [ ] ,
mutation : false ,
ownership : {
valid : labels [ "app.kubernetes.io/managed-by" ] === MANAGED_BY && labels [ TARGET_LABEL ] === target . id && labels [ INSTANCE_LABEL ] === context . instanceId ,
managedBy : labels [ "app.kubernetes.io/managed-by" ] ? ? null ,
target : labels [ TARGET_LABEL ] ? ? null ,
instance : labels [ INSTANCE_LABEL ] ? ? null ,
runtime : {
queried : runtime.queried === true ,
exists : runtime.exists === true ,
code : runtime.code ? ? "runtime-status-unknown" ,
ownership : runtime.exists !== true ? null : {
valid : labels [ "app.kubernetes.io/managed-by" ] === MANAGED_BY && labels [ TARGET_LABEL ] === target . id && labels [ INSTANCE_LABEL ] === context . instanceId ,
managedBy : labels [ "app.kubernetes.io/managed-by" ] ? ? null ,
target : labels [ TARGET_LABEL ] ? ? null ,
instance : labels [ INSTANCE_LABEL ] ? ? null ,
} ,
resources : items.map ( ( item ) = > {
const itemMetadata = optionalRecord ( item . metadata , "resource.metadata" ) ? ? { } ;
const status = optionalRecord ( item . status , "resource.status" ) ? ? { } ;
const spec = optionalRecord ( item . spec , "resource.spec" ) ? ? { } ;
return {
kind : item.kind ? ? null ,
name : itemMetadata.name ? ? null ,
desired : spec.replicas ? ? null ,
ready : status.readyReplicas ? ? status . numberReady ? ? null ,
phase : status.phase ? ? null ,
} ;
} ) ,
} ,
resources : items.map ( ( item ) = > {
const itemMetadata = optionalRecord ( item . metadata , "resource.metadata" ) ? ? { } ;
const status = optionalRecord ( item . status , "resource.status" ) ? ? { } ;
const spec = optionalRecord ( item . spec , "resource.spec" ) ? ? { } ;
return {
kind : item.kind ? ? null ,
name : itemMetadata.name ? ? null ,
desired : spec.replicas ? ? null ,
ready : status.readyReplicas ? ? status . numberReady ? ? null ,
phase : status.phase ? ? null ,
} ;
} ) ,
valuesPrinted : false ,
} ;
}
@@ -926,12 +1264,14 @@ function renderText(payload: Record<string, unknown>): string {
const warnings = Array . isArray ( payload . warnings ) ? payload . warnings . filter ( isRecord ) : [ ] ;
const blockers = Array . isArray ( payload . blockers ) ? payload . blockers . filter ( isRecord ) : [ ] ;
const status = optionalRecord ( payload . status , "status" ) ;
const task = status === null ? null : optionalRecord ( status . task , "status.task" ) ;
return [
` PIKAOA TEST TARGET ${ String ( payload . action ? ? "" ) . split ( " " ) . at ( - 1 ) ? . toUpperCase ( ) ? ? "" } ( ${ payload . ok === false ? "blocked" : "ok" } ) ` ,
` configured= ${ String ( payload . configured === true ) } mutation= ${ String ( payload . mutation === true ) } config= ${ String ( payload . configPath ? ? "-" ) } ` ,
` target= ${ String ( target ? . id ? ? "-" ) } node= ${ String ( target ? . node ? ? "-" ) } validationOnly= ${ String ( target ? . validationOnly ? ? "-" ) } ` ,
` instance= ${ String ( instance ? . id ? ? "-" ) } namespace= ${ String ( instance ? . namespace ? ? "-" ) } ` ,
. . . ( status === null ? [ ] : [ ` status= ${ status . remoteQueried === false ? String ( status . reason ? ? "local" ) : status . exists === false ? "absent" : status . ok === false ? "failed" : "present" } ` ] ) ,
. . . ( payload . code === undefined ? [ ] : [ ` code= ${ String ( payload . code ) } jobId= ${ String ( payload . jobId ? ? "-" ) } ` ] ) ,
. . . ( status === null ? [ ] : [ ` status= ${ status . remoteQueried === false ? String ( status . reason ? ? "local" ) : String ( status . code ? ? "unknown" ) } phase= ${ String ( task ? . phase ? ? "-" ) } ` ] ) ,
` warnings= ${ warnings . length } blockers= ${ blockers . length } valuesPrinted=false ` ,
. . . warnings . map ( ( item ) = > ` WARNING ${ String ( item . code ? ? "warning" ) } blocking=false ${ String ( item . message ? ? "" ) } ` ) ,
. . . blockers . map ( ( item ) = > ` BLOCKER ${ String ( item . code ? ? "blocked" ) } field= ${ String ( item . field ? ? "-" ) } ${ String ( item . message ? ? "" ) } ` ) ,
@@ -950,6 +1290,7 @@ function targetSummary(target: TestTargetSpec): Record<string, unknown> {
validationOnly : target.validationOnly ,
namespacePrefix : target.namespacePrefix ,
ttlSeconds : target.ttlSeconds ,
taskState : target.taskState ,
cleanup : target.cleanup ,
} ;
}
@@ -1172,6 +1513,12 @@ function absolutePath(value: unknown, path: string): string {
return result ;
}
function taskStateRoot ( value : unknown , path : string ) : string {
const result = absolutePath ( value , path ) . replace ( / \ / + $ / u , " " ) ;
if ( result . length === 0 ) throw new Error ( ` ${ path } 不能是文件系统根目录 ` ) ;
return result ;
}
function nonEmptyStringList ( value : unknown , path : string ) : string [ ] {
if ( ! Array . isArray ( value ) || value . length === 0 ) throw new Error ( ` ${ path } 必须是非空字符串数组 ` ) ;
return value . map ( ( item , index ) = > nonEmpty ( item , ` ${ path } [ ${ index } ] ` ) ) ;