diff --git a/.ainav/operations/index.md b/.ainav/operations/index.md index 39d9caca4..ba927b665 100644 --- a/.ainav/operations/index.md +++ b/.ainav/operations/index.md @@ -37,7 +37,7 @@ Manual operations, policies, CSI, and driver management. | File | Purpose | |------|---------| -| `discover_node.go` | Node discovery | +| `discover_node.go` | Node discovery; recreates discovery container on owner-spec drift (image/tolerations/pullSecret/serviceAccount) | | `ensure_nics.go` | NIC configuration | | `trace_session.go` | Remote trace collection | | `cleanup_persistent_dir.go` | Cleanup operations | diff --git a/internal/controllers/operations/discover_node.go b/internal/controllers/operations/discover_node.go index 0e2dc2854..63881246d 100644 --- a/internal/controllers/operations/discover_node.go +++ b/internal/controllers/operations/discover_node.go @@ -4,9 +4,11 @@ import ( "context" "encoding/json" "fmt" + "reflect" "github.com/pkg/errors" "github.com/weka/go-steps-engine/lifecycle" + k8sutil "github.com/weka/weka-k8s-api/util" "github.com/weka/go-weka-observability/instrumentation" weka "github.com/weka/weka-k8s-api/api/v1alpha1" corev1 "k8s.io/api/core/v1" @@ -170,10 +172,34 @@ func (o *DiscoverNodeOperation) GetContainer(ctx context.Context) error { return nil } +// isContainerSpecChanged reports whether the existing discovery container no longer matches the owner-derived spec +func (o *DiscoverNodeOperation) isContainerSpecChanged() bool { + spec := o.container.Spec + return spec.Image != o.image || + spec.ImagePullSecret != o.pullSecret || + spec.ServiceAccountName != o.serviceAccount || + !reflect.DeepEqual(k8sutil.NormalizeTolerations(spec.Tolerations), k8sutil.NormalizeTolerations(o.tolerations)) +} + func (o *DiscoverNodeOperation) EnsureContainers(ctx context.Context) error { if o.container != nil { - return nil + if o.container.GetDeletionTimestamp() != nil { + return lifecycle.NewWaitError(fmt.Errorf("discovery container %s is being deleted", o.container.Name)) + } + // the discovery container is a shared per-node singleton; only its controller-owner + // enforces spec drift, otherwise owners with different specs delete each other's container in a loop + if !metav1.IsControlledBy(o.container, o.ownerRef) { + return nil + } + if !o.isContainerSpecChanged() { + return nil + } + // recreate on a later pass — a same-name Create would fail while the old container is terminating + if err := o.DeleteContainers(ctx); err != nil { + return err + } + return lifecycle.NewWaitError(fmt.Errorf("discovery container spec changed, recreating")) } // If we already have valid discovery information, skip container creation diff --git a/internal/controllers/operations/discover_node_test.go b/internal/controllers/operations/discover_node_test.go new file mode 100644 index 000000000..98598f59c --- /dev/null +++ b/internal/controllers/operations/discover_node_test.go @@ -0,0 +1,217 @@ +package operations + +import ( + "context" + "errors" + "testing" + + weka "github.com/weka/weka-k8s-api/api/v1alpha1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + 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/weka/go-steps-engine/lifecycle" +) + +func newDiscoverNodeTestScheme(t *testing.T) *runtime.Scheme { + t.Helper() + scheme := runtime.NewScheme() + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + if err := weka.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + return scheme +} + +func newDiscoverNodeTestOp(scheme *runtime.Scheme, existing ...*weka.WekaContainer) *DiscoverNodeOperation { + builder := fake.NewClientBuilder().WithScheme(scheme) + for _, c := range existing { + builder = builder.WithObjects(c) + } + kclient := builder.Build() + + owner := &weka.WekaClient{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-client", + Namespace: "default", + UID: "test-client-uid", + }, + } + return &DiscoverNodeOperation{ + client: kclient, + scheme: scheme, + nodeName: "test-node", + image: "quay.io/weka.io/weka-in-container:4.5.0", + pullSecret: "pull-secret", + serviceAccount: "weka-sa", + tolerations: []corev1.Toleration{{Key: "gpu", Operator: corev1.TolerationOpExists}}, + ownerRef: owner, + node: &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: "test-node"}, + }, + } +} + +func (o *DiscoverNodeOperation) desiredTestContainer() *weka.WekaContainer { + controller := true + return &weka.WekaContainer{ + ObjectMeta: metav1.ObjectMeta{ + Name: o.getContainerName(), + Namespace: o.ownerRef.GetNamespace(), + OwnerReferences: []metav1.OwnerReference{{ + APIVersion: "weka.weka.io/v1alpha1", + Kind: "WekaClient", + Name: o.ownerRef.GetName(), + UID: o.ownerRef.GetUID(), + Controller: &controller, + }}, + }, + Spec: weka.WekaContainerSpec{ + Mode: weka.WekaContainerModeDiscovery, + NodeAffinity: weka.NodeName(o.node.Name), + Image: o.image, + ImagePullSecret: o.pullSecret, + Tolerations: o.tolerations, + ServiceAccountName: o.serviceAccount, + }, + } +} + +func TestIsContainerSpecChanged(t *testing.T) { + scheme := newDiscoverNodeTestScheme(t) + + cases := []struct { + name string + mutate func(c *weka.WekaContainer) + want bool + }{ + {"identical", func(c *weka.WekaContainer) {}, false}, + {"image drift", func(c *weka.WekaContainer) { c.Spec.Image = "quay.io/weka.io/weka-in-container:4.6.0" }, true}, + {"pull secret drift", func(c *weka.WekaContainer) { c.Spec.ImagePullSecret = "other-secret" }, true}, + {"service account drift", func(c *weka.WekaContainer) { c.Spec.ServiceAccountName = "other-sa" }, true}, + {"tolerations drift", func(c *weka.WekaContainer) { c.Spec.Tolerations = nil }, true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + op := newDiscoverNodeTestOp(scheme) + container := op.desiredTestContainer() + tc.mutate(container) + op.container = container + if got := op.isContainerSpecChanged(); got != tc.want { + t.Errorf("isContainerSpecChanged() = %v, want %v", got, tc.want) + } + }) + } + + t.Run("nil vs empty tolerations is not drift", func(t *testing.T) { + op := newDiscoverNodeTestOp(scheme) + op.tolerations = []corev1.Toleration{} + container := op.desiredTestContainer() + container.Spec.Tolerations = nil + op.container = container + if op.isContainerSpecChanged() { + t.Error("nil vs empty tolerations must not count as drift") + } + }) +} + +func TestEnsureContainers_ExistingMatching(t *testing.T) { + scheme := newDiscoverNodeTestScheme(t) + op := newDiscoverNodeTestOp(scheme) + container := op.desiredTestContainer() + op = newDiscoverNodeTestOp(scheme, container) + op.container = container + + if err := op.EnsureContainers(context.Background()); err != nil { + t.Fatalf("expected nil error, got %v", err) + } + if op.container == nil { + t.Fatal("matching container must be kept") + } + got := &weka.WekaContainer{} + if err := op.client.Get(context.Background(), client.ObjectKey{Namespace: "default", Name: container.Name}, got); err != nil { + t.Fatalf("matching container must not be deleted: %v", err) + } +} + +func TestEnsureContainers_SpecChangedDeletesAndWaits(t *testing.T) { + scheme := newDiscoverNodeTestScheme(t) + op := newDiscoverNodeTestOp(scheme) + container := op.desiredTestContainer() + container.Spec.Image = "quay.io/weka.io/weka-in-container:old" + op = newDiscoverNodeTestOp(scheme, container) + op.container = container + + err := op.EnsureContainers(context.Background()) + waitErr := &lifecycle.WaitError{} + if !errors.As(err, &waitErr) { + t.Fatalf("expected WaitError, got %v", err) + } + got := &weka.WekaContainer{} + getErr := op.client.Get(context.Background(), client.ObjectKey{Namespace: "default", Name: container.Name}, got) + if !apierrors.IsNotFound(getErr) { + t.Fatalf("drifted container must be deleted, got %v", getErr) + } +} + +func TestEnsureContainers_ForeignOwnerNotTouched(t *testing.T) { + scheme := newDiscoverNodeTestScheme(t) + op := newDiscoverNodeTestOp(scheme) + container := op.desiredTestContainer() + container.Spec.Image = "quay.io/weka.io/weka-in-container:old" + container.OwnerReferences[0].Name = "other-owner" + container.OwnerReferences[0].UID = "other-owner-uid" + op = newDiscoverNodeTestOp(scheme, container) + op.container = container + + if err := op.EnsureContainers(context.Background()); err != nil { + t.Fatalf("expected nil error for foreign-owned container, got %v", err) + } + got := &weka.WekaContainer{} + if err := op.client.Get(context.Background(), client.ObjectKey{Namespace: "default", Name: container.Name}, got); err != nil { + t.Fatalf("foreign-owned container must not be deleted: %v", err) + } +} + +func TestEnsureContainers_TerminatingWaitsWithoutDelete(t *testing.T) { + scheme := newDiscoverNodeTestScheme(t) + op := newDiscoverNodeTestOp(scheme) + container := op.desiredTestContainer() + container.Spec.Image = "quay.io/weka.io/weka-in-container:old" + now := metav1.Now() + container.DeletionTimestamp = &now + container.Finalizers = []string{"weka.io/test"} + op.container = container + + err := op.EnsureContainers(context.Background()) + waitErr := &lifecycle.WaitError{} + if !errors.As(err, &waitErr) { + t.Fatalf("expected WaitError for terminating container, got %v", err) + } + if op.container == nil { + t.Fatal("terminating container must not be re-deleted (DeleteContainers resets o.container)") + } +} + +func TestEnsureContainers_CreatesWhenMissing(t *testing.T) { + scheme := newDiscoverNodeTestScheme(t) + op := newDiscoverNodeTestOp(scheme) + + if err := op.EnsureContainers(context.Background()); err != nil { + t.Fatalf("expected nil error, got %v", err) + } + if op.container == nil { + t.Fatal("container must be created") + } + if op.container.Spec.Image != op.image { + t.Errorf("created with image %q, want %q", op.container.Spec.Image, op.image) + } + if op.container.Spec.ServiceAccountName != op.serviceAccount { + t.Errorf("created with serviceAccount %q, want %q", op.container.Spec.ServiceAccountName, op.serviceAccount) + } +}