Support declared graceful stop and steam.update probe

- LifecycleActionTemplate 新增 gracefulStop:stop 动作可声明插件自有的优雅关闭命令、
  参数、环境、超时与 fallback;超时且 fallback=report 时任务失败,避免默认杀进程。
- 依赖探针新增 steam.update:通过 steamcmd +app_info_print 读取公开分支 buildid,
  与本地 appmanifest_<appid>.acf 比较,输出 installed/latest/update=yes|no|unknown。
- 依赖执行输入新增 ServerRoot,供探针定位插件声明的服务器安装目录。
This commit is contained in:
npc0-hue
2026-09-15 13:31:36 +08:00
parent a695aafd0f
commit 4e88199d18
7 changed files with 464 additions and 26 deletions
+5 -1
View File
@@ -193,7 +193,11 @@ func (worker *Worker) runAutonomousDependencies(ctx context.Context, plan protoc
return nil
}
for _, probe := range plan.DependencyProbes {
state, evidence, err := worker.executor.runDependencyProbe(ctx, probe, plan.RuntimeBindings)
installRoot := ""
if plan.Deployment != nil {
installRoot = plan.Deployment.ServerRoot
}
state, evidence, err := worker.executor.runDependencyProbe(ctx, probe, plan.RuntimeBindings, installRoot)
if err != nil {
if probe.Required {
log.Printf("RUN phase=autonomous_lifecycle.dependencies status=failed probe=%s error=%s", safeOptional(probe.Key), err.Error())
+126 -3
View File
@@ -30,6 +30,8 @@ const (
var (
dependencyTokenPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9_.:+@/-]{0,119}$`)
steamAppIDPattern = regexp.MustCompile(`^[0-9]{1,10}$`)
steamBuildIDLinePattern = regexp.MustCompile(`^"buildid"\s+"([0-9]{1,20})"$`)
dependencyVersionPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9_.:+~-]{0,79}$`)
)
@@ -139,7 +141,7 @@ func (worker *Worker) executeDependencyJob(ctx context.Context, assignment proto
}
if assignment.Capability == protocol.RunCapabilityDependenciesCheck {
state, evidence, probeErr := worker.executor.runDependencyProbe(ctx, input.Probe, input.Bindings)
state, evidence, probeErr := worker.executor.runDependencyProbe(ctx, input.Probe, input.Bindings, input.ServerRoot)
if probeErr != nil {
if errors.Is(probeErr, context.Canceled) || errors.Is(probeErr, context.DeadlineExceeded) {
return LifecycleExecutionResult{State: lifecycleResultStateCancelled, Progress: protocol.RunJobProgressReport{Percent: 100, Message: "dependency probe cancelled"}, Message: "dependency probe cancelled", ErrorCode: "dependency_probe_cancelled"}
@@ -178,7 +180,7 @@ func (worker *Worker) executeDependencyJob(ctx context.Context, assignment proto
return lifecycleFailure("dependency_journal_failed", err.Error())
}
}
state, evidence, probeErr := worker.executor.runDependencyProbe(ctx, input.Probe, input.Bindings)
state, evidence, probeErr := worker.executor.runDependencyProbe(ctx, input.Probe, input.Bindings, input.ServerRoot)
if probeErr != nil {
return lifecycleFailure("dependency_verify_failed", probeErr.Error())
}
@@ -206,6 +208,11 @@ func validateDependencyInput(assignment protocol.RunJobAssignment, input protoco
if !protocol.ValidLogicalFileKey(input.Probe.Key) || !protocol.ValidLogicalFileKey(input.Probe.TargetKey) {
return fmt.Errorf("dependency probe is unsafe")
}
if root := strings.TrimSpace(input.ServerRoot); root != "" {
if !filepath.IsAbs(root) || strings.ContainsRune(root, 0) {
return fmt.Errorf("dependency install root is unsafe")
}
}
if assignment.Capability == protocol.RunCapabilityDependenciesCheck {
if assignment.TargetKey != "dependencies/"+input.Probe.Key || input.Plan.Key != "" {
return fmt.Errorf("dependency check declaration does not match job")
@@ -232,7 +239,7 @@ func validDependencyInstallStepType(value string) bool {
}
}
func (executor LifecycleExecutor) runDependencyProbe(ctx context.Context, probe protocol.DependencyProbe, bindings map[string]string) (string, string, error) {
func (executor LifecycleExecutor) runDependencyProbe(ctx context.Context, probe protocol.DependencyProbe, bindings map[string]string, installRoot string) (string, string, error) {
target := strings.TrimSpace(bindings[probe.TargetKey])
if target == "" {
target = probe.TargetKey
@@ -262,6 +269,8 @@ func (executor LifecycleExecutor) runDependencyProbe(ctx context.Context, probe
return "missing", "declared package executable is not present", nil
}
return "present", "declared package executable is present", nil
case "steam.update":
return executor.runSteamUpdateProbe(ctx, probe, target, installRoot)
case "command.version", "java.version", "docker.available":
if err := validateDependencyExecutable(target); err != nil {
return "", "", err
@@ -303,6 +312,120 @@ func (executor LifecycleExecutor) runDependencyProbe(ctx context.Context, probe
}
}
// runSteamUpdateProbe compares the installed Steam app build with the build
// published on the app's public branch. The declared probe target is the
// SteamCMD executable, the app id is plugin-declared, and the installed app
// manifest is read from the server's declared install root before falling back
// to the SteamCMD directory. All three values are generic Steam facts, so no
// game-specific behavior lives here.
func (executor LifecycleExecutor) runSteamUpdateProbe(ctx context.Context, probe protocol.DependencyProbe, target string, installRoot string) (string, string, error) {
if err := validateDependencyExecutable(target); err != nil {
return "", "", err
}
if !steamAppIDPattern.MatchString(probe.SteamAppID) {
return "", "", fmt.Errorf("dependency steam app id is invalid")
}
executable := target
if !filepath.IsAbs(executable) {
resolved, err := exec.LookPath(executable)
if err != nil {
return "missing", "declared SteamCMD executable is not available", nil
}
executable = resolved
}
installed, installedKnown := readInstalledSteamBuildID(steamInstallRoots(executable, installRoot), probe.SteamAppID)
result, err := executor.dependencyRunner.Run(ctx, ProcessCommand{
Args: []string{executable, "+login", "anonymous", "+app_info_update", "1", "+app_info_print", probe.SteamAppID, "+quit"},
Timeout: 60 * time.Second,
Capability: "dependency.probe",
Action: probe.Kind,
})
if err != nil || result.ExitCode != 0 {
if ctx.Err() != nil {
return "", "", ctx.Err()
}
return "missing", "declared SteamCMD app info query is not available", nil
}
latest, latestKnown := parseSteamPublicBuildID(processOutputEvidence(result.Stdout, result.Stderr))
switch {
case !installedKnown && !latestKnown:
return "unknown", "installed and published Steam builds are unknown", nil
case !installedKnown:
return "unknown", fmt.Sprintf("installed=none latest=%s update=yes", latest), nil
case !latestKnown:
return "unknown", fmt.Sprintf("installed=%s latest=none update=unknown", installed), nil
case installed == latest:
return "present", fmt.Sprintf("installed=%s latest=%s update=no", installed, latest), nil
default:
return "present", fmt.Sprintf("installed=%s latest=%s update=yes", installed, latest), nil
}
}
// steamInstallRoots lists the directories that can hold the installed Steam
// app manifest, most specific first. SteamCMD writes the manifest into the app
// install root when a force_install_dir is used, and into its own directory
// otherwise.
func steamInstallRoots(steamcmdExecutable string, installRoot string) []string {
roots := make([]string, 0, 2)
if trimmed := strings.TrimSpace(installRoot); trimmed != "" && filepath.IsAbs(trimmed) {
roots = append(roots, filepath.Clean(trimmed))
}
return append(roots, filepath.Dir(steamcmdExecutable))
}
func readInstalledSteamBuildID(roots []string, appID string) (string, bool) {
for _, root := range roots {
body, err := os.ReadFile(filepath.Join(root, "steamapps", "appmanifest_"+appID+".acf"))
if err != nil {
continue
}
if buildID, ok := steamBuildIDFromManifest(string(body)); ok {
return buildID, true
}
}
return "", false
}
func steamBuildIDFromManifest(body string) (string, bool) {
for _, line := range strings.Split(body, "\n") {
trimmed := strings.TrimSpace(line)
matches := steamBuildIDLinePattern.FindStringSubmatch(trimmed)
if matches == nil {
continue
}
return matches[1], true
}
return "", false
}
// parseSteamPublicBuildID reads the build id of the app's public branch from
// SteamCMD `app_info_print` output. Depot sections carry their own build ids,
// so the parser only accepts the build id that directly follows the public
// branch entry.
func parseSteamPublicBuildID(output string) (string, bool) {
inPublicBranch := false
for _, line := range strings.Split(output, "\n") {
trimmed := strings.TrimSpace(line)
if trimmed == "" {
continue
}
if trimmed == "\"public\"" {
inPublicBranch = true
continue
}
if !inPublicBranch {
continue
}
if matches := steamBuildIDLinePattern.FindStringSubmatch(trimmed); matches != nil {
return matches[1], true
}
if trimmed != "{" {
inPublicBranch = false
}
}
return "", false
}
func (executor LifecycleExecutor) runDependencyInstallStep(ctx context.Context, assignment protocol.RunJobAssignment, input protocol.DependencyExecutionInputResponse, step protocol.DependencyInstallStep, index int) error {
if !protocol.ValidLogicalFileKey(step.TargetKey) {
return fmt.Errorf("dependency install target is unsafe")
+72 -1
View File
@@ -92,7 +92,7 @@ func TestDependencyInstallExecutesTypedPlanAndResumesCompletedSteps(t *testing.T
first := worker.executeDependencyJob(context.Background(), assignment)
if first.State != lifecycleResultStateSucceeded || first.ExecutionResult.Kind != "dependency.install" || first.ExecutionResult.Checksum != dependencyTestDigest {
state, evidence, probeErr := worker.executor.runDependencyProbe(context.Background(), client.dependencyInput.Probe, client.dependencyInput.Bindings)
state, evidence, probeErr := worker.executor.runDependencyProbe(context.Background(), client.dependencyInput.Probe, client.dependencyInput.Bindings, client.dependencyInput.ServerRoot)
t.Fatalf("expected real typed dependency install, got %+v calls=%+v present=%v probe=%s evidence=%s err=%v", first, runner.calls, runner.present, state, evidence, probeErr)
}
second := worker.executeDependencyJob(context.Background(), assignment)
@@ -165,6 +165,77 @@ func TestDependencyInstallRejectsLegacySteamCMDAppAdapter(t *testing.T) {
}
}
type steamProbeSupervisor struct {
output string
exit int
calls [][]string
}
func (supervisor *steamProbeSupervisor) Run(_ context.Context, command ProcessCommand) (ProcessResult, error) {
supervisor.calls = append(supervisor.calls, append([]string(nil), command.Args...))
return ProcessResult{ExitCode: supervisor.exit, Stdout: supervisor.output}, nil
}
func TestSteamUpdateProbeReportsInstalledAndPublishedBuilds(t *testing.T) {
dir := t.TempDir()
steamcmdDir := filepath.Join(dir, "steamcmd")
if err := os.MkdirAll(steamcmdDir, 0o700); err != nil {
t.Fatalf("create steamcmd fixture dir: %v", err)
}
steamcmd := filepath.Join(steamcmdDir, "steamcmd.exe")
if err := os.WriteFile(steamcmd, []byte("steamcmd fixture"), 0o700); err != nil {
t.Fatalf("write steamcmd fixture: %v", err)
}
installRoot := filepath.Join(dir, "scumserver")
if err := os.MkdirAll(filepath.Join(installRoot, "steamapps"), 0o700); err != nil {
t.Fatalf("create install root fixture: %v", err)
}
manifest := "\"AppState\"\n{\n\t\"appid\"\t\t\"3792580\"\n\t\"buildid\"\t\t\"16912763\"\n}\n"
if err := os.WriteFile(filepath.Join(installRoot, "steamapps", "appmanifest_3792580.acf"), []byte(manifest), 0o600); err != nil {
t.Fatalf("write appmanifest fixture: %v", err)
}
published := "AppID : 3792580, change number : 1\n\"depots\"\n{\n\t\"buildid\"\t\t\"99999999\"\n}\n\"branches\"\n{\n\t\"public\"\n\t{\n\t\t\"buildid\"\t\t\"16919999\"\n\t}\n}\n"
runner := &steamProbeSupervisor{output: published}
executor := NewLifecycleExecutor(WithLifecycleWorkspaceRoot(t.TempDir()), WithDependencyCommandRunner(runner))
probe := protocol.DependencyProbe{Key: "scum-server-build", Kind: "steam.update", TargetKey: "steamcmd", SteamAppID: "3792580"}
state, evidence, err := executor.runDependencyProbe(context.Background(), probe, map[string]string{"steamcmd": steamcmd}, installRoot)
if err != nil {
t.Fatalf("steam update probe failed: %v", err)
}
if state != "present" || evidence != "installed=16912763 latest=16919999 update=yes" {
t.Fatalf("unexpected steam update evidence: state=%s evidence=%s", state, evidence)
}
if len(runner.calls) != 1 || !strings.Contains(strings.Join(runner.calls[0], " "), "+app_info_print 3792580") {
t.Fatalf("unexpected steamcmd invocation: %+v", runner.calls)
}
runner.output = strings.Replace(published, "16919999", "16912763", 1)
state, evidence, err = executor.runDependencyProbe(context.Background(), probe, map[string]string{"steamcmd": steamcmd}, installRoot)
if err != nil {
t.Fatalf("steam update probe failed: %v", err)
}
if state != "present" || evidence != "installed=16912763 latest=16912763 update=no" {
t.Fatalf("unexpected up-to-date evidence: state=%s evidence=%s", state, evidence)
}
if err := os.RemoveAll(filepath.Join(installRoot, "steamapps")); err != nil {
t.Fatalf("remove install root manifest: %v", err)
}
state, evidence, err = executor.runDependencyProbe(context.Background(), probe, map[string]string{"steamcmd": steamcmd}, "")
if err != nil {
t.Fatalf("steam update probe without install root failed: %v", err)
}
if state != "unknown" || evidence != "installed=none latest=16912763 update=yes" {
t.Fatalf("unexpected unknown installed build evidence: state=%s evidence=%s", state, evidence)
}
probe.SteamAppID = "3792580;rm"
if _, _, err := executor.runDependencyProbe(context.Background(), probe, map[string]string{"steamcmd": steamcmd}, installRoot); err == nil {
t.Fatal("expected invalid steam app id rejection")
}
}
func dependencyAssignment(capability string) protocol.RunJobAssignment {
assignment := workerJobAssignment(capability)
assignment.LeaseToken = "lease-dependency"
+112
View File
@@ -7,6 +7,7 @@ import (
"os"
"os/exec"
"path/filepath"
"strconv"
"strings"
"testing"
"time"
@@ -50,6 +51,17 @@ func TestRunHelperProcess(t *testing.T) {
time.Sleep(100 * time.Millisecond)
}
}
if pidText := os.Getenv("RUN_GRACEFUL_STOP_PID"); pidText != "" {
// Stands in for a game-owned graceful shutdown: the plugin-declared
// stop command asks the running game to exit instead of Run
// terminating the supervised process.
if target, convErr := strconv.Atoi(pidText); convErr == nil {
if process, findErr := os.FindProcess(target); findErr == nil {
_ = process.Kill()
}
}
return
}
if os.Getenv("RUN_EXIT_NOW") == "1" {
return
}
@@ -132,6 +144,106 @@ func TestTypedProcessOutputCaptureResumesAfterRunRestart(t *testing.T) {
waitForManagedTailers(t, restarted.managed.(*OSManagedProcessSupervisor))
}
func TestGracefulStopRunsDeclaredShutdownBeforeTermination(t *testing.T) {
root := t.TempDir()
start := executionAssignment(protocol.RunCapabilityProcessStart)
setupProcessWorkspace(t, root, start, false)
scope := processScope(root, start)
executor := NewLifecycleExecutor(WithLifecycleWorkspaceRoot(root), WithProcessLogSink(&recordingLogSink{}))
if result := executor.Execute(start); result.State != lifecycleResultStateSucceeded {
t.Fatalf("start managed process: %+v", result)
}
managed := executor.managed.(*OSManagedProcessSupervisor)
identity := managed.Status(ProcessIdentity{Scope: scope})
writeJSONFixture(t, filepath.Join(scope, "actions", "stop.json"), map[string]any{
"version": 1,
"action": "stop",
"mode": "control",
"stopTimeoutMs": 30000,
"gracefulStop": map[string]any{
"executableKey": "bin/game-server",
"arguments": []string{"-test.run=TestRunHelperProcess"},
"environment": map[string]string{"RUN_TEST_HELPER": "1", "RUN_GRACEFUL_STOP_PID": strconv.Itoa(identity.PID)},
"timeoutMs": 30000,
"fallback": "report",
},
})
stop := executionAssignment(protocol.RunCapabilityProcessStop)
stop.TargetKey = "actions/stop.json"
result := executor.Execute(stop)
if result.State != lifecycleResultStateSucceeded {
t.Fatalf("graceful stop result: %+v", result)
}
if result.ExecutionResult.ProcessState != "stopped" {
t.Fatalf("graceful stop process state: %+v", result.ExecutionResult)
}
if result.ExecutionResult.ExitClassification == "forced-stop" {
t.Fatalf("declared graceful stop terminated the process: %+v", result.ExecutionResult)
}
if _, err := os.FindProcess(identity.PID); err == nil && processAliveForTest(identity.PID) {
t.Fatalf("declared shutdown did not stop process %d", identity.PID)
}
}
func TestGracefulStopReportsWhenDeclaredShutdownDoesNotStopProcess(t *testing.T) {
root := t.TempDir()
start := executionAssignment(protocol.RunCapabilityProcessStart)
setupProcessWorkspace(t, root, start, false)
scope := processScope(root, start)
executor := NewLifecycleExecutor(WithLifecycleWorkspaceRoot(root), WithProcessLogSink(&recordingLogSink{}))
if result := executor.Execute(start); result.State != lifecycleResultStateSucceeded {
t.Fatalf("start managed process: %+v", result)
}
managed := executor.managed.(*OSManagedProcessSupervisor)
identity := managed.Status(ProcessIdentity{Scope: scope})
writeStopAction := func(fallback string) {
writeJSONFixture(t, filepath.Join(scope, "actions", "stop.json"), map[string]any{
"version": 1,
"action": "stop",
"mode": "control",
"stopTimeoutMs": 30000,
"gracefulStop": map[string]any{
"executableKey": "bin/game-server",
"arguments": []string{"-test.run=TestRunHelperProcess"},
"environment": map[string]string{"RUN_TEST_HELPER": "1", "RUN_LOG_LINES": "1"},
"timeoutMs": 30000,
"fallback": fallback,
},
})
}
stop := executionAssignment(protocol.RunCapabilityProcessStop)
stop.TargetKey = "actions/stop.json"
t.Cleanup(func() {
writeStopAction("terminate")
_ = executor.Execute(stop)
})
writeStopAction("report")
reported := executor.Execute(stop)
if reported.State != lifecycleResultStateFailed || reported.ErrorCode != "graceful_stop_timeout" {
t.Fatalf("expected unverified graceful stop failure, got %+v", reported)
}
if current := managed.Status(ProcessIdentity{Scope: scope}); current.State != "running" {
t.Fatalf("report-only graceful stop terminated the process: %+v", current)
}
writeStopAction("terminate")
terminated := executor.Execute(stop)
if terminated.State != lifecycleResultStateSucceeded || terminated.ExecutionResult.ProcessState != "stopped" {
t.Fatalf("expected declared fallback termination, got %+v", terminated)
}
if current := managed.Status(ProcessIdentity{Scope: scope}); current.State == "running" {
t.Fatalf("declared fallback termination left the process running: %+v", current)
}
_ = identity
}
func processAliveForTest(pid int) bool {
return processAlivePID(pid)
}
func TestManagedProcessOutputAfterCanceledRunContextIsNotSpooledOnRestart(t *testing.T) {
root := t.TempDir()
assignment := executionAssignment(protocol.RunCapabilityProcessStart)
+134 -20
View File
@@ -534,18 +534,32 @@ func (executor LifecycleExecutor) writeProcessLogs(ctx context.Context, assignme
}
type LifecycleActionTemplate struct {
Version int `json:"version,omitempty"`
Action string `json:"action,omitempty"`
Mode string `json:"mode,omitempty"`
ExecutableKey string `json:"executableKey,omitempty"`
TargetExecutableKey string `json:"targetExecutableKey,omitempty"`
Arguments []string `json:"arguments,omitempty"`
Environment map[string]string `json:"environment,omitempty"`
OutputMode string `json:"outputMode,omitempty"`
StopTimeoutMS int `json:"stopTimeoutMs,omitempty"`
Command []string `json:"command"`
Env map[string]string `json:"env,omitempty"`
TimeoutMS int `json:"timeoutMs,omitempty"`
Version int `json:"version,omitempty"`
Action string `json:"action,omitempty"`
Mode string `json:"mode,omitempty"`
ExecutableKey string `json:"executableKey,omitempty"`
TargetExecutableKey string `json:"targetExecutableKey,omitempty"`
Arguments []string `json:"arguments,omitempty"`
Environment map[string]string `json:"environment,omitempty"`
OutputMode string `json:"outputMode,omitempty"`
StopTimeoutMS int `json:"stopTimeoutMs,omitempty"`
GracefulStop *GracefulStopActionTemplate `json:"gracefulStop,omitempty"`
Command []string `json:"command"`
Env map[string]string `json:"env,omitempty"`
TimeoutMS int `json:"timeoutMs,omitempty"`
}
// GracefulStopActionTemplate declares the plugin-owned shutdown step that runs
// before a supervised process is terminated. The command is ordinary
// plugin-declared game policy: it can announce a shutdown, ask the game to
// save, and wait for the game to exit on its own. Run only reports and keeps
// the bounded wait here; it never inspects the command text or its output.
type GracefulStopActionTemplate struct {
ExecutableKey string `json:"executableKey,omitempty"`
Arguments []string `json:"arguments,omitempty"`
Environment map[string]string `json:"environment,omitempty"`
TimeoutMS int `json:"timeoutMs,omitempty"`
Fallback string `json:"fallback,omitempty"`
}
func (executor LifecycleExecutor) loadLifecycleTemplate(assignment protocol.RunJobAssignment) (LifecycleActionTemplate, string, error) {
@@ -608,14 +622,7 @@ func (executor LifecycleExecutor) executeManaged(ctx context.Context, assignment
return processExecutionResult(item, "process started")
}
if assignment.Capability == protocol.RunCapabilityProcessStop {
log.Printf("RUN phase=lifecycle.managed status=stopping job=%s scope=%s", assignment.JobID, scope)
item, err := executor.managed.Stop(ctx, identity)
if err != nil {
log.Printf("RUN phase=lifecycle.managed status=stop_failed job=%s error=%s", assignment.JobID, err.Error())
return lifecycleFailure("lifecycle_stop_failed", err.Error())
}
log.Printf("RUN phase=lifecycle.managed status=stopped job=%s pid=%d state=%s classification=%s", assignment.JobID, item.PID, item.State, safeOptional(item.ExitClassification))
return processExecutionResult(item, "process stopped")
return executor.stopManaged(ctx, assignment, template, scope, identity)
}
log.Printf("RUN phase=lifecycle.managed status=querying job=%s scope=%s", assignment.JobID, scope)
item := executor.managed.Status(identity)
@@ -623,6 +630,113 @@ func (executor LifecycleExecutor) executeManaged(ctx context.Context, assignment
return processExecutionResult(item, "process status queried")
}
// stopManaged executes the plugin-declared shutdown step for a supervised
// process, then applies the plugin-declared fallback policy. Run keeps the
// generic supervision rules here; the declared command owns all game policy
// (player announcement, save request, shutdown command, exit polling).
func (executor LifecycleExecutor) stopManaged(ctx context.Context, assignment protocol.RunJobAssignment, template LifecycleActionTemplate, scope string, identity ProcessIdentity) LifecycleExecutionResult {
declared := template.GracefulStop
if declared == nil || strings.TrimSpace(declared.ExecutableKey) == "" {
log.Printf("RUN phase=lifecycle.managed status=stopping job=%s scope=%s", assignment.JobID, scope)
return executor.terminateManagedProcess(ctx, assignment, identity)
}
fallback := strings.TrimSpace(declared.Fallback)
if fallback == "" {
fallback = "report"
}
if fallback != "report" && fallback != "terminate" {
log.Printf("RUN phase=lifecycle.managed status=graceful_stop_invalid job=%s fallback=%s", assignment.JobID, fallback)
return lifecycleFailure("unsafe_lifecycle_command", "graceful stop fallback policy is unsupported")
}
command, err := declared.toProcessCommand(template, executor.workspaceRoot, scope, assignment)
if err != nil {
log.Printf("RUN phase=lifecycle.managed status=graceful_stop_invalid job=%s error=%s", assignment.JobID, err.Error())
return lifecycleFailure("unsafe_graceful_stop_command", err.Error())
}
log.Printf("RUN phase=lifecycle.managed status=graceful_stop_requested job=%s scope=%s fallback=%s command=%s timeoutMs=%d", assignment.JobID, scope, fallback, quotedCommandLine(command.Args), command.Timeout.Milliseconds())
command.OutputLine = func(stream string, line string) {
_ = executor.logSink.Append(ctx, assignment, stream, line)
}
result, err := executor.supervisor.Run(ctx, command)
if !result.OutputRelayed {
executor.writeProcessLogs(ctx, assignment, result)
}
if ctx.Err() != nil {
log.Printf("RUN phase=lifecycle.managed status=graceful_stop_cancelled job=%s error=%s", assignment.JobID, ctx.Err().Error())
return lifecycleExecutionFailure("lifecycle_cancelled", "lifecycle action cancelled", false)
}
if err != nil || result.ExitCode != 0 {
log.Printf("RUN phase=lifecycle.managed status=graceful_stop_failed job=%s exitCode=%d error=%s", assignment.JobID, result.ExitCode, errorSummary(err))
return executor.gracefulStopFallback(ctx, assignment, identity, fallback, "graceful_stop_failed")
}
current := executor.managed.Status(identity)
if current.State != "running" {
item, failure := executor.terminateManagedProcessResult(ctx, assignment, identity)
if failure.State != "" {
return failure
}
log.Printf("RUN phase=lifecycle.managed status=graceful_stop_completed job=%s pid=%d state=%s classification=%s", assignment.JobID, item.PID, item.State, safeOptional(item.ExitClassification))
result := processExecutionResult(item, "process stopped gracefully")
result.Progress.Message = "process stopped gracefully"
return result
}
log.Printf("RUN phase=lifecycle.managed status=graceful_stop_alive job=%s pid=%d fallback=%s", assignment.JobID, current.PID, fallback)
return executor.gracefulStopFallback(ctx, assignment, identity, fallback, "graceful_stop_timeout")
}
func (executor LifecycleExecutor) gracefulStopFallback(ctx context.Context, assignment protocol.RunJobAssignment, identity ProcessIdentity, fallback string, failureCode string) LifecycleExecutionResult {
if fallback != "terminate" {
log.Printf("RUN phase=lifecycle.managed status=graceful_stop_unverified job=%s code=%s", assignment.JobID, failureCode)
return lifecycleFailure(failureCode, "declared graceful stop did not stop the process and the plugin does not allow termination")
}
log.Printf("RUN phase=lifecycle.managed status=graceful_stop_escalated job=%s code=%s", assignment.JobID, failureCode)
return executor.terminateManagedProcess(ctx, assignment, identity)
}
func (executor LifecycleExecutor) terminateManagedProcessResult(ctx context.Context, assignment protocol.RunJobAssignment, identity ProcessIdentity) (ProcessIdentity, LifecycleExecutionResult) {
item, err := executor.managed.Stop(ctx, identity)
if err != nil {
log.Printf("RUN phase=lifecycle.managed status=stop_failed job=%s error=%s", assignment.JobID, err.Error())
return ProcessIdentity{}, lifecycleFailure("lifecycle_stop_failed", err.Error())
}
log.Printf("RUN phase=lifecycle.managed status=stopped job=%s pid=%d state=%s classification=%s", assignment.JobID, item.PID, item.State, safeOptional(item.ExitClassification))
return item, LifecycleExecutionResult{}
}
func (executor LifecycleExecutor) terminateManagedProcess(ctx context.Context, assignment protocol.RunJobAssignment, identity ProcessIdentity) LifecycleExecutionResult {
item, failure := executor.terminateManagedProcessResult(ctx, assignment, identity)
if item.State == "" && failure.State != "" {
return failure
}
return processExecutionResult(item, "process stopped")
}
func (declaration GracefulStopActionTemplate) toProcessCommand(template LifecycleActionTemplate, workspaceRoot string, scope string, assignment protocol.RunJobAssignment) (ProcessCommand, error) {
stopTemplate := LifecycleActionTemplate{
Action: template.Action,
ExecutableKey: declaration.ExecutableKey,
Arguments: declaration.Arguments,
Environment: mergeLifecycleEnvironment(template.Environment, declaration.Environment),
OutputMode: "pipes",
TimeoutMS: declaration.TimeoutMS,
}
return stopTemplate.ToExecutableProcessCommand(NewWorkspaceResolver(workspaceRoot), scope, assignment)
}
func mergeLifecycleEnvironment(base map[string]string, override map[string]string) map[string]string {
if len(base) == 0 && len(override) == 0 {
return nil
}
merged := make(map[string]string, len(base)+len(override))
for key, value := range base {
merged[key] = value
}
for key, value := range override {
merged[key] = value
}
return merged
}
func (executor LifecycleExecutor) ResumeManagedProcessLogs(ctx context.Context) {
if executor.managed == nil {
return