diff --git a/.coverage-baseline b/.coverage-baseline index 106c8e6c..9274442d 100644 --- a/.coverage-baseline +++ b/.coverage-baseline @@ -1 +1 @@ -78.7 +78.9 diff --git a/internal/controller/gittarget_branch_target_metric_test.go b/internal/controller/gittarget_branch_target_metric_test.go index 279f3741..9fa599c4 100644 --- a/internal/controller/gittarget_branch_target_metric_test.go +++ b/internal/controller/gittarget_branch_target_metric_test.go @@ -3,6 +3,7 @@ package controller import ( + "context" "testing" "github.com/stretchr/testify/assert" @@ -38,7 +39,11 @@ func TestGitTargetCleanup_ForgetsTheBranchTargetMapping(t *testing.T) { require.True(t, ok, "precondition: the mapping is published") r := &GitTargetReconciler{} - r.cleanupDeletedGitTarget(k8stypes.NamespacedName{Namespace: "team-a", Name: "mirror"}, logf.Log) + _ = r.cleanupDeletedGitTarget( + context.Background(), + k8stypes.NamespacedName{Namespace: "team-a", Name: "mirror"}, + logf.Log, + ) _, ok = telemetry.CollectInt64Sum(reader, branchTargetsMetricName, map[string]string{ "gittarget_namespace": "team-a", diff --git a/internal/controller/gittarget_controller.go b/internal/controller/gittarget_controller.go index fd6e0655..3dc53313 100644 --- a/internal/controller/gittarget_controller.go +++ b/internal/controller/gittarget_controller.go @@ -167,7 +167,7 @@ func (r *GitTargetReconciler) Reconcile(ctx context.Context, req ctrl.Request) ( var target configbutleraiv1alpha3.GitTarget if err := r.Get(ctx, req.NamespacedName, &target); err != nil { - return r.handleFetchError(err, log, req.NamespacedName) + return r.handleFetchError(ctx, err, log, req.NamespacedName) } st := beginStatus(r.Client, r.Recorder, &target) @@ -1141,23 +1141,37 @@ func hasAgeKeyEntry(data map[string][]byte) bool { // handleFetchError handles errors from fetching GitTarget. func (r *GitTargetReconciler) handleFetchError( + ctx context.Context, err error, log logr.Logger, namespacedName k8stypes.NamespacedName, ) (ctrl.Result, error) { if client.IgnoreNotFound(err) == nil { - r.cleanupDeletedGitTarget(namespacedName, log) + cleanupErr := r.cleanupDeletedGitTarget(ctx, namespacedName, log) log.Info("GitTarget not found, was likely deleted", "namespacedName", namespacedName) - return ctrl.Result{}, nil + // A deleted object still has a reconcile key, so returning the error requeues this same + // pass with backoff. That matters because the worker sweep is the ONLY thing that retires + // a worker: if its read of the GitTargets failed, swallowing the error here would leave + // the worker and its clone standing until another GitTarget somewhere is deleted, which + // nothing guarantees will ever happen. + return ctrl.Result{}, cleanupErr } log.Error(err, "unable to fetch GitTarget", "namespacedName", namespacedName) return ctrl.Result{}, err } +// cleanupDeletedGitTarget releases everything this reconciler and its data plane held for a +// GitTarget that is gone. +// +// Every step is idempotent, and they all run before the error is returned: the in-memory releases +// cannot fail, and the one step that can — the worker sweep, which reads the API — must not stop +// the others from happening. The error it returns is the caller's signal to come back, not a +// report that nothing was cleaned up. func (r *GitTargetReconciler) cleanupDeletedGitTarget( + ctx context.Context, namespacedName k8stypes.NamespacedName, log logr.Logger, -) { +) error { gitDest := types.NewResourceReference(namespacedName.Name, namespacedName.Namespace) // Unconditionally, and before the EventRouter check below: the tracker is the reconciler's own // memory, so it must be released for a deleted target whether or not a data plane is wired. The @@ -1169,8 +1183,23 @@ func (r *GitTargetReconciler) cleanupDeletedGitTarget( // branch's push failures to an object that no longer exists. telemetry.ForgetBranchTarget(namespacedName.Namespace, namespacedName.Name) + // The branch worker outlives its GitTarget unless something says otherwise, and this reconcile + // — the one the delete watch event produced — is the moment the set of needed workers can have + // shrunk. The sweep decides from the API, so a branch several GitTargets share keeps its worker + // while any of them is still listed. It runs ahead of the EventRouter check below because it is + // the WorkerManager's business, not the data plane's. + var sweepErr error + if r.WorkerManager != nil { + if err := r.WorkerManager.ReconcileWorkers(ctx); err != nil { + log.V(1).Info("Could not sweep branch workers after a GitTarget was deleted", + "error", err.Error()) + sweepErr = fmt.Errorf("sweep branch workers after %s was deleted: %w", + namespacedName.String(), err) + } + } + if r.EventRouter == nil { - return + return sweepErr } r.EventRouter.UnregisterGitTargetEventStream(gitDest) @@ -1183,6 +1212,7 @@ func (r *GitTargetReconciler) cleanupDeletedGitTarget( } log.V(1).Info("Cleaned up in-memory state for deleted GitTarget", "gitDest", gitDest.String()) + return sweepErr } func clampIntToInt32(value int) int32 { diff --git a/internal/controller/gittarget_worker_cleanup_test.go b/internal/controller/gittarget_worker_cleanup_test.go new file mode 100644 index 00000000..1c36ed12 --- /dev/null +++ b/internal/controller/gittarget_worker_cleanup_test.go @@ -0,0 +1,159 @@ +// SPDX-License-Identifier: Apache-2.0 + +package controller + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/go-logr/logr" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + k8stypes "k8s.io/apimachinery/pkg/types" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" + + configbutleraiv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3" + "github.com/ConfigButler/gitops-reverser/internal/git" + "github.com/ConfigButler/gitops-reverser/internal/types" +) + +// TestCleanupDeletedGitTarget_StopsAWorkerNothingNeedsAnyMore is the wiring, and the wiring is the +// whole bug: the sweep existed and was never called, so a branch worker — its goroutine, its event +// queue and its on-disk clone — outlived every GitTarget that justified it, until the process +// restarted. The reconcile that the delete watch event produces is where the set of needed workers +// can have shrunk, so that is where the sweep belongs. +func TestCleanupDeletedGitTarget_StopsAWorkerNothingNeedsAnyMore(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, clientgoscheme.AddToScheme(scheme)) + require.NoError(t, configbutleraiv1alpha3.AddToScheme(scheme)) + k8sClient := fake.NewClientBuilder().WithScheme(scheme).Build() + + workers := git.NewWorkerManager(k8sClient, logr.Discard(), git.BranchWorkerLimits{}, + types.SensitiveResourcePolicy{}) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + go func() { _ = workers.Start(ctx) }() + time.Sleep(100 * time.Millisecond) + + require.NoError(t, workers.EnsureWorker(ctx, "repo1", "shop", "main")) + _, exists := workers.GetWorkerForTarget("repo1", "shop", "main") + require.True(t, exists) + + r := &GitTargetReconciler{Client: k8sClient, WorkerManager: workers} + _ = r.cleanupDeletedGitTarget(ctx, + k8stypes.NamespacedName{Name: "checkout", Namespace: "shop"}, logr.Discard()) + + _, exists = workers.GetWorkerForTarget("repo1", "shop", "main") + assert.False(t, exists, "the GitTarget is gone and nothing else lists this branch") +} + +// TestCleanupDeletedGitTarget_LeavesAWorkerAnotherGitTargetStillUses. The sweep reads the API +// rather than counting registrations, and this is why: one worker serves every GitTarget on its +// (provider, branch), so one deletion says nothing about the rest. +func TestCleanupDeletedGitTarget_LeavesAWorkerAnotherGitTargetStillUses(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, clientgoscheme.AddToScheme(scheme)) + require.NoError(t, configbutleraiv1alpha3.AddToScheme(scheme)) + + sibling := &configbutleraiv1alpha3.GitTarget{ + ObjectMeta: metav1.ObjectMeta{Name: "invoices", Namespace: "shop"}, + } + sibling.Spec.GitProviderRef.Name = "repo1" + sibling.Spec.Branch = "main" + k8sClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(sibling).Build() + + workers := git.NewWorkerManager(k8sClient, logr.Discard(), git.BranchWorkerLimits{}, + types.SensitiveResourcePolicy{}) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + go func() { _ = workers.Start(ctx) }() + time.Sleep(100 * time.Millisecond) + require.NoError(t, workers.EnsureWorker(ctx, "repo1", "shop", "main")) + + r := &GitTargetReconciler{Client: k8sClient, WorkerManager: workers} + _ = r.cleanupDeletedGitTarget(ctx, + k8stypes.NamespacedName{Name: "checkout", Namespace: "shop"}, logr.Discard()) + + _, exists := workers.GetWorkerForTarget("repo1", "shop", "main") + assert.True(t, exists, "invoices still writes to this branch") +} + +// TestCleanupDeletedGitTarget_WithoutAWorkerManager covers the CLI and the tests: the cleanup runs +// on a reconciler with no data plane wired, and must not reach for one. +func TestCleanupDeletedGitTarget_WithoutAWorkerManager(t *testing.T) { + r := &GitTargetReconciler{} + assert.NotPanics(t, func() { + _ = r.cleanupDeletedGitTarget(context.Background(), + k8stypes.NamespacedName{Name: "checkout", Namespace: "shop"}, logr.Discard()) + }) +} + +// TestCleanupDeletedGitTarget_RetriesWhenTheSweepCannotRead. The sweep is the only thing that ever +// retires a worker, so a read that failed must come back. A deleted object still has a reconcile +// key, so returning the error requeues this same pass; swallowing it would leave the worker and +// its clone standing until some other GitTarget happens to be deleted, which nothing guarantees. +func TestCleanupDeletedGitTarget_RetriesWhenTheSweepCannotRead(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, clientgoscheme.AddToScheme(scheme)) + require.NoError(t, configbutleraiv1alpha3.AddToScheme(scheme)) + k8sClient := fake.NewClientBuilder().WithScheme(scheme). + WithInterceptorFuncs(interceptor.Funcs{ + List: func(context.Context, client.WithWatch, client.ObjectList, ...client.ListOption) error { + return errors.New("the cache is not having it") + }, + }).Build() + + workers := git.NewWorkerManager(k8sClient, logr.Discard(), git.BranchWorkerLimits{}, + types.SensitiveResourcePolicy{}) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + go func() { _ = workers.Start(ctx) }() + time.Sleep(100 * time.Millisecond) + require.NoError(t, workers.EnsureWorker(ctx, "repo1", "shop", "main")) + + r := &GitTargetReconciler{Client: k8sClient, WorkerManager: workers} + err := r.cleanupDeletedGitTarget(ctx, + k8stypes.NamespacedName{Name: "checkout", Namespace: "shop"}, logr.Discard()) + + require.Error(t, err, "the caller has to come back, or nothing ever retires this worker") + _, exists := workers.GetWorkerForTarget("repo1", "shop", "main") + assert.True(t, exists, "and the worker is still there, which is why the retry matters") +} + +// TestHandleFetchError_RequeuesAFailedCleanup is the wiring of the rule above: the reconcile must +// surface the error rather than reporting success for a pass that did not finish. +func TestHandleFetchError_RequeuesAFailedCleanup(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, clientgoscheme.AddToScheme(scheme)) + require.NoError(t, configbutleraiv1alpha3.AddToScheme(scheme)) + k8sClient := fake.NewClientBuilder().WithScheme(scheme). + WithInterceptorFuncs(interceptor.Funcs{ + List: func(context.Context, client.WithWatch, client.ObjectList, ...client.ListOption) error { + return errors.New("the cache is not having it") + }, + }).Build() + + workers := git.NewWorkerManager(k8sClient, logr.Discard(), git.BranchWorkerLimits{}, + types.SensitiveResourcePolicy{}) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + go func() { _ = workers.Start(ctx) }() + time.Sleep(100 * time.Millisecond) + require.NoError(t, workers.EnsureWorker(ctx, "repo1", "shop", "main")) + + r := &GitTargetReconciler{Client: k8sClient, WorkerManager: workers} + _, err := r.handleFetchError(ctx, apierrors.NewNotFound( + schema.GroupResource{Resource: "gittargets"}, "checkout"), + logr.Discard(), k8stypes.NamespacedName{Name: "checkout", Namespace: "shop"}) + + assert.Error(t, err) +} diff --git a/internal/git/branch_worker.go b/internal/git/branch_worker.go index ea9a557f..72eeef2d 100644 --- a/internal/git/branch_worker.go +++ b/internal/git/branch_worker.go @@ -8,6 +8,7 @@ import ( "encoding/hex" "errors" "fmt" + "os" "path/filepath" "strings" "sync" @@ -327,18 +328,58 @@ func NewBranchWorker( } } +// workerStateRoot is where every worker's on-disk state lives. It is a variable rather than a +// constant so a test can point it at its own directory: the tests that exercise reclaiming a +// retired worker's checkout delete directories, and doing that under the shared root would take +// out the state of anything else using it, including a developer's running operator. +// +//nolint:gochecknoglobals // a test seam, like pushAtomicFn above +var workerStateRoot = filepath.Join("/tmp", "gitops-reverser-workers") + func repoCacheKey(remoteURL string) string { sum := sha256.Sum256([]byte(strings.TrimSpace(remoteURL))) return hex.EncodeToString(sum[:16]) } +// branchPathComponent renders a branch name as ONE directory component. +// +// Git branch names contain slashes — `release/v1` is ordinary — and joining one into a path makes +// its worker's directory a CHILD of the `release` worker's. That is harmless while nothing deletes +// anything, and it is a data-loss bug the moment something does: retiring `release` would take the +// live `release/v1` worker's checkout with it. +// +// The readable prefix is for whoever is looking at the directory; the hash of the raw name is what +// makes it unambiguous, including for the names sanitising would otherwise collapse together +// (`release/v1` and `release-v1`). +func branchPathComponent(branch string) string { + sum := sha256.Sum256([]byte(branch)) + digest := hex.EncodeToString(sum[:8]) + + readable := make([]rune, 0, len(branch)) + for _, r := range branch { + switch { + case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '-', r == '_': + readable = append(readable, r) + default: + readable = append(readable, '-') + } + } + if len(readable) > branchComponentReadableMax { + readable = readable[:branchComponentReadableMax] + } + return string(readable) + "-" + digest +} + +// branchComponentReadableMax caps the readable half so a long branch name cannot push the path +// past what a filesystem accepts; the digest that follows is what carries the identity. +const branchComponentReadableMax = 40 + func (w *BranchWorker) repoRootPath() string { return filepath.Join( - "/tmp", - "gitops-reverser-workers", + workerStateRoot, w.GitProviderNamespace, w.GitProviderRef, - w.Branch, + branchPathComponent(w.Branch), "repos", ) } @@ -347,6 +388,33 @@ func (w *BranchWorker) repoPathForRemote(remoteURL string) string { return filepath.Join(w.repoRootPath(), repoCacheKey(remoteURL)) } +// removeLocalState deletes everything this worker kept on disk: the clone of every remote it +// worked with, under its own (provider namespace, provider, branch) directory. +// +// It is called only when a worker is being retired because nothing needs it any more, never on +// shutdown — a restart wants the clones it left behind, and re-cloning every branch on every +// restart is exactly the cost the on-disk cache exists to avoid. +// +// The identity guard is not defensive programming. repoRootPath joins the three identity fields +// into a fixed prefix, so a worker with empty fields — which tests build — resolves to the ROOT of +// every worker's state, and deleting that would take out the live workers next to it. +func (w *BranchWorker) removeLocalState() { + if w.GitProviderNamespace == "" || w.GitProviderRef == "" || w.Branch == "" { + w.Log.V(1).Info("Not removing worker state: the worker has no complete identity") + return + } + // The branch directory rather than the repos/ tree inside it, so nothing is left behind to + // accumulate one empty directory per branch the operator ever mirrored. + path := filepath.Dir(w.repoRootPath()) + if err := os.RemoveAll(path); err != nil { + // Worth knowing and not worth failing for: the worker is gone either way, and what is + // left is disk rather than anything that can produce a wrong answer. + w.Log.Error(err, "Could not remove the retired worker's checkout", "path", path) + return + } + w.Log.Info("Removed the retired worker's checkout", "path", path) +} + // Start begins processing events. func (w *BranchWorker) Start(parentCtx context.Context) error { w.mu.Lock() diff --git a/internal/git/branch_worker_queue_depth_test.go b/internal/git/branch_worker_queue_depth_test.go index 264f27dc..a931088b 100644 --- a/internal/git/branch_worker_queue_depth_test.go +++ b/internal/git/branch_worker_queue_depth_test.go @@ -40,7 +40,7 @@ func TestReleaseHandledItem_PublishesRetainedWorkBeforeReleasingInflight(t *test // The git_queue_depth source must not wait on the worker lifecycle lock. // -// queueDepthSamples runs on the metric SDK's collection goroutine. UnregisterTarget used to hold +// queueDepthSamples runs on the metric SDK's collection goroutine. Worker removal used to hold // m.mu across worker.Stop(), which waits for the loop goroutine to finish, so a slow shutdown // blocked collection for its whole duration — the staleness the observable gauge exists to remove, // reintroduced through the manager's own lock. @@ -58,16 +58,17 @@ func TestQueueDepthSamples_NotBlockedByASlowWorkerShutdown(t *testing.T) { release := func() { once.Do(w.wg.Done) } t.Cleanup(release) - var unregistered sync.WaitGroup - unregistered.Add(1) + var removed sync.WaitGroup + removed.Add(1) go func() { - defer unregistered.Done() - _ = manager.UnregisterTarget("", "", "test-provider", "test-ns", "main") + defer removed.Done() + manager.removeWorkers([]BranchKey{{RepoNamespace: "test-ns", RepoName: "test-provider", Branch: "main"}}, + "test") }() sampled := make(chan int, 1) go func() { - // Give the unregister a moment to reach Stop() before sampling. + // Give the removal a moment to reach Stop() before sampling. time.Sleep(50 * time.Millisecond) sampled <- len(manager.queueDepthSamples()) }() @@ -79,7 +80,7 @@ func TestQueueDepthSamples_NotBlockedByASlowWorkerShutdown(t *testing.T) { } release() - unregistered.Wait() + removed.Wait() } // A replacement worker must not start while the one it replaces is still stopping. @@ -120,10 +121,10 @@ func TestEnsureWorker_WaitsForTheWorkerItReplacesToStop(t *testing.T) { stopping := make(chan struct{}) go func() { close(stopping) - _ = manager.UnregisterTarget("", "", "test-provider", "test-ns", "main") + manager.removeWorkers([]BranchKey{key}, "test") }() <-stopping - time.Sleep(50 * time.Millisecond) // let the unregister reach Stop() + time.Sleep(50 * time.Millisecond) // let the removal reach Stop() ensured := make(chan struct{}) go func() { @@ -144,3 +145,58 @@ func TestEnsureWorker_WaitsForTheWorkerItReplacesToStop(t *testing.T) { t.Fatal("the replacement never started after its predecessor stopped") } } + +// A sweep must not stop a worker EnsureWorker has just handed to a caller. +// +// ReconcileWorkers chooses its orphans from the map and then removes them. If EnsureWorker can run +// between those two steps it finds the key present, returns "already there" without creating +// anything, and the sweep then stops the worker that GitTarget was just told it has — leaving it +// with no worker until its next reconcile, and its live events failing to route in the meantime. +// Holding lifecycleMu across the selection AND the removal is what closes it, and the seam is what +// makes that observable: without the lock the EnsureWorker below completes inside the window and +// the key ends up with no worker at all. +func TestReconcileWorkers_DoesNotStopAWorkerEnsureWorkerJustHandedOut(t *testing.T) { + withTemporaryWorkerStateRoot(t) + key := BranchKey{RepoNamespace: "test-ns", RepoName: "test-provider", Branch: "main"} + scheme := runtime.NewScheme() + require.NoError(t, configv1alpha3.AddToScheme(scheme)) + manager := &WorkerManager{ + Log: logr.Discard(), + ctx: context.Background(), + Client: fake.NewClientBuilder().WithScheme(scheme).Build(), // no GitTargets: every worker is an orphan + workers: map[BranchKey]*BranchWorker{}, + } + retiring := newMetricsTestWorker() + manager.workers[key] = retiring + + ensured := make(chan struct{}) + afterOrphanSelection = func() { + go func() { + defer close(ensured) + _ = manager.EnsureWorker(context.Background(), "test-provider", "test-ns", "main") + }() + // Long enough for an unguarded EnsureWorker to finish inside the window; a guarded one is + // blocked on lifecycleMu and finishes after the sweep instead. + time.Sleep(200 * time.Millisecond) + } + t.Cleanup(func() { afterOrphanSelection = nil }) + + require.NoError(t, manager.ReconcileWorkers(context.Background())) + + select { + case <-ensured: + case <-time.After(5 * time.Second): + t.Fatal("EnsureWorker never returned") + } + + manager.mu.RLock() + got, exists := manager.workers[key] + manager.mu.RUnlock() + t.Cleanup(func() { + if exists && got != retiring { + got.Stop() + } + }) + require.True(t, exists, "the target that asked for a worker must have one, not a slot the sweep emptied") + assert.NotSame(t, retiring, got, "and it must be a fresh one, not the worker the sweep retired") +} diff --git a/internal/git/worker_manager.go b/internal/git/worker_manager.go index a0edf856..bc6b816c 100644 --- a/internal/git/worker_manager.go +++ b/internal/git/worker_manager.go @@ -234,31 +234,6 @@ func (m *WorkerManager) SetRemoteReporter(reporter RemoteReporter) { m.remoteReporter = reporter } -// RegisterTarget ensures a worker exists for the target's (provider, branch) -// and registers the target with that worker. -// This is called by GitTarget controller when a target becomes Ready. -func (m *WorkerManager) RegisterTarget( - ctx context.Context, - targetName, targetNamespace string, - providerName, providerNamespace string, - branch, path string, -) error { - if err := m.EnsureWorker(ctx, providerName, providerNamespace, branch); err != nil { - return err - } - - m.Log.Info("GitTarget registered with branch worker", - "target", fmt.Sprintf("%s/%s", targetNamespace, targetName), - "workerKey", BranchKey{ - RepoNamespace: providerNamespace, - RepoName: providerName, - Branch: branch, - }.String(), - "path", path) - - return nil -} - // EnsureWorker ensures a worker exists for the given (provider, branch). // Worker creation/start is protected by the manager lock. func (m *WorkerManager) EnsureWorker( @@ -314,41 +289,47 @@ func (m *WorkerManager) EnsureWorker( return nil } -// UnregisterTarget removes a GitTarget from its worker. -// Destroys the worker if it was the last target using it. -// This is called by GitTarget controller when a target is deleted. -func (m *WorkerManager) UnregisterTarget( - _, _ string, - providerName, providerNamespace string, - branch string, -) error { - key := BranchKey{ - RepoNamespace: providerNamespace, - RepoName: providerName, - Branch: branch, - } - - // lifecycleMu is held across the Stop so EnsureWorker cannot start a replacement while this - // worker is still draining. m.mu is only held to detach: queueDepthSamples reads it on the - // scrape goroutine, so holding THAT across Stop() stalls collection instead of reporting it. +// removeWorkers detaches the named workers and stops them. +// +// The lock discipline is the whole reason this is one function rather than a loop at each call +// site. lifecycleMu is held across the Stops so EnsureWorker cannot start a replacement while the +// worker it replaces is still draining — they would share an on-disk clone. m.mu is held only to +// detach: queueDepthSamples reads it on the metric SDK's collection goroutine, so holding THAT +// across Stop(), which waits for the loop goroutine, stalls collection for the whole shutdown. +// +// A key that names no worker is skipped, so a caller may pass a key it is not sure about. +func (m *WorkerManager) removeWorkers(keys []BranchKey, reason string) { m.lifecycleMu.Lock() defer m.lifecycleMu.Unlock() + m.removeWorkersLocked(keys, reason) +} + +// removeWorkersLocked is removeWorkers for a caller that already holds lifecycleMu, so a sweep can +// hold it across choosing the workers to retire and retiring them. +func (m *WorkerManager) removeWorkersLocked(keys []BranchKey, reason string) { + if len(keys) == 0 { + return + } m.mu.Lock() - worker, exists := m.workers[key] - delete(m.workers, key) + detached := make(map[BranchKey]*BranchWorker, len(keys)) + for _, key := range keys { + if worker, exists := m.workers[key]; exists { + detached[key] = worker + delete(m.workers, key) + } + } m.mu.Unlock() - if !exists { - return nil + for key, worker := range detached { + m.Log.Info("Stopping branch worker", "key", key.String(), "reason", reason) + worker.Stop() + // After Stop, never before: Stop waits for the loop goroutine, so until it returns the + // worktree can still be written. A worker being retired takes its clone with it — the + // goroutine was only half the leak, and the checkout is the half that survives a + // restart. + worker.removeLocalState() } - - // Worker no longer tracks targets internally - always destroy worker - // since WorkerManager handles all lifecycle decisions - m.Log.Info("Unregistering target, destroying worker", "key", key.String()) - worker.Stop() - - return nil } // GetWorkerForTarget finds the worker for a target's (provider, branch). @@ -371,10 +352,31 @@ func (m *WorkerManager) GetWorkerForTarget( return worker, exists } -// ReconcileWorkers checks active GitTargets and cleans up orphaned workers. -// This ensures workers are removed when their GitTargets are deleted. +// afterOrphanSelection runs between choosing the workers to retire and retiring them. It is nil in +// production and exists because that gap is the ONLY place the lock discipline in ReconcileWorkers +// can be observed: with lifecycleMu held across both steps a concurrent EnsureWorker blocks here, +// and without it that EnsureWorker hands a caller the very worker about to be stopped. The seam +// follows pushAtomicFn, which the push path already uses for the same reason. +// +//nolint:gochecknoglobals // a test seam, like pushAtomicFn +var afterOrphanSelection func() + +// ReconcileWorkers stops every worker no live GitTarget still needs. +// +// It is the ONLY thing that removes a worker before shutdown, and it is driven from the GitTarget +// reconcile's deleted-object path: that reconcile is what the delete watch event produces, so the +// sweep runs exactly when the set of needed workers can have shrunk. A worker that outlives its +// last GitTarget is not free — it holds a goroutine, an event queue and an on-disk clone until the +// process restarts. +// +// It decides from the API rather than from a count of registrations, which is what makes it +// correct for a branch SHARED by several GitTargets: deleting one of them leaves the others +// listed, so the worker they share is still needed and is left alone. +// +// It fails safe. A List that errors stops nothing: the alternative — treating "I could not read +// the targets" as "there are no targets" — would take down every live worker in the process. func (m *WorkerManager) ReconcileWorkers(ctx context.Context) error { - // The List and the Stops stay outside m.mu, for the reason in UnregisterTarget. + // The List and the Stops stay outside m.mu, for the reason in removeWorkers. // Get all GitTargets var targetList configv1alpha3.GitTargetList @@ -401,30 +403,39 @@ func (m *WorkerManager) ReconcileWorkers(ctx context.Context) error { neededWorkers[key] = true } - // Detach the orphans under m.mu; stop them after releasing it, with lifecycleMu held so no - // replacement starts while one is draining. + // lifecycleMu is held across the SELECTION as well as the removal, and the two must not be + // separated. EnsureWorker takes this lock, finds the key present and returns "already there" + // without creating anything; if it ran between a selection made outside the lock and the + // removal, this sweep would then stop the worker that GitTarget had just been told it has, + // leaving it with no worker until its next reconcile. m.lifecycleMu.Lock() defer m.lifecycleMu.Unlock() - m.mu.Lock() - orphans := make(map[BranchKey]*BranchWorker) - for key, worker := range m.workers { + m.mu.RLock() + orphans := make([]BranchKey, 0, len(m.workers)) + for key := range m.workers { if !neededWorkers[key] { - orphans[key] = worker - delete(m.workers, key) + orphans = append(orphans, key) } } - remaining := len(m.workers) - m.mu.Unlock() + m.mu.RUnlock() - for key, worker := range orphans { - m.Log.Info("Cleaning up orphaned worker", "key", key.String()) - worker.Stop() + // The window the lock above closes, opened on request so a test can stand in it. Nil in + // production; see afterOrphanSelection. + if afterOrphanSelection != nil { + afterOrphanSelection() } + m.removeWorkersLocked(orphans, "no GitTarget needs this worker any more") + + m.mu.RLock() + remaining := len(m.workers) + m.mu.RUnlock() + m.Log.V(1).Info("Worker reconciliation complete", "activeWorkers", remaining, - "neededWorkers", len(neededWorkers)) + "neededWorkers", len(neededWorkers), + "stopped", len(orphans)) return nil } diff --git a/internal/git/worker_manager_test.go b/internal/git/worker_manager_test.go index b04f6571..da3f760f 100644 --- a/internal/git/worker_manager_test.go +++ b/internal/git/worker_manager_test.go @@ -4,7 +4,10 @@ package git import ( "context" + "errors" + "os" "path/filepath" + "strings" "testing" "time" @@ -16,6 +19,7 @@ import ( clientgoscheme "k8s.io/client-go/kubernetes/scheme" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" configv1alpha3 "github.com/ConfigButler/gitops-reverser/api/v1alpha3" "github.com/ConfigButler/gitops-reverser/internal/types" @@ -97,8 +101,8 @@ func createTargetForRegister( require.NoError(t, k8sClient.Create(ctx, target)) } -// TestWorkerManagerRegisterTarget verifies worker registration. -func TestWorkerManagerRegisterTarget(t *testing.T) { +// TestEnsureWorker_CreatesAWorkerWithTheIdentityItWasAskedFor. +func TestEnsureWorker_CreatesAWorkerWithTheIdentityItWasAskedFor(t *testing.T) { scheme := setupScheme() client := fake.NewClientBuilder().WithScheme(scheme).Build() log := logr.Discard() @@ -116,10 +120,7 @@ func TestWorkerManagerRegisterTarget(t *testing.T) { createTargetForRegister(ctx, t, client, "target1", "repo1", "main", "clusters/prod") // Register first target - err := manager.RegisterTarget(ctx, - "target1", "default", - "repo1", "gitops-system", - "main", "clusters/prod") + err := manager.EnsureWorker(ctx, "repo1", "gitops-system", "main") if err != nil { t.Fatalf("Failed to register target: %v", err) } @@ -171,18 +172,12 @@ func TestWorkerManagerMultipleTargetsSameBranch(t *testing.T) { createTargetForRegister(ctx, t, client, "target-infra", "shared-repo", "main", "infra/") // Register two targets for same repo+branch, different paths - err := manager.RegisterTarget(ctx, - "target-apps", "default", - "shared-repo", "gitops-system", - "main", "apps/") + err := manager.EnsureWorker(ctx, "shared-repo", "gitops-system", "main") if err != nil { t.Fatalf("Failed to register target-apps: %v", err) } - err = manager.RegisterTarget(ctx, - "target-infra", "default", - "shared-repo", "gitops-system", - "main", "infra/") + err = manager.EnsureWorker(ctx, "shared-repo", "gitops-system", "main") if err != nil { t.Fatalf("Failed to register target-infra: %v", err) } @@ -225,18 +220,12 @@ func TestWorkerManagerDifferentBranches(t *testing.T) { createTargetForRegister(ctx, t, client, "target-dev", "repo1", "develop", "base/") // Register targets for same repo, different branches - err := manager.RegisterTarget(ctx, - "target-main", "default", - "repo1", "gitops-system", - "main", "base/") + err := manager.EnsureWorker(ctx, "repo1", "gitops-system", "main") if err != nil { t.Fatalf("Failed to register target-main: %v", err) } - err = manager.RegisterTarget(ctx, - "target-dev", "default", - "repo1", "gitops-system", - "develop", "base/") + err = manager.EnsureWorker(ctx, "repo1", "gitops-system", "develop") if err != nil { t.Fatalf("Failed to register target-dev: %v", err) } @@ -265,74 +254,73 @@ func TestWorkerManagerDifferentBranches(t *testing.T) { time.Sleep(100 * time.Millisecond) } -// TestWorkerManagerUnregisterTarget verifies target unregistration. -func TestWorkerManagerUnregisterTarget(t *testing.T) { +// targetForRegister addresses a GitTarget createTargetForRegister made, for a delete. +func targetForRegister(name string) *configv1alpha3.GitTarget { + target := &configv1alpha3.GitTarget{} + target.Name = name + target.Namespace = testTargetNamespace + return target +} + +// TestReconcileWorkers_KeepsAWorkerASiblingStillNeeds is the rule a per-target count gets wrong. +// One worker serves every GitTarget on its (provider, branch), so deleting one of them says +// nothing about the others: the sweep decides from the API, where the siblings are still listed. +func TestReconcileWorkers_KeepsAWorkerASiblingStillNeeds(t *testing.T) { scheme := setupScheme() - client := fake.NewClientBuilder().WithScheme(scheme).Build() - log := logr.Discard() + k8sClient := fake.NewClientBuilder().WithScheme(scheme).Build() - manager := NewWorkerManager(client, log, BranchWorkerLimits{}, types.SensitiveResourcePolicy{}) - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + manager := NewWorkerManager(k8sClient, logr.Discard(), BranchWorkerLimits{}, types.SensitiveResourcePolicy{}) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - - go func() { - _ = manager.Start(ctx) - }() + go func() { _ = manager.Start(ctx) }() time.Sleep(100 * time.Millisecond) - createProviderWithLocalRepo(ctx, t, client, "repo1") - createTargetForRegister(ctx, t, client, "target1", "repo1", "main", "apps/") - createTargetForRegister(ctx, t, client, "target2", "repo1", "main", "infra/") - - // Register two targets - _ = manager.RegisterTarget(ctx, - "target1", "default", - "repo1", "gitops-system", - "main", "apps/") - _ = manager.RegisterTarget(ctx, - "target2", "default", - "repo1", "gitops-system", - "main", "infra/") - - // Verify worker exists - _, exists := manager.GetWorkerForTarget("repo1", "gitops-system", "main") - if !exists { - t.Fatal("Worker should exist") - } - - // Unregister first target - err := manager.UnregisterTarget("target1", "default", "repo1", "gitops-system", "main") - if err != nil { - t.Fatalf("Failed to unregister target1: %v", err) - } - // Verify worker was destroyed (WorkerManager now destroys on any unregister) - _, exists = manager.GetWorkerForTarget("repo1", "gitops-system", "main") - if exists { - t.Error("Worker should be destroyed when target unregistered") - } + createProviderWithLocalRepo(ctx, t, k8sClient, "repo1") + createTargetForRegister(ctx, t, k8sClient, "target1", "repo1", "main", "apps/") + createTargetForRegister(ctx, t, k8sClient, "target2", "repo1", "main", "infra/") + // The provider lives beside its targets, which is the namespace the sweep derives. + require.NoError(t, manager.EnsureWorker(ctx, "repo1", testTargetNamespace, "main")) - // Unregister last target - err = manager.UnregisterTarget("target2", "default", "repo1", "gitops-system", "main") - if err != nil { - t.Fatalf("Failed to unregister target2: %v", err) - } + // One of the two goes. + require.NoError(t, k8sClient.Delete(ctx, targetForRegister("target1"))) + require.NoError(t, manager.ReconcileWorkers(ctx)) - // Verify worker was destroyed - _, exists = manager.GetWorkerForTarget("repo1", "gitops-system", "main") - if exists { - t.Error("Worker should be destroyed when last target unregistered") - } + _, exists := manager.GetWorkerForTarget("repo1", testTargetNamespace, "main") + assert.True(t, exists, "target2 still writes to this branch, so its worker must survive") - manager.mu.RLock() - finalWorkerCount := len(manager.workers) - manager.mu.RUnlock() + // And now the last one. + require.NoError(t, k8sClient.Delete(ctx, targetForRegister("target2"))) + require.NoError(t, manager.ReconcileWorkers(ctx)) - if finalWorkerCount != 0 { - t.Errorf("Manager should have 0 workers, got %d", finalWorkerCount) - } + _, exists = manager.GetWorkerForTarget("repo1", testTargetNamespace, "main") + assert.False(t, exists, + "nothing needs this worker now, and it holds a goroutine, a queue and a clone until it stops") +} - cancel() +// TestReconcileWorkers_StopsNothingWhenTheTargetsCannotBeRead. Treating "I could not read the +// GitTargets" as "there are no GitTargets" would stop every live worker in the process, which is +// the one outcome a cleanup sweep must never produce. +func TestReconcileWorkers_StopsNothingWhenTheTargetsCannotBeRead(t *testing.T) { + scheme := setupScheme() + k8sClient := fake.NewClientBuilder().WithScheme(scheme). + WithInterceptorFuncs(interceptor.Funcs{ + List: func(context.Context, client.WithWatch, client.ObjectList, ...client.ListOption) error { + return errors.New("the cache is not having it") + }, + }).Build() + + manager := NewWorkerManager(k8sClient, logr.Discard(), BranchWorkerLimits{}, types.SensitiveResourcePolicy{}) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + go func() { _ = manager.Start(ctx) }() time.Sleep(100 * time.Millisecond) + require.NoError(t, manager.EnsureWorker(ctx, "repo1", testTargetNamespace, "main")) + + err := manager.ReconcileWorkers(ctx) + + require.Error(t, err) + _, exists := manager.GetWorkerForTarget("repo1", testTargetNamespace, "main") + assert.True(t, exists, "a read that failed proves nothing about which workers are needed") } // TestWorkerManagerConcurrentRegistration verifies thread safety. @@ -356,13 +344,9 @@ func TestWorkerManagerConcurrentRegistration(t *testing.T) { done := make(chan bool, 10) for i := range 10 { go func(index int) { - targetName := "target" - err := manager.RegisterTarget(ctx, - targetName, "default", - "repo1", "gitops-system", - "main", "base/") + err := manager.EnsureWorker(ctx, "repo1", "gitops-system", "main") if err != nil { - t.Errorf("Failed to register target %d: %v", index, err) + t.Errorf("Failed to ensure worker %d: %v", index, err) } done <- true }(i) @@ -403,17 +387,96 @@ func TestWorkerManagerGetNonexistentWorker(t *testing.T) { } } -// TestWorkerManagerUnregisterNonexistent verifies unregistering nonexistent target is safe. -func TestWorkerManagerUnregisterNonexistent(t *testing.T) { +// TestRemoveWorkers_IgnoresAKeyThatNamesNoWorker keeps the sweep idempotent: it computes its +// orphan list outside the lifecycle lock, so a worker can have gone by the time it is removed. +func TestRemoveWorkers_IgnoresAKeyThatNamesNoWorker(t *testing.T) { scheme := setupScheme() - client := fake.NewClientBuilder().WithScheme(scheme).Build() - log := logr.Discard() + k8sClient := fake.NewClientBuilder().WithScheme(scheme).Build() - manager := NewWorkerManager(client, log, BranchWorkerLimits{}, types.SensitiveResourcePolicy{}) + manager := NewWorkerManager(k8sClient, logr.Discard(), BranchWorkerLimits{}, types.SensitiveResourcePolicy{}) - // Unregister should be idempotent and not error - err := manager.UnregisterTarget("nonexistent", "default", "repo1", "gitops-system", "main") - if err != nil { - t.Errorf("Unregister nonexistent should not error: %v", err) - } + assert.NotPanics(t, func() { + manager.removeWorkers([]BranchKey{{ + RepoNamespace: "gitops-system", RepoName: "repo1", Branch: "main", + }}, "test") + }) +} + +// TestReconcileWorkers_RemovesTheRetiredWorkersCheckout. Stopping the goroutine was only half the +// leak. Each worker keeps a clone per remote under its own (provider, branch) directory, and that +// half survives a restart, so a cluster that creates and deletes targets accumulates checkouts +// until somebody notices the disk. +func TestReconcileWorkers_RemovesTheRetiredWorkersCheckout(t *testing.T) { + withTemporaryWorkerStateRoot(t) + scheme := setupScheme() + k8sClient := fake.NewClientBuilder().WithScheme(scheme).Build() + + manager := NewWorkerManager(k8sClient, logr.Discard(), BranchWorkerLimits{}, types.SensitiveResourcePolicy{}) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + go func() { _ = manager.Start(ctx) }() + time.Sleep(100 * time.Millisecond) + + createProviderWithLocalRepo(ctx, t, k8sClient, "repo1") + createTargetForRegister(ctx, t, k8sClient, "target1", "repo1", "main", "apps/") + require.NoError(t, manager.EnsureWorker(ctx, "repo1", testTargetNamespace, "main")) + + worker, exists := manager.GetWorkerForTarget("repo1", testTargetNamespace, "main") + require.True(t, exists) + // Stand in for the clone a first publication would leave behind. + checkout := worker.repoPathForRemote("https://example.invalid/repo.git") + require.NoError(t, os.MkdirAll(checkout, 0o750)) + require.NoError(t, os.WriteFile(filepath.Join(checkout, "HEAD"), []byte("ref: refs/heads/main\n"), 0o600)) + + require.NoError(t, k8sClient.Delete(ctx, targetForRegister("target1"))) + require.NoError(t, manager.ReconcileWorkers(ctx)) + + _, err := os.Stat(checkout) + assert.True(t, os.IsNotExist(err), "the retired worker's checkout must go with it") +} + +// TestRemoveLocalState_RefusesAWorkerWithNoIdentity. repoRootPath joins the three identity fields +// into a fixed prefix, so a worker with empty fields resolves to the ROOT of every worker's state. +// Deleting that would take out the live workers beside it, which is a far worse outcome than the +// leak this reclaims. +func TestRemoveLocalState_RefusesAWorkerWithNoIdentity(t *testing.T) { + root := withTemporaryWorkerStateRoot(t) + require.NoError(t, os.MkdirAll(filepath.Join(root, "someone-elses-state"), 0o750)) + + (&BranchWorker{Log: logr.Discard()}).removeLocalState() + + _, err := os.Stat(filepath.Join(root, "someone-elses-state")) + assert.NoError(t, err, "a worker with no identity must never delete the shared root") +} + +// TestBranchPathComponent_KeepsOneBranchOutOfAnothersDirectory. Git branch names contain slashes, +// so joining one into a path made `release/v1`'s worker a child of `release`'s — and once a +// retired worker deletes its directory, retiring `release` would take the live worker's checkout +// with it. +func TestBranchPathComponent_KeepsOneBranchOutOfAnothersDirectory(t *testing.T) { + withTemporaryWorkerStateRoot(t) + parent := &BranchWorker{GitProviderNamespace: "shop", GitProviderRef: "repo1", Branch: "release"} + child := &BranchWorker{GitProviderNamespace: "shop", GitProviderRef: "repo1", Branch: "release/v1"} + + assert.NotContains(t, child.repoRootPath(), filepath.Dir(parent.repoRootPath())+string(filepath.Separator)+"repos", + "one branch's tree must not sit inside another's") + rel, err := filepath.Rel(filepath.Dir(parent.repoRootPath()), child.repoRootPath()) + require.NoError(t, err) + assert.True(t, strings.HasPrefix(rel, ".."), "release/v1 must not be reachable from inside release") + + // And names that sanitising alone would collapse stay apart. + lookalike := &BranchWorker{GitProviderNamespace: "shop", GitProviderRef: "repo1", Branch: "release-v1"} + assert.NotEqual(t, child.repoRootPath(), lookalike.repoRootPath()) +} + +// withTemporaryWorkerStateRoot points every worker's on-disk state at this test's own directory, +// so a test that DELETES worker state cannot reach the shared root — where another test's +// fixtures, or a developer's running operator, keep theirs. +func withTemporaryWorkerStateRoot(t *testing.T) string { + t.Helper() + previous := workerStateRoot + root := t.TempDir() + workerStateRoot = root + t.Cleanup(func() { workerStateRoot = previous }) + return root }