diff --git a/cmd/agent.go b/cmd/agent.go index 75c0bc0..18a10b4 100644 --- a/cmd/agent.go +++ b/cmd/agent.go @@ -37,13 +37,11 @@ import ( sdktypes "github.com/compliance-framework/api/sdk/types" "github.com/coreos/go-systemd/v22/daemon" oscalTypes_1_1_3 "github.com/defenseunicorns/go-oscal/src/types/oscal-1-1-3" - "github.com/fsnotify/fsnotify" v1 "github.com/google/go-containerregistry/pkg/v1" "github.com/google/go-containerregistry/pkg/v1/remote" "github.com/hashicorp/go-hclog" "github.com/hashicorp/go-plugin" "github.com/spf13/cobra" - "github.com/spf13/viper" "golang.org/x/sync/singleflight" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -288,39 +286,22 @@ func configureRunner(name string, runnerInstance runner.RunnerV2, config agentPl return err } -// loadConfig reads and validates the config file and builds its runtime form. Tolerated file -// problems (R34) are logged; the plugins they affect are skipped. -func loadConfig(cmd *cobra.Command, configPath string, logger hclog.Logger) (*agentConfig, error) { - base, err := loadBase(cmd, configPath) - if err != nil { - return nil, err - } - for _, w := range base.warnings { - if isToleratedFileRule(w) { - logger.Warn("Ignoring a problem in the config file; the plugin is skipped", "path", w.Path, "error", w.Message) - continue - } - logger.Warn("Ignoring a problem in the config file; the value is kept unchanged", "path", w.Path, "error", w.Message) - } - return toRuntime(base.declared, base.skip) -} - // Main the entrypoint for the `agent` command // // It will read the configuration file, and then run the agent. Various command line flags can // be used to override the config file. func agentRunner(cmd *cobra.Command, args []string) error { - configPath, err := filepath.Abs(cmd.Flag("config").Value.String()) - if err != nil { - return err - } - logger := hclog.New(&hclog.LoggerOptions{ Name: "agent", Output: os.Stdout, Level: hclog.Debug, }) + configPath, err := filepath.Abs(cmd.Flag("config").Value.String()) + if err != nil { + return err + } + stateDir, stateDirSource, err := stateDirFrom(cmd, configPath) if err != nil { return err @@ -335,46 +316,27 @@ func agentRunner(cmd *cobra.Command, args []string) error { // file silently creates a new instance. Say where state lives. logger.Info("Agent state", "state_dir", stateDir, "state_dir_source", stateDirSource, "instance_id", id.String(), "instance_id_persisted", persisted) - agentRun := NewAgentRunner(WithInstanceID(id)) + ar := NewAgentRunner(WithInstanceID(id)) + rc := newReconciler(cmd, configPath, store, ar, logger) + rc.onStartupFailure = ar.ReportStartupFailure - ctx, configCancel := context.WithCancel(context.Background()) - defer configCancel() - - // The watcher only signals; every load builds a fresh viper (loadBase). - w := viper.New() - w.SetConfigFile(configPath) - w.OnConfigChange(func(in fsnotify.Event) { - // We want to wait for any running agent processes to finish first. - logger.Debug("config file changed", "path", in.Name) - configCancel() - }) - w.WatchConfig() - - // For the daemon, we run the agent continuously. - // It will exit as soon as the config changes, and then start again with new configs set. - for { - ctx, configCancel = context.WithCancel(context.Background()) - config, err := loadConfig(cmd, configPath, logger) - if err != nil { - logger.Error("Error loading new config", "error", err) - panic(err) - } - agentRun.UpdateConfig(config) - err = agentRun.Run(ctx) - - if err != nil { - logger.Error("Error running agent", "error", err) - os.Exit(1) - } - - if !config.Daemon { - break - } + active, err := rc.startup(context.Background()) + if err != nil { + // An unusable local configuration at startup exits 1, as it always has. + return err } - configCancel() + rootCtx, stopLoop := context.WithCancel(context.Background()) + defer stopLoop() + if active.runtime.Daemon { + defer rc.watchFile()() + go rc.loop(rootCtx) + } - return nil + return rc.run(active, func(ctx context.Context, cfg *agentConfig) error { + ar.UpdateConfig(cfg) + return ar.Run(ctx) + }) } // stateDirFrom resolves the state directory: --state-dir, then CCF_STATE_DIR, then the diff --git a/cmd/config.go b/cmd/config.go index 232e10e..eccbc86 100644 --- a/cmd/config.go +++ b/cmd/config.go @@ -28,6 +28,8 @@ type baseSnapshot struct { warnings []agentconfig.FieldError // skip holds the plugins dropped from the runtime because of a tolerated problem. skip map[string]string + // fingerprint identifies the base for the failed backoff. + fingerprint string } // toleratedFileRules are the validation rules whose failure is non-fatal when the value comes @@ -210,6 +212,7 @@ func baseFromViper(cmd *cobra.Command, v *viper.Viper, raw []byte) (*baseSnapsho } base.warnings = part.warnings base.skip = part.skip + base.fingerprint = agentconfig.Digest(declared) return base, nil } diff --git a/cmd/reconciler.go b/cmd/reconciler.go new file mode 100644 index 0000000..273237c --- /dev/null +++ b/cmd/reconciler.go @@ -0,0 +1,466 @@ +package cmd + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "math/rand" + "sync" + "sync/atomic" + "time" + + "github.com/compliance-framework/agent/internal/agentstate" + "github.com/compliance-framework/api/pkg/agentconfig" + "github.com/fsnotify/fsnotify" + "github.com/hashicorp/go-hclog" + "github.com/spf13/cobra" + "github.com/spf13/viper" +) + +var ( + // failedRetryMin / failedRetryMax bound the retry of a failed/* revision. + failedRetryMin = time.Minute + failedRetryMax = 10 * time.Minute + // prepareNetworkTimeout bounds the network step of prepare (plugin/policy prefetch), so a + // hung registry cannot stall the reconciler. A timeout is a failed/download-failed, which + // is retried with the failed backoff. + prepareNetworkTimeout = 5 * time.Minute +) + +// candidate is a complete, validated configuration that is ready to run. The reconciler builds +// it BEFORE cancelling the running configuration (prepare-then-cancel, R32). It is immutable +// once built. +type candidate struct { + base *baseSnapshot + declared agentconfig.Config // the declared config the runtime was built from + runtime *agentConfig // resolved, enabled-only, skipped plugins removed + // identity changes whenever anything that affects the runtime changes, including the + // values the digest masks or omits (api block, secrets). It never leaves the process. + identity string +} + +// applyError is why a candidate could not be prepared. Status is agentconfig.StatusRejected +// or agentconfig.StatusFailed and Reason is one of agentconfig.Reasons. +type applyError struct { + Status string + Reason string + Err error + // runtime is the prepared runtime of a download-failed candidate: startup hands it to + // onStartupFailure so the startup-failure evidence describes it, as on main. + runtime *agentConfig +} + +func (e *applyError) Error() string { + if e == nil { + return "" + } + if e.Err == nil { + return fmt.Sprintf("%s: %s", e.Status, e.Reason) + } + return fmt.Sprintf("%s: %s: %v", e.Status, e.Reason, e.Err) +} + +func (e *applyError) Unwrap() error { return e.Err } + +func failed(reason string, err error) *applyError { + return &applyError{Status: agentconfig.StatusFailed, Reason: reason, Err: err} +} + +// prefetcher is the part of AgentRunner the reconciler drives (a test seam). +type prefetcher interface { + Prefetch(ctx context.Context, cfg *agentConfig) error +} + +// runFunc runs one configuration until it is cancelled (daemon) or completes (one-shot). +type runFunc func(ctx context.Context, cfg *agentConfig) error + +type trigger int + +const ( + triggerFile trigger = iota + triggerPoll +) + +// reconciler is the single writer of the configuration state. Triggers are serialized in one +// goroutine: a trigger builds a complete candidate and only then cancels the running +// configuration; a failure tears nothing down. +type reconciler struct { + cmd *cobra.Command + configPath string + store *agentstate.Store + runner prefetcher + logger hclog.Logger + fileEvents chan struct{} + runFailed chan *candidate + // onStartupFailure records a startup download failure of the file-only configuration + // (plugin run state + startup-failure agent evidence, as AgentRunner.Run always did). + onStartupFailure func(ctx context.Context, cfg *agentConfig, err error) + // debounce coalesces bursts of config file events (editors write in several steps). + debounce time.Duration + + now func() time.Time + + mu sync.Mutex // guards active, pending, cancelRun + active *candidate + pending *candidate + cancelRun context.CancelFunc + + // Everything below is owned by the reconciler goroutine (startup runs before loop). + base *baseSnapshot + + failedBase string + failedRetryAt time.Time + failedInterval time.Duration +} + +func newReconciler(cmd *cobra.Command, configPath string, store *agentstate.Store, runner prefetcher, logger hclog.Logger) *reconciler { + if logger == nil { + logger = hclog.NewNullLogger() + } + rc := &reconciler{ + cmd: cmd, + configPath: configPath, + store: store, + runner: runner, + logger: logger.Named("reconciler"), + fileEvents: make(chan struct{}, 1), + runFailed: make(chan *candidate, 1), + debounce: 500 * time.Millisecond, + now: time.Now, + } + return rc +} + +func (rc *reconciler) rcfg() agentconfig.RemoteConfig { + return rc.base.declared.EffectiveRemoteConfig() +} + +// setBase installs a new base and logs its warnings. +func (rc *reconciler) setBase(base *baseSnapshot) { + rc.base = base + rc.logWarnings(base.warnings) +} + +// startup loads the file and prepares it (R32). Only an unusable local configuration is an +// error (exit 1, as before). +func (rc *reconciler) startup(ctx context.Context) (*candidate, error) { + base, err := loadBase(rc.cmd, rc.configPath) + if err != nil { + return nil, fmt.Errorf("config file: %w", err) + } + rc.setBase(base) + + active, aerr := rc.prepare(ctx, base) + if aerr != nil { + if aerr.runtime != nil && rc.onStartupFailure != nil { + rc.onStartupFailure(ctx, aerr.runtime, aerr.Err) + } + return nil, aerr + } + return active, nil +} + +// startFailedBackoff starts (or doubles, for the same base) the retry delay of a candidate that +// failed to prepare or to run on base baseFingerprint. +func (rc *reconciler) startFailedBackoff(baseFingerprint string) { + if rc.failedBase == baseFingerprint && rc.failedInterval > 0 { + rc.failedInterval = min(rc.failedInterval*2, failedRetryMax) + } else { + rc.failedInterval = failedRetryMin + } + rc.failedBase = baseFingerprint + rc.failedRetryAt = rc.now().Add(rc.failedInterval) +} + +func (rc *reconciler) inFailedBackoff() bool { + return rc.failedInterval > 0 && rc.failedBase == rc.base.fingerprint && rc.now().Before(rc.failedRetryAt) +} + +// clearFailedBackoff forgets the failed backoff after a candidate prepared on the current base, +// unless it is the base in backoff: that one may still fail to RUN, and the retry delay must +// keep growing instead of restarting at failedRetryMin (no flapping every poll). +func (rc *reconciler) clearFailedBackoff() { + if rc.failedBase == rc.base.fingerprint { + return + } + rc.failedBase, rc.failedInterval = "", 0 +} + +// prepare builds a candidate from a base. It never touches the running configuration. +func (rc *reconciler) prepare(ctx context.Context, base *baseSnapshot) (*candidate, *applyError) { + declared := base.declared + runtime, err := toRuntime(declared, base.skip) + if err != nil { + return nil, failed(agentconfig.ReasonInvalidConfig, err) + } + prefetchCtx, cancelPrefetch := context.WithTimeout(ctx, prepareNetworkTimeout) + err = rc.runner.Prefetch(prefetchCtx, runtime) + cancelPrefetch() + if err != nil { + aerr := failed(agentconfig.ReasonDownloadFailed, err) + aerr.runtime = runtime + return nil, aerr + } + return &candidate{ + base: base, + declared: declared, + runtime: runtime, + identity: candidateIdentity(declared), + }, nil +} + +// candidateIdentity hashes the declared config, including the values the digest masks or +// omits, so any change that affects the runtime is a different configuration. +func candidateIdentity(c agentconfig.Config) string { + raw, err := agentconfig.CanonicalJSON(struct { + Config agentconfig.Config `json:"config"` + }{c}) + if err != nil { + raw = []byte(err.Error()) + } + sum := sha256.Sum256(raw) + return hex.EncodeToString(sum[:]) +} + +// bind records the running candidate and how to cancel it. If a swap raced in between two +// runs, the new run is cancelled at once so the pending candidate is picked up. +func (rc *reconciler) bind(active *candidate, cancel context.CancelFunc) { + rc.mu.Lock() + defer rc.mu.Unlock() + rc.active = active + rc.cancelRun = cancel + if rc.pending != nil { + cancel() + } +} + +// adopt records cand, whose identity equals old's, as the running (or pending) configuration +// WITHOUT a restart: the runtime old runs is kept, and sameAsActive holds for cand's base on +// the next trigger (no re-prepare every poll). +func (rc *reconciler) adopt(old, cand *candidate) *candidate { + adopted := *cand + if old.runtime != nil { + adopted.runtime = old.runtime + } + rc.mu.Lock() + defer rc.mu.Unlock() + switch { + case rc.pending == old: + rc.pending = &adopted + case rc.active == old: + rc.active = &adopted + } + return &adopted +} + +// record returns the reconciler's current record of the candidate running c.runtime: adopt +// may have replaced it in place. +func (rc *reconciler) record(c *candidate) *candidate { + rc.mu.Lock() + defer rc.mu.Unlock() + if rc.active != nil && c != nil && rc.active.runtime == c.runtime { + return rc.active + } + return c +} + +// swap makes next the pending candidate and cancels the running one. +func (rc *reconciler) swap(next *candidate) { + rc.mu.Lock() + defer rc.mu.Unlock() + rc.pending = next + if rc.cancelRun != nil { + rc.cancelRun() + } +} + +func (rc *reconciler) takePending() *candidate { + rc.mu.Lock() + defer rc.mu.Unlock() + next := rc.pending + rc.pending = nil + return next +} + +// current returns the candidate that is running, or about to run when a swap is pending. +func (rc *reconciler) current() *candidate { + rc.mu.Lock() + defer rc.mu.Unlock() + if rc.pending != nil { + return rc.pending + } + return rc.active +} + +// run drives run with the active candidate. A cancelled run (swap) picks up the pending +// candidate; a run that fails on its own falls back to the previous candidate once. +func (rc *reconciler) run(active *candidate, run runFunc) error { + var previous, failedRun *candidate + for { + runCtx, cancel := context.WithCancel(context.Background()) + rc.bind(active, cancel) + if failedRun != nil { + // Notify only once the fallback is bound, so the reconciler's current() is the + // fallback, never the candidate that failed. + rc.notifyRunFailed(failedRun) + failedRun = nil + } + runErr := run(runCtx, active.runtime) + reload := runCtx.Err() != nil + cancel() + active = rc.record(active) + if runErr != nil && !reload { + if previous != nil { + rc.logger.Error("Configuration failed to run; falling back to the previous configuration", "error", runErr) + failedRun = active + active, previous = previous, nil + continue + } + return runErr + } + if !active.runtime.Daemon { + return runErr + } + next := rc.takePending() + if next == nil { + continue + } + previous, active = active, next + } +} + +// notifyRunFailed hands a failed-to-run candidate to the reconciler goroutine. +func (rc *reconciler) notifyRunFailed(c *candidate) { + select { + case rc.runFailed <- c: + default: + } +} + +// pollDelay is the next poll delay: poll_interval ±10% jitter, at least MinPollInterval. +func (rc *reconciler) pollDelay() time.Duration { + interval, err := time.ParseDuration(rc.rcfg().PollInterval) + if err != nil { + interval = agentconfig.DefaultPollInterval + } + interval = max(interval, agentconfig.MinPollInterval) + jitter := time.Duration((rand.Float64()*0.2 - 0.1) * float64(interval)) + return max(interval+jitter, agentconfig.MinPollInterval) +} + +// loop is the daemon's reconcile goroutine: config file events (debounced) and the poll +// ticker, which retries a candidate whose failed backoff expired. It returns when ctx is done. +func (rc *reconciler) loop(ctx context.Context) { + var debounce <-chan time.Time + poll := time.NewTimer(rc.pollDelay()) + defer poll.Stop() + for { + select { + case <-ctx.Done(): + return + case <-rc.fileEvents: + if debounce == nil { + debounce = time.After(rc.debounce) + } + case <-debounce: + debounce = nil + rc.reconcile(ctx, triggerFile) + case c := <-rc.runFailed: + rc.onRunFailed(ctx, c) + case <-poll.C: + rc.reconcile(ctx, triggerPoll) + poll.Reset(rc.pollDelay()) + } + } +} + +// onRunFailed records that a prepared candidate failed to run (the run loop already fell back). +func (rc *reconciler) onRunFailed(_ context.Context, c *candidate) { + if c == nil { + return + } + // Back the candidate off: otherwise every poll re-prepares the new base, cancels the + // healthy configuration, fails and falls back again. + baseFingerprint := rc.base.fingerprint + if c.base != nil { + baseFingerprint = c.base.fingerprint + } + rc.startFailedBackoff(baseFingerprint) +} + +// reconcile handles one trigger (G3.4). All work runs in the reconciler goroutine, so two +// prepares never overlap. +func (rc *reconciler) reconcile(ctx context.Context, t trigger) { + if t == triggerFile { + base, err := loadBase(rc.cmd, rc.configPath) + if err != nil { + rc.logger.Error("Config file is invalid; keeping the running configuration", "error", err) + return + } + rc.setBase(base) + } + + active := rc.current() + if rc.sameAsActive(active) || rc.inFailedBackoff() { + return + } + + cand, aerr := rc.prepare(ctx, rc.base) + if aerr != nil { + rc.logger.Warn("Could not apply the configuration; keeping the running configuration", "status", aerr.Status, "reason", aerr.Reason, "error", aerr.Err) + rc.startFailedBackoff(rc.base.fingerprint) + return + } + rc.clearFailedBackoff() + if active != nil && active.identity == cand.identity { + rc.logger.Debug("Trigger did not change the effective configuration; recording it without a restart") + rc.adopt(active, cand) + return + } + rc.logger.Info("Applying the new configuration") + rc.swap(cand) +} + +// sameAsActive reports whether the active candidate already runs this base. +func (rc *reconciler) sameAsActive(active *candidate) bool { + return active != nil && active.base != nil && + bytes.Equal(active.base.raw, rc.base.raw) && active.base.fingerprint == rc.base.fingerprint +} + +func (rc *reconciler) logWarnings(warnings []agentconfig.FieldError) { + for _, w := range warnings { + if isToleratedFileRule(w) { + rc.logger.Warn("Ignoring a problem in the config file; the plugin is skipped", "path", w.Path, "error", w.Message) + continue + } + rc.logger.Warn("Ignoring a problem in the config file; the value is kept unchanged", "path", w.Path, "error", w.Message) + } +} + +// signalFile queues a file event without blocking. +func (rc *reconciler) signalFile() { + select { + case rc.fileEvents <- struct{}{}: + default: + } +} + +// watchFile watches the config file on a dedicated viper instance whose OnConfigChange only +// signals; loading always happens in the reconciler goroutine on a fresh viper. Known limit +// (R32 follow-up): viper stops watching after a Remove event. +func (rc *reconciler) watchFile() (stop func()) { + var stopped atomic.Bool + w := viper.New() + w.SetConfigFile(rc.configPath) + w.OnConfigChange(func(in fsnotify.Event) { + if stopped.Load() { + return + } + rc.logger.Debug("config file changed", "path", in.Name) + rc.signalFile() + }) + w.WatchConfig() + return func() { stopped.Store(true) } +} diff --git a/cmd/reconciler_test.go b/cmd/reconciler_test.go index 365cb5b..d89ddc0 100644 --- a/cmd/reconciler_test.go +++ b/cmd/reconciler_test.go @@ -6,6 +6,7 @@ import ( "io" "net/http" "os" + "path/filepath" "slices" "strings" "sync" @@ -14,10 +15,312 @@ import ( "testing" "time" + "github.com/compliance-framework/agent/internal/agentstate" "github.com/google/go-containerregistry/pkg/v1/remote" "github.com/google/uuid" ) +// fakePrefetcher records Prefetch calls and can be told to fail or to hang. +type fakePrefetcher struct { + mu sync.Mutex + calls int + err error + block bool +} + +func (f *fakePrefetcher) Prefetch(ctx context.Context, _ *agentConfig) error { + f.mu.Lock() + f.calls++ + err, block := f.err, f.block + f.mu.Unlock() + if block { + <-ctx.Done() + return ctx.Err() + } + return err +} + +func (f *fakePrefetcher) setBlock(block bool) { + f.mu.Lock() + f.block = block + f.mu.Unlock() +} + +// errStopRun ends a test's run func (and so rc.run) on a test-owned channel. +var errStopRun = errors.New("test stopped the run") + +func (f *fakePrefetcher) setErr(err error) { + f.mu.Lock() + f.err = err + f.mu.Unlock() +} + +func (f *fakePrefetcher) callCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.calls +} + +const reconcilerTestConfig = ` +daemon: true +api: + url: http://localhost:8080 +plugins: + ssh: + source: ./plugin-ssh + schedule: "%s" +` + +func newTestReconciler(t *testing.T, content string) (*reconciler, *fakePrefetcher, string) { + t.Helper() + dir := t.TempDir() + path := filepath.Join(dir, "config.yaml") + if err := os.WriteFile(path, []byte(content), 0o600); err != nil { + t.Fatal(err) + } + pf := &fakePrefetcher{} + rc := newReconciler(AgentCmd(), path, agentstate.Open(filepath.Join(dir, "state"), nil), pf, nil) + return rc, pf, path +} + +func configWithSchedule(schedule string) string { + return strings.Replace(reconcilerTestConfig, "%s", schedule, 1) +} + +// runningReconciler starts rc.run with a fake run func that blocks until cancelled and +// records every config it was given. +type runRecorder struct { + mu sync.Mutex + configs []*agentConfig + started chan *agentConfig + fail func(cfg *agentConfig) error +} + +func newRunRecorder() *runRecorder { + return &runRecorder{started: make(chan *agentConfig, 100)} +} + +func (r *runRecorder) run(ctx context.Context, cfg *agentConfig) error { + r.mu.Lock() + r.configs = append(r.configs, cfg) + fail := r.fail + r.mu.Unlock() + r.started <- cfg + if fail != nil { + if err := fail(cfg); err != nil { + return err + } + } + <-ctx.Done() + return nil +} + +func (r *runRecorder) runCount() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.configs) +} + +func waitStarted(t *testing.T, r *runRecorder) *agentConfig { + t.Helper() + select { + case cfg := <-r.started: + return cfg + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for a run to start") + return nil + } +} + +func expectNoStart(t *testing.T, r *runRecorder, within time.Duration) { + t.Helper() + select { + case cfg := <-r.started: + t.Fatalf("unexpected run started with %#v", cfg) + case <-time.After(within): + } +} + +func startReconciler(t *testing.T, rc *reconciler, rec *runRecorder) { + t.Helper() + active, err := rc.startup(context.Background()) + if err != nil { + t.Fatalf("startup: %v", err) + } + go func() { _ = rc.run(active, rec.run) }() + waitStarted(t, rec) +} + +func TestReconciler_InvalidFileAtStartupReturnsError(t *testing.T) { + rc, _, _ := newTestReconciler(t, "daemon: true\nplugins:\n ssh:\n source: ./x\n") + if _, err := rc.startup(context.Background()); err == nil || !strings.Contains(err.Error(), "/api") { + t.Fatalf("expected a config file error, got %v", err) + } +} + +func TestReconciler_InvalidEditKeepsRunning(t *testing.T) { + rc, pf, path := newTestReconciler(t, configWithSchedule("* * * * *")) + rec := newRunRecorder() + startReconciler(t, rc, rec) + + if err := os.WriteFile(path, []byte("daemon: true\nplugins: [\n"), 0o600); err != nil { + t.Fatal(err) + } + rc.reconcile(context.Background(), triggerFile) + expectNoStart(t, rec, 200*time.Millisecond) + if pf.callCount() != 1 { + t.Fatalf("an invalid file must not be prefetched, got %d calls", pf.callCount()) + } +} + +func TestReconciler_ValidEditCancelsOnce(t *testing.T) { + rc, _, path := newTestReconciler(t, configWithSchedule("* * * * *")) + rec := newRunRecorder() + startReconciler(t, rc, rec) + + if err := os.WriteFile(path, []byte(configWithSchedule("*/5 * * * *")), 0o600); err != nil { + t.Fatal(err) + } + rc.reconcile(context.Background(), triggerFile) + cfg := waitStarted(t, rec) + if got := *cfg.Plugins["ssh"].Schedule; got != "*/5 * * * *" { + t.Fatalf("new run has schedule %q", got) + } + // The same content again is not a change. + rc.reconcile(context.Background(), triggerFile) + expectNoStart(t, rec, 200*time.Millisecond) + if rec.runCount() != 2 { + t.Fatalf("expected exactly one reload, got %d runs", rec.runCount()) + } +} + +func TestReconciler_PrefetchFailureDoesNotCancel(t *testing.T) { + rc, pf, path := newTestReconciler(t, configWithSchedule("* * * * *")) + rec := newRunRecorder() + startReconciler(t, rc, rec) + + pf.setErr(errors.New("registry down")) + if err := os.WriteFile(path, []byte(configWithSchedule("*/5 * * * *")), 0o600); err != nil { + t.Fatal(err) + } + rc.reconcile(context.Background(), triggerFile) + expectNoStart(t, rec, 200*time.Millisecond) +} + +func TestReconciler_RunFailureFallsBackToPrevious(t *testing.T) { + rc, _, path := newTestReconciler(t, configWithSchedule("* * * * *")) + rec := newRunRecorder() + rec.fail = func(cfg *agentConfig) error { + if *cfg.Plugins["ssh"].Schedule == "*/5 * * * *" { + return errors.New("cache wiped after prepare") + } + return nil + } + startReconciler(t, rc, rec) + + if err := os.WriteFile(path, []byte(configWithSchedule("*/5 * * * *")), 0o600); err != nil { + t.Fatal(err) + } + rc.reconcile(context.Background(), triggerFile) + if got := *waitStarted(t, rec).Plugins["ssh"].Schedule; got != "*/5 * * * *" { + t.Fatalf("expected the new config to be tried, got %q", got) + } + if got := *waitStarted(t, rec).Plugins["ssh"].Schedule; got != "* * * * *" { + t.Fatalf("expected a fallback to the previous config, got %q", got) + } +} + +func TestReconciler_RapidFileEventsRace(t *testing.T) { + rc, _, path := newTestReconciler(t, configWithSchedule("* * * * *")) + rec := newRunRecorder() + rc.debounce = 5 * time.Millisecond + startReconciler(t, rc, rec) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go rc.loop(ctx) + + const last = "*/9 * * * *" // unique: no intermediate reload can satisfy the check + for i := 0; i < 50; i++ { + schedule := "*/5 * * * *" + switch { + case i == 49: + schedule = last + case i%2 == 0: + schedule = "*/7 * * * *" + } + if err := os.WriteFile(path, []byte(configWithSchedule(schedule)), 0o600); err != nil { + t.Fatal(err) + } + rc.signalFile() + time.Sleep(time.Millisecond) + } + // The last write wins: it is started, and it is still what runs once reloads quiesce. + deadline := time.After(5 * time.Second) + for applied := false; !applied; { + select { + case cfg := <-rec.started: + applied = *cfg.Plugins["ssh"].Schedule == last + case <-deadline: + t.Fatal("the last file write was never applied") + } + } + expectNoStart(t, rec, 300*time.Millisecond) + if got := *rc.current().runtime.Plugins["ssh"].Schedule; got != last { + t.Fatalf("the final running config has schedule %q, want %q", got, last) + } +} + +// TestReconciler_FileOnlyRunFailureBacksOff: a file edit that prepares but fails to run is +// not re-applied on every poll (it would cancel the healthy config and fall back each time). +func TestReconciler_FileOnlyRunFailureBacksOff(t *testing.T) { + rc, pf, path := newTestReconciler(t, configWithSchedule("* * * * *")) + rec := newRunRecorder() + rec.fail = func(cfg *agentConfig) error { + if *cfg.Plugins["ssh"].Schedule == "*/5 * * * *" { + return errors.New("cron setup failed") + } + return nil + } + startReconciler(t, rc, rec) + + if err := os.WriteFile(path, []byte(configWithSchedule("*/5 * * * *")), 0o600); err != nil { + t.Fatal(err) + } + rc.reconcile(context.Background(), triggerFile) + waitStarted(t, rec) // the new config, which fails + waitStarted(t, rec) // the fallback + select { + case c := <-rc.runFailed: + rc.onRunFailed(context.Background(), c) + case <-time.After(5 * time.Second): + t.Fatal("the run failure was never notified") + } + calls := pf.callCount() + for i := 0; i < 3; i++ { + rc.reconcile(context.Background(), triggerPoll) + } + expectNoStart(t, rec, 200*time.Millisecond) + if pf.callCount() != calls { + t.Fatalf("a file-only run failure must be backed off, got %d prepares", pf.callCount()-calls) + } +} + +// TestReconciler_StartupDownloadFailureReported: a download failure of the file-only config +// at startup is handed to onStartupFailure (startup-failure evidence, as Run did on main). +func TestReconciler_StartupDownloadFailureReported(t *testing.T) { + rc, pf, _ := newTestReconciler(t, configWithSchedule("* * * * *")) + pf.setErr(&downloadError{source: "./plugin-ssh", err: errors.New("registry down")}) + var gotCfg *agentConfig + var gotErr error + rc.onStartupFailure = func(_ context.Context, cfg *agentConfig, err error) { gotCfg, gotErr = cfg, err } + if _, err := rc.startup(context.Background()); err == nil { + t.Fatal("startup must fail") + } + if gotCfg == nil || gotCfg.Plugins["ssh"] == nil || !strings.Contains(gotErr.Error(), "registry down") { + t.Fatalf("onStartupFailure was not called with the runtime and error: %v %v", gotCfg, gotErr) + } +} + func TestReportStartupFailureMarksPluginsAndSendsEvidence(t *testing.T) { var mu sync.Mutex var bodies []string @@ -42,6 +345,16 @@ func TestReportStartupFailureMarksPluginsAndSendsEvidence(t *testing.T) { } } +func TestReconciler_BindCancelsWhenSwapRacedIn(t *testing.T) { + rc, _, _ := newTestReconciler(t, configWithSchedule("* * * * *")) + rc.swap(&candidate{}) + ctx, cancel := context.WithCancel(context.Background()) + rc.bind(&candidate{}, cancel) + if ctx.Err() == nil { + t.Fatal("bind must cancel a run when a candidate is already pending") + } +} + func TestPluginEnvironDropsAPICredentials(t *testing.T) { got := pluginEnviron([]string{ "CCF_API_AUTH_CLIENT_SECRET=s", diff --git a/docs/configuration.md b/docs/configuration.md index 6503483..5a271af 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -182,7 +182,8 @@ cannot contain dots. A plugin `schedule` in the file that does not parse does not stop the agent: that plugin is skipped, the others run, and the problem is logged as a warning (R34). A few other file values that always loaded are also only warnings, and are kept unchanged: a negative `verbosity` (`-1` logs WARN and above) and a literal `${env:...}` outside -`plugins.*.config`. Every other invalid value in the file (for example a missing `api.url`) still fails startup. +`plugins.*.config`. Every other invalid value in the file (for example a missing `api.url`) still fails startup, and on +a live reload the agent keeps running its last good configuration. ## State directory and instance ID