From 659f6d22c0487dc09b55e1ceeb7dcdecc01196d9 Mon Sep 17 00:00:00 2001 From: "ccf-lisa[bot]" <286799724+ccf-lisa[bot]@users.noreply.github.com> Date: Mon, 5 Oct 2026 14:49:52 -0300 Subject: [PATCH] feat(agent): AgentRunner primitives for prepare-then-cancel reloads The building blocks the reconciler in the next PR drives: - Prefetch downloads every plugin and policy source of a candidate config and resolves plugin protocols without touching the running config's locations, so a download failure never tears a working agent down (R32). ReportStartupFailure keeps main's startup-failure behaviour (plugins marked failed, startup-failure agent evidence) for a failed prefetch. - Plugin protocol versions read from OCI annotations are cached per source and survive reloads, so a registry outage cannot turn a v2 plugin into v1. - runDaemon returns setup errors instead of exiting. On a reload, in-flight plugin runs drain for up to 5 minutes (R33); a SIGINT/SIGTERM during the drain still exits within 30s. Each cron job runs with the config, client and logger of its own setup. - UpdateConfig reuses the SDK client when the api block is unchanged, so its token cache survives reloads. - Plugins get the host environment minus CCF_API_AUTH_* (R26). - make test runs with -race. Co-Authored-By: Claude Opus 5.5 --- Makefile | 2 +- cmd/agent.go | 280 ++++++++++++++++++++++++++++++++++++----- cmd/agent_test.go | 24 ++++ cmd/reconciler_test.go | 181 ++++++++++++++++++++++++++ docs/configuration.md | 2 + 5 files changed, 456 insertions(+), 33 deletions(-) diff --git a/Makefile b/Makefile index 28fab22..18c6baa 100644 --- a/Makefile +++ b/Makefile @@ -38,7 +38,7 @@ proto-gen: ## Generate objects from proto definitions ##@ Test .PHONY: test test: ## Run tests - @if ! go test ./... -coverprofile cover.out -v; then \ + @if ! go test ./... -race -coverprofile cover.out -v; then \ $(WARN) "Tests failed"; \ exit 1; \ fi ; \ diff --git a/cmd/agent.go b/cmd/agent.go index 678235f..75c0bc0 100644 --- a/cmd/agent.go +++ b/cmd/agent.go @@ -6,15 +6,19 @@ import ( "crypto/sha256" "encoding/base64" "encoding/json" + "errors" "fmt" "io" + "maps" "math/rand" "net/http" "os" "os/exec" "os/signal" "path/filepath" + "reflect" "runtime" + "slices" "sort" "strconv" "strings" @@ -38,7 +42,6 @@ import ( "github.com/google/go-containerregistry/pkg/v1/remote" "github.com/hashicorp/go-hclog" "github.com/hashicorp/go-plugin" - "github.com/open-policy-agent/opa/rego" "github.com/spf13/cobra" "github.com/spf13/viper" "golang.org/x/sync/singleflight" @@ -151,7 +154,15 @@ const AgentPolicyDir = ".compliance-framework/policies" const DefaultProtocolVersion int32 = 1 const RunnerV2ProtocolVersion int32 = 2 const AnnotationProtocolVersionKey = "org.ccf.plugin.protocol.version" -const daemonCronStopTimeout = 30 * time.Second + +// daemonCronStopTimeout bounds the cron stop on SIGINT/SIGTERM before plugins are killed, +// also when the signal arrives during a reload drain (R33). +var daemonCronStopTimeout = 30 * time.Second + +// reloadDrainTimeout bounds how long in-flight plugin runs may finish when a new +// configuration replaces the running one (R33). SIGTERM keeps daemonCronStopTimeout. +var reloadDrainTimeout = 5 * time.Minute + const agentEvidenceErrorArtifactMaxBytes = 1024 * 1024 type pluginRunStatus string @@ -415,15 +426,23 @@ type AgentRunner struct { fetchAnnotations func(ctx context.Context, source string, option ...remote.Option) (map[string]string, error) runPluginFunc func(ctx context.Context, name string, pluginConfig *agentPlugin) error sendHeartbeatFunc func(ctx context.Context, instanceID uuid.UUID) error + // notifySignals and exitFunc are test seams over signal.Notify(SIGINT, SIGTERM) and + // os.Exit (nil = the real ones). + notifySignals func(c chan<- os.Signal) + exitFunc func(code int) pluginRunMu sync.RWMutex pluginRuns map[string]pluginRunRecord firstAgentEvidenceSendStarted bool - queryBundles []*rego.Rego - // instanceID is this agent instance's stable ID (R31); set once at construction. instanceID uuid.UUID + + // protocolCache maps a plugin source to the protocol version its OCI annotations + // declared. It survives reloads so a registry outage during a reload cannot silently + // turn a v2 plugin into v1 (R32). + protocolCacheMu sync.Mutex + protocolCache map[string]int32 } // AgentRunnerOption configures an AgentRunner. @@ -443,6 +462,7 @@ func NewAgentRunner(opts ...AgentRunnerOption) *AgentRunner { fetchAnnotations: internal.GetAnnotations, httpClient: http.DefaultClient, instanceID: uuid.New(), + protocolCache: map[string]int32{}, } for _, opt := range opts { opt(ar) @@ -459,9 +479,14 @@ func (ar *AgentRunner) UpdateConfig(config *agentConfig) { Output: os.Stdout, Level: hclog.Level(config.logVerbosity()), }) - client := ar.buildAPIClient(config, logger) ar.stateMu.Lock() + // Reuse the SDK client when the api block is unchanged, so its token cache survives + // overlay reloads. + client := ar.apiClient + if client == nil || ar.config == nil || !reflect.DeepEqual(ar.config.ApiConfig, config.ApiConfig) { + client = ar.buildAPIClient(config, logger) + } ar.config = config ar.logger = logger ar.apiClient = client @@ -975,26 +1000,43 @@ func (ar *AgentRunner) Run(ctx context.Context) error { } logger.Debug("Pessimistically downloading plugins and policies worked successfully. Starting the agent.") - if config.Daemon == true { - ar.runDaemon(ctx) - return nil + if config.Daemon { + return ar.runDaemon(ctx) } return ar.runAllPlugins(ctx) } func (ar *AgentRunner) resolvePluginProtocols(ctx context.Context) { + ar.resolveProtocolsFor(ctx, ar.getConfig(), ar.getLogger()) +} + +// resolveProtocolsFor sets the protocol version of every implicit-protocol OCI plugin in config +// from its annotations, consulting protocolCache first. Only a cache miss fetches annotations. +func (ar *AgentRunner) resolveProtocolsFor(ctx context.Context, config *agentConfig, logger hclog.Logger) { if ctx == nil { ctx = context.Background() } + if config == nil { + return + } + if logger == nil { + logger = hclog.NewNullLogger() + } - config := ar.getConfig() - logger := ar.getLogger() for pluginName, pluginConfig := range config.Plugins { if pluginConfig == nil || pluginConfig.protocolSet || !internal.IsOCI(pluginConfig.Source) { continue } + ar.protocolCacheMu.Lock() + cached, hit := ar.protocolCache[pluginConfig.Source] + ar.protocolCacheMu.Unlock() + if hit { + pluginConfig.ProtocolVersion = cached + continue + } + func() { annotationCtx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() @@ -1017,32 +1059,37 @@ func (ar *AgentRunner) resolvePluginProtocols(ctx context.Context) { } pluginConfig.ProtocolVersion = protocolVersion + ar.protocolCacheMu.Lock() + ar.protocolCache[pluginConfig.Source] = protocolVersion + ar.protocolCacheMu.Unlock() }() } } -// Should never return, either handles any error or panics. -func (ar *AgentRunner) runDaemon(ctx context.Context) { +// runDaemon runs the plugin crons until ctx is cancelled (a reload: in-flight runs drain for +// up to reloadDrainTimeout, R33) or the process receives SIGINT/SIGTERM (exit). Setup errors +// are returned rather than exiting the process. +func (ar *AgentRunner) runDaemon(ctx context.Context) error { logger := ar.getLogger() sigs := make(chan os.Signal, 1) - signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM) + ar.signalNotify(sigs) defer signal.Stop(sigs) agentCron, err := ar.setupCron(ctx) if err != nil { logger.Error("Error setting up agent cron", "error", err) - os.Exit(1) + return err } heartbeatCron, err := ar.setupHeartbeatCron(ctx) if err != nil { logger.Error("Error setting up heartbeat", "error", err) - os.Exit(1) + return err } agentEvidenceCron, err := ar.setupAgentEvidenceCron(ctx) if err != nil { logger.Error("Error setting up agent evidence", "error", err) - os.Exit(1) + return err } // Start the cron and notify readiness @@ -1070,23 +1117,59 @@ func (ar *AgentRunner) runDaemon(ctx context.Context) { logger.Debug("Shutting down plugins") ar.closePluginClients() logger.Debug("Exiting") - os.Exit(0) + ar.exitProcess(0) + return nil case <-ctx.Done(): logger.Debug("received cancel signal to return from daemon") logger.Debug("Stopping crons") agentCronStopCtx := agentCron.Stop() heartbeatCronStopCtx := heartbeatCron.Stop() agentEvidenceCronStopCtx := agentEvidenceCron.Stop() - if !waitForCronStop(daemonCronStopTimeout, agentCronStopCtx, heartbeatCronStopCtx, agentEvidenceCronStopCtx) { - logger.Warn("Timed out waiting for cron jobs to stop before plugin cleanup", "timeout", daemonCronStopTimeout) + drained, sig := waitForCronStopOrSignal(reloadDrainTimeout, sigs, agentCronStopCtx, heartbeatCronStopCtx, agentEvidenceCronStopCtx) + if sig != nil { + // A SIGINT/SIGTERM during the reload drain is not lost: it keeps its 30s (R33). + logger.Info("received signal during the reload drain; terminating plugins and exiting", "signal", sig) + if !waitForCronStop(daemonCronStopTimeout, agentCronStopCtx, heartbeatCronStopCtx, agentEvidenceCronStopCtx) { + logger.Warn("Timed out waiting for cron jobs to stop before plugin cleanup", "timeout", daemonCronStopTimeout) + } + ar.closePluginClients() + logger.Debug("Exiting") + ar.exitProcess(0) + return nil + } + if !drained { + logger.Warn("Timed out waiting for in-flight plugin runs to drain before reload", "timeout", reloadDrainTimeout) } logger.Debug("Shutting down plugins") ar.closePluginClients() + return nil + } +} + +func (ar *AgentRunner) signalNotify(c chan<- os.Signal) { + if ar.notifySignals != nil { + ar.notifySignals(c) return } + signal.Notify(c, syscall.SIGINT, syscall.SIGTERM) +} + +func (ar *AgentRunner) exitProcess(code int) { + if ar.exitFunc != nil { + ar.exitFunc(code) + return + } + os.Exit(code) } func waitForCronStop(timeout time.Duration, stopContexts ...context.Context) bool { + done, _ := waitForCronStopOrSignal(timeout, nil, stopContexts...) + return done +} + +// waitForCronStopOrSignal waits until every stop context is done (true), the timeout expires +// (false) or a signal arrives on sigs (false and the signal; a nil sigs never fires). +func waitForCronStopOrSignal(timeout time.Duration, sigs <-chan os.Signal, stopContexts ...context.Context) (bool, os.Signal) { allDone := make(chan struct{}) waitCtx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -1114,13 +1197,15 @@ func waitForCronStop(timeout time.Duration, stopContexts ...context.Context) boo select { case <-allDone: - return true + return true, nil + case sig := <-sigs: + return false, sig case <-timer.C: select { case <-allDone: - return true + return true, nil default: - return false + return false, nil } } } @@ -1215,10 +1300,18 @@ func (ar *AgentRunner) setupCron(ctx context.Context) (*cron.Cron, error) { parserOptions, ))) config := ar.getConfig() - runPlugin := ar.runPlugin + // Each job runs with the config, API client and logger of THIS setup: a job still running + // when a reload's drain times out must not pick up the next configuration's state. + snap := ar.snapshot() + runPlugin := func(ctx context.Context, name string, plugin *agentPlugin) error { + return ar.runPluginWith(ctx, snap, name, plugin) + } if ar.runPluginFunc != nil { runPlugin = ar.runPluginFunc } + // Plugin runs are not cut short by a reload: runDaemon stops the cron and lets in-flight + // runs drain for up to reloadDrainTimeout (R33) before killing the plugin processes. + jobCtx := context.WithoutCancel(ctx) for pluginName, pluginConfig := range config.Plugins { currentPluginName := pluginName @@ -1233,14 +1326,14 @@ func (ar *AgentRunner) setupCron(ctx context.Context) (*cron.Cron, error) { jobLogger := logger.With("plugin", currentPluginName, "schedule", schedule) job := cron.NewChain(cron.SkipIfStillRunning(cronLogger{logger: jobLogger})).Then(cron.FuncJob(func() { ar.markPluginRunStarted(currentPluginName) - err := runPlugin(ctx, currentPluginName, currentPluginConfig) + err := runPlugin(jobCtx, currentPluginName, currentPluginConfig) ar.markPluginRunFinished(currentPluginName, err) if err != nil { // TODO how will we handle these errors ? jobLogger.Error("Error running plugin", "error", err, "protocol_version", currentPluginConfig.ProtocolVersion) } if ar.reserveFirstAgentEvidenceSend() { - if evidenceErr := ar.SendAgentRunEvidence(ctx); evidenceErr != nil { + if evidenceErr := ar.SendAgentRunEvidence(jobCtx); evidenceErr != nil { ar.releaseFirstAgentEvidenceSend() jobLogger.Error("Failed to send agent run evidence", "error", evidenceErr) } @@ -1411,15 +1504,23 @@ func (ar *AgentRunner) sendAgentRunEvidenceOnStartupFailure(ctx context.Context) return ar.SendAgentRunEvidence(ctx) } -// Run the agent as an instance, this is a single run of the agent that will check the -// policies against the plugins. +// runSnapshot is the state one plugin run uses from start to end. +type runSnapshot struct { + config *agentConfig + client *sdk.Client + logger hclog.Logger +} + +func (ar *AgentRunner) snapshot() runSnapshot { + return runSnapshot{config: ar.getConfig(), client: ar.getAPIClient(), logger: ar.getLogger()} +} + +// runPluginWith runs one plugin once with the state in snap. // // Returns: // - error: any error that occurred during the run -func (ar *AgentRunner) runPlugin(ctx context.Context, name string, plugin *agentPlugin) error { - config := ar.getConfig() - client := ar.getAPIClient() - logger := ar.getLogger() +func (ar *AgentRunner) runPluginWith(ctx context.Context, snap runSnapshot, name string, plugin *agentPlugin) error { + config, client, logger := snap.config, snap.client, snap.logger logger.Debug("Running single plugin with shared API SDK client", "plugin", name, "auth_enabled", hasAPIAuth(config), @@ -1763,12 +1864,23 @@ func safePluginErrorFilename(pluginName string) string { return b.String() + "-error.txt" } +// pluginCommand is the command that starts the plugin binary at path. Plugins get the host +// environment minus the agent's own API credentials (R26); go-plugin would otherwise append +// the whole environment. +func pluginCommand(path string) *exec.Cmd { + cmd := exec.Command(path) + cmd.Env = pluginEnviron(os.Environ()) + return cmd +} + func (ar *AgentRunner) getRunnerInstance(logger hclog.Logger, path string, protocolVersion int32) (runner.RunnerV2, func(), error) { // We're a host! Start by launching the plugin process. + cmd := pluginCommand(path) client := plugin.NewClient(&plugin.ClientConfig{ HandshakeConfig: runner.HandshakeConfig, Plugins: runner.PluginMap, - Cmd: exec.Command(path), + Cmd: cmd, + SkipHostEnv: true, Logger: logger, AllowedProtocols: []plugin.Protocol{plugin.ProtocolGRPC}, }) @@ -1868,6 +1980,110 @@ func (ar *AgentRunner) DownloadPolicies(ctx context.Context) error { return nil } +// pluginEnviron returns environ without the variables whose name starts with CCF_API_AUTH_ +// (case-insensitive), so plugins never see the agent's API credentials (R26). +func pluginEnviron(environ []string) []string { + out := make([]string, 0, len(environ)) + for _, kv := range environ { + name, _, _ := strings.Cut(kv, "=") + if strings.HasPrefix(strings.ToUpper(name), "CCF_API_AUTH_") { + continue + } + out = append(out, kv) + } + return out +} + +// Prefetch downloads every plugin and policy source of cfg and resolves plugin +// protocol versions, WITHOUT touching the running configuration's pluginLocations or +// policyLocations. The reconciler calls it before cancelling the running configuration, so a +// download failure never tears down a working agent (prepare-then-cancel, R32). +func (ar *AgentRunner) Prefetch(ctx context.Context, cfg *agentConfig) error { + logger := ar.getLogger() + if logger == nil { + logger = hclog.NewNullLogger() + } + pluginSources := map[string]struct{}{} + policySources := map[string]struct{}{} + for _, pluginConfig := range cfg.Plugins { + pluginSources[pluginConfig.Source] = struct{}{} + for _, policy := range pluginConfig.Policies { + policySources[string(policy)] = struct{}{} + } + } + for _, source := range slices.Sorted(maps.Keys(pluginSources)) { + if _, err := ar.downloadPlugin(ctx, source, logger); err != nil { + return &downloadError{source: source, err: err} + } + } + ar.resolveProtocolsFor(ctx, cfg, logger) + for _, source := range slices.Sorted(maps.Keys(policySources)) { + if _, err := ar.downloadPolicy(ctx, source, logger); err != nil { + return &downloadError{source: source, policy: true, err: err} + } + } + return nil +} + +// downloadError is a Prefetch failure: which plugin or policy source could not be fetched. +type downloadError struct { + source string + policy bool + err error +} + +func (e *downloadError) Error() string { + kind := "plugin" + if e.policy { + kind = "policy" + } + return fmt.Sprintf("download %s %s: %v", kind, e.source, e.err) +} + +func (e *downloadError) Unwrap() error { return e.err } + +// ReportStartupFailure records that the configuration the agent starts with could not be +// downloaded, exactly as Run always has on a startup download failure: the plugins using the +// failed source are marked failed and the startup-failure agent evidence is sent (when agent +// evidence and emit_on_run_completion are enabled). The agent then exits 1. +func (ar *AgentRunner) ReportStartupFailure(ctx context.Context, cfg *agentConfig, err error) { + ar.UpdateConfig(cfg) + var dl *downloadError + switch { + case errors.As(err, &dl) && dl.policy: + ar.markPluginsWithPolicyFailed(agentPolicy(dl.source), dl.err) + case errors.As(err, &dl): + ar.markPluginsWithSourceFailed(dl.source, dl.err) + } + logger := ar.getLogger() + logger.Error("Error downloading plugins and policies", "error", err) + if evidenceErr := ar.sendAgentRunEvidenceOnStartupFailure(ctx); evidenceErr != nil { + logger.Error("Error sending agent run evidence", "error", evidenceErr) + } +} + +// downloadPlugin returns the plugin binary of source for this platform, downloading it into +// the shared plugin cache when it is not there yet. Prefetch uses it, so the plugins report +// (R76) reads the agent library version from the binary Prefetch fetched. +func (ar *AgentRunner) downloadPlugin(ctx context.Context, source string, logger hclog.Logger) (string, error) { + if logger == nil { + logger = hclog.NewNullLogger() + } + platform := v1.Platform{ + Architecture: runtime.GOARCH, + OS: runtime.GOOS, + } + return ar.download(ctx, source, AgentPluginDir, "plugin", platformDownloadKey(platform), logger, remote.WithPlatform(platform)) +} + +// downloadPolicy fetches one policy source into the shared policy cache. +func (ar *AgentRunner) downloadPolicy(ctx context.Context, source string, logger hclog.Logger) (string, error) { + if logger == nil { + logger = hclog.NewNullLogger() + } + return ar.download(ctx, source, AgentPolicyDir, "policies", "", logger) +} + func platformDownloadKey(platform v1.Platform) string { return strings.Join([]string{platform.OS, platform.Architecture, platform.Variant}, "/") } diff --git a/cmd/agent_test.go b/cmd/agent_test.go index 45294a0..b04ab13 100644 --- a/cmd/agent_test.go +++ b/cmd/agent_test.go @@ -2165,3 +2165,27 @@ func jsonResponse(statusCode int, body string) *http.Response { Header: make(http.Header), } } + +// TestPluginCommandStripsAPICredentials: plugins get the host environment without the agent's +// API credentials (R26). +func TestPluginCommandStripsAPICredentials(t *testing.T) { + t.Setenv("CCF_API_AUTH_CLIENT_SECRET", "s3cret") + t.Setenv("ccf_api_auth_client_id", "id") + t.Setenv("CCF_PLUGIN_SETTING", "kept") + + cmd := pluginCommand("/bin/plugin") + if cmd.Path != "/bin/plugin" || cmd.Dir != "" { + t.Fatalf("path = %q, dir = %q", cmd.Path, cmd.Dir) + } + var kept bool + for _, kv := range cmd.Env { + name, _, _ := strings.Cut(kv, "=") + if strings.HasPrefix(strings.ToUpper(name), "CCF_API_AUTH_") { + t.Fatalf("the plugin must not see %s", name) + } + kept = kept || kv == "CCF_PLUGIN_SETTING=kept" + } + if !kept { + t.Fatal("other variables must be passed through") + } +} diff --git a/cmd/reconciler_test.go b/cmd/reconciler_test.go index 5de8019..365cb5b 100644 --- a/cmd/reconciler_test.go +++ b/cmd/reconciler_test.go @@ -2,12 +2,89 @@ package cmd import ( "context" + "errors" + "io" + "net/http" + "os" + "slices" + "strings" "sync" + "sync/atomic" + "syscall" "testing" + "time" + "github.com/google/go-containerregistry/pkg/v1/remote" "github.com/google/uuid" ) +func TestReportStartupFailureMarksPluginsAndSendsEvidence(t *testing.T) { + var mu sync.Mutex + var bodies []string + ar := NewAgentRunner() + ar.httpClient = newTestHTTPClient(func(r *http.Request) (*http.Response, error) { + raw, _ := io.ReadAll(r.Body) + mu.Lock() + bodies = append(bodies, string(raw)) + mu.Unlock() + return jsonResponse(http.StatusCreated, ""), nil + }) + cfg := newTestAgentConfig("http://example.test", nil) + ar.ReportStartupFailure(context.Background(), cfg, &downloadError{source: "ghcr.io/some-plugin:v1", err: errors.New("registry down")}) + + if snap := ar.pluginRunSnapshot(); !slices.Contains(snap.Failed, "test-plugin") || !strings.Contains(snap.Errors["test-plugin"], "registry down") { + t.Fatalf("the plugin using the failed source must be marked failed, got %+v", snap) + } + mu.Lock() + defer mu.Unlock() + if len(bodies) != 1 || !strings.Contains(bodies[0], "Plugins with errors: test-plugin") { + t.Fatalf("expected one startup-failure evidence naming the failed plugin, got %d: %v", len(bodies), bodies) + } +} + +func TestPluginEnvironDropsAPICredentials(t *testing.T) { + got := pluginEnviron([]string{ + "CCF_API_AUTH_CLIENT_SECRET=s", + "ccf_api_auth_client_id=i", + "AWS_REGION=eu-west-1", + "CCF_INSTANCE_ID=abc", + "PATH=/bin", + }) + want := []string{"AWS_REGION=eu-west-1", "CCF_INSTANCE_ID=abc", "PATH=/bin"} + if strings.Join(got, ",") != strings.Join(want, ",") { + t.Fatalf("pluginEnviron = %v, want %v", got, want) + } +} + +func TestResolvePluginProtocols_CacheSurvivesLookupFailure(t *testing.T) { + ar := NewAgentRunner() + calls := 0 + ar.fetchAnnotations = func(context.Context, string, ...remote.Option) (map[string]string, error) { + calls++ + if calls == 1 { + return map[string]string{AnnotationProtocolVersionKey: "2"}, nil + } + return nil, errors.New("registry down") + } + newCfg := func() *agentConfig { + return &agentConfig{Plugins: map[string]*agentPlugin{ + "p": {Source: "ghcr.io/example/plugin:v1", ProtocolVersion: DefaultProtocolVersion}, + }} + } + first := newCfg() + ar.UpdateConfig(first) + ar.resolvePluginProtocols(context.Background()) + second := newCfg() + ar.UpdateConfig(second) + ar.resolvePluginProtocols(context.Background()) + if got := second.Plugins["p"].ProtocolVersion; got != RunnerV2ProtocolVersion { + t.Fatalf("expected the cached protocol 2 after a failing lookup, got %d", got) + } + if calls != 1 { + t.Fatalf("expected one annotation lookup, got %d", calls) + } +} + func TestHeartbeatUsesStableInstanceIDAcrossRuns(t *testing.T) { id := uuid.New() ar := NewAgentRunner(WithInstanceID(id)) @@ -33,3 +110,107 @@ func TestHeartbeatUsesStableInstanceIDAcrossRuns(t *testing.T) { t.Fatalf("heartbeats used %v, want %s twice", seen, id) } } + +func TestRunDaemonDrainsInFlightRunsOnReload(t *testing.T) { + oldDrain := reloadDrainTimeout + reloadDrainTimeout = 10 * time.Second + t.Cleanup(func() { reloadDrainTimeout = oldDrain }) + + schedule := "@every 1s" + disabled := false + ar := NewAgentRunner() + ar.UpdateConfig(&agentConfig{ + Daemon: true, + ApiConfig: &apiConfig{Url: "http://127.0.0.1:1"}, + AgentEvidence: &agentEvidenceConfig{Enabled: &disabled}, + Plugins: map[string]*agentPlugin{"slow": {Source: "/tmp/slow", Schedule: &schedule}}, + }) + started := make(chan struct{}, 1) + var finishedCleanly atomic.Bool + ar.runPluginFunc = func(ctx context.Context, _ string, _ *agentPlugin) error { + select { + case started <- struct{}{}: + default: + } + time.Sleep(300 * time.Millisecond) + finishedCleanly.Store(ctx.Err() == nil) + return nil + } + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- ar.runDaemon(ctx) }() + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("plugin never started") + } + cancel() + select { + case err := <-done: + if err != nil { + t.Fatalf("runDaemon: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("runDaemon did not return after the drain") + } + if !finishedCleanly.Load() { + t.Fatal("the in-flight run was cut short by the reload") + } +} + +// TestRunDaemonSignalDuringReloadDrainExits: a SIGTERM that arrives while a reload drains is +// not lost; it exits (R33: SIGTERM keeps its 30s) instead of waiting out the 5m drain. +func TestRunDaemonSignalDuringReloadDrainExits(t *testing.T) { + oldStop := daemonCronStopTimeout + daemonCronStopTimeout = 100 * time.Millisecond + t.Cleanup(func() { daemonCronStopTimeout = oldStop }) + + schedule := "@every 1s" + disabled := false + ar := NewAgentRunner() + ar.UpdateConfig(&agentConfig{ + Daemon: true, + ApiConfig: &apiConfig{Url: "http://127.0.0.1:1"}, + AgentEvidence: &agentEvidenceConfig{Enabled: &disabled}, + Plugins: map[string]*agentPlugin{"slow": {Source: "/tmp/slow", Schedule: &schedule}}, + }) + sigCh := make(chan chan<- os.Signal, 1) + ar.notifySignals = func(c chan<- os.Signal) { sigCh <- c } + exited := make(chan int, 1) + ar.exitFunc = func(code int) { exited <- code } + started := make(chan struct{}, 1) + release := make(chan struct{}) + ar.runPluginFunc = func(context.Context, string, *agentPlugin) error { + select { + case started <- struct{}{}: + default: + } + <-release // outlives the test's patience: only the signal can end the drain + return nil + } + defer close(release) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- ar.runDaemon(ctx) }() + sigs := <-sigCh + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("plugin never started") + } + cancel() // reload: the drain waits for the blocked run + time.Sleep(50 * time.Millisecond) + sigs <- syscall.SIGTERM + + select { + case code := <-exited: + if code != 0 { + t.Fatalf("exit code %d, want 0", code) + } + case <-time.After(5 * time.Second): + t.Fatal("a SIGTERM during the reload drain was lost") + } + <-done +} diff --git a/docs/configuration.md b/docs/configuration.md index 712b0ea..6503483 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -199,3 +199,5 @@ Because the default key depends on the config file's path, **moving or renaming instance** (R52). The agent logs the state directory, where it came from and the instance ID at startup. Containers and Helm deployments should pin `CCF_STATE_DIR` to a mounted volume; ephemeral one-shot runs (CI, Kubernetes jobs) can set `CCF_INSTANCE_ID` so repeated runs report as one instance. + +Plugins receive the agent's environment except `CCF_API_AUTH_*` (R26).