diff --git a/cmd/provider/main.go b/cmd/provider/main.go index 359ec95..16b6a01 100644 --- a/cmd/provider/main.go +++ b/cmd/provider/main.go @@ -21,6 +21,7 @@ import ( "log" "os" "path/filepath" + "strings" "time" "github.com/alecthomas/kingpin/v2" @@ -34,13 +35,22 @@ import ( "github.com/crossplane/crossplane-runtime/v2/pkg/reconciler/customresourcesgate" "github.com/crossplane/crossplane-runtime/v2/pkg/reconciler/managed" "github.com/crossplane/crossplane-runtime/v2/pkg/statemetrics" + sourcev1 "github.com/fluxcd/source-controller/api/v1" + sourcev1beta2 "github.com/fluxcd/source-controller/api/v1beta2" + apiscluster "github.com/upbound/provider-terraform/apis/cluster" + apisnamespaced "github.com/upbound/provider-terraform/apis/namespaced" zapuber "go.uber.org/zap" "go.uber.org/zap/zapcore" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" + "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/manager" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" "k8s.io/client-go/tools/leaderelection/resourcelock" ctrl "sigs.k8s.io/controller-runtime" @@ -48,19 +58,16 @@ import ( "sigs.k8s.io/controller-runtime/pkg/log/zap" "sigs.k8s.io/controller-runtime/pkg/metrics" - sourcev1 "github.com/fluxcd/source-controller/api/v1" - sourcev1beta2 "github.com/fluxcd/source-controller/api/v1beta2" - authv1 "k8s.io/api/authorization/v1" - apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" - - apiscluster "github.com/upbound/provider-terraform/apis/cluster" - apisnamespaced "github.com/upbound/provider-terraform/apis/namespaced" + clusterv1beta1 "github.com/upbound/provider-terraform/apis/cluster/v1beta1" + namespacedv1beta1 "github.com/upbound/provider-terraform/apis/namespaced/v1beta1" "github.com/upbound/provider-terraform/internal/bootcheck" + "github.com/upbound/provider-terraform/internal/claims" clusterworkspace "github.com/upbound/provider-terraform/internal/controller/cluster" "github.com/upbound/provider-terraform/internal/controller/cluster/workspace" "github.com/upbound/provider-terraform/internal/controller/gc" namespacedworkspace "github.com/upbound/provider-terraform/internal/controller/namespaced" "github.com/upbound/provider-terraform/internal/features" + authv1 "k8s.io/api/authorization/v1" ) func init() { @@ -70,6 +77,11 @@ func init() { } } +// defaultLeaderElectionID is today's hardcoded lease name, predating the +// --leader-election-id flag. A deployment that leaves the flag unset must +// keep getting exactly this value (Req 1.4 parity). +const defaultLeaderElectionID = "crossplane-leader-election-provider-terraform" + func main() { var ( app = kingpin.New(filepath.Base(os.Args[0]), "Terraform support for Crossplane.").DefaultEnvars() @@ -80,14 +92,34 @@ func main() { pollJitter = app.Flag("poll-jitter", "If non-zero, varies the poll interval by a random amount up to plus-or-minus this value.").Default("1m").Duration() timeout = app.Flag("timeout", "Controls how long Terraform processes may run before they are killed.").Default("20m").Duration() leaderElection = app.Flag("leader-election", "Use leader election for the controller manager.").Short('l').Default("false").Envar("LEADER_ELECTION").Bool() + leaderElectionID = app.Flag("leader-election-id", "Name of the leader election lease. Set a distinct value per instance so leaders of different instances run concurrently.").Default(defaultLeaderElectionID).Envar("LEADER_ELECTION_ID").String() + watchLabelSelector = app.Flag("watch-label-selector", "Restrict the manager cache to Workspaces matching this label selector. Empty (default) watches all Workspaces.").Default("").Envar("WATCH_LABEL_SELECTOR").String() maxReconcileRate = app.Flag("max-reconcile-rate", "The maximum number of concurrent reconciliation operations.").Default("1").Int() enableManagementPolicies = app.Flag("enable-management-policies", "Enable support for Management Policies.").Default("true").Envar("ENABLE_MANAGEMENT_POLICIES").Bool() enableChangeLogs = app.Flag("enable-changelogs", "Enable support for capturing change logs during reconciliation.").Default("false").Envar("ENABLE_CHANGE_LOGS").Bool() changelogsSocketPath = app.Flag("changelogs-socket-path", "Path for changelogs socket (if enabled)").Default("/var/run/changelogs/changelogs.sock").Envar("CHANGELOGS_SOCKET_PATH").String() logEncoding = app.Flag("log-encoding", "Container logging output ending. Possible values: console, json").Default("console").Enum("console", "json") + enableOwnershipClaims = app.Flag("enable-ownership-claims", "Guard every Terraform run with a per-Workspace ownership claim, for safe handover when a Workspace is relabeled between instances.").Default("false").Envar("ENABLE_OWNERSHIP_CLAIMS").Bool() + ownershipClaimTTL = app.Flag("ownership-claim-ttl", "How long an ownership claim's heartbeat may go stale before another instance may steal it.").Default("90s").Envar("OWNERSHIP_CLAIM_TTL").Duration() + ownershipHeartbeat = app.Flag("ownership-heartbeat-interval", "How often a held ownership claim's heartbeat is renewed while Terraform is running.").Default("30s").Envar("OWNERSHIP_HEARTBEAT_INTERVAL").Duration() ) kingpin.MustParse(app.Parse(os.Args[1:])) + workspaceCache, err := workspaceCacheByObject(*watchLabelSelector) + kingpin.FatalIfError(err, "Cannot parse --watch-label-selector value %q", *watchLabelSelector) + + holderIdentity, claimNamespace := resolveClaimIdentity(*leaderElectionID) + if *enableOwnershipClaims && claimNamespace == "" { + kingpin.Fatalf("Cannot resolve this instance's namespace for --enable-ownership-claims: set $POD_NAMESPACE via the pod's downward API, or run in-cluster") + } + claimCfg := claims.Config{ + Enabled: *enableOwnershipClaims, + TTL: *ownershipClaimTTL, + HeartbeatInterval: *ownershipHeartbeat, + HolderIdentity: holderIdentity, + Namespace: claimNamespace, + } + var logEncoder zap.Opts switch *logEncoding { case "json": @@ -112,32 +144,30 @@ func main() { cfg, err := ctrl.GetConfig() kingpin.FatalIfError(err, "Cannot get API server rest config") + scheme := buildScheme() + mgr, err := ctrl.NewManager(ratelimiter.LimitRESTConfig(cfg, *maxReconcileRate), ctrl.Options{ + Scheme: scheme, Cache: cache.Options{ SyncPeriod: syncInterval, + ByObject: workspaceCache, }, // controller-runtime uses both ConfigMaps and Leases for leader // election by default. Leases expire after 15 seconds, with a - // 10 second renewal deadline. We've observed leader loss due to + // 10 seconds renewal deadline. We've observed leader loss due to // renewal deadlines being exceeded when under high load - i.e. // hundreds of reconciles per second and ~200rps to the API // server. Switching to Leases only and longer leases appears to // alleviate this. LeaderElection: *leaderElection, - LeaderElectionID: "crossplane-leader-election-provider-terraform", + LeaderElectionID: *leaderElectionID, LeaderElectionResourceLock: resourcelock.LeasesResourceLock, LeaseDuration: func() *time.Duration { d := 60 * time.Second; return &d }(), RenewDeadline: func() *time.Duration { d := 50 * time.Second; return &d }(), }) kingpin.FatalIfError(err, "Cannot create controller manager") - kingpin.FatalIfError(apiscluster.AddToScheme(mgr.GetScheme()), "Cannot add terraform APIs to scheme") - kingpin.FatalIfError(apisnamespaced.AddToScheme(mgr.GetScheme()), "Cannot add terraform APIs to scheme") - kingpin.FatalIfError(sourcev1.AddToScheme(mgr.GetScheme()), "Cannot add flux gitrepository APIs to scheme") - kingpin.FatalIfError(sourcev1beta2.AddToScheme(mgr.GetScheme()), "Cannot add flux ocirepository APIs to scheme") - kingpin.FatalIfError(apiextensionsv1.AddToScheme(mgr.GetScheme()), "Cannot register k8s apiextensions APIs to scheme") - metricRecorder := managed.NewMRMetricRecorder() stateMetrics := statemetrics.NewMRStateMetrics() @@ -201,16 +231,91 @@ func main() { clusterOpts.Gate = crdGate namespacedOpts.Gate = crdGate kingpin.FatalIfError(customresourcesgate.Setup(mgr, namespacedOpts), "Cannot setup CRD gate") - kingpin.FatalIfError(clusterworkspace.SetupGated(mgr, clusterOpts, *timeout, *pollJitter), "Cannot setup cluster-scoped Workspace controllers") - kingpin.FatalIfError(namespacedworkspace.SetupGated(mgr, namespacedOpts, *timeout, *pollJitter), "Cannot setup namespaced Workspace controllers") + kingpin.FatalIfError(clusterworkspace.SetupGated(mgr, clusterOpts, *timeout, *pollJitter, claimCfg), "Cannot setup cluster-scoped Workspace controllers") + kingpin.FatalIfError(namespacedworkspace.SetupGated(mgr, namespacedOpts, *timeout, *pollJitter, claimCfg), "Cannot setup namespaced Workspace controllers") } else { log.Info("Provider has missing RBAC permissions for watching CRDs, controller SafeStart capability will be disabled") - kingpin.FatalIfError(clusterworkspace.Setup(mgr, clusterOpts, *timeout, *pollJitter), "Cannot setup cluster-scoped Workspace controllers") - kingpin.FatalIfError(namespacedworkspace.Setup(mgr, namespacedOpts, *timeout, *pollJitter), "Cannot setup namespaced Workspace controllers") + kingpin.FatalIfError(clusterworkspace.Setup(mgr, clusterOpts, *timeout, *pollJitter, claimCfg), "Cannot setup cluster-scoped Workspace controllers") + kingpin.FatalIfError(namespacedworkspace.Setup(mgr, namespacedOpts, *timeout, *pollJitter, claimCfg), "Cannot setup namespaced Workspace controllers") } kingpin.FatalIfError(mgr.Start(ctrl.SetupSignalHandler()), "Cannot start controller manager") } +// buildScheme assembles the scheme before the manager exists, so the cache's +// ByObject GVK resolution (which runs inside ctrl.NewManager) can find the +// Workspace kinds. A supplied scheme replaces controller-runtime's default +// client-go scheme, so the built-ins (Leases for leader election, +// SelfSubjectAccessReview for the CRD precheck) must be re-added here too. +func buildScheme() *runtime.Scheme { + s := runtime.NewScheme() + kingpin.FatalIfError(clientgoscheme.AddToScheme(s), "Cannot add client-go APIs to scheme") + kingpin.FatalIfError(apiscluster.AddToScheme(s), "Cannot add terraform APIs to scheme") + kingpin.FatalIfError(apisnamespaced.AddToScheme(s), "Cannot add terraform APIs to scheme") + kingpin.FatalIfError(sourcev1.AddToScheme(s), "Cannot add flux gitrepository APIs to scheme") + kingpin.FatalIfError(sourcev1beta2.AddToScheme(s), "Cannot add flux ocirepository APIs to scheme") + kingpin.FatalIfError(apiextensionsv1.AddToScheme(s), "Cannot register k8s apiextensions APIs to scheme") + kingpin.FatalIfError(authv1.AddToScheme(s), "Cannot register k8s authorization APIs to scheme") + return s +} + +// workspaceCacheByObject scopes the manager cache to Workspaces matching +// selector, restricting *only* the cluster-scoped and namespaced Workspace +// types. An empty selector returns a nil map, leaving the cache unrestricted +// for every type (today's behavior). A non-empty selector must not be +// applied via cache.Options.DefaultLabelSelector: that field filters every +// cached type (ProviderConfigs, Secrets, Leases, CRDs), not just Workspaces. +func workspaceCacheByObject(selector string) (map[client.Object]cache.ByObject, error) { + if selector == "" { + return nil, nil + } + sel, err := labels.Parse(selector) + if err != nil { + return nil, err + } + return map[client.Object]cache.ByObject{ + &clusterv1beta1.Workspace{}: {Label: sel}, + &namespacedv1beta1.Workspace{}: {Label: sel}, + }, nil +} + +// inClusterNamespacePath is where a pod's own namespace is projected by the +// service account token volume -- the same source controller-runtime itself +// reads for LeaderElectionNamespace when that option is left unset. +const inClusterNamespacePath = "/var/run/secrets/kubernetes.io/serviceaccount/namespace" + +// resolveClaimIdentity determines this instance's ownership-claim holder +// identity and the namespace its claim Leases for *cluster-scoped* +// Workspaces live in (namespaced Workspaces' claims live in their own +// namespace instead -- see claims.Config.Namespace). +// +// HolderIdentity precedence: $POD_NAME (downward API) > hostname > +// leaderElectionID. The last resort falls back to the lease ID rather than +// an empty string because --leader-election-id is already required to be +// unique per instance (TD-2), which makes it a reasonable identity when no +// pod identity is available (e.g. running outside a pod). +// +// Namespace precedence: $POD_NAMESPACE (downward API) > the in-cluster +// service account namespace file. +func resolveClaimIdentity(leaderElectionID string) (holderIdentity, namespace string) { + holderIdentity = os.Getenv("POD_NAME") + if holderIdentity == "" { + if h, err := os.Hostname(); err == nil && h != "" { + holderIdentity = h + } + } + if holderIdentity == "" { + holderIdentity = leaderElectionID + } + + namespace = os.Getenv("POD_NAMESPACE") + if namespace == "" { + if b, err := os.ReadFile(inClusterNamespacePath); err == nil { + namespace = strings.TrimSpace(string(b)) + } + } + return holderIdentity, namespace +} + // UseISO8601 sets the logger to use ISO8601 timestamp format func UseISO8601() zap.Opts { return func(o *zap.Options) { @@ -228,9 +333,6 @@ func UseJSON() zap.Opts { } func canWatchCRD(ctx context.Context, mgr manager.Manager) (bool, error) { - if err := authv1.AddToScheme(mgr.GetScheme()); err != nil { - return false, err - } verbs := []string{"get", "list", "watch"} for _, verb := range verbs { sar := &authv1.SelfSubjectAccessReview{ diff --git a/cmd/provider/main_test.go b/cmd/provider/main_test.go new file mode 100644 index 0000000..ef24f8a --- /dev/null +++ b/cmd/provider/main_test.go @@ -0,0 +1,243 @@ +/* +Copyright 2020 The Crossplane Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package main + +import ( + "os" + "testing" + + "github.com/alecthomas/kingpin/v2" + authv1 "k8s.io/api/authorization/v1" + coordinationv1 "k8s.io/api/coordination/v1" + "k8s.io/apimachinery/pkg/labels" + "sigs.k8s.io/controller-runtime/pkg/client" + + clusterv1beta1 "github.com/upbound/provider-terraform/apis/cluster/v1beta1" + namespacedv1beta1 "github.com/upbound/provider-terraform/apis/namespaced/v1beta1" +) + +// TestDefaultLeaderElectionID pins the literal lease name a deployment gets +// when --leader-election-id is left unset. This is the Req 1.4 parity +// guard: today's hardcoded lease ID must not silently change. +func TestDefaultLeaderElectionID(t *testing.T) { + want := "crossplane-leader-election-provider-terraform" + if defaultLeaderElectionID != want { + t.Fatalf("defaultLeaderElectionID = %q, want %q", defaultLeaderElectionID, want) + } + + app := kingpin.New("test", "") + id := app.Flag("leader-election-id", "").Default(defaultLeaderElectionID).String() + if _, err := app.Parse(nil); err != nil { + t.Fatalf("app.Parse(nil): unexpected error: %v", err) + } + if *id != want { + t.Errorf("parsed --leader-election-id default = %q, want %q", *id, want) + } +} + +// TestBuildSchemeRegistersWorkspaceTypes reproduces the --watch-label-selector +// startup crash if the scheme ordering ever regresses. ctrl.NewManager resolves +// each cache.ByObject key's GVK against the scheme (via s.ObjectKinds); if the +// Workspace types aren't registered by then it fails with "no kind is +// registered for the type v1beta1.Workspace". buildScheme must register them +// up front. It must also re-add the built-ins that controller-runtime's default +// scheme would otherwise supply -- Leases (leader election) and +// SelfSubjectAccessReview (the CRD precheck) -- since supplying our own scheme +// replaces that default. +func TestBuildSchemeRegistersWorkspaceTypes(t *testing.T) { + s := buildScheme() + + // The exact lookup ctrl.NewManager makes for each ByObject key. + for _, obj := range []client.Object{ + &clusterv1beta1.Workspace{}, + &namespacedv1beta1.Workspace{}, + } { + if _, _, err := s.ObjectKinds(obj); err != nil { + t.Errorf("scheme missing kind for %T: %v", obj, err) + } + } + + if !s.Recognizes(coordinationv1.SchemeGroupVersion.WithKind("Lease")) { + t.Error("built-in Lease missing; leader election would break") + } + if !s.Recognizes(authv1.SchemeGroupVersion.WithKind("SelfSubjectAccessReview")) { + t.Error("built-in SelfSubjectAccessReview missing; the CRD precheck would break") + } +} + +func TestWorkspaceCacheByObject(t *testing.T) { + cases := map[string]struct { + selector string + wantNil bool + wantErr bool + }{ + "EmptyWatchesAll": { + // Empty selector must leave the cache unrestricted -- today's + // behavior (Req 1.4 parity). + selector: "", + wantNil: true, + }, + "ValidEquality": { + selector: "sharding.example.com/shard=1", + }, + "ValidNegation": { + selector: "!sharding.example.com/shard", + }, + "ValidSetBasedIn": { + selector: "sharding.example.com/shard in (0,1)", + }, + "ValidSetBasedNotIn": { + selector: "sharding.example.com/shard notin (1,2)", + }, + "InvalidSelectorSyntax": { + selector: "===bad===", + wantErr: true, + }, + "InvalidORSyntaxIsRejected": { + // Kubernetes selectors have no OR combinator (see design doc + // TD-8 correction). Confirm it's rejected outright rather than + // silently accepted or misparsed. + selector: "sharding.example.com/shard==0 OR !sharding.example.com/shard", + wantErr: true, + }, + } + + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + got, err := workspaceCacheByObject(tc.selector) + + if tc.wantErr { + if err == nil { + t.Fatalf("workspaceCacheByObject(%q): want error, got nil", tc.selector) + } + return + } + if err != nil { + t.Fatalf("workspaceCacheByObject(%q): unexpected error: %v", tc.selector, err) + } + + if tc.wantNil { + if got != nil { + t.Errorf("workspaceCacheByObject(%q) = %v, want nil (watch-all)", tc.selector, got) + } + return + } + + if len(got) != 2 { + t.Fatalf("workspaceCacheByObject(%q): got %d entries, want 2 (cluster + namespaced Workspace)", tc.selector, len(got)) + } + + var sawCluster, sawNamespaced bool + for obj, byObj := range got { + if byObj.Label == nil { + t.Errorf("entry for %T has a nil Label selector", obj) + continue + } + switch obj.(type) { + case *clusterv1beta1.Workspace: + sawCluster = true + case *namespacedv1beta1.Workspace: + sawNamespaced = true + default: + t.Errorf("unexpected object type in cache ByObject map: %T", obj) + } + } + if !sawCluster || !sawNamespaced { + t.Errorf("workspaceCacheByObject(%q): expected entries for both cluster and namespaced Workspace, sawCluster=%v sawNamespaced=%v", tc.selector, sawCluster, sawNamespaced) + } + }) + } +} + +// TestWorkspaceCacheByObjectDefaultShardCatchAll proves the documented +// default-instance selector "!shard" (DoesNotExist) matches only Workspaces +// with no shard label at all. There is no "shard 0": any Workspace carrying +// the label -- including shard=0 or the empty string "" -- is excluded from +// the catch-all, per the contract in docs/monolith/Configuration.md. +func TestWorkspaceCacheByObjectDefaultShardCatchAll(t *testing.T) { + got, err := workspaceCacheByObject("!sharding.example.com/shard") + if err != nil { + t.Fatalf("workspaceCacheByObject(%q): unexpected error: %v", "!sharding.example.com/shard", err) + } + + var sel labels.Selector + for _, byObj := range got { + sel = byObj.Label + break + } + if sel == nil { + t.Fatal("no selector found in workspaceCacheByObject result") + } + + cases := map[string]struct { + set labels.Set + want bool + }{ + "AbsentLabelMatches": {set: labels.Set{}, want: true}, + "Shard0Excluded": {set: labels.Set{"sharding.example.com/shard": "0"}, want: false}, + "Shard1Excluded": {set: labels.Set{"sharding.example.com/shard": "1"}, want: false}, + "Shard2Excluded": {set: labels.Set{"sharding.example.com/shard": "2"}, want: false}, + "EmptyStringExcluded": {set: labels.Set{"sharding.example.com/shard": ""}, want: false}, + } + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + if got := sel.Matches(tc.set); got != tc.want { + t.Errorf("selector.Matches(%v) = %v, want %v", tc.set, got, tc.want) + } + }) + } +} + +func TestResolveClaimIdentity(t *testing.T) { + t.Run("PodNameEnvTakesPrecedence", func(t *testing.T) { + t.Setenv("POD_NAME", "provider-terraform-shard-1-abc123") + holder, _ := resolveClaimIdentity("fallback-lease-id") + if holder != "provider-terraform-shard-1-abc123" { + t.Errorf("resolveClaimIdentity(%q) holder = %q, want %q", "fallback-lease-id", holder, "provider-terraform-shard-1-abc123") + } + }) + + t.Run("HolderIdentityNeverEmpty", func(t *testing.T) { + // Can't portably force os.Hostname() to fail, but the documented + // precedence guarantees a non-empty result either way: hostname, or + // (as a last resort) the caller-supplied leaderElectionID. + t.Setenv("POD_NAME", "") + holder, _ := resolveClaimIdentity("fallback-lease-id") + if holder == "" { + t.Error("holderIdentity should never be empty") + } + }) + + t.Run("PodNamespaceEnvTakesPrecedence", func(t *testing.T) { + t.Setenv("POD_NAMESPACE", "tenant-a") + _, ns := resolveClaimIdentity("fallback-lease-id") + if ns != "tenant-a" { + t.Errorf("resolveClaimIdentity(%q) namespace = %q, want %q", "fallback-lease-id", ns, "tenant-a") + } + }) + + t.Run("EmptyNamespaceWhenNotInCluster", func(t *testing.T) { + t.Setenv("POD_NAMESPACE", "") + if _, err := os.Stat(inClusterNamespacePath); err == nil { + t.Skip("running somewhere with an in-cluster namespace file; precedence is already covered above") + } + _, ns := resolveClaimIdentity("fallback-lease-id") + if ns != "" { + t.Errorf("namespace = %q, want empty (not running in-cluster, no override set)", ns) + } + }) +} diff --git a/docs/monolith/Configuration.md b/docs/monolith/Configuration.md index 8131189..4ec04bf 100644 --- a/docs/monolith/Configuration.md +++ b/docs/monolith/Configuration.md @@ -376,4 +376,85 @@ spec: ... ``` -- `enableTerraformCLILogging`: Specifies whether logging is enabled (`true`) or disabled (`false`). When enabled, Terraform CLI command output will be written to the container logs. Default is `false` \ No newline at end of file +- `enableTerraformCLILogging`: Specifies whether logging is enabled (`true`) or disabled (`false`). When enabled, Terraform CLI command output will be written to the container logs. Default is `false` + +## Horizontal Scaling: Sharding Workspaces Across Instances + +A single provider-terraform pod reconciles every `Workspace` it can see. When the number of Workspaces on a cluster grows large enough that one pod's `terraform` throughput becomes the bottleneck, you can run **N instances** of the provider, each watching a disjoint subset ("shard") of Workspaces. Every instance is the same binary, deployed as a separate `Provider`/`ControllerConfig` (or equivalent) with different flag values — there is no separate "sharding" binary or CRD. + +All three flags below default to today's single-instance behavior, so an existing deployment is unaffected until you opt in. + +### `--watch-label-selector`: partition Workspaces by label + +Restricts an instance's manager cache to only the Workspaces matching a label selector. A Workspace whose labels don't match never enters that instance's cache — it is structurally invisible, not filtered after the fact. + +```yaml +apiVersion: pkg.crossplane.io/v1alpha1 +kind: ControllerConfig +metadata: + name: terraform-shard-1 +spec: + args: + - --watch-label-selector=sharding.example.com/shard=1 +``` + +- **Default:** `""` (empty) — watch every Workspace, exactly like today. +- **At most one running instance may use this default.** An empty selector means *no filtering at all* — it watches every Workspace, including ones explicitly labeled for another shard. Leaving more than one instance at the default (or a flag simply omitted, which is identical to `""`), or leaving an old single-instance deployment's empty-selector instance running alongside a newly-added sharded fleet, means every empty-selector instance reconciles the *entire* Workspace set concurrently with whichever shard instances also match — full overlap on every Workspace, not a partial or edge-case one. This is the same failure class as leaving `--leader-election-id` at its default across instances (below); the provider can't detect it itself since no instance knows what selector any other instance is running — getting this right is a deployment-time responsibility, not something the binary enforces. +- **Assignment contract:** a Workspace is reconciled by instance `k` if and only if it carries the label `sharding.example.com/shard=k`. Every shard instance uses a plain equality selector (`shard=1`, `shard=2`, ...). This label is additive and optional — it requires no change to the Workspace CRD schema. +- **The default (catch-all) instance watches unlabelled Workspaces only — nothing else.** There is no "shard 0." A Workspace that carries the `shard` label — even if the value doesn't match any currently-running instance (a typo, a decommissioned shard) — is **not** picked up by the default instance. That's deliberate: an instance should only ever run `terraform` on Workspaces it was actually assigned, not absorb whatever nobody else claims. The default instance's selector is: + + ``` + --watch-label-selector=!sharding.example.com/shard + ``` + + `!key` (`DoesNotExist`) is standard Kubernetes selector syntax and matches only Workspaces where the label is **absent entirely**. +- **`shard=""` is not the same as "no label" — do not use it.** The `DoesNotExist` check is pure key-presence: a Workspace labeled `shard=""` (key present, empty value) does **not** match `!shard`, and matches no real shard's equality selector either. It becomes an orphan — reconciled by nobody. If you need to represent "not yet assigned," omit the `shard` label entirely; never set it to an empty string. +- **Operational consequence:** because a mislabeled or orphaned-shard Workspace is now deliberately left unpicked rather than silently absorbed, monitor for it explicitly — e.g. a periodic audit alerting on any Workspace whose `shard` label (including `shard=""`) doesn't match a currently-deployed instance's selector. A Workspace nobody is watching should be a loud signal, not a silent one. +- Selector syntax accepts the full Kubernetes label selector grammar (`=`, `!=`, `in (...)`, `notin (...)`, `key`, `!key`); an invalid selector fails the provider at startup rather than at reconcile time. +- **Relabelling mid-apply is supported, including during a Workspace's first apply.** A filtered cache drops a Workspace the instant its label stops matching — the API server delivers "was matching, now isn't" as a `DELETED` watch event — so the outgoing instance finishes its `terraform` run against a Workspace it can no longer read from cache. It still has to record the result of that run: Crossplane refuses to reconcile a Workspace whose `crossplane.io/external-create-pending` annotation was never resolved, and does not requeue, so an unrecorded result strands the Workspace on **every** instance, including the one that just inherited it. The reconcilers therefore persist that bookkeeping through a client that reads from the API server rather than the cache. The outgoing instance records its result on the relabelled Workspace without reverting the relabel, and the incoming instance is woken by that write. +- **If the owning instance *dies* mid-apply, the Workspace is left for you to resolve.** A crash, OOM kill, eviction or `--timeout` kill leaves no process to record the create result, so the Workspace stays `create-pending` and no instance will touch it — including for later spec changes. This is upstream Crossplane behaviour, not something sharding introduces, and it is deliberately not bypassed: `Observe` answers "does this exist?" from `terraform state list` and `terraform output`, so a run that died before its resources reached persisted state leaves nothing to observe, and re-applying provisions them a second time. Whether that is a risk depends on your module's backend, which the provider cannot know. Check the state, then release the Workspace: + + ```bash + kubectl annotate workspace crossplane.io/external-create-pending- + ``` + +### `--leader-election-id`: give each instance its own leader lease + +```yaml + - --leader-election + - --leader-election-id=crossplane-leader-election-provider-terraform-shard-1 +``` + +- **Default:** `crossplane-leader-election-provider-terraform` — today's hardcoded lease name. +- **Why you must set this per instance:** the lease name has nothing to do with the Deployment's name. If two instances share a namespace and both leave this flag at its default (or set it to the same value), all of their pods race for **one** lease, and only one pod cluster-wide ever becomes leader. Every other instance's pods sit as permanent standbys with no leader, so their entire shard silently stops being reconciled — not degraded throughput, a total and silent outage for that shard. +- **Deployment rule of thumb:** derive `--leader-election-id` from the *same* shard index used in `--watch-label-selector` (e.g. append `-shard-1` to both), so the two settings can never drift apart in your deployment templates. +- Only matters when `--leader-election` (`-l`) is on with `replicas >= 2`; with a single replica per instance the ID is inert. + +### `--enable-ownership-claims` (optional): safe handover when a Workspace is relabeled + +When a Workspace's shard label changes while a `terraform apply` is still running on the old instance, a bare relabel is already safe by default (no destroy is triggered, and the Terraform state lock serializes any overlap) — this flag is **optional polish**, not a correctness requirement. Turning it on adds: + +1. **Noise suppression** — the incoming instance backs off quietly instead of repeatedly failing against the held state lock during the handover window. +2. **Automated crash recovery** — if the old owner crashed mid-apply, the new owner automatically runs `terraform force-unlock` once the old claim goes stale, instead of requiring a manual runbook. + +```yaml + - --enable-ownership-claims + - --ownership-claim-ttl=90s + - --ownership-heartbeat-interval=30s +``` + +- `--enable-ownership-claims`: default `false`. Off = identical to today's behavior. +- `--ownership-claim-ttl`: how long a claim's heartbeat may go stale before another instance is allowed to steal it. Default `90s`. +- `--ownership-heartbeat-interval`: how often the current owner refreshes its claim while `terraform` is running. Default `30s`. +- **Requires downward-API wiring:** the provider resolves its own identity from `$POD_NAME` (falls back to hostname, then to `--leader-election-id`) and its claim namespace from `$POD_NAMESPACE` (falls back to the in-cluster service account namespace). If you enable this flag, set both env vars via the pod's downward API: + + ```yaml + env: + - name: POD_NAME + valueFrom: { fieldRef: { fieldPath: metadata.name } } + - name: POD_NAMESPACE + valueFrom: { fieldRef: { fieldPath: metadata.namespace } } + ``` + + The provider fails fast at startup if `--enable-ownership-claims` is set and no namespace can be resolved. +- **RBAC:** the provider's ServiceAccount needs `get`, `list`, `watch`, `create`, `update`, `delete` on `leases.coordination.k8s.io` wherever this flag is enabled (claims are stored in a dedicated `Lease` per Workspace, separate from the leader-election lease). **This must be a `ClusterRole`, not a namespace-scoped `Role`.** A cluster-scoped Workspace's claim Lease lives in the provider's own namespace, but a *namespaced* Workspace's claim lives in that Workspace's own (tenant) namespace — so the grant can't be limited to the provider's namespace alone; it needs to cover every namespace a namespaced Workspace could exist in. \ No newline at end of file diff --git a/internal/claims/claims.go b/internal/claims/claims.go new file mode 100644 index 0000000..6dd04ab --- /dev/null +++ b/internal/claims/claims.go @@ -0,0 +1,226 @@ +/* +Copyright 2020 The Crossplane Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Package claims implements the optional per-Workspace ownership-claim +// handover protocol (design doc §8, TD-6). A claim is stored in a dedicated +// coordination.k8s.io/v1 Lease named after the Workspace's UID, separate +// from the Workspace's assignment label (the label says who *should* own a +// Workspace; the claim says who *actually* runs terraform on it). +package claims + +import ( + "context" + "time" + + "github.com/pkg/errors" + coordinationv1 "k8s.io/api/coordination/v1" + kerrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// Outcome describes the result of an Acquire call (design §8 R2-R4). +type Outcome int + +const ( + // Acquired means the caller now holds the claim and may run terraform + // (the claim was empty or already held by this instance). + Acquired Outcome = iota + // Backoff means a foreign, fresh claim is held by another instance (R3), + // or this instance lost a concurrent acquire race (R2); the caller must + // not run and should requeue. + Backoff + // Stolen means a foreign, stale claim was overwritten (R4). The caller + // now holds the claim but must verify/force-unlock any stale Terraform + // state lock left by the presumed-dead previous owner before running. + Stolen +) + +// Errors returned by Manager methods. +const ( + errGetLease = "cannot get ownership claim lease" + errCreateLease = "cannot create ownership claim lease" + errUpdateLease = "cannot update ownership claim lease" + errClearLease = "cannot clear ownership claim lease" + errNotHolder = "cannot heartbeat: this instance no longer holds the claim" +) + +// Config configures a Manager (design §7.1 flags). +type Config struct { + // Enabled turns on ownership-claim enforcement + // (--enable-ownership-claims). When false, callers should bypass the + // Manager entirely; today's bare-relabel behavior is already correct + // per design §5.4. + Enabled bool + + // TTL is the staleness threshold: a claim not renewed within TTL is + // presumed abandoned by a dead owner (--ownership-claim-ttl, default + // 90s -- three missed heartbeats at the default interval). + TTL time.Duration + + // HeartbeatInterval is how often a held claim's renewTime is refreshed + // while a run executes (--ownership-heartbeat-interval, default 30s). + HeartbeatInterval time.Duration + + // HolderIdentity uniquely identifies this instance (e.g. its pod name). + HolderIdentity string + + // Namespace holds claim Leases for cluster-scoped Workspaces, which have + // no namespace of their own. Namespaced Workspaces' claims live in the + // Workspace's own namespace instead, so the Lease's owner reference + // stays same-namespace and valid -- Kubernetes silently drops (and + // never garbage-collects on) a cross-namespace owner reference. + Namespace string +} + +// Manager guards Terraform runs with a per-Workspace ownership claim stored +// in a coordination.k8s.io/v1 Lease (design §8). +type Manager struct { + kube client.Client + cfg Config + now func() time.Time +} + +// NewManager returns a claim Manager backed by kube. +func NewManager(kube client.Client, cfg Config) *Manager { + return &Manager{kube: kube, cfg: cfg, now: time.Now} +} + +// claimKey returns the claim Lease's namespace and name for ws. +func claimKey(cfg Config, ws client.Object) client.ObjectKey { + ns := ws.GetNamespace() + if ns == "" { + ns = cfg.Namespace + } + return client.ObjectKey{Namespace: ns, Name: string(ws.GetUID())} +} + +// Acquire attempts to take the ownership claim for ws (R2-R4), creating its +// Lease if it doesn't exist yet. ownerGVK identifies ws's concrete type +// (cluster-scoped or namespaced Workspace) so the new Lease can carry a +// correct owner reference; Manager itself has no dependency on either +// Workspace API package. +func (m *Manager) Acquire(ctx context.Context, ws client.Object, ownerGVK schema.GroupVersionKind) (Outcome, error) { + key := claimKey(m.cfg, ws) + + lease := &coordinationv1.Lease{} + err := m.kube.Get(ctx, key, lease) + if kerrors.IsNotFound(err) { + return m.create(ctx, key, ws, ownerGVK) + } + if err != nil { + return Backoff, errors.Wrap(err, errGetLease) + } + + holder := "" + if lease.Spec.HolderIdentity != nil { + holder = *lease.Spec.HolderIdentity + } + foreign := holder != "" && holder != m.cfg.HolderIdentity + + if foreign && m.isFresh(lease) { + return Backoff, nil // R3 + } + + m.stamp(lease) + if err := m.kube.Update(ctx, lease); err != nil { + if kerrors.IsConflict(err) { + return Backoff, nil // R2: another racer's write won the resourceVersion race + } + return Backoff, errors.Wrap(err, errUpdateLease) + } + //Signifies that the lease is acquired from another owner with a stale claim. + if foreign { + return Stolen, nil // R4 + } + return Acquired, nil // R2: claim was empty or already ours +} + +func (m *Manager) create(ctx context.Context, key client.ObjectKey, ws client.Object, ownerGVK schema.GroupVersionKind) (Outcome, error) { + lease := &coordinationv1.Lease{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: key.Namespace, + Name: key.Name, + OwnerReferences: []metav1.OwnerReference{*metav1.NewControllerRef(ws, ownerGVK)}, + }, + } + m.stamp(lease) + if err := m.kube.Create(ctx, lease); err != nil { + if kerrors.IsAlreadyExists(err) { + return Backoff, nil // R2: another racer created it first + } + return Backoff, errors.Wrap(err, errCreateLease) + } + return Acquired, nil +} + +// isFresh reports whether lease's last heartbeat is within TTL. +func (m *Manager) isFresh(lease *coordinationv1.Lease) bool { + if lease.Spec.RenewTime == nil { + return false + } + return m.now().Sub(lease.Spec.RenewTime.Time) < m.cfg.TTL +} + +// stamp sets lease's holder, renew time, and duration to this instance's +// current claim. +func (m *Manager) stamp(lease *coordinationv1.Lease) { + now := metav1.NewMicroTime(m.now()) + holder := m.cfg.HolderIdentity + dur := int32(m.cfg.TTL / time.Second) + lease.Spec.HolderIdentity = &holder + lease.Spec.RenewTime = &now + lease.Spec.LeaseDurationSeconds = &dur +} + +// Heartbeat renews the claim's renewTime while a run is executing (R5). It +// returns an error if this instance no longer holds the claim -- the +// zombie-window case (design §8 residual risk: a paused-then-resumed owner +// may briefly believe it still holds the claim; its next heartbeat write +// must fail so it can abort). +func (m *Manager) Heartbeat(ctx context.Context, ws client.Object) error { + key := claimKey(m.cfg, ws) + lease := &coordinationv1.Lease{} + if err := m.kube.Get(ctx, key, lease); err != nil { + return errors.Wrap(err, errGetLease) + } + if lease.Spec.HolderIdentity == nil || *lease.Spec.HolderIdentity != m.cfg.HolderIdentity { + return errors.New(errNotHolder) + } + m.stamp(lease) + return errors.Wrap(m.kube.Update(ctx, lease), errUpdateLease) +} + +// Release clears the claim when a reconcile ends, success or failure (R6). +// It is a no-op if the claim was already cleared, already stolen by another +// instance, or its Lease no longer exists. +func (m *Manager) Release(ctx context.Context, ws client.Object) error { + key := claimKey(m.cfg, ws) + lease := &coordinationv1.Lease{} + if err := m.kube.Get(ctx, key, lease); err != nil { + if kerrors.IsNotFound(err) { + return nil + } + return errors.Wrap(err, errGetLease) + } + if lease.Spec.HolderIdentity == nil || *lease.Spec.HolderIdentity != m.cfg.HolderIdentity { + return nil + } + lease.Spec.HolderIdentity = nil + lease.Spec.RenewTime = nil + return errors.Wrap(m.kube.Update(ctx, lease), errClearLease) +} diff --git a/internal/claims/claims_test.go b/internal/claims/claims_test.go new file mode 100644 index 0000000..def6818 --- /dev/null +++ b/internal/claims/claims_test.go @@ -0,0 +1,363 @@ +/* +Copyright 2020 The Crossplane Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package claims + +import ( + "context" + "testing" + "time" + + "github.com/crossplane/crossplane-runtime/v2/pkg/test" + "github.com/pkg/errors" + coordinationv1 "k8s.io/api/coordination/v1" + corev1 "k8s.io/api/core/v1" + kerrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +var testGVK = schema.GroupVersionKind{Group: "tf.upbound.io", Version: "v1beta1", Kind: "Workspace"} + +// testWorkspace returns a stand-in client.Object. The claims package has no +// dependency on either Workspace API, so any client.Object with a UID and +// (optionally) a namespace exercises it faithfully. +func testWorkspace(ns string) client.Object { + return &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "ws-a", Namespace: ns, UID: types.UID("uid-a")}} +} + +func holderPtr(s string) *string { return &s } + +func microTime(t time.Time) *metav1.MicroTime { + mt := metav1.NewMicroTime(t) + return &mt +} + +func TestClaimKey(t *testing.T) { + cfg := Config{Namespace: "provider-ns"} + + cases := map[string]struct { + ws client.Object + want client.ObjectKey + }{ + "ClusterScopedFallsBackToProviderNamespace": { + ws: testWorkspace(""), + want: client.ObjectKey{Namespace: "provider-ns", Name: "uid-a"}, + }, + "NamespacedUsesWorkspaceNamespace": { + // Keeps the Lease's owner reference same-namespace and therefore + // valid -- Kubernetes silently drops a cross-namespace owner + // reference and never garbage-collects across the boundary. + ws: testWorkspace("tenant-a"), + want: client.ObjectKey{Namespace: "tenant-a", Name: "uid-a"}, + }, + } + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + if got := claimKey(cfg, tc.ws); got != tc.want { + t.Errorf("claimKey(...) = %+v, want %+v", got, tc.want) + } + }) + } +} + +func TestAcquire(t *testing.T) { + fixedNow := time.Date(2026, 7, 3, 12, 0, 0, 0, time.UTC) + cfg := Config{HolderIdentity: "instance-1", Namespace: "provider-ns", TTL: 90 * time.Second} + + cases := map[string]struct { + client *test.MockClient + want Outcome + wantErr bool + }{ + "CreatesWhenAbsent": { + // No Lease exists yet: this instance creates and claims it (R2). + client: &test.MockClient{ + MockGet: test.NewMockGetFn(kerrors.NewNotFound(coordinationv1.Resource("leases"), "uid-a")), + MockCreate: test.NewMockCreateFn(nil), + }, + want: Acquired, + }, + "CreateRaceLoses": { + // Two instances raced to create the Lease for a brand-new + // Workspace; another instance's Create won (R2 loser). + client: &test.MockClient{ + MockGet: test.NewMockGetFn(kerrors.NewNotFound(coordinationv1.Resource("leases"), "uid-a")), + MockCreate: test.NewMockCreateFn(kerrors.NewAlreadyExists(coordinationv1.Resource("leases"), "uid-a")), + }, + want: Backoff, + }, + "AcquiresEmptyClaim": { + // Lease exists but is unclaimed (e.g. after a Release): take it. + client: &test.MockClient{ + MockGet: test.NewMockGetFn(nil), + MockUpdate: test.NewMockUpdateFn(nil), + }, + want: Acquired, + }, + "ReacquiresOwnClaim": { + // Already ours: re-acquiring (e.g. a retried reconcile) succeeds. + client: &test.MockClient{ + MockGet: test.NewMockGetFn(nil, func(obj client.Object) error { + l := obj.(*coordinationv1.Lease) //nolint:forcetypeassert // test double, always a Lease + l.Spec.HolderIdentity = holderPtr("instance-1") + l.Spec.RenewTime = microTime(fixedNow) + return nil + }), + MockUpdate: test.NewMockUpdateFn(nil), + }, + want: Acquired, + }, + "UpdateRaceLoses": { + // Claim was empty when read, but another instance's write won + // the resourceVersion race before ours landed (R2 loser). + client: &test.MockClient{ + MockGet: test.NewMockGetFn(nil), + MockUpdate: test.NewMockUpdateFn(kerrors.NewConflict(coordinationv1.Resource("leases"), "uid-a", errors.New("conflict"))), + }, + want: Backoff, + }, + "ForeignFreshBacksOff": { + // R3: a foreign, fresh (10s old, TTL 90s) claim -- do not run. + client: &test.MockClient{ + MockGet: test.NewMockGetFn(nil, func(obj client.Object) error { + l := obj.(*coordinationv1.Lease) //nolint:forcetypeassert // test double, always a Lease + l.Spec.HolderIdentity = holderPtr("instance-2") + l.Spec.RenewTime = microTime(fixedNow.Add(-10 * time.Second)) + return nil + }), + MockUpdate: func(_ context.Context, _ client.Object, _ ...client.UpdateOption) error { + t.Fatal("Update must not be called when a foreign claim is fresh (R3)") + return nil + }, + }, + want: Backoff, + }, + "ForeignStaleIsStolen": { + // R4: a foreign claim 100s old (> 90s TTL) is presumed abandoned. + client: &test.MockClient{ + MockGet: test.NewMockGetFn(nil, func(obj client.Object) error { + l := obj.(*coordinationv1.Lease) //nolint:forcetypeassert // test double, always a Lease + l.Spec.HolderIdentity = holderPtr("instance-2") + l.Spec.RenewTime = microTime(fixedNow.Add(-100 * time.Second)) + return nil + }), + MockUpdate: test.NewMockUpdateFn(nil), + }, + want: Stolen, + }, + "MissingRenewTimeIsTreatedAsStale": { + // A held claim with no renewTime yet must not be treated as + // fresh -- isFresh returns false whenever RenewTime is nil. + client: &test.MockClient{ + MockGet: test.NewMockGetFn(nil, func(obj client.Object) error { + l := obj.(*coordinationv1.Lease) //nolint:forcetypeassert // test double, always a Lease + l.Spec.HolderIdentity = holderPtr("instance-2") + return nil + }), + MockUpdate: test.NewMockUpdateFn(nil), + }, + want: Stolen, + }, + "GetErrorIsBackoff": { + client: &test.MockClient{ + MockGet: test.NewMockGetFn(errors.New("boom")), + }, + want: Backoff, + wantErr: true, + }, + } + + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + m := NewManager(tc.client, cfg) + m.now = func() time.Time { return fixedNow } + + got, err := m.Acquire(context.Background(), testWorkspace(""), testGVK) + if (err != nil) != tc.wantErr { + t.Fatalf("Acquire(...): err = %v, wantErr = %v", err, tc.wantErr) + } + if got != tc.want { + t.Errorf("Acquire(...) = %v, want %v", got, tc.want) + } + }) + } +} + +func TestAcquireSetsOwnerReference(t *testing.T) { + fixedNow := time.Date(2026, 7, 3, 12, 0, 0, 0, time.UTC) + cfg := Config{HolderIdentity: "instance-1", Namespace: "provider-ns", TTL: 90 * time.Second} + + var created *coordinationv1.Lease + c := &test.MockClient{ + MockGet: test.NewMockGetFn(kerrors.NewNotFound(coordinationv1.Resource("leases"), "uid-a")), + MockCreate: test.NewMockCreateFn(nil, func(obj client.Object) error { + created = obj.(*coordinationv1.Lease) //nolint:forcetypeassert // test double, always a Lease + return nil + }), + } + m := NewManager(c, cfg) + m.now = func() time.Time { return fixedNow } + + ws := testWorkspace("tenant-a") + if _, err := m.Acquire(context.Background(), ws, testGVK); err != nil { + t.Fatalf("Acquire(...): unexpected error: %v", err) + } + + if created == nil { + t.Fatal("Create was not called") + } + if created.Namespace != "tenant-a" { + t.Errorf("Lease namespace = %q, want %q (same-namespace as the Workspace, for a valid owner reference)", created.Namespace, "tenant-a") + } + if len(created.OwnerReferences) != 1 { + t.Fatalf("got %d owner references, want 1", len(created.OwnerReferences)) + } + if got := created.OwnerReferences[0].UID; got != ws.GetUID() { + t.Errorf("owner reference UID = %q, want %q", got, ws.GetUID()) + } + if got := created.OwnerReferences[0].Kind; got != testGVK.Kind { + t.Errorf("owner reference Kind = %q, want %q", got, testGVK.Kind) + } +} + +func TestHeartbeat(t *testing.T) { + fixedNow := time.Date(2026, 7, 3, 12, 5, 0, 0, time.UTC) + cfg := Config{HolderIdentity: "instance-1", Namespace: "provider-ns", TTL: 90 * time.Second} + + cases := map[string]struct { + client *test.MockClient + wantErr bool + }{ + "RenewsOwnClaim": { + client: &test.MockClient{ + MockGet: test.NewMockGetFn(nil, func(obj client.Object) error { + l := obj.(*coordinationv1.Lease) //nolint:forcetypeassert // test double, always a Lease + l.Spec.HolderIdentity = holderPtr("instance-1") + l.Spec.RenewTime = microTime(fixedNow.Add(-30 * time.Second)) + return nil + }), + MockUpdate: test.NewMockUpdateFn(nil), + }, + }, + "ZombieWindowAbortsWhenClaimWasStolen": { + // Design §8 residual risk: a paused-then-resumed owner may + // briefly believe it still holds the claim. Its heartbeat must + // fail once another instance has taken over, so it aborts + // instead of running concurrently. + client: &test.MockClient{ + MockGet: test.NewMockGetFn(nil, func(obj client.Object) error { + l := obj.(*coordinationv1.Lease) //nolint:forcetypeassert // test double, always a Lease + l.Spec.HolderIdentity = holderPtr("instance-2") + l.Spec.RenewTime = microTime(fixedNow) + return nil + }), + MockUpdate: func(_ context.Context, _ client.Object, _ ...client.UpdateOption) error { + t.Fatal("Update must not be called once the claim has been stolen by another instance") + return nil + }, + }, + wantErr: true, + }, + "ClaimClearedAbortsToo": { + client: &test.MockClient{ + MockGet: test.NewMockGetFn(nil), + MockUpdate: func(_ context.Context, _ client.Object, _ ...client.UpdateOption) error { + t.Fatal("Update must not be called once the claim has been cleared") + return nil + }, + }, + wantErr: true, + }, + } + + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + m := NewManager(tc.client, cfg) + m.now = func() time.Time { return fixedNow } + + err := m.Heartbeat(context.Background(), testWorkspace("")) + if (err != nil) != tc.wantErr { + t.Fatalf("Heartbeat(...): err = %v, wantErr = %v", err, tc.wantErr) + } + }) + } +} + +func TestRelease(t *testing.T) { + cfg := Config{HolderIdentity: "instance-1", Namespace: "provider-ns", TTL: 90 * time.Second} + + cases := map[string]struct { + client *test.MockClient + wantErr bool + }{ + "ClearsOwnClaim": { + client: &test.MockClient{ + MockGet: test.NewMockGetFn(nil, func(obj client.Object) error { + l := obj.(*coordinationv1.Lease) //nolint:forcetypeassert // test double, always a Lease + l.Spec.HolderIdentity = holderPtr("instance-1") + l.Spec.RenewTime = microTime(time.Now()) + return nil + }), + MockUpdate: test.NewMockUpdateFn(nil, func(obj client.Object) error { + l := obj.(*coordinationv1.Lease) //nolint:forcetypeassert // test double, always a Lease + if l.Spec.HolderIdentity != nil { + t.Errorf("HolderIdentity = %q, want cleared (nil)", *l.Spec.HolderIdentity) + } + if l.Spec.RenewTime != nil { + t.Errorf("RenewTime = %v, want cleared (nil)", l.Spec.RenewTime) + } + return nil + }), + }, + }, + "NoOpWhenNotHeldByUs": { + client: &test.MockClient{ + MockGet: test.NewMockGetFn(nil, func(obj client.Object) error { + l := obj.(*coordinationv1.Lease) //nolint:forcetypeassert // test double, always a Lease + l.Spec.HolderIdentity = holderPtr("instance-2") + return nil + }), + MockUpdate: func(_ context.Context, _ client.Object, _ ...client.UpdateOption) error { + t.Fatal("Update must not be called releasing a claim this instance doesn't hold") + return nil + }, + }, + }, + "NoOpWhenLeaseAlreadyGone": { + client: &test.MockClient{ + MockGet: test.NewMockGetFn(kerrors.NewNotFound(coordinationv1.Resource("leases"), "uid-a")), + }, + }, + "GetErrorPropagates": { + client: &test.MockClient{ + MockGet: test.NewMockGetFn(errors.New("boom")), + }, + wantErr: true, + }, + } + + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + m := NewManager(tc.client, cfg) + if err := m.Release(context.Background(), testWorkspace("")); (err != nil) != tc.wantErr { + t.Fatalf("Release(...): err = %v, wantErr = %v", err, tc.wantErr) + } + }) + } +} diff --git a/internal/claims/guard.go b/internal/claims/guard.go new file mode 100644 index 0000000..9d566f6 --- /dev/null +++ b/internal/claims/guard.go @@ -0,0 +1,108 @@ +/* +Copyright 2020 The Crossplane Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package claims + +import ( + "context" + "time" + + "github.com/pkg/errors" + "k8s.io/apimachinery/pkg/runtime/schema" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// ErrBackoff is returned by Guard when a foreign, fresh claim prevented a +// run (R3). Callers should surface this as a quiet, expected condition -- +// not a Terraform failure -- and let the normal reconcile requeue retry. +var ErrBackoff = errors.New("workspace deferred: ownership claim held by another instance") + +// GuardFunc is work Guard performs on ws's behalf. +type GuardFunc func(ctx context.Context) error + +// Guard runs fn only while this instance holds ws's ownership claim (R1) -- +// the single seam both the cluster-scoped and namespaced Workspace +// reconcilers call before a state-mutating Terraform run (Apply or +// Destroy). It implements design doc §8: +// +// - R2 (empty/own claim): acquire, then run fn immediately. +// - R3 (foreign, fresh claim): fn is not called; Guard returns ErrBackoff +// so the caller requeues with backoff instead of running. +// - R4 (foreign, stale claim): the claim is stolen, unlock runs first to +// clear any state lock left by the presumed-dead owner, then fn runs. +// - R5: while fn runs, the claim is heartbeated every cfg.HeartbeatInterval. +// - R6: the claim is released when fn returns, success or failure. +// - R7 (a label flip never interrupts a run): Guard has no part in this -- +// it holds only for the fn call, which the caller's existing synchronous +// reconcile + exec.CommandContext binding already protects. +// +// If cfg.Enabled is false, Guard is a transparent pass-through to fn -- +// this is the parity guarantee for the (default) disabled case. +func Guard(ctx context.Context, mgr *Manager, cfg Config, ws client.Object, ownerGVK schema.GroupVersionKind, unlock, fn GuardFunc) error { + if !cfg.Enabled { + return fn(ctx) + } + + outcome, err := mgr.Acquire(ctx, ws, ownerGVK) + if err != nil { + return err + } + + switch outcome { + case Backoff: + return ErrBackoff + case Stolen: + if err := unlock(ctx); err != nil { + return err + } + case Acquired: + } + + stop := make(chan struct{}) + done := make(chan struct{}) + go heartbeat(ctx, mgr, ws, cfg.HeartbeatInterval, stop, done) + + runErr := fn(ctx) + + close(stop) + <-done + + if relErr := mgr.Release(ctx, ws); relErr != nil && runErr == nil { + return relErr + } + return runErr +} + +// heartbeat renews ws's claim every interval until stop is closed, then +// closes done. Heartbeat failures are not fatal to the in-flight run -- +// R7 relies on the run completing regardless (the zombie-window residual +// risk noted in design §8 is bounded by the state lock, not by aborting +// heartbeats mid-run). +func heartbeat(ctx context.Context, mgr *Manager, ws client.Object, interval time.Duration, stop, done chan struct{}) { + defer close(done) + + t := time.NewTicker(interval) + defer t.Stop() + + for { + select { + case <-t.C: + _ = mgr.Heartbeat(ctx, ws) + case <-stop: + return + } + } +} diff --git a/internal/claims/guard_test.go b/internal/claims/guard_test.go new file mode 100644 index 0000000..400ec28 --- /dev/null +++ b/internal/claims/guard_test.go @@ -0,0 +1,225 @@ +/* +Copyright 2020 The Crossplane Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package claims + +import ( + "context" + "sync/atomic" + "testing" + "time" + + "github.com/crossplane/crossplane-runtime/v2/pkg/test" + "github.com/pkg/errors" + coordinationv1 "k8s.io/api/coordination/v1" + kerrors "k8s.io/apimachinery/pkg/api/errors" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +func TestGuardDisabledIsPassthrough(t *testing.T) { + // cfg.Enabled=false must never touch mgr at all -- a nil Manager must + // not panic, proving the short-circuit happens before any claim I/O. + cfg := Config{Enabled: false} + + var fnCalled, unlockCalled bool + err := Guard(context.Background(), nil, cfg, testWorkspace(""), testGVK, + func(ctx context.Context) error { unlockCalled = true; return nil }, + func(ctx context.Context) error { fnCalled = true; return nil }, + ) + if err != nil { + t.Fatalf("Guard(...): unexpected error: %v", err) + } + if !fnCalled { + t.Error("fn was not called") + } + if unlockCalled { + t.Error("unlock must not be called when claims are disabled") + } +} + +func TestGuardAcquiredRunsFnAndReleases(t *testing.T) { + // A tiny fake backing store: Get/Update read and write the same Lease, + // so Release's independent Get sees what Acquire's Update wrote. + var stored coordinationv1.Lease + c := &test.MockClient{ + MockGet: func(_ context.Context, _ client.ObjectKey, obj client.Object) error { + l := obj.(*coordinationv1.Lease) //nolint:forcetypeassert // test double, always a Lease + stored.DeepCopyInto(l) + return nil + }, + MockUpdate: func(_ context.Context, obj client.Object, _ ...client.UpdateOption) error { + l := obj.(*coordinationv1.Lease) //nolint:forcetypeassert // test double, always a Lease + l.DeepCopyInto(&stored) + return nil + }, + } + mgr := NewManager(c, Config{HolderIdentity: "instance-1", Namespace: "provider-ns", TTL: 90 * time.Second}) + + var fnCalled bool + err := Guard(context.Background(), mgr, Config{Enabled: true, HeartbeatInterval: time.Hour}, testWorkspace(""), testGVK, + func(ctx context.Context) error { t.Fatal("unlock must not be called on a clean acquire"); return nil }, + func(ctx context.Context) error { fnCalled = true; return nil }, + ) + if err != nil { + t.Fatalf("Guard(...): unexpected error: %v", err) + } + if !fnCalled { + t.Error("fn was not called") + } + if stored.Spec.HolderIdentity != nil { + t.Errorf("claim was not released after fn succeeded: HolderIdentity = %q", *stored.Spec.HolderIdentity) + } +} + +func TestGuardBackoffDoesNotRunFn(t *testing.T) { + holder := "instance-2" + fresh := time.Now() + c := &test.MockClient{ + MockGet: test.NewMockGetFn(nil, func(obj client.Object) error { + l := obj.(*coordinationv1.Lease) //nolint:forcetypeassert // test double, always a Lease + l.Spec.HolderIdentity = &holder + mt := microTime(fresh) + l.Spec.RenewTime = mt + return nil + }), + } + mgr := NewManager(c, Config{HolderIdentity: "instance-1", Namespace: "provider-ns", TTL: 90 * time.Second}) + + err := Guard(context.Background(), mgr, Config{Enabled: true, HeartbeatInterval: time.Hour}, testWorkspace(""), testGVK, + func(ctx context.Context) error { t.Fatal("unlock must not be called on backoff"); return nil }, + func(ctx context.Context) error { t.Fatal("fn must not run while a foreign claim is fresh"); return nil }, + ) + if !errors.Is(err, ErrBackoff) { + t.Errorf("err = %v, want ErrBackoff", err) + } +} + +func TestGuardStolenCallsUnlockThenFn(t *testing.T) { + holder := "instance-2" + stale := time.Now().Add(-100 * time.Second) + c := &test.MockClient{ + MockGet: test.NewMockGetFn(nil, func(obj client.Object) error { + l := obj.(*coordinationv1.Lease) //nolint:forcetypeassert // test double, always a Lease + l.Spec.HolderIdentity = &holder + mt := microTime(stale) + l.Spec.RenewTime = mt + return nil + }), + MockUpdate: test.NewMockUpdateFn(nil), + } + mgr := NewManager(c, Config{HolderIdentity: "instance-1", Namespace: "provider-ns", TTL: 90 * time.Second}) + + var order []string + err := Guard(context.Background(), mgr, Config{Enabled: true, HeartbeatInterval: time.Hour}, testWorkspace(""), testGVK, + func(ctx context.Context) error { order = append(order, "unlock"); return nil }, + func(ctx context.Context) error { order = append(order, "fn"); return nil }, + ) + if err != nil { + t.Fatalf("Guard(...): unexpected error: %v", err) + } + if len(order) != 2 || order[0] != "unlock" || order[1] != "fn" { + t.Errorf("call order = %v, want [unlock fn]", order) + } +} + +func TestGuardUnlockErrorPreventsFn(t *testing.T) { + holder := "instance-2" + stale := time.Now().Add(-100 * time.Second) + c := &test.MockClient{ + MockGet: test.NewMockGetFn(nil, func(obj client.Object) error { + l := obj.(*coordinationv1.Lease) //nolint:forcetypeassert // test double, always a Lease + l.Spec.HolderIdentity = &holder + mt := microTime(stale) + l.Spec.RenewTime = mt + return nil + }), + // Acquire's steal path writes the new holder via Update *before* + // Guard ever calls unlock -- this must succeed for Stolen to be + // returned at all. + MockUpdate: test.NewMockUpdateFn(nil), + } + mgr := NewManager(c, Config{HolderIdentity: "instance-1", Namespace: "provider-ns", TTL: 90 * time.Second}) + + wantErr := errors.New("force-unlock failed") + err := Guard(context.Background(), mgr, Config{Enabled: true, HeartbeatInterval: time.Hour}, testWorkspace(""), testGVK, + func(ctx context.Context) error { return wantErr }, + func(ctx context.Context) error { t.Fatal("fn must not run when unlock fails"); return nil }, + ) + if !errors.Is(err, wantErr) { + t.Errorf("err = %v, want %v", err, wantErr) + } +} + +func TestGuardFnErrorTakesPrecedenceOverReleaseError(t *testing.T) { + // First Update (Acquire) succeeds; second Update (Release) fails -- Guard + // must still surface fn's error, not Release's. + var calls int32 + c := &test.MockClient{ + MockGet: test.NewMockGetFn(nil), + MockUpdate: func(_ context.Context, _ client.Object, _ ...client.UpdateOption) error { + if atomic.AddInt32(&calls, 1) == 1 { + return nil + } + return kerrors.NewConflict(coordinationv1.Resource("leases"), "uid-a", errors.New("conflict")) + }, + } + mgr := NewManager(c, Config{HolderIdentity: "instance-1", Namespace: "provider-ns", TTL: 90 * time.Second}) + + fnErr := errors.New("apply failed") + err := Guard(context.Background(), mgr, Config{Enabled: true, HeartbeatInterval: time.Hour}, testWorkspace(""), testGVK, + func(ctx context.Context) error { return nil }, + func(ctx context.Context) error { return fnErr }, + ) + if !errors.Is(err, fnErr) { + t.Errorf("err = %v, want the fn error (%v), not the release error", err, fnErr) + } +} + +func TestGuardHeartbeatsWhileFnRuns(t *testing.T) { + // A fake backing store, as in TestGuardAcquiredRunsFnAndReleases: each + // heartbeat's Get must observe what the prior Update (acquire, or an + // earlier heartbeat) wrote, or Heartbeat bails out as "not the holder". + var stored coordinationv1.Lease + var updates int32 + c := &test.MockClient{ + MockGet: func(_ context.Context, _ client.ObjectKey, obj client.Object) error { + l := obj.(*coordinationv1.Lease) //nolint:forcetypeassert // test double, always a Lease + stored.DeepCopyInto(l) + return nil + }, + MockUpdate: func(_ context.Context, obj client.Object, _ ...client.UpdateOption) error { + atomic.AddInt32(&updates, 1) + l := obj.(*coordinationv1.Lease) //nolint:forcetypeassert // test double, always a Lease + l.DeepCopyInto(&stored) + return nil + }, + } + mgr := NewManager(c, Config{HolderIdentity: "instance-1", Namespace: "provider-ns", TTL: 90 * time.Second}) + + err := Guard(context.Background(), mgr, Config{Enabled: true, HeartbeatInterval: 5 * time.Millisecond}, testWorkspace(""), testGVK, + func(ctx context.Context) error { return nil }, + func(ctx context.Context) error { time.Sleep(40 * time.Millisecond); return nil }, + ) + if err != nil { + t.Fatalf("Guard(...): unexpected error: %v", err) + } + // The first Update acquires the claim and the last releases it; + // sleeping 8x the interval in between should yield at least one + // genuine heartbeat beyond those two. + if atomic.LoadInt32(&updates) < 3 { + t.Errorf("Update called %d times, want at least 3 (acquire + >=1 heartbeat + release)", updates) + } +} diff --git a/internal/controller/cluster/terraform.go b/internal/controller/cluster/terraform.go index b0f3a37..0d4f0f5 100644 --- a/internal/controller/cluster/terraform.go +++ b/internal/controller/cluster/terraform.go @@ -23,34 +23,25 @@ import ( "github.com/crossplane/crossplane-runtime/v2/pkg/controller" + "github.com/upbound/provider-terraform/internal/claims" "github.com/upbound/provider-terraform/internal/controller/cluster/config" "github.com/upbound/provider-terraform/internal/controller/cluster/workspace" ) // Setup creates all TF controllers with the supplied logger and adds them // to the supplied manager. -func Setup(mgr ctrl.Manager, o controller.Options, timeout time.Duration, pollJitter time.Duration) error { - for _, setup := range []func(ctrl.Manager, controller.Options, time.Duration, time.Duration) error{ - config.Setup, - workspace.Setup, - } { - if err := setup(mgr, o, timeout, pollJitter); err != nil { - return err - } +func Setup(mgr ctrl.Manager, o controller.Options, timeout time.Duration, pollJitter time.Duration, claimCfg claims.Config) error { + if err := config.Setup(mgr, o, timeout, pollJitter); err != nil { + return err } - return nil + return workspace.Setup(mgr, o, timeout, pollJitter, claimCfg) } // SetupGated creates all controllers with the supplied logger and adds them to // the supplied manager gated. -func SetupGated(mgr ctrl.Manager, o controller.Options, timeout time.Duration, pollJitter time.Duration) error { - for _, setup := range []func(ctrl.Manager, controller.Options, time.Duration, time.Duration) error{ - config.SetupGated, - workspace.SetupGated, - } { - if err := setup(mgr, o, timeout, pollJitter); err != nil { - return err - } +func SetupGated(mgr ctrl.Manager, o controller.Options, timeout time.Duration, pollJitter time.Duration, claimCfg claims.Config) error { + if err := config.SetupGated(mgr, o, timeout, pollJitter); err != nil { + return err } - return nil + return workspace.SetupGated(mgr, o, timeout, pollJitter, claimCfg) } diff --git a/internal/controller/cluster/workspace/workspace.go b/internal/controller/cluster/workspace/workspace.go index 37d977c..25d6c77 100644 --- a/internal/controller/cluster/workspace/workspace.go +++ b/internal/controller/cluster/workspace/workspace.go @@ -46,8 +46,10 @@ import ( "github.com/crossplane/crossplane-runtime/v2/pkg/resource" "github.com/upbound/provider-terraform/apis/cluster/v1beta1" + "github.com/upbound/provider-terraform/internal/claims" tfClient "github.com/upbound/provider-terraform/internal/clients" "github.com/upbound/provider-terraform/internal/features" + "github.com/upbound/provider-terraform/internal/handover" "github.com/upbound/provider-terraform/internal/terraform" sourcev1 "github.com/fluxcd/source-controller/api/v1" @@ -119,18 +121,21 @@ type tfclient interface { Destroy(ctx context.Context, o ...terraform.Option) error DeleteCurrentWorkspace(ctx context.Context) error GenerateChecksum(ctx context.Context) (string, error) + ForceUnlock(ctx context.Context, o ...terraform.Option) error } // Setup adds a controller that reconciles Workspace managed resources. -func Setup(mgr ctrl.Manager, o controller.Options, timeout, pollJitter time.Duration) error { +func Setup(mgr ctrl.Manager, o controller.Options, timeout, pollJitter time.Duration, claimCfg claims.Config) error { name := managed.ControllerName(v1beta1.WorkspaceGroupKind) fs := afero.Afero{Fs: afero.NewOsFs()} c := &connector{ - kube: mgr.GetClient(), - usage: resource.NewLegacyProviderConfigUsageTracker(mgr.GetClient(), &v1beta1.ProviderConfigUsage{}), - logger: o.Logger, - fs: fs, + kube: mgr.GetClient(), + usage: resource.NewLegacyProviderConfigUsageTracker(mgr.GetClient(), &v1beta1.ProviderConfigUsage{}), + logger: o.Logger, + fs: fs, + claimCfg: claimCfg, + claimMgr: claims.NewManager(mgr.GetClient(), claimCfg), terraform: func(dir string, usePluginCache bool, enableTerraformCLILogging bool, logger logging.Logger, envs ...string) tfclient { return terraform.Harness{Path: tfPath, Dir: dir, UsePluginCache: usePluginCache, EnableTerraformCLILogging: enableTerraformCLILogging, Logger: logger, Envs: envs} }, @@ -146,6 +151,8 @@ func Setup(mgr ctrl.Manager, o controller.Options, timeout, pollJitter time.Dura managed.WithMetricRecorder(o.MetricOptions.MRMetrics), } + opts = append(opts, handover.ReconcilerOptions(mgr.GetClient(), mgr.GetAPIReader())...) + if o.Features.Enabled(features.EnableBetaManagementPolicies) { opts = append(opts, managed.WithManagementPolicies()) } @@ -169,9 +176,9 @@ func Setup(mgr ctrl.Manager, o controller.Options, timeout, pollJitter time.Dura // SetupGated adds a controller that reconciles ProviderConfigs by accounting for // their current usage. -func SetupGated(mgr ctrl.Manager, o controller.Options, timeout time.Duration, pollJitter time.Duration) error { +func SetupGated(mgr ctrl.Manager, o controller.Options, timeout time.Duration, pollJitter time.Duration, claimCfg claims.Config) error { o.Gate.Register(func() { - if err := Setup(mgr, o, timeout, pollJitter); err != nil { + if err := Setup(mgr, o, timeout, pollJitter, claimCfg); err != nil { mgr.GetLogger().Error(err, "unable to setup reconciler", "gvk", v1beta1.WorkspaceGroupVersionKind.String()) } }, v1beta1.WorkspaceGroupVersionKind) @@ -183,6 +190,8 @@ type connector struct { usage tfClient.LegacyTracker logger logging.Logger fs afero.Afero + claimCfg claims.Config + claimMgr *claims.Manager terraform func(dir string, usePluginCache bool, enableTerraformCLILogging bool, logger logging.Logger, envs ...string) tfclient } @@ -439,7 +448,7 @@ func (c *connector) Connect(ctx context.Context, mg resource.Managed) (managed.E } if cr.Status.AtProvider.Checksum == checksum { l.Debug("Checksums match - skip running terraform init") - return &external{tf: tf, kube: c.kube, logger: c.logger}, errors.Wrap(tf.Workspace(ctx, meta.GetExternalName(cr)), errWorkspace) + return &external{tf: tf, kube: c.kube, logger: c.logger, claimCfg: c.claimCfg, claimMgr: c.claimMgr}, errors.Wrap(tf.Workspace(ctx, meta.GetExternalName(cr)), errWorkspace) } l.Debug("Checksums don't match so run terraform init:", "old", cr.Status.AtProvider.Checksum, "new", checksum) } @@ -452,7 +461,7 @@ func (c *connector) Connect(ctx context.Context, mg resource.Managed) (managed.E if err := tf.Init(ctx, o...); err != nil { return nil, errors.Wrap(err, errInit) } - return &external{tf: tf, kube: c.kube}, errors.Wrap(tf.Workspace(ctx, meta.GetExternalName(cr)), errWorkspace) + return &external{tf: tf, kube: c.kube, claimCfg: c.claimCfg, claimMgr: c.claimMgr}, errors.Wrap(tf.Workspace(ctx, meta.GetExternalName(cr)), errWorkspace) } func (c *connector) getFluxArtefactURL(ctx context.Context, fluxSourceName string) (string, error) { @@ -496,9 +505,11 @@ func (c *connector) getFluxArtefactURL(ctx context.Context, fluxSourceName strin } type external struct { - tf tfclient - kube client.Client - logger logging.Logger + tf tfclient + kube client.Client + logger logging.Logger + claimCfg claims.Config + claimMgr *claims.Manager } func (c *external) checkDiff(ctx context.Context, cr *v1beta1.Workspace) (bool, error) { @@ -589,7 +600,14 @@ func (c *external) Update(ctx context.Context, mg resource.Managed) (managed.Ext } o = append(o, terraform.WithArgs(cr.Spec.ForProvider.ApplyArgs)) - if err := c.tf.Apply(ctx, o...); err != nil { + err = claims.Guard(ctx, c.claimMgr, c.claimCfg, cr, v1beta1.WorkspaceGroupVersionKind, + func(ctx context.Context) error { return c.tf.ForceUnlock(ctx, o...) }, + func(ctx context.Context) error { return c.tf.Apply(ctx, o...) }, + ) + if errors.Is(err, claims.ErrBackoff) { + return managed.ExternalUpdate{}, err + } + if err != nil { return managed.ExternalUpdate{}, errors.Wrap(err, errApply) } @@ -629,7 +647,14 @@ func (c *external) Delete(ctx context.Context, mg resource.Managed) (managed.Ext } o = append(o, terraform.WithArgs(cr.Spec.ForProvider.DestroyArgs)) - return managed.ExternalDelete{}, errors.Wrap(c.tf.Destroy(ctx, o...), errDestroy) + err = claims.Guard(ctx, c.claimMgr, c.claimCfg, cr, v1beta1.WorkspaceGroupVersionKind, + func(ctx context.Context) error { return c.tf.ForceUnlock(ctx, o...) }, + func(ctx context.Context) error { return c.tf.Destroy(ctx, o...) }, + ) + if errors.Is(err, claims.ErrBackoff) { + return managed.ExternalDelete{}, err + } + return managed.ExternalDelete{}, errors.Wrap(err, errDestroy) } func (c *external) Disconnect(ctx context.Context) error { diff --git a/internal/controller/cluster/workspace/workspace_test.go b/internal/controller/cluster/workspace/workspace_test.go index 65b004c..dc5c66b 100644 --- a/internal/controller/cluster/workspace/workspace_test.go +++ b/internal/controller/cluster/workspace/workspace_test.go @@ -81,6 +81,7 @@ type MockTf struct { MockDestroy func(ctx context.Context, o ...terraform.Option) error MockDeleteCurrentWorkspace func(ctx context.Context) error MockGenerateChecksum func(ctx context.Context) (string, error) + MockForceUnlock func(ctx context.Context, o ...terraform.Option) error } func (tf *MockTf) Init(ctx context.Context, o ...terraform.InitOption) error { @@ -119,6 +120,10 @@ func (tf *MockTf) DeleteCurrentWorkspace(ctx context.Context) error { return tf.MockDeleteCurrentWorkspace(ctx) } +func (tf *MockTf) ForceUnlock(ctx context.Context, o ...terraform.Option) error { + return tf.MockForceUnlock(ctx, o...) +} + func TestConnect(t *testing.T) { errBoom := errors.New("boom") errNoProviderConfig := errors.New(errProviderConfigNotSet) diff --git a/internal/controller/namespaced/terraform.go b/internal/controller/namespaced/terraform.go index f723a09..424beab 100644 --- a/internal/controller/namespaced/terraform.go +++ b/internal/controller/namespaced/terraform.go @@ -23,34 +23,25 @@ import ( "github.com/crossplane/crossplane-runtime/v2/pkg/controller" + "github.com/upbound/provider-terraform/internal/claims" "github.com/upbound/provider-terraform/internal/controller/namespaced/config" "github.com/upbound/provider-terraform/internal/controller/namespaced/workspace" ) // Setup creates all TF controllers with the supplied logger and adds them // to the supplied manager. -func Setup(mgr ctrl.Manager, o controller.Options, timeout time.Duration, pollJitter time.Duration) error { - for _, setup := range []func(ctrl.Manager, controller.Options, time.Duration, time.Duration) error{ - config.Setup, - workspace.Setup, - } { - if err := setup(mgr, o, timeout, pollJitter); err != nil { - return err - } +func Setup(mgr ctrl.Manager, o controller.Options, timeout time.Duration, pollJitter time.Duration, claimCfg claims.Config) error { + if err := config.Setup(mgr, o, timeout, pollJitter); err != nil { + return err } - return nil + return workspace.Setup(mgr, o, timeout, pollJitter, claimCfg) } // SetupGated creates all controllers with the supplied logger and adds them to // the supplied manager gated. -func SetupGated(mgr ctrl.Manager, o controller.Options, timeout time.Duration, pollJitter time.Duration) error { - for _, setup := range []func(ctrl.Manager, controller.Options, time.Duration, time.Duration) error{ - config.SetupGated, - workspace.SetupGated, - } { - if err := setup(mgr, o, timeout, pollJitter); err != nil { - return err - } +func SetupGated(mgr ctrl.Manager, o controller.Options, timeout time.Duration, pollJitter time.Duration, claimCfg claims.Config) error { + if err := config.SetupGated(mgr, o, timeout, pollJitter); err != nil { + return err } - return nil + return workspace.SetupGated(mgr, o, timeout, pollJitter, claimCfg) } diff --git a/internal/controller/namespaced/workspace/workspace.go b/internal/controller/namespaced/workspace/workspace.go index 27700bb..a8b7a07 100644 --- a/internal/controller/namespaced/workspace/workspace.go +++ b/internal/controller/namespaced/workspace/workspace.go @@ -46,8 +46,10 @@ import ( "github.com/crossplane/crossplane-runtime/v2/pkg/resource" "github.com/upbound/provider-terraform/apis/namespaced/v1beta1" + "github.com/upbound/provider-terraform/internal/claims" tfClient "github.com/upbound/provider-terraform/internal/clients" "github.com/upbound/provider-terraform/internal/features" + "github.com/upbound/provider-terraform/internal/handover" "github.com/upbound/provider-terraform/internal/terraform" sourcev1 "github.com/fluxcd/source-controller/api/v1" @@ -119,18 +121,21 @@ type tfclient interface { Destroy(ctx context.Context, o ...terraform.Option) error DeleteCurrentWorkspace(ctx context.Context) error GenerateChecksum(ctx context.Context) (string, error) + ForceUnlock(ctx context.Context, o ...terraform.Option) error } // Setup adds a controller that reconciles Workspace managed resources. -func Setup(mgr ctrl.Manager, o controller.Options, timeout, pollJitter time.Duration) error { +func Setup(mgr ctrl.Manager, o controller.Options, timeout, pollJitter time.Duration, claimCfg claims.Config) error { name := managed.ControllerName(v1beta1.WorkspaceGroupKind) fs := afero.Afero{Fs: afero.NewOsFs()} c := &connector{ - kube: mgr.GetClient(), - usage: resource.NewProviderConfigUsageTracker(mgr.GetClient(), &v1beta1.ProviderConfigUsage{}), - logger: o.Logger, - fs: fs, + kube: mgr.GetClient(), + usage: resource.NewProviderConfigUsageTracker(mgr.GetClient(), &v1beta1.ProviderConfigUsage{}), + logger: o.Logger, + fs: fs, + claimCfg: claimCfg, + claimMgr: claims.NewManager(mgr.GetClient(), claimCfg), terraform: func(dir string, usePluginCache bool, enableTerraformCLILogging bool, logger logging.Logger, envs ...string) tfclient { return terraform.Harness{Path: tfPath, Dir: dir, UsePluginCache: usePluginCache, EnableTerraformCLILogging: enableTerraformCLILogging, Logger: logger, Envs: envs} }, @@ -146,6 +151,8 @@ func Setup(mgr ctrl.Manager, o controller.Options, timeout, pollJitter time.Dura managed.WithMetricRecorder(o.MetricOptions.MRMetrics), } + opts = append(opts, handover.ReconcilerOptions(mgr.GetClient(), mgr.GetAPIReader())...) + if o.Features.Enabled(features.EnableBetaManagementPolicies) { opts = append(opts, managed.WithManagementPolicies()) } @@ -169,9 +176,9 @@ func Setup(mgr ctrl.Manager, o controller.Options, timeout, pollJitter time.Dura // SetupGated adds a controller that reconciles ProviderConfigs by accounting for // their current usage. -func SetupGated(mgr ctrl.Manager, o controller.Options, timeout time.Duration, pollJitter time.Duration) error { +func SetupGated(mgr ctrl.Manager, o controller.Options, timeout time.Duration, pollJitter time.Duration, claimCfg claims.Config) error { o.Gate.Register(func() { - if err := Setup(mgr, o, timeout, pollJitter); err != nil { + if err := Setup(mgr, o, timeout, pollJitter, claimCfg); err != nil { mgr.GetLogger().Error(err, "unable to setup reconciler", "gvk", v1beta1.WorkspaceGroupVersionKind.String()) } }, v1beta1.WorkspaceGroupVersionKind) @@ -183,6 +190,8 @@ type connector struct { usage tfClient.ModernTracker logger logging.Logger fs afero.Afero + claimCfg claims.Config + claimMgr *claims.Manager terraform func(dir string, usePluginCache bool, enableTerraformCLILogging bool, logger logging.Logger, envs ...string) tfclient } @@ -439,7 +448,7 @@ func (c *connector) Connect(ctx context.Context, mg resource.Managed) (managed.E } if cr.Status.AtProvider.Checksum == checksum { l.Debug("Checksums match - skip running terraform init") - return &external{tf: tf, kube: c.kube, logger: c.logger}, errors.Wrap(tf.Workspace(ctx, meta.GetExternalName(cr)), errWorkspace) + return &external{tf: tf, kube: c.kube, logger: c.logger, claimCfg: c.claimCfg, claimMgr: c.claimMgr}, errors.Wrap(tf.Workspace(ctx, meta.GetExternalName(cr)), errWorkspace) } l.Debug("Checksums don't match so run terraform init:", "old", cr.Status.AtProvider.Checksum, "new", checksum) } @@ -452,7 +461,7 @@ func (c *connector) Connect(ctx context.Context, mg resource.Managed) (managed.E if err := tf.Init(ctx, o...); err != nil { return nil, errors.Wrap(err, errInit) } - return &external{tf: tf, kube: c.kube}, errors.Wrap(tf.Workspace(ctx, meta.GetExternalName(cr)), errWorkspace) + return &external{tf: tf, kube: c.kube, claimCfg: c.claimCfg, claimMgr: c.claimMgr}, errors.Wrap(tf.Workspace(ctx, meta.GetExternalName(cr)), errWorkspace) } func (c *connector) getFluxArtefactURL(ctx context.Context, fluxSourceName string) (string, error) { @@ -496,9 +505,11 @@ func (c *connector) getFluxArtefactURL(ctx context.Context, fluxSourceName strin } type external struct { - tf tfclient - kube client.Client - logger logging.Logger + tf tfclient + kube client.Client + logger logging.Logger + claimCfg claims.Config + claimMgr *claims.Manager } func (c *external) checkDiff(ctx context.Context, cr *v1beta1.Workspace) (bool, error) { @@ -589,7 +600,14 @@ func (c *external) Update(ctx context.Context, mg resource.Managed) (managed.Ext } o = append(o, terraform.WithArgs(cr.Spec.ForProvider.ApplyArgs)) - if err := c.tf.Apply(ctx, o...); err != nil { + err = claims.Guard(ctx, c.claimMgr, c.claimCfg, cr, v1beta1.WorkspaceGroupVersionKind, + func(ctx context.Context) error { return c.tf.ForceUnlock(ctx, o...) }, + func(ctx context.Context) error { return c.tf.Apply(ctx, o...) }, + ) + if errors.Is(err, claims.ErrBackoff) { + return managed.ExternalUpdate{}, err + } + if err != nil { return managed.ExternalUpdate{}, errors.Wrap(err, errApply) } @@ -629,7 +647,14 @@ func (c *external) Delete(ctx context.Context, mg resource.Managed) (managed.Ext } o = append(o, terraform.WithArgs(cr.Spec.ForProvider.DestroyArgs)) - return managed.ExternalDelete{}, errors.Wrap(c.tf.Destroy(ctx, o...), errDestroy) + err = claims.Guard(ctx, c.claimMgr, c.claimCfg, cr, v1beta1.WorkspaceGroupVersionKind, + func(ctx context.Context) error { return c.tf.ForceUnlock(ctx, o...) }, + func(ctx context.Context) error { return c.tf.Destroy(ctx, o...) }, + ) + if errors.Is(err, claims.ErrBackoff) { + return managed.ExternalDelete{}, err + } + return managed.ExternalDelete{}, errors.Wrap(err, errDestroy) } func (c *external) Disconnect(ctx context.Context) error { diff --git a/internal/controller/namespaced/workspace/workspace_test.go b/internal/controller/namespaced/workspace/workspace_test.go index ea04ae7..3f5ee1a 100644 --- a/internal/controller/namespaced/workspace/workspace_test.go +++ b/internal/controller/namespaced/workspace/workspace_test.go @@ -81,6 +81,7 @@ type MockTf struct { MockDestroy func(ctx context.Context, o ...terraform.Option) error MockDeleteCurrentWorkspace func(ctx context.Context) error MockGenerateChecksum func(ctx context.Context) (string, error) + MockForceUnlock func(ctx context.Context, o ...terraform.Option) error } func (tf *MockTf) Init(ctx context.Context, o ...terraform.InitOption) error { @@ -119,6 +120,10 @@ func (tf *MockTf) DeleteCurrentWorkspace(ctx context.Context) error { return tf.MockDeleteCurrentWorkspace(ctx) } +func (tf *MockTf) ForceUnlock(ctx context.Context, o ...terraform.Option) error { + return tf.MockForceUnlock(ctx, o...) +} + func TestConnect(t *testing.T) { errBoom := errors.New("boom") errNoProviderConfig := errors.New(errProviderConfigNotSet) diff --git a/internal/handover/handover.go b/internal/handover/handover.go new file mode 100644 index 0000000..ed50bb3 --- /dev/null +++ b/internal/handover/handover.go @@ -0,0 +1,112 @@ +/* +Copyright 2020 The Crossplane Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Package handover keeps a Workspace reconcilable when its assignment label is +// flipped to another instance while a terraform run is still in flight. +// +// Sharding a Workspace onto another instance mid-apply does two things at once. +// It bumps the Workspace's resourceVersion, so the copy the running reconcile +// has been holding since before the relabel is stale and its next write +// conflicts. And it evicts the Workspace from the outgoing instance's cache, +// because --watch-label-selector makes that cache label-filtered and the API +// server delivers "was matching, now isn't" as a DELETED watch event. +// +// Between them those two effects strand the Workspace. The reconciler stamps +// crossplane.io/external-create-pending before calling Create, and records +// external-create-succeeded once Create returns. If that second write cannot +// land, creation looks incomplete forever, and every instance -- including the +// one that just inherited the Workspace -- refuses to reconcile it at all: +// crossplane-runtime bails on ExternalCreateIncomplete before it ever calls +// Observe or Create, and does not requeue. +// +// ReconcilerOptions repairs the write. It deliberately leaves the refusal alone. +// +// # Why the create-pending gate is left in place +// +// managed.WithDeterministicExternalName(true) would make the reconciler proceed +// through an undeterminable create result rather than refuse. It looks +// applicable here, because a Workspace's external name really is deterministic: +// it is the Workspace's own name, assigned by the default NameAsExternalName +// initializer and only ever read by Create. +// +// That is the wrong reading of the option. What it asserts is that a create +// whose result was never recorded is safe to repeat, because Observe can still +// find whatever that create made. For a Workspace, Observe answers +// ResourceExists from `terraform state list` and `terraform output` -- so it can +// only find what the Terraform state records. A run killed between provisioning +// a resource and that resource reaching persisted state leaves nothing for +// Observe to find, and the next apply provisions it a second time. Terraform +// workspace names are deterministic; the cloud resources a module creates are +// not. +// +// Whether that risk is real depends on the backend each module declares, which +// this provider cannot know. So the call belongs to whoever owns the Workspace, +// made against the actual state: +// +// kubectl annotate workspace crossplane.io/external-create-pending- +// +// That only arises when the owning instance dies mid-apply and records no +// outcome at all -- a crash, an OOM kill, an eviction, a --timeout kill. It is +// upstream behaviour rather than anything sharding introduced, and an instance +// that merely loses a Workspace to another shard still records its result. +package handover + +import ( + "context" + + "github.com/crossplane/crossplane-runtime/v2/pkg/reconciler/managed" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// ReconcilerOptions returns the managed reconciler options that let a Workspace +// survive a mid-apply relabel. c is the manager's client and reader reads +// straight from the API server -- manager.GetAPIReader(). +func ReconcilerOptions(c client.Client, reader client.Reader) []managed.ReconcilerOption { + return []managed.ReconcilerOption{ + // Persist critical annotations through a client whose reads bypass the + // cache. RetryingCriticalAnnotationUpdater recovers from a stale + // resourceVersion by re-Getting the object, re-applying the + // annotations, and retrying the write -- which is exactly the recovery + // a relabel needs, and exactly the read a label-filtered cache can no + // longer serve. Reading from the API server instead lets the outgoing + // instance record its create result on a Workspace it no longer owns, + // without clobbering the relabel that handed it over. + managed.WithCriticalAnnotationUpdater( + managed.NewRetryingCriticalAnnotationUpdater(Uncached(c, reader)), + ), + } +} + +// Uncached returns c with its reads redirected to reader. Writes still go +// through c: a client.Reader cannot write, and writes never went through the +// cache to begin with. +func Uncached(c client.Client, reader client.Reader) client.Client { + return &uncachedReads{Client: c, reader: reader} +} + +// uncachedReads is a client.Client whose Get and List bypass the cache. +type uncachedReads struct { + client.Client + reader client.Reader +} + +func (c *uncachedReads) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + return c.reader.Get(ctx, key, obj, opts...) +} + +func (c *uncachedReads) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error { + return c.reader.List(ctx, list, opts...) +} diff --git a/internal/handover/handover_test.go b/internal/handover/handover_test.go new file mode 100644 index 0000000..544a971 --- /dev/null +++ b/internal/handover/handover_test.go @@ -0,0 +1,330 @@ +/* +Copyright 2020 The Crossplane Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package handover + +import ( + "context" + "reflect" + "strconv" + "testing" + "time" + "unsafe" + + "github.com/crossplane/crossplane-runtime/v2/pkg/meta" + "github.com/crossplane/crossplane-runtime/v2/pkg/reconciler/managed" + "github.com/crossplane/crossplane-runtime/v2/pkg/test" + "github.com/pkg/errors" + corev1 "k8s.io/api/core/v1" + kerrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/upbound/provider-terraform/apis/cluster/v1beta1" +) + +// shardLabel is the assignment label documented for --watch-label-selector. +const shardLabel = "sharding.example.com/shard" + +var ( + errCacheServedRead = errors.New("read served by the cache") + + workspaceResource = schema.GroupResource{Group: v1beta1.Group, Resource: "workspaces"} +) + +// evictedCache serves reads the way a label-filtered informer does once an +// object stops matching this instance's selector: it no longer has it. +func evictedCache() *test.MockClient { + return &test.MockClient{ + MockGet: func(_ context.Context, _ client.ObjectKey, _ client.Object) error { return errCacheServedRead }, + MockList: func(_ context.Context, _ client.ObjectList, _ ...client.ListOption) error { return errCacheServedRead }, + } +} + +func TestUncachedReadsBypassTheCache(t *testing.T) { + var gotGet, gotList bool + reader := &test.MockClient{ + MockGet: func(_ context.Context, _ client.ObjectKey, _ client.Object) error { gotGet = true; return nil }, + MockList: func(_ context.Context, _ client.ObjectList, _ ...client.ListOption) error { gotList = true; return nil }, + } + + c := Uncached(evictedCache(), reader) + + if err := c.Get(context.Background(), client.ObjectKey{Name: "ws"}, &corev1.ConfigMap{}); err != nil { + t.Errorf("Get(...): unexpected error: %v", err) + } + if !gotGet { + t.Error("Get(...) was served by the wrapped client, want the reader") + } + + if err := c.List(context.Background(), &corev1.ConfigMapList{}); err != nil { + t.Errorf("List(...): unexpected error: %v", err) + } + if !gotList { + t.Error("List(...) was served by the wrapped client, want the reader") + } +} + +func TestUncachedWritesUseTheWrappedClient(t *testing.T) { + var gotCreate, gotUpdate, gotDelete bool + wrapped := evictedCache() + wrapped.MockCreate = func(_ context.Context, _ client.Object, _ ...client.CreateOption) error { gotCreate = true; return nil } + wrapped.MockUpdate = func(_ context.Context, _ client.Object, _ ...client.UpdateOption) error { gotUpdate = true; return nil } + wrapped.MockDelete = func(_ context.Context, _ client.Object, _ ...client.DeleteOption) error { gotDelete = true; return nil } + + reader := &test.MockClient{ + MockGet: func(_ context.Context, _ client.ObjectKey, _ client.Object) error { + t.Error("a write was routed to the reader") + return nil + }, + } + + c := Uncached(wrapped, reader) + ctx := context.Background() + + if err := c.Create(ctx, &corev1.ConfigMap{}); err != nil { + t.Errorf("Create(...): unexpected error: %v", err) + } + if err := c.Update(ctx, &corev1.ConfigMap{}); err != nil { + t.Errorf("Update(...): unexpected error: %v", err) + } + if err := c.Delete(ctx, &corev1.ConfigMap{}); err != nil { + t.Errorf("Delete(...): unexpected error: %v", err) + } + + if !gotCreate || !gotUpdate || !gotDelete { + t.Errorf("writes did not reach the wrapped client: create=%v update=%v delete=%v", gotCreate, gotUpdate, gotDelete) + } +} + +// TestReconcilerOptionsLeavesCreatePendingGateIntact pins a deliberate omission. +// +// managed.WithDeterministicExternalName(true) would wave the reconciler through +// a create whose result was never recorded, which makes the create-pending wedge +// disappear and is therefore a tempting thing to reach for. It is not safe here: +// the option asserts that Observe can find whatever an unrecorded create made, +// and a Workspace's Observe can only find what the Terraform state records. If +// an apply died before its resources reached persisted state, re-applying +// provisions them again. See the package doc. +// +// Whoever owns the Workspace decides that, by removing the annotation. +func TestReconcilerOptionsLeavesCreatePendingGateIntact(t *testing.T) { + r := applyOptions(&test.MockClient{}, &test.MockClient{}) + + f := reflect.ValueOf(r).Elem().FieldByName("deterministicExternalName") + if !f.IsValid() || f.Kind() != reflect.Bool { + t.Fatal("managed.Reconciler no longer has a bool deterministicExternalName field: check whether " + + "WithDeterministicExternalName still exists after the crossplane-runtime bump, and update this test") + } + if f.Bool() { + t.Error("ReconcilerOptions() declared the external name deterministic, bypassing the create-pending gate; " + + "a terraform apply that died before persisting state will now be silently repeated") + } +} + +// TestCriticalAnnotationsSurviveMidApplyRelabel is the regression test for the +// shard-handover deadlock. +// +// A Workspace is created under shard-1. The reconciler stamps +// external-create-pending, persists it, and starts a long terraform apply. +// Mid-apply the Workspace is relabelled onto shard-2, which does two things at +// once: it bumps the resourceVersion, so shard-1's in-flight copy goes stale +// and its next write conflicts; and it evicts the Workspace from shard-1's +// label-filtered cache, because the API server delivers "was matching, now +// isn't" as a DELETED watch event. +// +// When the apply finishes shard-1 has to record external-create-succeeded. +// Recovering from the conflict through the cache cannot work -- the object is +// no longer there -- so the annotation is lost and creation looks permanently +// incomplete, which makes shard-2 refuse to reconcile the Workspace ever again: +// crossplane-runtime bails on ExternalCreateIncomplete before it ever calls +// Observe or Create, and does not requeue. Recovering through the API server +// lets the write land, and the handover completes. +// +// The updater is type-agnostic, so one concrete type is enough. It uses the +// real cluster-scoped Workspace so the assertions are the same predicate that +// gates the reconciler in production. +func TestCriticalAnnotationsSurviveMidApplyRelabel(t *testing.T) { + pending := time.Date(2026, 7, 30, 17, 24, 0, 0, time.UTC) + succeeded := pending.Add(10 * time.Minute) + + // relabelled is the Workspace as it exists on the API server once the + // shard label has been flipped: shard-2, resourceVersion 2, create-pending. + relabelled := func() *fakeAPI { + ws := &v1beta1.Workspace{ObjectMeta: metav1.ObjectMeta{ + Name: "ws-s1-long-01-test", + ResourceVersion: "2", + Labels: map[string]string{shardLabel: "2"}, + }} + meta.SetExternalCreatePending(ws, pending) + return &fakeAPI{ws: ws} + } + + // inflight is the copy shard-1's reconcile has held since before the + // relabel -- resourceVersion 1, shard-1 -- with the succeeded annotation + // just stamped on it by the reconciler. + inflight := func() *v1beta1.Workspace { + ws := &v1beta1.Workspace{ObjectMeta: metav1.ObjectMeta{ + Name: "ws-s1-long-01-test", + ResourceVersion: "1", + Labels: map[string]string{shardLabel: "1"}, + }} + meta.SetExternalCreatePending(ws, pending) + meta.SetExternalCreateSucceeded(ws, succeeded) + return ws + } + + // shardOneCache is the outgoing instance's label-filtered cache: it can + // still write, but it can no longer read the relabelled Workspace. + shardOneCache := func(a *fakeAPI) *test.MockClient { + return &test.MockClient{ + MockGet: func(_ context.Context, key client.ObjectKey, _ client.Object) error { + return kerrors.NewNotFound(workspaceResource, key.Name) + }, + MockUpdate: func(_ context.Context, obj client.Object, _ ...client.UpdateOption) error { + return a.update(obj) + }, + } + } + + cases := map[string]struct { + reason string + updater func(t *testing.T, a *fakeAPI) managed.CriticalAnnotationUpdater + wantErr bool + wantIncomplete bool + }{ + "CachedReadsWedgeTheWorkspace": { + reason: "Recovering from the relabel's conflict via the label-filtered cache cannot work, and leaves creation permanently incomplete.", + updater: func(_ *testing.T, a *fakeAPI) managed.CriticalAnnotationUpdater { + return managed.NewRetryingCriticalAnnotationUpdater(shardOneCache(a)) + }, + wantErr: true, + wantIncomplete: true, + }, + "ReconcilerOptionsCompleteTheHandover": { + reason: "The updater ReconcilerOptions installs recovers via the API server, so the outgoing instance records its create result and releases the Workspace to its new shard.", + updater: func(t *testing.T, a *fakeAPI) managed.CriticalAnnotationUpdater { + reader := &test.MockClient{ + MockGet: func(_ context.Context, _ client.ObjectKey, obj client.Object) error { return a.get(obj) }, + } + return criticalAnnotationUpdater(t, applyOptions(shardOneCache(a), reader)) + }, + wantErr: false, + wantIncomplete: false, + }, + } + + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + a := relabelled() + + err := tc.updater(t, a).UpdateCriticalAnnotations(context.Background(), inflight()) + + if tc.wantErr && err == nil { + t.Fatalf("UpdateCriticalAnnotations(...): want error, got none\n%s", tc.reason) + } + if !tc.wantErr && err != nil { + t.Fatalf("UpdateCriticalAnnotations(...): unexpected error: %v\n%s", err, tc.reason) + } + + if got := meta.ExternalCreateIncomplete(a.ws); got != tc.wantIncomplete { + t.Errorf("ExternalCreateIncomplete(...): want %v, got %v\n%s", tc.wantIncomplete, got, tc.reason) + } + + if tc.wantIncomplete { + return + } + + // The outgoing instance writes the create result without stomping + // the relabel that handed the Workspace over: it re-read the + // object, then re-applied only its annotations. + if got := a.ws.GetLabels()[shardLabel]; got != "2" { + t.Errorf("shard label: want %q, got %q -- the outgoing instance reverted the handover", "2", got) + } + }) + } +} + +// applyOptions builds the Reconciler that ReconcilerOptions would configure. +// managed.Reconciler's fields are unexported, so a zero value plus the options +// is the only way to observe them without a live manager. +func applyOptions(c client.Client, reader client.Reader) *managed.Reconciler { + r := &managed.Reconciler{} + for _, opt := range ReconcilerOptions(c, reader) { + opt(r) + } + return r +} + +func criticalAnnotationUpdater(t *testing.T, r *managed.Reconciler) managed.CriticalAnnotationUpdater { + t.Helper() + + f := reflect.ValueOf(r).Elem().FieldByName("managed") + if f.IsValid() { + f = f.FieldByName("CriticalAnnotationUpdater") + } + if !f.IsValid() { + t.Fatal("managed.Reconciler no longer holds a CriticalAnnotationUpdater where this test expects it: " + + "check whether WithCriticalAnnotationUpdater still exists after the crossplane-runtime bump, and update this test") + } + + // Reading an unexported field's interface value needs NewAt; the vet-safe + // form is a direct unsafe.Pointer(UnsafeAddr()) conversion. + u, ok := reflect.NewAt(f.Type(), unsafe.Pointer(f.UnsafeAddr())).Elem().Interface().(managed.CriticalAnnotationUpdater) + if !ok || u == nil { + t.Fatal("ReconcilerOptions() did not set a CriticalAnnotationUpdater") + } + return u +} + +// fakeAPI is a minimal stand-in for the API server's optimistic concurrency: an +// Update carrying a stale resourceVersion is rejected with a conflict, and a +// successful one bumps the stored resourceVersion and is written back into the +// caller's object, as a real client does. +type fakeAPI struct { + ws *v1beta1.Workspace +} + +func (a *fakeAPI) get(obj client.Object) error { + ws, ok := obj.(*v1beta1.Workspace) + if !ok { + return errors.Errorf("unexpected object type %T", obj) + } + a.ws.DeepCopyInto(ws) + return nil +} + +func (a *fakeAPI) update(obj client.Object) error { + ws, ok := obj.(*v1beta1.Workspace) + if !ok { + return errors.Errorf("unexpected object type %T", obj) + } + if ws.GetResourceVersion() != a.ws.GetResourceVersion() { + return kerrors.NewConflict(workspaceResource, ws.GetName(), + errors.New("the object has been modified; please apply your changes to the latest version and try again")) + } + + rv, err := strconv.Atoi(a.ws.GetResourceVersion()) + if err != nil { + return err + } + stored := ws.DeepCopy() + stored.SetResourceVersion(strconv.Itoa(rv + 1)) + a.ws = stored + stored.DeepCopyInto(ws) + return nil +} diff --git a/internal/terraform/terraform.go b/internal/terraform/terraform.go index 5df759b..e490837 100644 --- a/internal/terraform/terraform.go +++ b/internal/terraform/terraform.go @@ -676,6 +676,84 @@ func (h Harness) Destroy(ctx context.Context, o ...Option) error { return Classify(err) } +// forceUnlockProbeTimeout bounds the read-only lock probe ForceUnlock runs +// before touching anything. It only needs to be long enough for the backend +// to respond that the lock is (or isn't) held -- not for a real plan. +const forceUnlockProbeTimeout = "1s" + +// lockInfoID matches the "ID:" line inside Terraform's "Lock Info" block, +// emitted on stderr when a state-lock acquisition fails. +var lockInfoID = regexp.MustCompile(`(?m)^\s*ID:\s+(\S+)\s*$`) + +// parseLockID extracts a Terraform state lock ID from a failed operation's +// stderr, if the failure was a lock-acquisition conflict. +func parseLockID(stderr []byte) (string, bool) { + m := lockInfoID.FindSubmatch(stderr) + if m == nil { + return "", false + } + return string(m[1]), true +} + +// ForceUnlock clears a Terraform state lock left behind by a dead owner. +// Unlike Apply and Destroy it does not assume a lock is actually held: it +// first probes with a short, real (locking) plan -- unlike Diff, which +// intentionally passes -lock=false -- and only force-unlocks if that probe +// fails with a lock-acquisition conflict. A probe failure for any other +// reason (e.g. a genuine configuration error) is returned unmodified rather +// than treated as a lock. +// +// Callers must independently confirm the presumed-dead owner's era (design +// doc §8 R4) before calling this: unconditionally force-unlocking a lock +// that a live process is legitimately holding can corrupt state. +func (h Harness) ForceUnlock(ctx context.Context, o ...Option) error { + po := &options{} + for _, fn := range o { + fn(po) + } + + for _, vf := range po.varFiles { + if err := os.WriteFile(filepath.Join(h.Dir, vf.filename), vf.data, 0600); err != nil { + return errors.Wrap(err, errWriteVarFile) + } + } + + args := append([]string{"plan", "-no-color", "-input=false", "-detailed-exitcode", "-lock=true", "-lock-timeout=" + forceUnlockProbeTimeout}, po.args...) + cmd := exec.CommandContext(ctx, h.Path, args...) //nolint:gosec + cmd.Dir = h.Dir + if len(h.Envs) > 0 { + cmd.Env = append(os.Environ(), h.Envs...) + } + + _, err := runCommand(ctx, cmd) + // The -detailed-exitcode flag makes plan return 0 (no changes) or 2 + // (changes present) on success, 1 on error -- see Diff, above. + switch cmd.ProcessState.ExitCode() { + case 0, 2: + return nil // lock is free + } + + ee := &exec.ExitError{} + if !errors.As(err, &ee) { + return Classify(err) + } + lockID, ok := parseLockID(ee.Stderr) + if !ok { + // Plan failed for a reason other than a held lock -- surface the + // real failure rather than attempting an unlock. + return Classify(err) + } + + uArgs := []string{"force-unlock", "-force", "-no-color", lockID} + uCmd := exec.CommandContext(ctx, h.Path, uArgs...) //nolint:gosec + uCmd.Dir = h.Dir + if len(h.Envs) > 0 { + uCmd.Env = append(os.Environ(), h.Envs...) + } + _, err = runCommand(ctx, uCmd) + return Classify(err) +} + // cmdResult represents the result of the command execution type cmdResult struct { out []byte diff --git a/internal/terraform/terraform_test.go b/internal/terraform/terraform_test.go index 3cb93bf..030625c 100644 --- a/internal/terraform/terraform_test.go +++ b/internal/terraform/terraform_test.go @@ -210,6 +210,59 @@ func TestClassify(t *testing.T) { } } +func TestParseLockID(t *testing.T) { + cases := map[string]struct { + stderr string + want string + wantOk bool + }{ + "LockHeld": { + stderr: heredoc.Doc(` + Error: Error acquiring the state lock + + Error message: leases.coordination.k8s.io "some-workspace-lock" already exists + Lock Info: + ID: d3a1f8c2-1234-5678-9abc-def012345678 + Path: some-workspace + Operation: OperationTypeApply + Who: instance-2@provider-terraform + Version: 1.5.7 + Created: 2026-07-03 12:00:00.000000 +0000 UTC + Info: + `), + want: "d3a1f8c2-1234-5678-9abc-def012345678", + wantOk: true, + }, + "NoLockInfo": { + stderr: heredoc.Doc(` + Error: Unsupported argument + + on test.tf line 10, in resource "aws_s3_bucket" "example": + 10: name = "cp-example" + + An argument named "name" is not expected here. + `), + wantOk: false, + }, + "Empty": { + stderr: "", + wantOk: false, + }, + } + + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + got, ok := parseLockID([]byte(tc.stderr)) + if ok != tc.wantOk { + t.Fatalf("parseLockID(...): ok = %v, want %v", ok, tc.wantOk) + } + if got != tc.want { + t.Errorf("parseLockID(...) = %q, want %q", got, tc.want) + } + }) + } +} + func TestFormatTerraformErrorOutput(t *testing.T) { tferrs := make(map[string]string) expectedOutput := make(map[string]map[string]string)