diff --git a/pkg/aksmachine/client_armapi.go b/pkg/aksmachine/client_armapi.go index c91cb0e9..f6c2327b 100644 --- a/pkg/aksmachine/client_armapi.go +++ b/pkg/aksmachine/client_armapi.go @@ -77,9 +77,10 @@ func (c *armMachineClient) Create(ctx context.Context, desired GoalState) (*Mach if err := c.validateMachineIdentity(resp.Machine); err != nil { return nil, err } - result := machineFromARM(resp.Machine, desired) - result.ID = c.machineID.String() - result.Name = c.machineID.Name + result := machineFromARM(resp.Machine, c.machineID.String(), c.machineID.Name) + if err := result.Validate(); err != nil { + return nil, fmt.Errorf("validate create machine response: %w", err) + } return result, nil } @@ -103,9 +104,10 @@ func (c *armMachineClient) Get(ctx context.Context) (*Machine, error) { if err := c.validateMachineIdentity(resp.Machine); err != nil { return nil, err } - result := machineFromARM(resp.Machine, GoalState{}) - result.ID = c.machineID.String() - result.Name = c.machineID.Name + result := machineFromARM(resp.Machine, c.machineID.String(), c.machineID.Name) + if err := result.Validate(); err != nil { + return nil, fmt.Errorf("validate get machine response: %w", err) + } return result, nil } @@ -256,8 +258,8 @@ func (c *armMachineClient) validateMachineIdentity(machine armcontainerservice.M return nil } -func machineFromARM(machine armcontainerservice.Machine, fallback GoalState) *Machine { - result := &Machine{Goal: fallback} +func machineFromARM(machine armcontainerservice.Machine, defaultID, defaultName string) *Machine { + result := &Machine{ID: defaultID, Name: defaultName} if machine.ID != nil { result.ID = *machine.ID } @@ -274,9 +276,6 @@ func machineFromARM(machine armcontainerservice.Machine, fallback GoalState) *Ma if kubernetes.OrchestratorVersion != nil { result.Goal.KubernetesVersion = *kubernetes.OrchestratorVersion } - if result.Goal.KubernetesVersion == "" && kubernetes.CurrentOrchestratorVersion != nil { - result.Goal.KubernetesVersion = *kubernetes.CurrentOrchestratorVersion - } if kubernetes.MaxPods != nil { result.Goal.MaxPods = int(*kubernetes.MaxPods) } @@ -298,9 +297,6 @@ func machineFromARM(machine armcontainerservice.Machine, fallback GoalState) *Ma if properties.ETag != nil { result.Goal.SettingsVersion = *properties.ETag } - if result.Goal.SettingsVersion == "" { - result.Goal.SettingsVersion = result.Goal.KubernetesVersion - } if properties.ProvisioningState != nil { result.Status.ProvisioningState = ProvisioningState(*properties.ProvisioningState) } diff --git a/pkg/aksmachine/client_armapi_test.go b/pkg/aksmachine/client_armapi_test.go index ef3f61e1..0644171d 100644 --- a/pkg/aksmachine/client_armapi_test.go +++ b/pkg/aksmachine/client_armapi_test.go @@ -317,7 +317,7 @@ func TestMachineFromARM(t *testing.T) { }, ProvisioningState: ptr("Succeeded"), }, - }, GoalState{SettingsVersion: "fallback-settings"}) + }, "default-id", "default-name") if machine.ID != "machine-id" || machine.Name != "node1" { t.Fatalf("machine identity = %#v", machine) @@ -398,7 +398,7 @@ func TestValidateMachineIdentity(t *testing.T) { } } -func TestMachineFromARMUsesCurrentOrchestratorVersionFallback(t *testing.T) { +func TestMachineFromARMDoesNotUseCurrentOrchestratorVersionAsGoal(t *testing.T) { t.Parallel() currentVersion := "1.35.2" @@ -408,13 +408,36 @@ func TestMachineFromARMUsesCurrentOrchestratorVersionFallback(t *testing.T) { CurrentOrchestratorVersion: ¤tVersion, }, }, - }, GoalState{}) + }, "", "") - if machine.Goal.KubernetesVersion != currentVersion { - t.Fatalf("KubernetesVersion = %q, want %q", machine.Goal.KubernetesVersion, currentVersion) + if machine.Goal.KubernetesVersion != "" { + t.Fatalf("KubernetesVersion = %q, want empty without desired orchestratorVersion", machine.Goal.KubernetesVersion) } - if machine.Goal.SettingsVersion != currentVersion { - t.Fatalf("SettingsVersion = %q, want %q", machine.Goal.SettingsVersion, currentVersion) +} + +func TestMachineFromARMDoesNotSynthesizeSettingsVersion(t *testing.T) { + t.Parallel() + + machine := machineFromARM(armcontainerservice.Machine{ + Properties: &armcontainerservice.MachineProperties{ + Kubernetes: &armcontainerservice.MachineKubernetesProfile{ + OrchestratorVersion: ptr("1.35.2"), + }, + }, + }, "", "") + + if machine.Goal.SettingsVersion != "" { + t.Fatalf("SettingsVersion = %q, want empty without ETag", machine.Goal.SettingsVersion) + } +} + +func TestMachineFromARMBackfillsIdentity(t *testing.T) { + t.Parallel() + + machine := machineFromARM(armcontainerservice.Machine{}, "machine-id", "node1") + + if machine.ID != "machine-id" || machine.Name != "node1" { + t.Fatalf("machine identity = %#v, want backfilled identity", machine) } } diff --git a/pkg/aksmachine/client_incluster.go b/pkg/aksmachine/client_incluster.go index 27314eb1..3b518a7a 100644 --- a/pkg/aksmachine/client_incluster.go +++ b/pkg/aksmachine/client_incluster.go @@ -170,8 +170,8 @@ func (c *clusterEndpointClient) adoptExistingMachine(ctx context.Context, desire } func validateAdoptedMachine(machine *Machine, desired GoalState) error { - if machine == nil { - return fmt.Errorf("cluster endpoint returned nil machine") + if err := machine.Validate(); err != nil { + return fmt.Errorf("cluster endpoint returned invalid machine: %w", err) } if desired.KubernetesVersion != "" && machine.Goal.KubernetesVersion != "" && machine.Goal.KubernetesVersion != desired.KubernetesVersion { return fmt.Errorf("pre-created machine Kubernetes version %q does not match desired %q", machine.Goal.KubernetesVersion, desired.KubernetesVersion) @@ -279,7 +279,11 @@ func machineFromEndpointJSON(data []byte) (*Machine, error) { if err := json.Unmarshal(data, &armMachine); err != nil { return nil, fmt.Errorf("decode cluster endpoint machine response: %w", err) } - return machineFromARM(armMachine, GoalState{}), nil + machine := machineFromARM(armMachine, "", "") + if err := machine.Validate(); err != nil { + return nil, fmt.Errorf("validate cluster endpoint machine response: %w", err) + } + return machine, nil } func joinEndpointURLPath(base string, elem ...string) string { diff --git a/pkg/aksmachine/client_incluster_test.go b/pkg/aksmachine/client_incluster_test.go index b67b4a97..77df721d 100644 --- a/pkg/aksmachine/client_incluster_test.go +++ b/pkg/aksmachine/client_incluster_test.go @@ -88,6 +88,19 @@ func TestMachineFromEndpointJSONUsesARMModel(t *testing.T) { } } +func TestMachineFromEndpointJSONRejectsMissingETag(t *testing.T) { + t.Parallel() + + _, err := machineFromEndpointJSON([]byte(`{ + "properties": { + "kubernetes": {"orchestratorVersion": "1.34.0"} + } +}`)) + if err == nil || !strings.Contains(err.Error(), "goal settings version is empty") { + t.Fatalf("machineFromEndpointJSON() error = %v, want missing settings version", err) + } +} + func TestClusterEndpointClientNotFound(t *testing.T) { t.Parallel() diff --git a/pkg/aksmachine/ensure.go b/pkg/aksmachine/ensure.go index cdd3b336..f08178c6 100644 --- a/pkg/aksmachine/ensure.go +++ b/pkg/aksmachine/ensure.go @@ -28,56 +28,89 @@ func EnsureMachine(machines MachineClient, goal *GoalState, require bool, logger func (t *ensureMachineTask) Name() string { return "ensure-machine" } func (t *ensureMachineTask) Do(ctx context.Context) error { - machine, err := t.machines.Get(ctx) - if err == nil { - if machine != nil && machine.Goal.KubernetesVersion == t.goal.KubernetesVersion { - t.logger.Info("machine already registered, skipping") - return t.adoptSettingsVersion(machine, "get machine") - } + remoteMachine, err := t.fetchRemoteMachine(ctx) + if err != nil { + return t.handleError("get machine", err) + } - remoteVersion := "" - if machine != nil { - remoteVersion = machine.Goal.KubernetesVersion + switch { + case remoteMachine == nil: + remoteMachine, err = t.createRemoteMachineFromGoal(ctx) + if err != nil { + return t.handleError("create machine", err) } - t.logger.Info( - "updating registered machine from local bootstrap config", - "remoteKubernetesVersion", remoteVersion, - "localKubernetesVersion", t.goal.KubernetesVersion, - ) - machine, err = t.machines.Create(ctx, *t.goal) + case machineGoalHasDrift(remoteMachine.Goal, *t.goal): + remoteMachine, err = t.updateRemoteMachineFromGoal(ctx, remoteMachine) if err != nil { return t.handleError("update machine", err) } - return t.adoptSettingsVersion(machine, "update machine") + default: + t.logger.Info("machine already registered, skipping") } + t.applyGoalStateWithRemoteMachineSettingsVersion(remoteMachine) + return nil +} +func (t *ensureMachineTask) fetchRemoteMachine(ctx context.Context) (*Machine, error) { + machine, err := t.machines.Get(ctx) var notFound *NotFoundError - if !errors.As(err, ¬Found) { - return t.handleError("get machine", err) + if errors.As(err, ¬Found) { + return nil, nil } - machine, err = t.machines.Create(ctx, *t.goal) if err != nil { - return t.handleError("create machine", err) + return nil, err } - return t.adoptSettingsVersion(machine, "create machine") + if err := machine.Validate(); err != nil { + return nil, fmt.Errorf("AKS returned an invalid machine: %w", err) + } + return machine, nil } -func (t *ensureMachineTask) adoptSettingsVersion(machine *Machine, operation string) error { - if machine == nil { - return t.handleError(operation, fmt.Errorf("AKS returned a nil machine")) +func (t *ensureMachineTask) createRemoteMachineFromGoal(ctx context.Context) (*Machine, error) { + machine, err := t.machines.Create(ctx, *t.goal) + if err != nil { + return nil, err } - if machine.Goal.KubernetesVersion != t.goal.KubernetesVersion { - return t.handleError( - operation, - fmt.Errorf( - "AKS machine Kubernetes version %q does not match local bootstrap version %q", - machine.Goal.KubernetesVersion, - t.goal.KubernetesVersion, - ), - ) + if err := validateMachineForGoal(machine, *t.goal); err != nil { + return nil, err } - if machine.Goal.SettingsVersion != "" { - t.goal.SettingsVersion = machine.Goal.SettingsVersion + return machine, nil +} + +func (t *ensureMachineTask) updateRemoteMachineFromGoal(ctx context.Context, current *Machine) (*Machine, error) { + t.logger.Info( + "updating registered machine from local bootstrap config", + "remoteKubernetesVersion", current.Goal.KubernetesVersion, + "localKubernetesVersion", t.goal.KubernetesVersion, + ) + machine, err := t.machines.Create(ctx, *t.goal) + if err != nil { + return nil, err + } + if err := validateMachineForGoal(machine, *t.goal); err != nil { + return nil, err + } + return machine, nil +} + +func (t *ensureMachineTask) applyGoalStateWithRemoteMachineSettingsVersion(machine *Machine) { + t.goal.SettingsVersion = machine.Goal.SettingsVersion +} + +func machineGoalHasDrift(remote, desired GoalState) bool { + return remote.KubernetesVersion != desired.KubernetesVersion +} + +func validateMachineForGoal(machine *Machine, goal GoalState) error { + if err := machine.Validate(); err != nil { + return fmt.Errorf("AKS returned an invalid machine: %w", err) + } + if machine.Goal.KubernetesVersion != goal.KubernetesVersion { + return fmt.Errorf( + "AKS machine Kubernetes version %q does not match local bootstrap version %q", + machine.Goal.KubernetesVersion, + goal.KubernetesVersion, + ) } return nil } diff --git a/pkg/aksmachine/ensure_test.go b/pkg/aksmachine/ensure_test.go index 5e7f09bb..3e8b0750 100644 --- a/pkg/aksmachine/ensure_test.go +++ b/pkg/aksmachine/ensure_test.go @@ -84,7 +84,7 @@ func TestEnsureMachineGetFailure(t *testing.T) { func TestEnsureMachineCreatesAndAdoptsSettingsVersion(t *testing.T) { t.Parallel() - goal := GoalState{KubernetesVersion: "1.35.1", SettingsVersion: "1.35.1"} + goal := GoalState{KubernetesVersion: "1.35.1"} client := &ensureMachineClient{createResult: &Machine{Goal: GoalState{ KubernetesVersion: "1.35.1", SettingsVersion: "etag-created", @@ -107,7 +107,6 @@ func TestEnsureMachineAdoptsExistingSettingsVersionWithoutReplacingLocalGoal(t * goal := GoalState{ KubernetesVersion: "1.35.1", - SettingsVersion: "1.35.1", MaxPods: 30, NodeLabels: map[string]string{"source": "local"}, NodeTaints: []string{"local=true:NoSchedule"}, @@ -149,7 +148,7 @@ func TestEnsureMachineAdoptsExistingSettingsVersionWithoutReplacingLocalGoal(t * func TestEnsureMachineUpdatesMismatchedVersion(t *testing.T) { t.Parallel() - goal := GoalState{KubernetesVersion: "1.35.1", SettingsVersion: "1.35.1"} + goal := GoalState{KubernetesVersion: "1.35.1"} client := &ensureMachineClient{ machine: &Machine{Goal: GoalState{KubernetesVersion: "1.34.0", SettingsVersion: "etag-old"}}, createResult: &Machine{Goal: GoalState{ @@ -176,7 +175,7 @@ func TestEnsureMachineUpdatesMismatchedVersion(t *testing.T) { func TestEnsureMachineRejectsUnchangedRemoteVersionAfterUpdate(t *testing.T) { t.Parallel() - goal := GoalState{KubernetesVersion: "1.35.1", SettingsVersion: "1.35.1"} + goal := GoalState{KubernetesVersion: "1.35.1"} client := &ensureMachineClient{ machine: &Machine{Goal: GoalState{KubernetesVersion: "1.34.0", SettingsVersion: "etag-old"}}, createResult: &Machine{Goal: GoalState{ @@ -190,13 +189,63 @@ func TestEnsureMachineRejectsUnchangedRemoteVersionAfterUpdate(t *testing.T) { if err == nil || !strings.Contains(err.Error(), `AKS machine Kubernetes version "1.34.0" does not match local bootstrap version "1.35.1"`) { t.Fatalf("Do() error = %v, want version mismatch", err) } - if goal.SettingsVersion != "1.35.1" { - t.Fatalf("SettingsVersion = %q, want local fallback", goal.SettingsVersion) + if goal.SettingsVersion != "" { + t.Fatalf("SettingsVersion = %q, want empty before a valid Machine response", goal.SettingsVersion) + } +} + +func TestEnsureMachineRejectsInvalidExistingMachine(t *testing.T) { + t.Parallel() + + tests := map[string]struct { + machine *Machine + require bool + wantErr string + }{ + "best effort preserves local goal after missing settings version": { + machine: &Machine{Goal: GoalState{KubernetesVersion: "1.35.1"}}, + }, + "required rejects missing settings version": { + machine: &Machine{Goal: GoalState{KubernetesVersion: "1.35.1"}}, + require: true, + wantErr: "goal settings version is empty", + }, + "best effort preserves local goal after nil response": {}, + "required rejects nil response": { + require: true, + wantErr: "machine is nil", + }, + } + + for name, tt := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + + goal := GoalState{KubernetesVersion: "1.35.1", NodeLabels: map[string]string{"source": "local"}} + client := &ensureMachineClient{machine: tt.machine, getResultSet: true} + task := EnsureMachine(client, &goal, tt.require, slog.New(slog.NewTextHandler(io.Discard, nil))) + + err := task.Do(t.Context()) + if tt.wantErr != "" { + if err == nil || !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("Do() error = %v, want containing %q", err, tt.wantErr) + } + } else if err != nil { + t.Fatalf("Do() error = %v", err) + } + if goal.SettingsVersion != "" || goal.NodeLabels["source"] != "local" { + t.Fatalf("local goal changed after invalid response: %#v", goal) + } + if client.createCalls != 0 { + t.Fatalf("Create() calls = %d, want 0", client.createCalls) + } + }) } } type ensureMachineClient struct { machine *Machine + getResultSet bool createResult *Machine getErr error createErr error @@ -208,7 +257,7 @@ func (c *ensureMachineClient) Get(context.Context) (*Machine, error) { if c.getErr != nil { return nil, c.getErr } - if c.machine != nil { + if c.getResultSet || c.machine != nil { return c.machine, nil } return nil, &NotFoundError{Resource: "machine"} diff --git a/pkg/aksmachine/types.go b/pkg/aksmachine/types.go index 89659fe6..35b1f4b8 100644 --- a/pkg/aksmachine/types.go +++ b/pkg/aksmachine/types.go @@ -49,11 +49,8 @@ func (g GoalState) validate() error { // GoalStateFromConfig builds and validates the initial AKS machine goal state // from local agent configuration. func GoalStateFromConfig(cfg *config.Config) (GoalState, error) { - // Until the finalized Machine API exposes a settings version in all paths, - // use KubernetesVersion as the same stable fallback used by ARM reads. goal := GoalState{ KubernetesVersion: cfg.Components.Kubernetes, - SettingsVersion: cfg.Components.Kubernetes, MaxPods: cfg.Node.MaxPods, NodeLabels: maps.Clone(cfg.Node.Labels), NodeTaints: slices.Clone(cfg.Node.Taints), @@ -93,6 +90,21 @@ type Machine struct { Status Status `json:"status"` } +// Validate verifies that a Machine returned by AKS contains a goal suitable +// for bootstrap or reconciliation. +func (m *Machine) Validate() error { + if m == nil { + return fmt.Errorf("machine is nil") + } + if err := m.Goal.validate(); err != nil { + return fmt.Errorf("goal: %w", err) + } + if m.Goal.SettingsVersion == "" { + return fmt.Errorf("goal settings version is empty") + } + return nil +} + // MachineClient provides access to the AKS-side machine representation. // Production should use the official Azure SDK implementation once the public // SDK contains the finalized resource shape; tests can provide fake or remote diff --git a/pkg/aksmachine/types_test.go b/pkg/aksmachine/types_test.go index 1d19343d..b6ab345f 100644 --- a/pkg/aksmachine/types_test.go +++ b/pkg/aksmachine/types_test.go @@ -41,8 +41,8 @@ func TestGoalStateFromConfig(t *testing.T) { if goal.KubernetesVersion != "1.35.1" { t.Fatalf("KubernetesVersion = %q, want 1.35.1", goal.KubernetesVersion) } - if goal.SettingsVersion != "1.35.1" { - t.Fatalf("SettingsVersion = %q, want 1.35.1", goal.SettingsVersion) + if goal.SettingsVersion != "" { + t.Fatalf("SettingsVersion = %q, want empty before Machine persistence", goal.SettingsVersion) } if goal.MaxPods != 42 { t.Fatalf("MaxPods = %d, want 42", goal.MaxPods) @@ -82,3 +82,44 @@ func TestGoalStateFromConfigValidates(t *testing.T) { t.Fatalf("GoalStateFromConfig() error = %v, want Kubernetes version validation", err) } } + +func TestMachineValidate(t *testing.T) { + t.Parallel() + + tests := map[string]struct { + machine *Machine + wantErr string + }{ + "nil machine": { + wantErr: "machine is nil", + }, + "missing Kubernetes version": { + machine: &Machine{Goal: GoalState{SettingsVersion: "42"}}, + wantErr: "kubernetes version is empty", + }, + "missing settings version": { + machine: &Machine{Goal: GoalState{KubernetesVersion: "1.35.1"}}, + wantErr: "goal settings version is empty", + }, + "complete machine": { + machine: &Machine{Goal: GoalState{KubernetesVersion: "1.35.1", SettingsVersion: "42"}}, + }, + } + + for name, tt := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + + err := tt.machine.Validate() + if tt.wantErr == "" { + if err != nil { + t.Fatalf("Validate() error = %v", err) + } + return + } + if err == nil || !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("Validate() error = %v, want containing %q", err, tt.wantErr) + } + }) + } +} diff --git a/pkg/daemon/repave_reconciler.go b/pkg/daemon/repave_reconciler.go index 8d4f99c5..0b96f167 100644 --- a/pkg/daemon/repave_reconciler.go +++ b/pkg/daemon/repave_reconciler.go @@ -198,6 +198,9 @@ func (r *repaveReconciler) machineSnapshot(ctx context.Context) (machineSnapshot if err != nil { return machineSnapshot{}, err } + if err := machine.Validate(); err != nil { + return machineSnapshot{}, fmt.Errorf("validate AKS machine snapshot: %w", err) + } return machineSnapshot{machine: machine}, nil } diff --git a/pkg/daemon/repave_reconciler_test.go b/pkg/daemon/repave_reconciler_test.go index 633b2eac..f8323568 100644 --- a/pkg/daemon/repave_reconciler_test.go +++ b/pkg/daemon/repave_reconciler_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "log/slog" + "strings" "testing" corev1 "k8s.io/api/core/v1" @@ -73,6 +74,22 @@ func TestRepaveReconcilerStateLoadFailurePatchesFailed(t *testing.T) { } } +func TestRepaveReconcilerRejectsInvalidMachineGoal(t *testing.T) { + t.Parallel() + + machines := &fakeMachineClient{machine: &aksmachine.Machine{Goal: aksmachine.GoalState{KubernetesVersion: "1.34.0"}}} + operator := &fakeNodeOperator{state: &State{AppliedSettingsVersion: "41", AppliedKubernetesVersion: "1.33.0", ActiveMachine: "kube1"}} + repaves := newTestRepaveReconciler(t, machines, fakeClient(), operator) + + err := repaves.reconcileOnce(t.Context()) + if err == nil || !strings.Contains(err.Error(), "validate AKS machine snapshot") { + t.Fatalf("Reconcile error = %v, want invalid machine snapshot", err) + } + if operator.applied { + t.Fatal("ApplyGoalState was called for an invalid machine goal") + } +} + func newTestRepaveReconciler(t *testing.T, machines aksmachine.MachineClient, kubeClient client.Client, operator nodeOperator) *repaveReconciler { t.Helper() repaves, err := newRepaveReconciler(repaveReconcilerOptions{