From f0879b596d71294efea2d14d6bc6123145b464a0 Mon Sep 17 00:00:00 2001 From: Sylvain Rabot Date: Tue, 21 Jul 2026 10:03:37 +0200 Subject: [PATCH] feat(stacks): karpenter node isolation per organization/stack Provision dedicated Karpenter node pools per organization (shared "" pool) or per stack ("-" pool), driven by karpenter.* settings and the formance.com/organization + /customer Stack labels. EC2NodeClass/NodePool are cloned from named cluster-scoped reference objects (unstructured + injected tags/labels/taints) so EC2 instances carry customer/organization/stack tags for cost attribution. Stack workloads are pinned onto the dedicated nodes via a nodeSelector + tolerations mutator in the applications deployment path. Integrated into the existing stacks reconciler (no new CRD, no second controller). Cleanup is label-driven (app.kubernetes.io/managed-by= formance-operator) because the module cleanup path namespace-filters and cannot see cluster-scoped objects; org pools use additive non-controller owner refs so they survive until the last owning stack is gone. Constraint: cluster-scoped dependent may only reference a cluster-scoped owner (the Stack CR is cluster-scoped, so owner refs are valid) Constraint: Karpenter CRDs are optional; feature no-ops when absent and a CustomResourceDefinition watch re-enables it without an operator restart Rejected: WithOwn-based cleanup | removeAllModulesOwnedObjects namespace-filters and never sees cluster-scoped NodePool/EC2NodeClass Rejected: typed Karpenter Go deps | avoids v1beta1->v1 churn; feature only materializes a configured spec, so unstructured suffices Directive: taints/tolerations/nodeSelector key on pool identity (organization[/stack]) only, never customer -- a shared org pool may see differing customer values which would break scheduling; customer stays a tag/node-label for cost attribution only Confidence: medium Scope-risk: moderate Not-tested: envtest integration against real Karpenter CRDs; migration Jobs are not pinned (taints only repel, so they run on shared nodes) --- api/formance.com/v1beta1/settings_types.go | 37 +++ api/formance.com/v1beta1/shared.go | 5 + config/crd/bases/formance.com_settings.yaml | 39 ++- config/rbac/role.yaml | 25 ++ ..._v1_clusterrole_formance-manager-role.yaml | 25 ++ .../resources/applications/application.go | 44 ++++ .../applications/application_test.go | 55 ++++ internal/resources/nodeisolation/available.go | 52 ++++ internal/resources/nodeisolation/karpenter.go | 241 ++++++++++++++++++ .../resources/nodeisolation/karpenter_test.go | 226 ++++++++++++++++ internal/resources/nodeisolation/mock_test.go | 24 ++ internal/resources/nodeisolation/resolve.go | 200 +++++++++++++++ .../resources/nodeisolation/resolve_test.go | 150 +++++++++++ internal/resources/stacks/init.go | 44 +++- 14 files changed, 1155 insertions(+), 12 deletions(-) create mode 100644 internal/resources/nodeisolation/available.go create mode 100644 internal/resources/nodeisolation/karpenter.go create mode 100644 internal/resources/nodeisolation/karpenter_test.go create mode 100644 internal/resources/nodeisolation/mock_test.go create mode 100644 internal/resources/nodeisolation/resolve.go create mode 100644 internal/resources/nodeisolation/resolve_test.go diff --git a/api/formance.com/v1beta1/settings_types.go b/api/formance.com/v1beta1/settings_types.go index 29999dafa..79ff86b5f 100644 --- a/api/formance.com/v1beta1/settings_types.go +++ b/api/formance.com/v1beta1/settings_types.go @@ -96,6 +96,43 @@ type SettingsSpec struct { // Refer to the documentation of each module and resource to discover available Settings. // // ##### Global settings +// ###### Karpenter node isolation +// +// The operator can provision dedicated Karpenter node pools per organization (or per +// organization+stack) so that EC2 instances carry customer/organization/stack tags for +// cost attribution, and so a stack's workloads run only on its dedicated nodes. The +// organization and customer identity are read from the Stack labels +// `formance.com/organization` and `formance.com/customer` (customer defaults to the +// organization when absent). +// +// Available keys: +// - `karpenter.enabled` (bool): master switch. Requires the Karpenter CRDs to be installed. +// - `karpenter.isolation` (`organization` | `stack`, default `organization`): chooses the +// pool naming — a shared `` pool, or a dedicated `-` pool. +// - `karpenter.ec2-node-class.reference` (string, required when enabled): name of the +// cluster-scoped reference EC2NodeClass to clone. +// - `karpenter.node-pool.reference` (string, required when enabled): name of the +// cluster-scoped reference NodePool to clone. +// - `karpenter.api.ec2-node-class.group-version` (string, default `karpenter.k8s.aws/v1`) +// and `karpenter.api.node-pool.group-version` (string, default `karpenter.sh/v1`): +// override the Karpenter API group/version. +// +// ```yaml +// apiVersion: formance.com/v1beta1 +// kind: Settings +// metadata: +// +// name: karpenter-enabled +// +// spec: +// +// key: karpenter.enabled +// stacks: +// - '*' +// value: "true" +// +// ``` +// // ###### AWS account // // A stack can use an AWS account for authentication. diff --git a/api/formance.com/v1beta1/shared.go b/api/formance.com/v1beta1/shared.go index 5894e65c1..6a3cbaddb 100644 --- a/api/formance.com/v1beta1/shared.go +++ b/api/formance.com/v1beta1/shared.go @@ -392,4 +392,9 @@ const ( StackLabel = "formance.com/stack" SkipLabel = "formance.com/skip" CreatedByAgentLabel = "formance.com/created-by-agent" + // OrganizationLabel and CustomerLabel are set externally (by the control plane) + // on the Stack CR metadata. They are not guaranteed to be present and are used to + // tag/label/taint dedicated Karpenter node pools for cost attribution. + OrganizationLabel = "formance.com/organization" + CustomerLabel = "formance.com/customer" ) diff --git a/config/crd/bases/formance.com_settings.yaml b/config/crd/bases/formance.com_settings.yaml index 063f14673..11fbc7a10 100644 --- a/config/crd/bases/formance.com_settings.yaml +++ b/config/crd/bases/formance.com_settings.yaml @@ -48,17 +48,34 @@ spec: all the stacks\n\tvalue: postgresql://postgresql.formance.svc.cluster.local:5432\n\n```\n\nSome settings are really global, while some are used by specific module.\n\nRefer to the documentation of each module and resource to discover available Settings.\n\n##### - Global settings\n###### AWS account\n\nA stack can use an AWS account for - authentication.\n\nIt can be used to connect to any AWS service we could - use.\n\nIt includes RDS, OpenSearch and MSK. To do so, you can create the - following setting:\n```yaml\napiVersion: formance.com/v1beta1\nkind: Settings\nmetadata:\n\n\tname: - aws-service-account\n\nspec:\n\n\tkey: aws.service-account\n\tstacks:\n\t- - '*'\n\tvalue: aws-access\n\n```\nThis setting instruct the operator than - there is somewhere on the cluster a service account named `aws-access`.\n\nSo, - each time a service has the capability to use AWS, the operator will use - this service account.\n\nThe service account could look like that :\n```yaml\napiVersion: - v1\nkind: ServiceAccount\nmetadata:\n\n\tannotations:\n\t eks.amazonaws.com/role-arn: - arn:aws:iam::************:role/staging-eu-west-1-hosting-stack-access\n\tlabels:\n\t + Global settings\n###### Karpenter node isolation\n\nThe operator can provision + dedicated Karpenter node pools per organization (or per\norganization+stack) + so that EC2 instances carry customer/organization/stack tags for\ncost attribution, + and so a stack's workloads run only on its dedicated nodes. The\norganization + and customer identity are read from the Stack labels\n`formance.com/organization` + and `formance.com/customer` (customer defaults to the\norganization when + absent).\n\nAvailable keys:\n - `karpenter.enabled` (bool): master switch. + Requires the Karpenter CRDs to be installed.\n - `karpenter.isolation` + (`organization` | `stack`, default `organization`): chooses the\n pool + naming — a shared `` pool, or a dedicated `-` + pool.\n - `karpenter.ec2-node-class.reference` (string, required when enabled): + name of the\n cluster-scoped reference EC2NodeClass to clone.\n - `karpenter.node-pool.reference` + (string, required when enabled): name of the\n cluster-scoped reference + NodePool to clone.\n - `karpenter.api.ec2-node-class.group-version` (string, + default `karpenter.k8s.aws/v1`)\n and `karpenter.api.node-pool.group-version` + (string, default `karpenter.sh/v1`):\n override the Karpenter API group/version.\n\n```yaml\napiVersion: + formance.com/v1beta1\nkind: Settings\nmetadata:\n\n\tname: karpenter-enabled\n\nspec:\n\n\tkey: + karpenter.enabled\n\tstacks:\n\t- '*'\n\tvalue: \"true\"\n\n```\n\n###### + AWS account\n\nA stack can use an AWS account for authentication.\n\nIt + can be used to connect to any AWS service we could use.\n\nIt includes RDS, + OpenSearch and MSK. To do so, you can create the following setting:\n```yaml\napiVersion: + formance.com/v1beta1\nkind: Settings\nmetadata:\n\n\tname: aws-service-account\n\nspec:\n\n\tkey: + aws.service-account\n\tstacks:\n\t- '*'\n\tvalue: aws-access\n\n```\nThis + setting instruct the operator than there is somewhere on the cluster a service + account named `aws-access`.\n\nSo, each time a service has the capability + to use AWS, the operator will use this service account.\n\nThe service account + could look like that :\n```yaml\napiVersion: v1\nkind: ServiceAccount\nmetadata:\n\n\tannotations:\n\t + \ eks.amazonaws.com/role-arn: arn:aws:iam::************:role/staging-eu-west-1-hosting-stack-access\n\tlabels:\n\t \ formance.com/stack: any\n\tname: aws-access\n\n```\nYou can note two things :\n 1. We have an annotation indicating the role arn used to connect to AWS. Refer to the AWS documentation to create this role\n 2. We have a label diff --git a/config/rbac/role.yaml b/config/rbac/role.yaml index bf8876117..69a412ced 100644 --- a/config/rbac/role.yaml +++ b/config/rbac/role.yaml @@ -29,6 +29,7 @@ rules: verbs: - get - list + - watch - apiGroups: - apps resources: @@ -189,6 +190,30 @@ rules: - patch - update - watch +- apiGroups: + - karpenter.k8s.aws + resources: + - ec2nodeclasses + verbs: + - create + - delete + - get + - list + - patch + - update + - watch +- apiGroups: + - karpenter.sh + resources: + - nodepools + verbs: + - create + - delete + - get + - list + - patch + - update + - watch - apiGroups: - networking.k8s.io resources: diff --git a/helm/operator/templates/gen/rbac.authorization.k8s.io_v1_clusterrole_formance-manager-role.yaml b/helm/operator/templates/gen/rbac.authorization.k8s.io_v1_clusterrole_formance-manager-role.yaml index 16f6c6e34..232ee2e1d 100644 --- a/helm/operator/templates/gen/rbac.authorization.k8s.io_v1_clusterrole_formance-manager-role.yaml +++ b/helm/operator/templates/gen/rbac.authorization.k8s.io_v1_clusterrole_formance-manager-role.yaml @@ -28,6 +28,7 @@ rules: verbs: - get - list + - watch - apiGroups: - apps resources: @@ -188,6 +189,30 @@ rules: - patch - update - watch +- apiGroups: + - karpenter.k8s.aws + resources: + - ec2nodeclasses + verbs: + - create + - delete + - get + - list + - patch + - update + - watch +- apiGroups: + - karpenter.sh + resources: + - nodepools + verbs: + - create + - delete + - get + - list + - patch + - update + - watch - apiGroups: - networking.k8s.io resources: diff --git a/internal/resources/applications/application.go b/internal/resources/applications/application.go index e6ce3a2fe..8419deb3c 100644 --- a/internal/resources/applications/application.go +++ b/internal/resources/applications/application.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "maps" + "slices" "strconv" "github.com/stoewer/go-strcase" @@ -20,6 +21,7 @@ import ( "github.com/formancehq/operator/v3/api/formance.com/v1beta1" "github.com/formancehq/operator/v3/internal/core" "github.com/formancehq/operator/v3/internal/resources/licence" + "github.com/formancehq/operator/v3/internal/resources/nodeisolation" "github.com/formancehq/operator/v3/internal/resources/settings" ) @@ -465,6 +467,47 @@ func (a Application) withNodeIP(_ core.Context) core.ObjectMutator[*appsv1.Deplo } } +// withNodeIsolation pins the deployment's pods onto the stack's dedicated Karpenter nodes +// by injecting a nodeSelector (enforces placement) and tolerations (permit the node +// taints). The operator-managed scheduling keys are stripped before re-applying so the +// result is idempotent (tolerations don't accumulate) and disable / mode-change / relabel +// removes stale rules. It also no-ops when the Karpenter CRDs are unavailable, to avoid +// pinning pods onto nodes that will never be provisioned. +func (a Application) withNodeIsolation(ctx core.Context) core.ObjectMutator[*appsv1.Deployment] { + return func(deployment *appsv1.Deployment) error { + stack := &v1beta1.Stack{} + if err := ctx.GetClient().Get(ctx, types.NamespacedName{Name: a.owner.GetStack()}, stack); err != nil { + return err + } + + cfg, err := nodeisolation.Resolve(ctx, stack) + if err != nil { + return err + } + + // Reconcile (not accumulate): drop previously-injected operator-managed keys first. + spec := &deployment.Spec.Template.Spec + for _, key := range nodeisolation.ManagedSchedulingKeys { + delete(spec.NodeSelector, key) + } + spec.Tolerations = slices.DeleteFunc(spec.Tolerations, func(t corev1.Toleration) bool { + return slices.Contains(nodeisolation.ManagedSchedulingKeys, t.Key) + }) + + if !cfg.Enabled || !nodeisolation.IsAvailable() { + return nil + } + + if spec.NodeSelector == nil { + spec.NodeSelector = map[string]string{} + } + maps.Copy(spec.NodeSelector, cfg.NodeSelector) + spec.Tolerations = append(spec.Tolerations, cfg.Tolerations...) + + return nil + } +} + func (a Application) withTerminationGracePeriod(ctx core.Context) core.ObjectMutator[*appsv1.Deployment] { return func(deployment *appsv1.Deployment) error { terminationGracePeriod, err := settings.GetInt64(ctx, a.owner.GetStack(), "deployments", deployment.Name, "spec", "template", "spec", "termination-grace-period-seconds") @@ -504,6 +547,7 @@ func (a Application) handleDeployment(ctx core.Context, deploymentLabels map[str a.withSemconvMetricsNames(ctx), a.withNodeIP(ctx), a.withTerminationGracePeriod(ctx), + a.withNodeIsolation(ctx), core.WithController[*appsv1.Deployment](ctx.GetScheme(), a.owner), ) diff --git a/internal/resources/applications/application_test.go b/internal/resources/applications/application_test.go index 6fcfd6bc6..e9f50636c 100644 --- a/internal/resources/applications/application_test.go +++ b/internal/resources/applications/application_test.go @@ -16,9 +16,64 @@ import ( "github.com/formancehq/operator/v3/api/formance.com/v1beta1" "github.com/formancehq/operator/v3/internal/core" + "github.com/formancehq/operator/v3/internal/resources/nodeisolation" "github.com/formancehq/operator/v3/internal/resources/settings" ) +func TestWithNodeIsolationIsIdempotent(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, v1beta1.AddToScheme(scheme)) + require.NoError(t, appsv1.AddToScheme(scheme)) + require.NoError(t, v1.AddToScheme(scheme)) + + nodeisolation.SetAvailable(true) + t.Cleanup(func() { nodeisolation.SetAvailable(false) }) + + stackName := "stack0" + enabled := &v1beta1.Settings{ObjectMeta: metav1.ObjectMeta{Name: "enabled"}, + Spec: v1beta1.SettingsSpec{Stacks: []string{"*"}, Key: "karpenter.enabled", Value: "true"}} + ec2 := &v1beta1.Settings{ObjectMeta: metav1.ObjectMeta{Name: "ec2"}, + Spec: v1beta1.SettingsSpec{Stacks: []string{"*"}, Key: "karpenter.ec2-node-class.reference", Value: "default"}} + np := &v1beta1.Settings{ObjectMeta: metav1.ObjectMeta{Name: "np"}, + Spec: v1beta1.SettingsSpec{Stacks: []string{"*"}, Key: "karpenter.node-pool.reference", Value: "default"}} + stack := &v1beta1.Stack{ObjectMeta: metav1.ObjectMeta{ + Name: stackName, + Labels: map[string]string{v1beta1.OrganizationLabel: "acme"}, + }} + + fakeClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(enabled, ec2, np, stack). + WithIndex(&v1beta1.Settings{}, "stack", func(obj client.Object) []string { + return obj.(*v1beta1.Settings).GetStacks() + }). + WithIndex(&v1beta1.Settings{}, "keylen", func(obj client.Object) []string { + return []string{fmt.Sprint(len(settings.SplitKeywordWithDot(obj.(*v1beta1.Settings).Spec.Key)))} + }). + Build() + + ctx := &mockContext{Context: context.Background(), client: fakeClient, scheme: scheme} + app := Application{owner: &v1beta1.Ledger{ + ObjectMeta: metav1.ObjectMeta{Name: "ledger"}, + Spec: v1beta1.LedgerSpec{StackDependency: v1beta1.StackDependency{Stack: stackName}}, + }} + deployment := &appsv1.Deployment{} + + // Applying the mutator repeatedly must not accumulate tolerations or selectors. + for i := 0; i < 3; i++ { + require.NoError(t, app.withNodeIsolation(ctx)(deployment)) + } + require.Equal(t, "acme", deployment.Spec.Template.Spec.NodeSelector[v1beta1.OrganizationLabel]) + require.Len(t, deployment.Spec.Template.Spec.Tolerations, 1) + + // Disabling must strip the operator-managed selector keys and tolerations. + enabled.Spec.Value = "false" + require.NoError(t, ctx.GetClient().Update(ctx, enabled)) + require.NoError(t, app.withNodeIsolation(ctx)(deployment)) + require.NotContains(t, deployment.Spec.Template.Spec.NodeSelector, v1beta1.OrganizationLabel) + require.Empty(t, deployment.Spec.Template.Spec.Tolerations) +} + func TestWithAnnotations(t *testing.T) { t.Parallel() type testCase struct { diff --git a/internal/resources/nodeisolation/available.go b/internal/resources/nodeisolation/available.go new file mode 100644 index 000000000..15886811e --- /dev/null +++ b/internal/resources/nodeisolation/available.go @@ -0,0 +1,52 @@ +package nodeisolation + +import ( + "sync/atomic" + + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/types" + + "github.com/formancehq/operator/v3/internal/core" +) + +// CRD names Karpenter installs for the two kinds this feature manages. +const ( + ec2NodeClassCRDName = "ec2nodeclasses.karpenter.k8s.aws" + nodePoolCRDName = "nodepools.karpenter.sh" +) + +// available reports whether both Karpenter CRDs are installed. It is read on every +// reconcile and refreshed by DetectCRDs at controller setup and whenever a Karpenter CRD +// appears/disappears (see the CustomResourceDefinition watch in the stacks reconciler), so +// installing Karpenter after the operator starts does not require a restart. +var available atomic.Bool + +// IsAvailable reports whether the Karpenter CRDs are currently installed. +func IsAvailable() bool { return available.Load() } + +// SetAvailable overrides the availability flag. Intended for tests. +func SetAvailable(v bool) { available.Store(v) } + +// IsKarpenterCRD reports whether the named CustomResourceDefinition is one of the +// Karpenter CRDs this feature depends on. +func IsKarpenterCRD(name string) bool { + return name == ec2NodeClassCRDName || name == nodePoolCRDName +} + +// DetectCRDs performs precise discovery (a targeted Get per CRD) and updates the +// availability flag. It sets available=true only if both Karpenter CRDs are present. +func DetectCRDs(ctx core.Context) error { + for _, name := range []string{ec2NodeClassCRDName, nodePoolCRDName} { + crd := &apiextensionsv1.CustomResourceDefinition{} + if err := ctx.GetAPIReader().Get(ctx, types.NamespacedName{Name: name}, crd); err != nil { + if apierrors.IsNotFound(err) { + available.Store(false) + return nil + } + return err + } + } + available.Store(true) + return nil +} diff --git a/internal/resources/nodeisolation/karpenter.go b/internal/resources/nodeisolation/karpenter.go new file mode 100644 index 000000000..649a612c4 --- /dev/null +++ b/internal/resources/nodeisolation/karpenter.go @@ -0,0 +1,241 @@ +package nodeisolation + +import ( + "github.com/pkg/errors" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/util/retry" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + + "github.com/formancehq/operator/v3/api/formance.com/v1beta1" + "github.com/formancehq/operator/v3/internal/core" +) + +// Reconcile materializes (or cleans up) the dedicated Karpenter EC2NodeClass and NodePool +// for a stack, based on its resolved isolation Config. It is a no-op when the Karpenter +// CRDs are not installed. +func Reconcile(ctx core.Context, stack *v1beta1.Stack) error { + if !IsAvailable() { + return nil + } + + cfg, err := Resolve(ctx, stack) + if err != nil { + return err + } + + if !cfg.Enabled { + return cleanup(ctx, stack, cfg) + } + + if err := reconcileEC2NodeClass(ctx, stack, cfg); err != nil { + return err + } + if err := reconcileNodePool(ctx, stack, cfg); err != nil { + return err + } + + return cleanup(ctx, stack, cfg) +} + +func reconcileEC2NodeClass(ctx core.Context, stack *v1beta1.Stack, cfg *Config) error { + refSpec, err := referenceSpec(ctx, cfg.EC2NodeClassGVK, cfg.EC2NodeClassRef) + if err != nil { + return err + } + + // Inject the customer/organization/stack tags onto the cloned spec. + tags, _, _ := unstructured.NestedStringMap(refSpec, "tags") + if tags == nil { + tags = map[string]string{} + } + for k, v := range cfg.Tags { + tags[k] = v + } + if err := unstructured.SetNestedStringMap(refSpec, tags, "tags"); err != nil { + return err + } + + return applyKarpenterObject(ctx, stack, cfg, cfg.EC2NodeClassGVK, refSpec) +} + +func reconcileNodePool(ctx core.Context, stack *v1beta1.Stack, cfg *Config) error { + refSpec, err := referenceSpec(ctx, cfg.NodePoolGVK, cfg.NodePoolRef) + if err != nil { + return err + } + + // Node labels on the provisioned nodes. + nodeLabels, _, _ := unstructured.NestedStringMap(refSpec, "template", "metadata", "labels") + if nodeLabels == nil { + nodeLabels = map[string]string{} + } + for k, v := range cfg.Tags { + nodeLabels[k] = v + } + if err := unstructured.SetNestedStringMap(refSpec, nodeLabels, "template", "metadata", "labels"); err != nil { + return err + } + + // Taints repel non-dedicated workloads. + if err := unstructured.SetNestedSlice(refSpec, taintsToUnstructured(cfg.Taints), "template", "spec", "taints"); err != nil { + return err + } + + // Point the NodePool at the cloned EC2NodeClass. + nodeClassRef := map[string]any{ + "group": cfg.EC2NodeClassGVK.Group, + "kind": cfg.EC2NodeClassGVK.Kind, + "name": cfg.PoolName, + } + if err := unstructured.SetNestedMap(refSpec, nodeClassRef, "template", "spec", "nodeClassRef"); err != nil { + return err + } + + return applyKarpenterObject(ctx, stack, cfg, cfg.NodePoolGVK, refSpec) +} + +// applyKarpenterObject creates or updates the Karpenter object named cfg.PoolName, setting +// its spec to the injected clone, stamping management labels, and adding the stack as a +// non-controller owner (so an organization pool shared by several stacks is only garbage +// collected once the last owning stack is gone). +func applyKarpenterObject(ctx core.Context, stack *v1beta1.Stack, cfg *Config, gvk schema.GroupVersionKind, spec map[string]any) error { + obj := &unstructured.Unstructured{} + obj.SetGroupVersionKind(gvk) + obj.SetName(cfg.PoolName) + + _, err := controllerutil.CreateOrUpdate(ctx, ctx.GetClient(), obj, func() error { + obj.Object["spec"] = runtime.DeepCopyJSON(spec) + + labels := obj.GetLabels() + if labels == nil { + labels = map[string]string{} + } + labels[ManagedByLabel] = ManagedByValue + labels[v1beta1.OrganizationLabel] = cfg.Organization + if cfg.Mode == ModeStack { + labels[v1beta1.StackLabel] = cfg.Stack + } + obj.SetLabels(labels) + + hasOwner, err := core.HasOwnerReference(ctx, stack, obj) + if err != nil { + return err + } + if !hasOwner { + if err := controllerutil.SetOwnerReference(stack, obj, ctx.GetScheme()); err != nil { + return err + } + } + return nil + }) + if err != nil { + return errors.Wrapf(err, "creating/updating %s %s", gvk.Kind, cfg.PoolName) + } + return nil +} + +// cleanup releases (and deletes when orphaned) any Karpenter object this stack participates +// in that is no longer the desired one — e.g. after an isolation-mode change, an +// organization relabel, or disabling the feature. It selects by management label rather +// than owner reference because the module cleanup path cannot see cluster-scoped objects. +func cleanup(ctx core.Context, stack *v1beta1.Stack, cfg *Config) error { + desired := "" + if cfg.Enabled { + desired = cfg.PoolName + } + + for _, gvk := range []schema.GroupVersionKind{cfg.EC2NodeClassGVK, cfg.NodePoolGVK} { + list := &unstructured.UnstructuredList{} + list.SetGroupVersionKind(gvk) + if err := ctx.GetClient().List(ctx, list, client.MatchingLabels{ManagedByLabel: ManagedByValue}); err != nil { + return errors.Wrapf(err, "listing %s for cleanup", gvk.Kind) + } + + for i := range list.Items { + item := &list.Items[i] + if item.GetName() == desired { + continue + } + // Cheap pre-filter on the (possibly stale) list item; releaseOrDelete re-fetches + // and re-checks under optimistic concurrency. + owned, err := core.HasOwnerReference(ctx, stack, item) + if err != nil { + return err + } + if !owned { + continue + } + if err := releaseOrDelete(ctx, stack, gvk, item.GetName()); err != nil { + return err + } + } + } + + return nil +} + +// releaseOrDelete drops this stack's owner reference from the named object and deletes it +// only if no owners remain. It re-fetches the latest object and uses optimistic concurrency +// (Update carries the resourceVersion; Delete is guarded by UID+resourceVersion +// preconditions) so a concurrent reconcile adding another owner cannot be clobbered or lose +// its pool to a stale-state deletion. +func releaseOrDelete(ctx core.Context, stack *v1beta1.Stack, gvk schema.GroupVersionKind, name string) error { + return retry.RetryOnConflict(retry.DefaultRetry, func() error { + obj := &unstructured.Unstructured{} + obj.SetGroupVersionKind(gvk) + if err := ctx.GetClient().Get(ctx, types.NamespacedName{Name: name}, obj); err != nil { + return client.IgnoreNotFound(err) + } + + owned, err := core.HasOwnerReference(ctx, stack, obj) + if err != nil { + return err + } + if !owned { + return nil + } + if err := controllerutil.RemoveOwnerReference(stack, obj, ctx.GetScheme()); err != nil { + return errors.Wrapf(err, "removing owner reference from %s", name) + } + + if len(obj.GetOwnerReferences()) > 0 { + return ctx.GetClient().Update(ctx, obj) + } + + core.LogDeletion(ctx, obj, "nodeisolation.cleanup") + uid := obj.GetUID() + resourceVersion := obj.GetResourceVersion() + err = ctx.GetClient().Delete(ctx, obj, client.Preconditions{UID: &uid, ResourceVersion: &resourceVersion}) + return client.IgnoreNotFound(err) + }) +} + +func referenceSpec(ctx core.Context, gvk schema.GroupVersionKind, name string) (map[string]any, error) { + ref := &unstructured.Unstructured{} + ref.SetGroupVersionKind(gvk) + if err := ctx.GetAPIReader().Get(ctx, types.NamespacedName{Name: name}, ref); err != nil { + return nil, errors.Wrapf(err, "getting reference %s %q", gvk.Kind, name) + } + spec, ok := core.GetSpecFromUnstructured(ref) + if !ok { + return nil, core.NewPendingError().WithMessage("reference %s %q has no spec", gvk.Kind, name) + } + return runtime.DeepCopyJSON(spec), nil +} + +func taintsToUnstructured(taints []corev1.Taint) []any { + out := make([]any, 0, len(taints)) + for _, t := range taints { + out = append(out, map[string]any{ + "key": t.Key, + "value": t.Value, + "effect": string(t.Effect), + }) + } + return out +} diff --git a/internal/resources/nodeisolation/karpenter_test.go b/internal/resources/nodeisolation/karpenter_test.go new file mode 100644 index 000000000..34b1cbdd7 --- /dev/null +++ b/internal/resources/nodeisolation/karpenter_test.go @@ -0,0 +1,226 @@ +package nodeisolation + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/require" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + "github.com/formancehq/operator/v3/api/formance.com/v1beta1" + "github.com/formancehq/operator/v3/internal/resources/settings" +) + +var ( + testEC2GVK = schema.GroupVersionKind{Group: "karpenter.k8s.aws", Version: "v1", Kind: "EC2NodeClass"} + testNodePoolGVK = schema.GroupVersionKind{Group: "karpenter.sh", Version: "v1", Kind: "NodePool"} +) + +func registerKarpenter(scheme *runtime.Scheme, gvk schema.GroupVersionKind) { + scheme.AddKnownTypeWithName(gvk, &unstructured.Unstructured{}) + listGVK := gvk + listGVK.Kind += "List" + scheme.AddKnownTypeWithName(listGVK, &unstructured.UnstructuredList{}) +} + +func reference(gvk schema.GroupVersionKind, name string, spec map[string]any) *unstructured.Unstructured { + obj := &unstructured.Unstructured{} + obj.SetGroupVersionKind(gvk) + obj.SetName(name) + obj.Object["spec"] = spec + return obj +} + +func newKarpenterContext(t *testing.T, objects ...client.Object) *mockContext { + t.Helper() + scheme := runtime.NewScheme() + require.NoError(t, v1beta1.AddToScheme(scheme)) + registerKarpenter(scheme, testEC2GVK) + registerKarpenter(scheme, testNodePoolGVK) + + fakeClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithIndex(&v1beta1.Settings{}, "stack", func(obj client.Object) []string { + return obj.(*v1beta1.Settings).GetStacks() + }). + WithIndex(&v1beta1.Settings{}, "keylen", func(obj client.Object) []string { + return []string{fmt.Sprint(len(settings.SplitKeywordWithDot(obj.(*v1beta1.Settings).Spec.Key)))} + }). + WithObjects(objects...). + Build() + + return &mockContext{Context: context.Background(), client: fakeClient, scheme: scheme} +} + +func karpenterEnabledSettings(mode string, stacks ...string) []client.Object { + objs := []client.Object{ + setting("enabled", "karpenter.enabled", "true", stacks...), + setting("ec2", "karpenter.ec2-node-class.reference", "default", stacks...), + setting("np", "karpenter.node-pool.reference", "default", stacks...), + } + if mode != "" { + objs = append(objs, setting("isolation", "karpenter.isolation", mode, stacks...)) + } + return objs +} + +func stackWithUID(name, uid string, labels map[string]string) *v1beta1.Stack { + s := stackWithLabels(name, labels) + s.SetUID(types.UID(uid)) + return s +} + +func getUnstructured(t *testing.T, ctx *mockContext, gvk schema.GroupVersionKind, name string) (*unstructured.Unstructured, bool) { + t.Helper() + obj := &unstructured.Unstructured{} + obj.SetGroupVersionKind(gvk) + err := ctx.GetClient().Get(ctx, types.NamespacedName{Name: name}, obj) + if apierrors.IsNotFound(err) { + return nil, false + } + require.NoError(t, err) + return obj, true +} + +func TestReconcileNoopWhenUnavailable(t *testing.T) { + SetAvailable(false) + ctx := newKarpenterContext(t, karpenterEnabledSettings("", "*")...) + require.NoError(t, Reconcile(ctx, stackWithUID("stack0", "uid-0", + map[string]string{v1beta1.OrganizationLabel: "acme"}))) + _, found := getUnstructured(t, ctx, testEC2GVK, "acme") + require.False(t, found) +} + +func TestReconcileCreatesOrganizationPool(t *testing.T) { + SetAvailable(true) + t.Cleanup(func() { SetAvailable(false) }) + + objs := append(karpenterEnabledSettings("", "*"), + reference(testEC2GVK, "default", map[string]any{"role": "KarpenterNodeRole"}), + reference(testNodePoolGVK, "default", map[string]any{ + "template": map[string]any{"spec": map[string]any{}}, + }), + ) + ctx := newKarpenterContext(t, objs...) + stack := stackWithUID("stack0", "uid-0", map[string]string{ + v1beta1.OrganizationLabel: "acme", + v1beta1.CustomerLabel: "acme-corp", + }) + + require.NoError(t, Reconcile(ctx, stack)) + + ec2, found := getUnstructured(t, ctx, testEC2GVK, "acme") + require.True(t, found) + tags, _, _ := unstructured.NestedStringMap(ec2.Object, "spec", "tags") + require.Equal(t, "acme-corp", tags[v1beta1.CustomerLabel]) + require.Equal(t, "acme", tags[v1beta1.OrganizationLabel]) + require.NotContains(t, tags, v1beta1.StackLabel) + // Reference field is preserved by the clone. + role, _, _ := unstructured.NestedString(ec2.Object, "spec", "role") + require.Equal(t, "KarpenterNodeRole", role) + require.Equal(t, ManagedByValue, ec2.GetLabels()[ManagedByLabel]) + require.Len(t, ec2.GetOwnerReferences(), 1) + + np, found := getUnstructured(t, ctx, testNodePoolGVK, "acme") + require.True(t, found) + nodeLabels, _, _ := unstructured.NestedStringMap(np.Object, "spec", "template", "metadata", "labels") + require.Equal(t, "acme", nodeLabels[v1beta1.OrganizationLabel]) + nodeClassName, _, _ := unstructured.NestedString(np.Object, "spec", "template", "spec", "nodeClassRef", "name") + require.Equal(t, "acme", nodeClassName) + taints, _, _ := unstructured.NestedSlice(np.Object, "spec", "template", "spec", "taints") + require.Len(t, taints, 1) // organization only in org mode (customer is not a scheduling key) +} + +func TestReconcileGCOnModeChange(t *testing.T) { + SetAvailable(true) + t.Cleanup(func() { SetAvailable(false) }) + + // Start in organization mode → pool "acme". + objs := append(karpenterEnabledSettings(ModeOrganization, "stack0"), + reference(testEC2GVK, "default", map[string]any{"role": "r"}), + reference(testNodePoolGVK, "default", map[string]any{"template": map[string]any{"spec": map[string]any{}}}), + ) + ctx := newKarpenterContext(t, objs...) + stack := stackWithUID("stack0", "uid-0", map[string]string{v1beta1.OrganizationLabel: "acme"}) + require.NoError(t, Reconcile(ctx, stack)) + _, found := getUnstructured(t, ctx, testEC2GVK, "acme") + require.True(t, found) + + // Switch to stack mode → new pool "acme-stack0"; old "acme" (owned only by this stack) + // must be released and deleted. + isolation := &v1beta1.Settings{} + require.NoError(t, ctx.GetClient().Get(ctx, types.NamespacedName{Name: "isolation"}, isolation)) + isolation.Spec.Value = ModeStack + require.NoError(t, ctx.GetClient().Update(ctx, isolation)) + require.NoError(t, Reconcile(ctx, stack)) + + _, found = getUnstructured(t, ctx, testEC2GVK, "acme-stack0") + require.True(t, found, "new stack-mode pool should exist") + _, found = getUnstructured(t, ctx, testEC2GVK, "acme") + require.False(t, found, "old organization pool should be garbage collected") + _, found = getUnstructured(t, ctx, testNodePoolGVK, "acme") + require.False(t, found, "old organization NodePool should be garbage collected") +} + +func TestReconcileGCOnDisable(t *testing.T) { + SetAvailable(true) + t.Cleanup(func() { SetAvailable(false) }) + + objs := append(karpenterEnabledSettings(ModeOrganization, "stack0"), + reference(testEC2GVK, "default", map[string]any{"role": "r"}), + reference(testNodePoolGVK, "default", map[string]any{"template": map[string]any{"spec": map[string]any{}}}), + ) + ctx := newKarpenterContext(t, objs...) + stack := stackWithUID("stack0", "uid-0", map[string]string{v1beta1.OrganizationLabel: "acme"}) + require.NoError(t, Reconcile(ctx, stack)) + _, found := getUnstructured(t, ctx, testEC2GVK, "acme") + require.True(t, found) + + // Disable karpenter → the cluster-scoped pool must be cleaned up (the regression the + // review flagged: module cleanup cannot see cluster-scoped objects). + disabled := &v1beta1.Settings{} + require.NoError(t, ctx.GetClient().Get(ctx, types.NamespacedName{Name: "enabled"}, disabled)) + disabled.Spec.Value = "false" + require.NoError(t, ctx.GetClient().Update(ctx, disabled)) + + require.NoError(t, Reconcile(ctx, stack)) + _, found = getUnstructured(t, ctx, testEC2GVK, "acme") + require.False(t, found) + _, found = getUnstructured(t, ctx, testNodePoolGVK, "acme") + require.False(t, found) +} + +func TestReconcileOrgPoolSurvivesWhileAnotherStackOwns(t *testing.T) { + SetAvailable(true) + t.Cleanup(func() { SetAvailable(false) }) + + objs := append(karpenterEnabledSettings(ModeOrganization, "*"), + reference(testEC2GVK, "default", map[string]any{"role": "r"}), + reference(testNodePoolGVK, "default", map[string]any{"template": map[string]any{"spec": map[string]any{}}}), + ) + ctx := newKarpenterContext(t, objs...) + + stackA := stackWithUID("stackA", "uid-a", map[string]string{v1beta1.OrganizationLabel: "acme"}) + stackB := stackWithUID("stackB", "uid-b", map[string]string{v1beta1.OrganizationLabel: "acme"}) + require.NoError(t, Reconcile(ctx, stackA)) + require.NoError(t, Reconcile(ctx, stackB)) + + ec2, found := getUnstructured(t, ctx, testEC2GVK, "acme") + require.True(t, found) + require.Len(t, ec2.GetOwnerReferences(), 2, "both stacks should own the shared org pool") + + // Disable for stackA only → pool must survive because stackB still owns it. + require.NoError(t, ctx.GetClient().Create(ctx, setting("disableA", "karpenter.enabled", "false", "stackA"))) + require.NoError(t, Reconcile(ctx, stackA)) + + ec2, found = getUnstructured(t, ctx, testEC2GVK, "acme") + require.True(t, found, "org pool must survive while stackB owns it") + require.Len(t, ec2.GetOwnerReferences(), 1) +} diff --git a/internal/resources/nodeisolation/mock_test.go b/internal/resources/nodeisolation/mock_test.go new file mode 100644 index 000000000..949a8099a --- /dev/null +++ b/internal/resources/nodeisolation/mock_test.go @@ -0,0 +1,24 @@ +package nodeisolation + +import ( + "context" + + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/formancehq/operator/v3/internal/core" +) + +var _ core.Context = (*mockContext)(nil) + +// mockContext is a minimal core.Context backed by a fake client, for unit tests. +type mockContext struct { + context.Context + client client.Client + scheme *runtime.Scheme +} + +func (m *mockContext) GetClient() client.Client { return m.client } +func (m *mockContext) GetScheme() *runtime.Scheme { return m.scheme } +func (m *mockContext) GetAPIReader() client.Reader { return m.client } +func (m *mockContext) GetPlatform() core.Platform { return core.Platform{} } diff --git a/internal/resources/nodeisolation/resolve.go b/internal/resources/nodeisolation/resolve.go new file mode 100644 index 000000000..a48664a8d --- /dev/null +++ b/internal/resources/nodeisolation/resolve.go @@ -0,0 +1,200 @@ +package nodeisolation + +import ( + "fmt" + "maps" + "slices" + "strings" + + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/util/validation" + + "github.com/formancehq/operator/v3/api/formance.com/v1beta1" + "github.com/formancehq/operator/v3/internal/core" + "github.com/formancehq/operator/v3/internal/resources/settings" +) + +const ( + // ModeOrganization isolates all stacks of an organization on a shared pool named + // after the organization. ModeStack gives every stack a dedicated pool named + // "-". + ModeOrganization = "organization" + ModeStack = "stack" + + // ManagedByLabel/ManagedByValue are stamped on every Karpenter object created by the + // operator so the cleanup sweep can select them without relying on owner references + // (which the module cleanup path cannot see for cluster-scoped objects). + ManagedByLabel = "app.kubernetes.io/managed-by" + ManagedByValue = "formance-operator" + + EC2NodeClassKind = "EC2NodeClass" + NodePoolKind = "NodePool" + + defaultEC2NodeClassGroupVersion = "karpenter.k8s.aws/v1" + defaultNodePoolGroupVersion = "karpenter.sh/v1" +) + +// ManagedSchedulingKeys are the nodeSelector/toleration keys the operator injects into +// workloads. They are the pool-identity dimensions (organization, and stack in +// dedicated-stack mode). Customer is deliberately NOT a scheduling key: in a shared +// organization pool, stacks may resolve different customer values, so tainting/tolerating +// on customer would break scheduling. Customer remains an EC2 tag / node label for cost +// attribution only. Consumers strip these keys before re-applying, so placement stays +// idempotent across mode changes, relabels, and disable. +var ManagedSchedulingKeys = []string{v1beta1.OrganizationLabel, v1beta1.StackLabel} + +// Config is the resolved node-isolation intent for a single stack. It is the single +// source of truth shared by the Karpenter reconciler (which materializes the pools) and +// the applications mutator (which pins workloads onto them). +type Config struct { + Enabled bool + Mode string + PoolName string + + Organization string + Customer string + Stack string + + // Tags is the map injected into EC2NodeClass spec.tags and NodePool node labels. + Tags map[string]string + Taints []corev1.Taint + Tolerations []corev1.Toleration + NodeSelector map[string]string + + EC2NodeClassGVK schema.GroupVersionKind + NodePoolGVK schema.GroupVersionKind + EC2NodeClassRef string + NodePoolRef string +} + +// Resolve reads the karpenter.* settings and the Stack identity labels and computes the +// desired isolation Config. The Karpenter GVKs are always populated (even when disabled) +// so the cleanup sweep can locate previously created objects. When disabled, only +// Enabled/GVKs are meaningful. +func Resolve(ctx core.Context, stack *v1beta1.Stack) (*Config, error) { + ec2GVK, err := resolveGVK(ctx, stack.Name, EC2NodeClassKind, defaultEC2NodeClassGroupVersion, "ec2-node-class") + if err != nil { + return nil, err + } + nodePoolGVK, err := resolveGVK(ctx, stack.Name, NodePoolKind, defaultNodePoolGroupVersion, "node-pool") + if err != nil { + return nil, err + } + + cfg := &Config{ + EC2NodeClassGVK: ec2GVK, + NodePoolGVK: nodePoolGVK, + } + + enabled, err := settings.GetBoolOrFalse(ctx, stack.Name, "karpenter", "enabled") + if err != nil { + return nil, err + } + cfg.Enabled = enabled + if !enabled { + return cfg, nil + } + + mode, err := settings.GetStringOrDefault(ctx, stack.Name, ModeOrganization, "karpenter", "isolation") + if err != nil { + return nil, err + } + if mode != ModeOrganization && mode != ModeStack { + return nil, core.NewPendingError().WithMessage( + "invalid karpenter.isolation %q, expected %q or %q", mode, ModeOrganization, ModeStack) + } + cfg.Mode = mode + + labels := stack.GetLabels() + org := labels[v1beta1.OrganizationLabel] + if org == "" { + return nil, core.NewPendingError().WithMessage( + "stack label %s is required for karpenter node isolation", v1beta1.OrganizationLabel) + } + customer := labels[v1beta1.CustomerLabel] + if customer == "" { + customer = org + } + cfg.Organization = org + cfg.Customer = customer + cfg.Stack = stack.Name + + switch mode { + case ModeOrganization: + cfg.PoolName = org + case ModeStack: + cfg.PoolName = fmt.Sprintf("%s-%s", org, stack.Name) + } + if errs := validation.IsDNS1123Subdomain(cfg.PoolName); len(errs) > 0 { + return nil, core.NewPendingError().WithMessage( + "computed karpenter pool name %q is invalid: %s", cfg.PoolName, strings.Join(errs, ", ")) + } + + // Tags/node-labels: customer + organization always; stack only in dedicated-stack mode + // (an organization pool is shared, so a single stack value would be misleading). + tags := map[string]string{ + v1beta1.CustomerLabel: customer, + v1beta1.OrganizationLabel: org, + } + if mode == ModeStack { + tags[v1beta1.StackLabel] = stack.Name + } + for k, v := range tags { + if errs := validation.IsValidLabelValue(v); len(errs) > 0 { + return nil, core.NewPendingError().WithMessage( + "label value for %s (%q) is invalid: %s", k, v, strings.Join(errs, ", ")) + } + } + cfg.Tags = tags + + // Scheduling identity = the pool-identity dimensions only (organization, plus stack in + // dedicated-stack mode). NodeSelector enforces placement; taints/tolerations (keyed on + // the same identity) repel non-dedicated workloads. Customer is intentionally excluded + // from scheduling (see ManagedSchedulingKeys). + cfg.NodeSelector = map[string]string{v1beta1.OrganizationLabel: org} + if mode == ModeStack { + cfg.NodeSelector[v1beta1.StackLabel] = stack.Name + } + for _, key := range slices.Sorted(maps.Keys(cfg.NodeSelector)) { + cfg.Taints = append(cfg.Taints, corev1.Taint{ + Key: key, + Value: cfg.NodeSelector[key], + Effect: corev1.TaintEffectNoSchedule, + }) + cfg.Tolerations = append(cfg.Tolerations, corev1.Toleration{ + Key: key, + Operator: corev1.TolerationOpEqual, + Value: cfg.NodeSelector[key], + Effect: corev1.TaintEffectNoSchedule, + }) + } + + cfg.EC2NodeClassRef, err = settings.GetStringOrEmpty(ctx, stack.Name, "karpenter", "ec2-node-class", "reference") + if err != nil { + return nil, err + } + cfg.NodePoolRef, err = settings.GetStringOrEmpty(ctx, stack.Name, "karpenter", "node-pool", "reference") + if err != nil { + return nil, err + } + if cfg.EC2NodeClassRef == "" || cfg.NodePoolRef == "" { + return nil, core.NewPendingError().WithMessage( + "karpenter.ec2-node-class.reference and karpenter.node-pool.reference settings are required") + } + + return cfg, nil +} + +func resolveGVK(ctx core.Context, stack, kind, defaultGroupVersion, settingSegment string) (schema.GroupVersionKind, error) { + gv, err := settings.GetStringOrDefault(ctx, stack, defaultGroupVersion, "karpenter", "api", settingSegment, "group-version") + if err != nil { + return schema.GroupVersionKind{}, err + } + parsed, err := schema.ParseGroupVersion(gv) + if err != nil { + return schema.GroupVersionKind{}, core.NewPendingError().WithMessage( + "invalid group-version %q for %s: %s", gv, kind, err) + } + return parsed.WithKind(kind), nil +} diff --git a/internal/resources/nodeisolation/resolve_test.go b/internal/resources/nodeisolation/resolve_test.go new file mode 100644 index 000000000..36240490b --- /dev/null +++ b/internal/resources/nodeisolation/resolve_test.go @@ -0,0 +1,150 @@ +package nodeisolation + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + "github.com/formancehq/operator/v3/api/formance.com/v1beta1" + "github.com/formancehq/operator/v3/internal/resources/settings" +) + +func newTestContext(t *testing.T, objects ...client.Object) *mockContext { + t.Helper() + scheme := runtime.NewScheme() + require.NoError(t, v1beta1.AddToScheme(scheme)) + + fakeClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithIndex(&v1beta1.Settings{}, "stack", func(obj client.Object) []string { + return obj.(*v1beta1.Settings).GetStacks() + }). + WithIndex(&v1beta1.Settings{}, "keylen", func(obj client.Object) []string { + return []string{fmt.Sprint(len(settings.SplitKeywordWithDot(obj.(*v1beta1.Settings).Spec.Key)))} + }). + WithObjects(objects...). + Build() + + return &mockContext{Context: context.Background(), client: fakeClient, scheme: scheme} +} + +func setting(name, key, value string, stacks ...string) *v1beta1.Settings { + return &v1beta1.Settings{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Spec: v1beta1.SettingsSpec{Stacks: stacks, Key: key, Value: value}, + } +} + +func stackWithLabels(name string, labels map[string]string) *v1beta1.Stack { + return &v1beta1.Stack{ObjectMeta: metav1.ObjectMeta{Name: name, Labels: labels}} +} + +func TestResolveDisabled(t *testing.T) { + t.Parallel() + ctx := newTestContext(t) + cfg, err := Resolve(ctx, stackWithLabels("stack0", nil)) + require.NoError(t, err) + require.False(t, cfg.Enabled) + // GVKs are always populated so cleanup can find previously created objects. + require.Equal(t, "karpenter.k8s.aws", cfg.EC2NodeClassGVK.Group) + require.Equal(t, "karpenter.sh", cfg.NodePoolGVK.Group) +} + +func TestResolveOrganizationMode(t *testing.T) { + t.Parallel() + ctx := newTestContext(t, + setting("enabled", "karpenter.enabled", "true", "*"), + setting("ec2", "karpenter.ec2-node-class.reference", "default", "*"), + setting("np", "karpenter.node-pool.reference", "default", "*"), + ) + stack := stackWithLabels("stack0", map[string]string{ + v1beta1.OrganizationLabel: "acme", + v1beta1.CustomerLabel: "acme-corp", + }) + + cfg, err := Resolve(ctx, stack) + require.NoError(t, err) + require.True(t, cfg.Enabled) + require.Equal(t, ModeOrganization, cfg.Mode) + require.Equal(t, "acme", cfg.PoolName) + + // Organization pool is shared: no stack tag/label. + require.Equal(t, map[string]string{ + v1beta1.CustomerLabel: "acme-corp", + v1beta1.OrganizationLabel: "acme", + }, cfg.Tags) + require.NotContains(t, cfg.Tags, v1beta1.StackLabel) + + // NodeSelector enforces org placement only. + require.Equal(t, map[string]string{v1beta1.OrganizationLabel: "acme"}, cfg.NodeSelector) + + // Scheduling keys on organization only (customer is not a scheduling key). + require.Len(t, cfg.Taints, 1) + require.Len(t, cfg.Tolerations, 1) + require.Equal(t, v1beta1.OrganizationLabel, cfg.Taints[0].Key) + for _, tol := range cfg.Tolerations { + require.Equal(t, corev1.TaintEffectNoSchedule, tol.Effect) + require.Equal(t, corev1.TolerationOpEqual, tol.Operator) + } +} + +func TestResolveStackMode(t *testing.T) { + t.Parallel() + ctx := newTestContext(t, + setting("enabled", "karpenter.enabled", "true", "stack0"), + setting("isolation", "karpenter.isolation", ModeStack, "stack0"), + setting("ec2", "karpenter.ec2-node-class.reference", "default", "stack0"), + setting("np", "karpenter.node-pool.reference", "default", "stack0"), + ) + stack := stackWithLabels("stack0", map[string]string{v1beta1.OrganizationLabel: "acme"}) + + cfg, err := Resolve(ctx, stack) + require.NoError(t, err) + require.Equal(t, ModeStack, cfg.Mode) + require.Equal(t, "acme-stack0", cfg.PoolName) + + // Customer defaults to organization when the label is absent. + require.Equal(t, "acme", cfg.Customer) + require.Equal(t, map[string]string{ + v1beta1.CustomerLabel: "acme", + v1beta1.OrganizationLabel: "acme", + v1beta1.StackLabel: "stack0", + }, cfg.Tags) + require.Equal(t, map[string]string{ + v1beta1.OrganizationLabel: "acme", + v1beta1.StackLabel: "stack0", + }, cfg.NodeSelector) + // Scheduling keys on organization + stack (customer excluded). + require.Len(t, cfg.Taints, 2) +} + +func TestResolveMissingOrganizationIsPending(t *testing.T) { + t.Parallel() + ctx := newTestContext(t, setting("enabled", "karpenter.enabled", "true", "*")) + _, err := Resolve(ctx, stackWithLabels("stack0", nil)) + require.Error(t, err) +} + +func TestResolveInvalidModeIsPending(t *testing.T) { + t.Parallel() + ctx := newTestContext(t, + setting("enabled", "karpenter.enabled", "true", "*"), + setting("isolation", "karpenter.isolation", "bogus", "*"), + ) + _, err := Resolve(ctx, stackWithLabels("stack0", map[string]string{v1beta1.OrganizationLabel: "acme"})) + require.Error(t, err) +} + +func TestResolveMissingReferencesIsPending(t *testing.T) { + t.Parallel() + ctx := newTestContext(t, setting("enabled", "karpenter.enabled", "true", "*")) + _, err := Resolve(ctx, stackWithLabels("stack0", map[string]string{v1beta1.OrganizationLabel: "acme"})) + require.Error(t, err) +} diff --git a/internal/resources/stacks/init.go b/internal/resources/stacks/init.go index d2171b279..f186eee8a 100644 --- a/internal/resources/stacks/init.go +++ b/internal/resources/stacks/init.go @@ -12,6 +12,7 @@ import ( "github.com/pkg/errors" corev1 "k8s.io/api/core/v1" networkingv1 "k8s.io/api/networking/v1" + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime" @@ -29,6 +30,7 @@ import ( "github.com/formancehq/operator/v3/api/formance.com/v1beta1" . "github.com/formancehq/operator/v3/internal/core" + "github.com/formancehq/operator/v3/internal/resources/nodeisolation" "github.com/formancehq/operator/v3/internal/resources/settings" ) @@ -50,6 +52,9 @@ import ( // +kubebuilder:rbac:groups=formance.com,resources=versions,verbs=get;list;watch;create;update;patch;delete // +kubebuilder:rbac:groups=formance.com,resources=versions/status,verbs=get;update;patch // +kubebuilder:rbac:groups=formance.com,resources=versions/finalizers,verbs=update +// +kubebuilder:rbac:groups=karpenter.sh,resources=nodepools,verbs=get;list;watch;create;update;patch;delete +// +kubebuilder:rbac:groups=karpenter.k8s.aws,resources=ec2nodeclasses,verbs=get;list;watch;create;update;patch;delete +// +kubebuilder:rbac:groups=apiextensions.k8s.io,resources=customresourcedefinitions,verbs=get;list;watch var ( ModuleReconciliation = "ModuleReconciliation" @@ -257,6 +262,10 @@ func Reconcile(ctx Context, stack *v1beta1.Stack) error { return err } + if err := nodeisolation.Reconcile(ctx, stack); err != nil { + return err + } + setLicenceCondition(ctx, stack) if err := setModulesCondition(ctx, stack); err != nil { @@ -466,7 +475,7 @@ func init() { b.Watches(&v1beta1.Settings{}, handler.EnqueueRequestsFromMapFunc( func(watchCtx context.Context, object client.Object) []reconcile.Request { s := object.(*v1beta1.Settings) - if s.Spec.Key != "networkpolicies.enabled" { + if s.Spec.Key != "networkpolicies.enabled" && !strings.HasPrefix(s.Spec.Key, "karpenter.") { return nil } requests := make([]reconcile.Request, 0) @@ -492,6 +501,39 @@ func init() { )) return nil }), + // Detect the Karpenter CRDs at setup, and re-detect + re-reconcile all stacks if + // they are installed/removed after the operator started (so a post-startup + // Karpenter install does not require an operator restart). + WithRaw[*v1beta1.Stack](func(ctx Context, b *builder.Builder) error { + if err := nodeisolation.DetectCRDs(ctx); err != nil { + return err + } + b.Watches(&apiextensionsv1.CustomResourceDefinition{}, handler.EnqueueRequestsFromMapFunc( + func(watchCtx context.Context, object client.Object) []reconcile.Request { + if !nodeisolation.IsKarpenterCRD(object.GetName()) { + return nil + } + if err := nodeisolation.DetectCRDs(ctx); err != nil { + return nil + } + stackList := &v1beta1.StackList{} + if err := ctx.GetClient().List(watchCtx, stackList); err != nil { + return nil + } + return collectionutils.Map(stackList.Items, func(s v1beta1.Stack) reconcile.Request { + return reconcile.Request{NamespacedName: types.NamespacedName{Name: s.Name}} + }) + })) + // The primary Stack predicate only fires on generation/owner-ref changes, but + // node isolation keys off the formance.com/organization and /customer labels + // (metadata, no generation bump). Watch label changes so relabeling a stack + // re-reconciles its pools and workload placement. + b.Watches(&v1beta1.Stack{}, handler.EnqueueRequestsFromMapFunc( + func(_ context.Context, object client.Object) []reconcile.Request { + return []reconcile.Request{{NamespacedName: types.NamespacedName{Name: object.GetName()}}} + }), builder.WithPredicates(predicate.LabelChangedPredicate{})) + return nil + }), // notes(gfyrag): Some resources need to be properly dropped before the stack is dropped WithFinalizer("delete", Clean), ),