From 5709c719ea21eebee7271dbba9491ef84117b065 Mon Sep 17 00:00:00 2001 From: wenxuanW Date: Tue, 28 Jul 2026 11:13:33 -0700 Subject: [PATCH 1/4] Apply aks managed labels to flex nodes --- pkg/config/adapter.go | 25 ++++++++++++++++++- pkg/config/adapter_test.go | 49 ++++++++++++++++++++++++++++++++++++++ pkg/config/config.go | 4 ---- 3 files changed, 73 insertions(+), 5 deletions(-) diff --git a/pkg/config/adapter.go b/pkg/config/adapter.go index 86fff893..92d8dafb 100644 --- a/pkg/config/adapter.go +++ b/pkg/config/adapter.go @@ -20,6 +20,13 @@ const ( // aksAADServerID is the Azure AD server application ID for AKS. aksAADServerID = "6dae42f8-4368-4678-94ff-3960e28e3630" + + managedNodeLabel = "kubernetes.azure.com/managed" + agentPoolNodeLabel = "kubernetes.azure.com/agentpool" + modeNodeLabel = "kubernetes.azure.com/mode" + nodePoolTypeNodeLabel = "kubernetes.azure.com/nodepool-type" + flexNodePoolType = "FlexNodes" + userNodeMode = "user" ) // ToAgentConfig converts a FlexNode Config to the shared agent library's @@ -44,7 +51,7 @@ func ToAgentConfig(cfg *Config, machineName string) *agentconfig.AgentConfig { Kubelet: agentconfig.AgentKubeletConfig{ ApiServer: cfg.APIServerURL(), NodeIP: cfg.Node.Kubelet.NodeIP, - Labels: cfg.Node.Labels, + Labels: kubeletNodeLabels(cfg), RegisterWithTaints: cfg.Node.Taints, Configuration: kubeletConfig, }, @@ -162,6 +169,22 @@ func kubeReservedOrDefault(cfg *Config, maxPods int) map[string]string { return defaultKubeReserved(runtime.NumCPU(), hostTotalMemoryMi(), maxPods) } +func kubeletNodeLabels(cfg *Config) map[string]string { + labels := maps.Clone(cfg.Node.Labels) + if labels == nil { + labels = make(map[string]string) + } + + // These labels describe how AKS should treat the registered Kubernetes + // Node. Keep them out of cfg.Node.Labels because that map is also sent to ARM + // as the Machine's custom node labels. + labels[managedNodeLabel] = "false" + labels[agentPoolNodeLabel] = cfg.Azure.TargetAgentPoolName + labels[modeNodeLabel] = userNodeMode + labels[nodePoolTypeNodeLabel] = flexNodePoolType + return labels +} + // ResolveMachineGoalState converts FlexNode config to the shared agent config // and resolves the nspawn machine goal state. Bootstrap and preflight both use // this helper so preflight validates the same sources that bootstrap consumes. diff --git a/pkg/config/adapter_test.go b/pkg/config/adapter_test.go index dacce958..f30fadb1 100644 --- a/pkg/config/adapter_test.go +++ b/pkg/config/adapter_test.go @@ -10,6 +10,55 @@ import ( "github.com/Azure/unbounded/pkg/agent/goalstates" ) +func TestToAgentConfigKubeletLabels(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + requireMachineRegistration bool + wantLabels map[string]string + }{ + { + name: "registration required", + requireMachineRegistration: true, + wantLabels: map[string]string{ + "workload": "edge", + managedNodeLabel: "false", + agentPoolNodeLabel: "flexnode-edge", + modeNodeLabel: userNodeMode, + nodePoolTypeNodeLabel: flexNodePoolType, + }, + }, + { + name: "registration optional", + wantLabels: map[string]string{"workload": "edge"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + wantCustomLabels := map[string]string{"workload": "edge"} + cfg := &Config{ + Azure: AzureConfig{TargetAgentPoolName: "flexnode-edge"}, + Agent: AgentConfig{RequireMachineRegistration: tt.requireMachineRegistration}, + Node: NodeConfig{Labels: map[string]string{"workload": "edge"}}, + } + cfg.setDefaults() + + agentCfg := ToAgentConfig(cfg, "kube1") + + if !maps.Equal(agentCfg.Kubelet.Labels, tt.wantLabels) { + t.Errorf("Kubelet.Labels = %#v, want %#v", agentCfg.Kubelet.Labels, tt.wantLabels) + } + if !maps.Equal(cfg.Node.Labels, wantCustomLabels) { + t.Errorf("Node.Labels mutated to %#v; ARM custom labels must remain %#v", cfg.Node.Labels, wantCustomLabels) + } + }) + } +} + func TestToAgentConfig_BootstrapToken(t *testing.T) { t.Parallel() diff --git a/pkg/config/config.go b/pkg/config/config.go index c4500af8..5a9f731b 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -456,10 +456,6 @@ func (c *Config) setNodeDefaults() { if c.Node.Labels == nil { c.Node.Labels = make(map[string]string) } - // Mark node as unmanaged by cloud controller manager by default, otherwise ccm will delete this node if node is not ready - // doc: https://cloud-provider-azure.sigs.k8s.io/topics/cross-resource-group-nodes/#unmanaged-nodes - c.Node.Labels["kubernetes.azure.com/managed"] = "false" - // Set default kubelet configuration if not provided if c.Node.Kubelet.Verbosity == 0 { c.Node.Kubelet.Verbosity = 2 From 269be0cbc2e5f7f5658b98adf50dcf2efca0fea9 Mon Sep 17 00:00:00 2001 From: wenxuanW Date: Wed, 29 Jul 2026 11:41:06 -0700 Subject: [PATCH 2/4] Add goal ack flow --- docs/design.md | 2 +- docs/design/in-cluster-machine.md | 12 +-- hack/e2e/lib/upgrade-drift.sh | 18 +++- pkg/aksmachine/client_armapi.go | 11 +- pkg/aksmachine/client_armapi_test.go | 8 +- pkg/aksmachine/client_incluster.go | 24 +---- pkg/aksmachine/client_incluster_test.go | 11 +- pkg/aksmachine/ensure.go | 50 ++------- pkg/aksmachine/ensure_test.go | 54 +++++----- pkg/aksmachine/types.go | 18 +++- pkg/aksmachine/types_test.go | 45 +++++++- pkg/cmd/start/start.go | 2 +- pkg/config/adapter_test.go | 38 ++++--- pkg/daemon/goalstate.go | 33 ++++++ pkg/daemon/goalstate_test.go | 44 ++++++++ pkg/daemon/nodeoperator.go | 41 ++++---- pkg/daemon/nodeoperator_test.go | 51 ++++++++- pkg/daemon/reconcile_test.go | 133 +++++++++++++++++++++++- pkg/daemon/repave_reconciler.go | 95 ++++++++++++++++- pkg/daemon/repave_reconciler_test.go | 124 +++++++++++++++++++--- pkg/daemon/state.go | 48 ++++++--- pkg/daemon/state_test.go | 116 +++++++++++++++++---- 22 files changed, 768 insertions(+), 210 deletions(-) create mode 100644 pkg/daemon/goalstate.go create mode 100644 pkg/daemon/goalstate_test.go diff --git a/docs/design.md b/docs/design.md index da72834b..6b4ec9d9 100644 --- a/docs/design.md +++ b/docs/design.md @@ -152,7 +152,7 @@ See [AKS RP And Flex Node Agent Interaction](design/agent-and-aks.md) for the de ## State And Idempotency -The agent persists local daemon state so it can recover after restart, reboot, or partial failure. Persisted state includes the applied Kubernetes/settings version and active nspawn machine side. +The agent persists local daemon state so it can recover after restart, reboot, or partial failure. Persisted state includes the current and previous applied goals and the active nspawn machine side. The current state model separates desired state, applied state, and runtime discovery: diff --git a/docs/design/in-cluster-machine.md b/docs/design/in-cluster-machine.md index 3d46a2ac..20df96f8 100644 --- a/docs/design/in-cluster-machine.md +++ b/docs/design/in-cluster-machine.md @@ -27,18 +27,18 @@ Status updates use a separate patch model because the agent operation status is ## Bootstrap flow -The local bootstrap configuration is authoritative while `aks-flex-node start` is running: +The local bootstrap configuration seeds a Machine when one does not already exist. Once the endpoint returns a Machine, its complete goal is authoritative for bootstrap: 1. `NewMachineClient` selects the in-cluster backend without a supplied Kubernetes REST config. 2. The client builds a REST config from the bootstrap token or configured exec credential. 3. `EnsureMachine` reads the machine through the Kubernetes service proxy. 4. If the machine is absent, the client sends a PUT using the local bootstrap goal. -5. If its Kubernetes version differs, the client sends a PUT that overwrites the remote goal with the local version. -6. If its Kubernetes version already matches, local bootstrap settings remain authoritative; remote settings other than the ETag do not replace them. -7. The returned ETag becomes the reconciliation baseline for the locally applied goal. -8. The daemon state is seeded from that ETag before host or nspawn state is mutated. A later ETag change is treated as a new remote goal. +5. Whether read or created, the returned Machine is validated and its complete goal replaces the local bootstrap goal. This includes the Kubernetes version, node settings, and ETag-backed settings version. +6. The daemon state is seeded from the accepted Machine goal before host or nspawn state is mutated. A later ETag change is treated as a new remote goal. -The ConfigMap-backed controller is read-only: it accepts mutation requests but returns the pre-created machine. Its fixture must therefore already match the local bootstrap version. When machine registration is required, a mismatch fails bootstrap before host mutation. +The ConfigMap-backed controller is read-only: it accepts mutation requests but returns the pre-created Machine. The agent adopts that returned goal even when it differs from local bootstrap configuration. The returned Machine must contain a valid Kubernetes version and settings version. + +When machine registration is required, a read, create, or validation failure stops bootstrap before host mutation. When registration is optional, the same failure is logged and bootstrap continues from the local goal; its settings version remains empty until a valid Machine goal is later observed. ## Daemon flow diff --git a/hack/e2e/lib/upgrade-drift.sh b/hack/e2e/lib/upgrade-drift.sh index dd7686ab..e1e6698f 100644 --- a/hack/e2e/lib/upgrade-drift.sh +++ b/hack/e2e/lib/upgrade-drift.sh @@ -56,8 +56,22 @@ _remote_active_machine_snapshot() { remote_exec "${vm_ip}" 'bash -s' <<'REMOTE' set +e state_file="/etc/aks-flex-node/daemon-state.json" -machine="$(sudo sed -n 's/.*"activeMachine"[[:space:]]*:[[:space:]]*"\([^"]*\)".*/\1/p' "${state_file}" 2>/dev/null)" -applied_settings_version="$(sudo sed -n 's/.*"appliedSettingsVersion"[[:space:]]*:[[:space:]]*"\([^"]*\)".*/\1/p' "${state_file}" 2>/dev/null)" +state_snapshot="$(sudo python3 - "${state_file}" <<'PY' 2>/dev/null +import json +import sys + +try: + with open(sys.argv[1], encoding="utf-8") as state_file: + state = json.load(state_file) + machine = state.get("activeMachine", "") + applied_goal = state.get("appliedGoal") or {} + settings_version = applied_goal.get("settingsVersion", "") + print(f"{machine}|{settings_version}") +except (OSError, json.JSONDecodeError, AttributeError): + print("|") +PY +)" +IFS='|' read -r machine applied_settings_version <<<"${state_snapshot}" state="" version="" if [[ -n "${machine}" ]]; then diff --git a/pkg/aksmachine/client_armapi.go b/pkg/aksmachine/client_armapi.go index c91cb0e9..b5676f30 100644 --- a/pkg/aksmachine/client_armapi.go +++ b/pkg/aksmachine/client_armapi.go @@ -77,7 +77,7 @@ 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 := machineFromARM(resp.Machine) result.ID = c.machineID.String() result.Name = c.machineID.Name return result, nil @@ -103,7 +103,7 @@ 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 := machineFromARM(resp.Machine) result.ID = c.machineID.String() result.Name = c.machineID.Name return result, nil @@ -256,8 +256,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) *Machine { + result := &Machine{} if machine.ID != nil { result.ID = *machine.ID } @@ -298,9 +298,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..87eaf9ac 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"}) + }) if machine.ID != "machine-id" || machine.Name != "node1" { t.Fatalf("machine identity = %#v", machine) @@ -408,13 +408,13 @@ 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.SettingsVersion != currentVersion { - t.Fatalf("SettingsVersion = %q, want %q", machine.Goal.SettingsVersion, currentVersion) + if machine.Goal.SettingsVersion != "" { + t.Fatalf("SettingsVersion = %q, want empty without an ETag", machine.Goal.SettingsVersion) } } diff --git a/pkg/aksmachine/client_incluster.go b/pkg/aksmachine/client_incluster.go index 27314eb1..d11885a5 100644 --- a/pkg/aksmachine/client_incluster.go +++ b/pkg/aksmachine/client_incluster.go @@ -130,7 +130,7 @@ func (c *clusterEndpointClient) Create(ctx context.Context, desired GoalState) ( if resp.StatusCode == http.StatusForbidden || resp.StatusCode == http.StatusMethodNotAllowed || resp.StatusCode == http.StatusNoContent { c.logger.Debug("cluster endpoint did not apply machine create request; verifying pre-created machine", "status", resp.Status) - return c.adoptExistingMachine(ctx, desired) + return c.adoptExistingMachine(ctx) } if resp.StatusCode < 200 || resp.StatusCode > 299 { return nil, clusterEndpointHTTPError("create machine through cluster endpoint", requestURL, resp) @@ -140,7 +140,7 @@ func (c *clusterEndpointClient) Create(ctx context.Context, desired GoalState) ( return nil, fmt.Errorf("read cluster endpoint machine create response: %w", err) } if strings.TrimSpace(string(data)) == "" { - return c.adoptExistingMachine(ctx, desired) + return c.adoptExistingMachine(ctx) } machine, err := machineFromEndpointJSON(data) if err != nil { @@ -152,33 +152,17 @@ func (c *clusterEndpointClient) Create(ctx context.Context, desired GoalState) ( if machine.Name == "" { machine.Name = c.nodeName } - if err := validateAdoptedMachine(machine, desired); err != nil { - return nil, err - } return machine, nil } -func (c *clusterEndpointClient) adoptExistingMachine(ctx context.Context, desired GoalState) (*Machine, error) { +func (c *clusterEndpointClient) adoptExistingMachine(ctx context.Context) (*Machine, error) { machine, err := c.Get(ctx) if err != nil { return nil, fmt.Errorf("verify pre-created machine from cluster endpoint: %w", err) } - if err := validateAdoptedMachine(machine, desired); err != nil { - return nil, err - } return machine, nil } -func validateAdoptedMachine(machine *Machine, desired GoalState) error { - if machine == nil { - return fmt.Errorf("cluster endpoint returned nil machine") - } - 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) - } - return nil -} - func (c *clusterEndpointClient) Get(ctx context.Context) (*Machine, error) { requestURL := c.machineURL(c.nodeName) req, err := http.NewRequestWithContext(ctx, http.MethodGet, requestURL.String(), nil) @@ -279,7 +263,7 @@ 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 + return machineFromARM(armMachine), 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..e8e6dfe3 100644 --- a/pkg/aksmachine/client_incluster_test.go +++ b/pkg/aksmachine/client_incluster_test.go @@ -134,7 +134,7 @@ func TestClusterEndpointCreateSendsMutation(t *testing.T) { } } -func TestClusterEndpointCreateVerifiesPrecreatedMachine(t *testing.T) { +func TestClusterEndpointCreateAdoptsReturnedMachine(t *testing.T) { t.Parallel() server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { @@ -144,9 +144,12 @@ func TestClusterEndpointCreateVerifiesPrecreatedMachine(t *testing.T) { defer server.Close() client := newTestClusterEndpointClient(t, server.URL, "node1") - _, err := client.Create(context.Background(), GoalState{KubernetesVersion: "1.35.0", SettingsVersion: "42"}) - if err == nil || !strings.Contains(err.Error(), "Kubernetes version") { - t.Fatalf("Create() error = %v, want version mismatch", err) + machine, err := client.Create(context.Background(), GoalState{KubernetesVersion: "1.35.0", SettingsVersion: "local"}) + if err != nil { + t.Fatalf("Create() error = %v", err) + } + if machine.Goal.KubernetesVersion != "1.34.0" || machine.Goal.SettingsVersion != "42" { + t.Fatalf("Create() machine goal = %#v, want returned machine goal", machine.Goal) } } diff --git a/pkg/aksmachine/ensure.go b/pkg/aksmachine/ensure.go index cdd3b336..9749df4d 100644 --- a/pkg/aksmachine/ensure.go +++ b/pkg/aksmachine/ensure.go @@ -17,10 +17,8 @@ type ensureMachineTask struct { } // EnsureMachine returns a task that ensures this machine is registered in AKS. -// Local configuration remains authoritative during bootstrap. When the remote -// Kubernetes version already matches, the task adopts only the remote ETag as -// the reconciliation baseline; other remote settings do not replace the local -// goal. Subsequent ETag changes are handled by the daemon as new remote goals. +// Local bootstrap configuration seeds a new Machine. Once a Machine exists, its +// complete goal is authoritative for Node creation and later reconciliation. func EnsureMachine(machines MachineClient, goal *GoalState, require bool, logger *slog.Logger) phases.Task { return &ensureMachineTask{machines: machines, goal: goal, require: require, logger: logger} } @@ -30,25 +28,8 @@ 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") - } - - remoteVersion := "" - if machine != nil { - remoteVersion = machine.Goal.KubernetesVersion - } - t.logger.Info( - "updating registered machine from local bootstrap config", - "remoteKubernetesVersion", remoteVersion, - "localKubernetesVersion", t.goal.KubernetesVersion, - ) - machine, err = t.machines.Create(ctx, *t.goal) - if err != nil { - return t.handleError("update machine", err) - } - return t.adoptSettingsVersion(machine, "update machine") + t.logger.Info("ARM machine already registered, adopting remote goal") + return t.adoptGoal(machine, "get machine") } var notFound *NotFoundError @@ -59,26 +40,15 @@ func (t *ensureMachineTask) Do(ctx context.Context) error { if err != nil { return t.handleError("create machine", err) } - return t.adoptSettingsVersion(machine, "create machine") + return t.adoptGoal(machine, "create machine") } -func (t *ensureMachineTask) adoptSettingsVersion(machine *Machine, operation string) error { - if machine == nil { - return t.handleError(operation, fmt.Errorf("AKS returned a nil machine")) - } - 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 machine.Goal.SettingsVersion != "" { - t.goal.SettingsVersion = machine.Goal.SettingsVersion +func (t *ensureMachineTask) adoptGoal(machine *Machine, operation string) error { + if err := machine.Validate(); err != nil { + return t.handleError(operation, fmt.Errorf("AKS returned an invalid machine: %w", err)) } + + *t.goal = machine.Goal return nil } diff --git a/pkg/aksmachine/ensure_test.go b/pkg/aksmachine/ensure_test.go index 5e7f09bb..f99d4338 100644 --- a/pkg/aksmachine/ensure_test.go +++ b/pkg/aksmachine/ensure_test.go @@ -102,7 +102,7 @@ func TestEnsureMachineCreatesAndAdoptsSettingsVersion(t *testing.T) { } } -func TestEnsureMachineAdoptsExistingSettingsVersionWithoutReplacingLocalGoal(t *testing.T) { +func TestEnsureMachineAdoptsExistingGoal(t *testing.T) { t.Parallel() goal := GoalState{ @@ -135,26 +135,27 @@ func TestEnsureMachineAdoptsExistingSettingsVersionWithoutReplacingLocalGoal(t * if client.createCalls != 0 { t.Fatalf("Create() calls = %d, want 0", client.createCalls) } - if goal.SettingsVersion != "etag-42" { - t.Fatalf("SettingsVersion = %q, want etag-42", goal.SettingsVersion) + if goal.SettingsVersion != "etag-42" || goal.KubernetesVersion != "1.35.1" { + t.Fatalf("version goal = %#v", goal) } - if goal.MaxPods != 30 || goal.NodeLabels["source"] != "local" || goal.NodeTaints[0] != "local=true:NoSchedule" { - t.Fatalf("local goal was replaced by remote settings: %#v", goal) + if goal.MaxPods != 110 || goal.NodeLabels["source"] != "remote" || goal.NodeTaints[0] != "remote=true:NoSchedule" { + t.Fatalf("remote goal was not adopted: %#v", goal) } - if goal.KubeletConfig.ImageGCHighThreshold != 85 || goal.KubeletConfig.ImageGCLowThreshold != 80 { - t.Fatalf("local kubelet config was replaced by remote settings: %#v", goal.KubeletConfig) + if goal.KubeletConfig.ImageGCHighThreshold != 70 || goal.KubeletConfig.ImageGCLowThreshold != 60 { + t.Fatalf("remote kubelet config was not adopted: %#v", goal.KubeletConfig) } } -func TestEnsureMachineUpdatesMismatchedVersion(t *testing.T) { +func TestEnsureMachineAdoptsExistingMismatchedVersion(t *testing.T) { t.Parallel() goal := GoalState{KubernetesVersion: "1.35.1", SettingsVersion: "1.35.1"} client := &ensureMachineClient{ - machine: &Machine{Goal: GoalState{KubernetesVersion: "1.34.0", SettingsVersion: "etag-old"}}, - createResult: &Machine{Goal: GoalState{ - KubernetesVersion: "1.35.1", - SettingsVersion: "etag-new", + machine: &Machine{Goal: GoalState{ + KubernetesVersion: "1.34.0", + SettingsVersion: "etag-remote", + NodeLabels: map[string]string{"source": "remote"}, + NodeTaints: []string{"remote=true:NoSchedule"}, }}, } task := EnsureMachine(client, &goal, true, slog.New(slog.NewTextHandler(io.Discard, nil))) @@ -162,36 +163,29 @@ func TestEnsureMachineUpdatesMismatchedVersion(t *testing.T) { if err := task.Do(context.Background()); err != nil { t.Fatalf("Do() error = %v", err) } - if client.createCalls != 1 { - t.Fatalf("Create() calls = %d, want 1", client.createCalls) + if client.createCalls != 0 { + t.Fatalf("Create() calls = %d, want 0", client.createCalls) } - if client.createdGoal.KubernetesVersion != "1.35.1" { - t.Fatalf("Create() goal = %#v", client.createdGoal) + if goal.KubernetesVersion != "1.34.0" || goal.SettingsVersion != "etag-remote" { + t.Fatalf("version goal = %#v", goal) } - if goal.SettingsVersion != "etag-new" { - t.Fatalf("SettingsVersion = %q, want etag-new", goal.SettingsVersion) + if goal.NodeLabels["source"] != "remote" || goal.NodeTaints[0] != "remote=true:NoSchedule" { + t.Fatalf("remote goal was not adopted: %#v", goal) } } -func TestEnsureMachineRejectsUnchangedRemoteVersionAfterUpdate(t *testing.T) { +func TestEnsureMachineRejectsMissingSettingsVersion(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{ - KubernetesVersion: "1.34.0", - SettingsVersion: "etag-old", - }}, + machine: &Machine{Goal: GoalState{KubernetesVersion: "1.35.1"}}, } task := EnsureMachine(client, &goal, true, slog.New(slog.NewTextHandler(io.Discard, nil))) err := task.Do(context.Background()) - 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 err == nil || !strings.Contains(err.Error(), "goal settings version is empty") { + t.Fatalf("Do() error = %v, want missing settings version", err) } } diff --git a/pkg/aksmachine/types.go b/pkg/aksmachine/types.go index 89659fe6..243e4dd0 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 complete 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/cmd/start/start.go b/pkg/cmd/start/start.go index 23e112c5..deb220e1 100644 --- a/pkg/cmd/start/start.go +++ b/pkg/cmd/start/start.go @@ -76,7 +76,7 @@ func runStart(ctx context.Context, cfg *config.Config, logger *slog.Logger) erro return err } - _, gs, containerImageArchives, err := config.ResolveMachineGoalState(ctx, logger, cfg, machineName) + _, gs, containerImageArchives, err := daemon.ResolveMachineGoalState(ctx, logger, cfg, machineName, &goal) if err != nil { return fmt.Errorf("bootstrap failed to resolve goal state: %w", err) } diff --git a/pkg/config/adapter_test.go b/pkg/config/adapter_test.go index f30fadb1..b478a56b 100644 --- a/pkg/config/adapter_test.go +++ b/pkg/config/adapter_test.go @@ -16,22 +16,13 @@ func TestToAgentConfigKubeletLabels(t *testing.T) { tests := []struct { name string requireMachineRegistration bool - wantLabels map[string]string }{ { name: "registration required", requireMachineRegistration: true, - wantLabels: map[string]string{ - "workload": "edge", - managedNodeLabel: "false", - agentPoolNodeLabel: "flexnode-edge", - modeNodeLabel: userNodeMode, - nodePoolTypeNodeLabel: flexNodePoolType, - }, }, { - name: "registration optional", - wantLabels: map[string]string{"workload": "edge"}, + name: "registration optional", }, } @@ -40,17 +31,24 @@ func TestToAgentConfigKubeletLabels(t *testing.T) { t.Parallel() wantCustomLabels := map[string]string{"workload": "edge"} + wantLabels := map[string]string{ + "workload": "edge", + managedNodeLabel: "false", + agentPoolNodeLabel: "flexnode-edge", + modeNodeLabel: userNodeMode, + nodePoolTypeNodeLabel: flexNodePoolType, + } cfg := &Config{ Azure: AzureConfig{TargetAgentPoolName: "flexnode-edge"}, - Agent: AgentConfig{RequireMachineRegistration: tt.requireMachineRegistration}, + Agent: AgentConfig{RequireMachineRegistration: &tt.requireMachineRegistration}, Node: NodeConfig{Labels: map[string]string{"workload": "edge"}}, } cfg.setDefaults() agentCfg := ToAgentConfig(cfg, "kube1") - if !maps.Equal(agentCfg.Kubelet.Labels, tt.wantLabels) { - t.Errorf("Kubelet.Labels = %#v, want %#v", agentCfg.Kubelet.Labels, tt.wantLabels) + if !maps.Equal(agentCfg.Kubelet.Labels, wantLabels) { + t.Errorf("Kubelet.Labels = %#v, want %#v", agentCfg.Kubelet.Labels, wantLabels) } if !maps.Equal(cfg.Node.Labels, wantCustomLabels) { t.Errorf("Node.Labels mutated to %#v; ARM custom labels must remain %#v", cfg.Node.Labels, wantCustomLabels) @@ -64,7 +62,8 @@ func TestToAgentConfig_BootstrapToken(t *testing.T) { cfg := &Config{ Azure: AzureConfig{ - BootstrapToken: &BootstrapTokenConfig{Token: "abcdef.0123456789abcdef"}, + BootstrapToken: &BootstrapTokenConfig{Token: "abcdef.0123456789abcdef"}, + TargetAgentPoolName: "flexnode-edge", }, Components: ComponentsConfig{Kubernetes: "1.30.0"}, Networking: NetworkingConfig{DNSServiceIP: "10.0.0.10"}, @@ -117,8 +116,15 @@ func TestToAgentConfig_BootstrapToken(t *testing.T) { if ac.Kubelet.Auth.ExecCredential != nil { t.Fatalf("Kubelet.Auth.ExecCredential should be nil for bootstrap token auth") } - if len(ac.Kubelet.Labels) != 1 || ac.Kubelet.Labels["env"] != "test" { - t.Fatalf("Kubelet.Labels=%v, want map[env:test]", ac.Kubelet.Labels) + wantLabels := map[string]string{ + "env": "test", + managedNodeLabel: "false", + agentPoolNodeLabel: "flexnode-edge", + modeNodeLabel: userNodeMode, + nodePoolTypeNodeLabel: flexNodePoolType, + } + if !maps.Equal(ac.Kubelet.Labels, wantLabels) { + t.Fatalf("Kubelet.Labels=%v, want %v", ac.Kubelet.Labels, wantLabels) } if len(ac.Kubelet.RegisterWithTaints) != 1 || ac.Kubelet.RegisterWithTaints[0] != "dedicated=infra:NoSchedule" { t.Fatalf("Kubelet.RegisterWithTaints=%v, want [dedicated=infra:NoSchedule]", ac.Kubelet.RegisterWithTaints) diff --git a/pkg/daemon/goalstate.go b/pkg/daemon/goalstate.go new file mode 100644 index 00000000..cc9d9502 --- /dev/null +++ b/pkg/daemon/goalstate.go @@ -0,0 +1,33 @@ +package daemon + +import ( + "context" + "log/slog" + "maps" + "slices" + + agentconfig "github.com/Azure/unbounded/pkg/agent/config" + "github.com/Azure/unbounded/pkg/agent/goalstates" + + "github.com/Azure/AKSFlexNode/pkg/aksmachine" + "github.com/Azure/AKSFlexNode/pkg/config" +) + +// ResolveMachineGoalState resolves an nspawn goal. This final nspawn goal remains a hybrid: +// only Kubernetes version, labels, and taints are overlaid from GoalState; +// networking, credentials, runtime versions, images, mounts, etc. still come from config. +func ResolveMachineGoalState( + ctx context.Context, + log *slog.Logger, + cfg *config.Config, + machineName string, + goal *aksmachine.GoalState, +) (*agentconfig.AgentConfig, *goalstates.MachineGoalState, *goalstates.ContainerImageArchiveStaging, error) { + resolvedConfig := cfg.DeepCopy() + if goal != nil { + resolvedConfig.Components.Kubernetes = goal.KubernetesVersion + resolvedConfig.Node.Labels = maps.Clone(goal.NodeLabels) + resolvedConfig.Node.Taints = slices.Clone(goal.NodeTaints) + } + return config.ResolveMachineGoalState(ctx, log, resolvedConfig, machineName) +} diff --git a/pkg/daemon/goalstate_test.go b/pkg/daemon/goalstate_test.go new file mode 100644 index 00000000..ce974019 --- /dev/null +++ b/pkg/daemon/goalstate_test.go @@ -0,0 +1,44 @@ +package daemon + +import ( + "log/slog" + "testing" + + "github.com/Azure/AKSFlexNode/pkg/aksmachine" + "github.com/Azure/AKSFlexNode/pkg/config" +) + +func TestResolveMachineGoalStateUsesMachineGoal(t *testing.T) { + t.Parallel() + + cfg := &config.Config{ + Azure: config.AzureConfig{TargetAgentPoolName: "flexnode-edge"}, + Components: config.ComponentsConfig{Kubernetes: "1.34.0"}, + Node: config.NodeConfig{ + Labels: map[string]string{"source": "bootstrap"}, + Taints: []string{"bootstrap=true:NoSchedule"}, + }, + } + goal := &aksmachine.GoalState{ + KubernetesVersion: "1.35.1", + NodeLabels: map[string]string{"source": "machine"}, + NodeTaints: []string{"machine=true:NoExecute"}, + } + + agentCfg, _, _, err := ResolveMachineGoalState(t.Context(), slog.Default(), cfg, "kube1", goal) + if err != nil { + t.Fatalf("ResolveMachineGoalState: %v", err) + } + if agentCfg.Cluster.Version != "1.35.1" { + t.Fatalf("Cluster.Version = %q, want 1.35.1", agentCfg.Cluster.Version) + } + if agentCfg.Kubelet.Labels["source"] != "machine" { + t.Fatalf("Kubelet.Labels = %#v, want Machine goal labels", agentCfg.Kubelet.Labels) + } + if len(agentCfg.Kubelet.RegisterWithTaints) != 1 || agentCfg.Kubelet.RegisterWithTaints[0] != "machine=true:NoExecute" { + t.Fatalf("Kubelet.RegisterWithTaints = %#v, want Machine goal taints", agentCfg.Kubelet.RegisterWithTaints) + } + if cfg.Components.Kubernetes != "1.34.0" || cfg.Node.Labels["source"] != "bootstrap" || cfg.Node.Taints[0] != "bootstrap=true:NoSchedule" { + t.Fatalf("bootstrap config was mutated: %#v", cfg) + } +} diff --git a/pkg/daemon/nodeoperator.go b/pkg/daemon/nodeoperator.go index 0a082cc7..ba847d60 100644 --- a/pkg/daemon/nodeoperator.go +++ b/pkg/daemon/nodeoperator.go @@ -23,6 +23,7 @@ type activeMachine struct { type nodeOperator interface { LoadState(ctx context.Context) (*State, error) ApplyGoalState(ctx context.Context, log *slog.Logger, goal aksmachine.GoalState) (*State, error) + AcknowledgeGoalState(ctx context.Context, goal aksmachine.GoalState) (*State, error) RestartNode(ctx context.Context, log *slog.Logger) error // ResetNode removes nspawn node runtime and persisted daemon state but must // not stop this daemon process. The controller publishes lifecycle completion @@ -38,11 +39,7 @@ func (o *nspawnNodeOperator) RestartNode(ctx context.Context, log *slog.Logger) return err } - cfg := o.cfg.DeepCopy() - if active.State.AppliedKubernetesVersion != "" { - cfg.Components.Kubernetes = active.State.AppliedKubernetesVersion - } - _, gs, containerImageArchives, err := config.ResolveMachineGoalState(ctx, log, cfg, active.Name) + _, gs, containerImageArchives, err := ResolveMachineGoalState(ctx, log, o.cfg, active.Name, active.State.AppliedGoal) if err != nil { return fmt.Errorf("resolve goal state for node restart: %w", err) } @@ -52,7 +49,7 @@ func (o *nspawnNodeOperator) RestartNode(ctx context.Context, log *slog.Logger) nodestop.StopNode(log, active.Name), nodestart.StartNode(log, gs.NodeStart), nodestart.WaitForKubelet(log, active.Name), - npd.Start(log, cfg, gs.NodeStart), + npd.Start(log, o.cfg, gs.NodeStart), ).Do(ctx) } @@ -96,14 +93,14 @@ func (o *nspawnNodeOperator) ApplyGoalState(ctx context.Context, log *slog.Logge "oldMachine", oldMachine, "newMachine", newMachine, "settingsVersion", goal.SettingsVersion, - "kubernetesVersion", cfg.Components.Kubernetes, + "kubernetesVersion", goal.KubernetesVersion, ) - _, gs, containerImageArchives, err := config.ResolveMachineGoalState(ctx, log, cfg, newMachine) + _, gs, containerImageArchives, err := ResolveMachineGoalState(ctx, log, cfg, newMachine, &goal) if err != nil { return nil, fmt.Errorf("resolve goal state for repave: %w", err) } - newState := nextAppliedState(active.State, goal, &activeMachine{Name: newMachine}) + newState := nextAppliedState(goal, &activeMachine{Name: newMachine, State: active.State}) tasks := phases.Serial(log, nodestop.StopNode(log, oldMachine), @@ -156,6 +153,18 @@ func (o *nspawnNodeOperator) configForGoalState(ctx context.Context, log *slog.L return cfg, nil } +func (o *nspawnNodeOperator) AcknowledgeGoalState(ctx context.Context, goal aksmachine.GoalState) (*State, error) { + active, err := o.findActiveMachine(ctx) + if err != nil { + return nil, err + } + newState := nextAppliedState(goal, active) + if err := o.state.Save(ctx, newState); err != nil { + return nil, fmt.Errorf("save acknowledged machine goal state: %w", err) + } + return newState, nil +} + func (o *nspawnNodeOperator) ResetNode(ctx context.Context, log *slog.Logger) error { return phases.ExecuteTask(ctx, log, ResetNode(log)) } @@ -164,19 +173,15 @@ func (o *nspawnNodeOperator) StopDaemon(ctx context.Context, log *slog.Logger) e return phases.ExecuteTask(ctx, log, UninstallService(log)) } -func nextAppliedState(current *State, goal aksmachine.GoalState, active *activeMachine) *State { +func nextAppliedState(goal aksmachine.GoalState, active *activeMachine) *State { next := &State{ - AppliedSettingsVersion: goal.SettingsVersion, - AppliedKubernetesVersion: goal.KubernetesVersion, - PreviousSettingsVersion: "", - PreviousKubernetesVersion: "", - } - if current != nil { - next.PreviousSettingsVersion = current.AppliedSettingsVersion - next.PreviousKubernetesVersion = current.AppliedKubernetesVersion + AppliedGoal: cloneGoalState(goal), } if active != nil { next.ActiveMachine = active.Name + if active.State != nil && active.State.AppliedGoal != nil { + next.PreviousAppliedGoal = cloneGoalState(*active.State.AppliedGoal) + } } return next } diff --git a/pkg/daemon/nodeoperator_test.go b/pkg/daemon/nodeoperator_test.go index c8f3f8a8..4b15ab2d 100644 --- a/pkg/daemon/nodeoperator_test.go +++ b/pkg/daemon/nodeoperator_test.go @@ -24,22 +24,26 @@ func TestFindActiveMachine(t *testing.T) { wantErr bool }{ "kube1": { - state: &State{ActiveMachine: goalstates.NSpawnMachineKube1}, + state: &State{AppliedGoal: &aksmachine.GoalState{KubernetesVersion: "1.34.0"}, ActiveMachine: goalstates.NSpawnMachineKube1}, want: goalstates.NSpawnMachineKube1, }, "kube2": { - state: &State{ActiveMachine: goalstates.NSpawnMachineKube2}, + state: &State{AppliedGoal: &aksmachine.GoalState{KubernetesVersion: "1.34.0"}, ActiveMachine: goalstates.NSpawnMachineKube2}, want: goalstates.NSpawnMachineKube2, }, "missing state": { wantErr: true, }, "missing active machine": { - state: &State{}, + state: &State{AppliedGoal: &aksmachine.GoalState{KubernetesVersion: "1.34.0"}}, + wantErr: true, + }, + "missing applied goal": { + state: &State{ActiveMachine: goalstates.NSpawnMachineKube1}, wantErr: true, }, "invalid active machine": { - state: &State{ActiveMachine: "kube3"}, + state: &State{AppliedGoal: &aksmachine.GoalState{KubernetesVersion: "1.34.0"}, ActiveMachine: "kube3"}, wantErr: true, }, } @@ -272,6 +276,42 @@ func discardLogger() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) } +func TestAcknowledgeGoalState(t *testing.T) { + t.Parallel() + + store := &testStateStore{state: &State{ + AppliedGoal: &aksmachine.GoalState{ + KubernetesVersion: "1.34.0", + SettingsVersion: "41", + }, + ActiveMachine: goalstates.NSpawnMachineKube1, + }} + operator := &nspawnNodeOperator{state: store} + goal := aksmachine.GoalState{ + KubernetesVersion: "1.34.0", + SettingsVersion: "42", + NodeLabels: map[string]string{"workload": "flex"}, + NodeTaints: []string{"dedicated=flex:NoSchedule"}, + } + + got, err := operator.AcknowledgeGoalState(t.Context(), goal) + if err != nil { + t.Fatalf("AcknowledgeGoalState: %v", err) + } + if got.AppliedGoal == nil || got.AppliedGoal.SettingsVersion != "42" || got.ActiveMachine != goalstates.NSpawnMachineKube1 { + t.Fatalf("state = %#v", got) + } + if got.PreviousAppliedGoal == nil || got.PreviousAppliedGoal.SettingsVersion != "41" { + t.Fatalf("PreviousAppliedGoal = %#v, want settings version 41", got.PreviousAppliedGoal) + } + if got.AppliedGoal == nil || got.AppliedGoal.NodeLabels["workload"] != "flex" || len(got.AppliedGoal.NodeTaints) != 1 { + t.Fatalf("AppliedGoal = %#v", got.AppliedGoal) + } + if store.state != got { + t.Fatal("acknowledged state was not persisted") + } +} + type testStateStore struct { state *State } @@ -280,7 +320,8 @@ func (s *testStateStore) Load(context.Context) (*State, error) { return s.state, nil } -func (s *testStateStore) Save(context.Context, *State) error { +func (s *testStateStore) Save(_ context.Context, state *State) error { + s.state = state return nil } diff --git a/pkg/daemon/reconcile_test.go b/pkg/daemon/reconcile_test.go index 43090acc..850d0900 100644 --- a/pkg/daemon/reconcile_test.go +++ b/pkg/daemon/reconcile_test.go @@ -4,6 +4,7 @@ import ( "testing" corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "github.com/Azure/AKSFlexNode/pkg/aksmachine" ) @@ -13,11 +14,37 @@ func TestDecide(t *testing.T) { goal := aksmachine.GoalState{KubernetesVersion: "1.34.0", SettingsVersion: "42"} machine := machineSnapshot{machine: &aksmachine.Machine{Goal: goal}} - applied := &State{AppliedSettingsVersion: "42", AppliedKubernetesVersion: "1.34.0"} - stale := &State{AppliedSettingsVersion: "41", AppliedKubernetesVersion: "1.33.0"} + appliedGoal := goal + applied := &State{AppliedGoal: &appliedGoal} + staleGoal := aksmachine.GoalState{KubernetesVersion: "1.33.0", SettingsVersion: "41"} + stale := &State{AppliedGoal: &staleGoal} node := nodeSnapshot{node: &corev1.Node{}} missingNode := nodeSnapshot{} deleteNode := nodeSnapshot{node: &corev1.Node{Spec: corev1.NodeSpec{Taints: []corev1.Taint{deletionTaint()}}}} + inPlaceAppliedGoal := aksmachine.GoalState{ + KubernetesVersion: "1.34.0", + SettingsVersion: "41", + NodeLabels: map[string]string{"workload": "old", "removed": "true"}, + NodeTaints: []string{"dedicated=old:NoSchedule", "removed=true:NoExecute"}, + } + inPlaceGoal := aksmachine.GoalState{ + KubernetesVersion: "1.34.0", + SettingsVersion: "42", + NodeLabels: map[string]string{"workload": "new"}, + NodeTaints: []string{"dedicated=new:NoSchedule"}, + } + inPlaceMachine := machineSnapshot{machine: &aksmachine.Machine{Goal: inPlaceGoal}} + inPlaceState := &State{AppliedGoal: &inPlaceAppliedGoal} + inPlaceNode := nodeSnapshot{node: &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{ + "workload": "new", + "unrelated.example/label": "preserved", + }}, + Spec: corev1.NodeSpec{Taints: []corev1.Taint{ + {Key: "dedicated", Value: "new", Effect: corev1.TaintEffectNoSchedule}, + {Key: "unrelated.example/taint", Effect: corev1.TaintEffectNoExecute}, + }}, + }} tests := map[string]struct { machine machineSnapshot @@ -73,6 +100,12 @@ func TestDecide(t *testing.T) { state: nil, want: decisionWaitForNodeSignal, }, + "present node acknowledges RP-applied labels and taints": { + machine: inPlaceMachine, + node: inPlaceNode, + state: inPlaceState, + want: decisionAcknowledgeGoalState, + }, } for name, tt := range tests { @@ -87,6 +120,102 @@ func TestDecide(t *testing.T) { } } +func TestGoalReflectedOnNode(t *testing.T) { + t.Parallel() + + appliedGoal := aksmachine.GoalState{ + KubernetesVersion: "1.34.0", + SettingsVersion: "41", + MaxPods: 110, + NodeLabels: map[string]string{"workload": "old", "removed": "true"}, + NodeTaints: []string{"dedicated=old:NoSchedule", "removed=true:NoExecute"}, + KubeletConfig: aksmachine.KubeletConfig{ImageGCHighThreshold: 85, ImageGCLowThreshold: 80}, + } + desiredGoal := aksmachine.GoalState{ + KubernetesVersion: "1.34.0", + SettingsVersion: "42", + MaxPods: 110, + NodeLabels: map[string]string{"workload": "new"}, + NodeTaints: []string{"dedicated=new:NoSchedule"}, + KubeletConfig: appliedGoal.KubeletConfig, + } + state := &State{AppliedGoal: &appliedGoal} + matchingNode := &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{ + "workload": "new", + "unrelated.example/label": "preserved", + }}, + Spec: corev1.NodeSpec{Taints: []corev1.Taint{ + {Key: "dedicated", Value: "new", Effect: corev1.TaintEffectNoSchedule}, + {Key: "unrelated.example/taint", Effect: corev1.TaintEffectNoExecute}, + }}, + } + + if !goalReflectedOnNode(desiredGoal, state, matchingNode) { + t.Fatal("goalReflectedOnNode returned false for matching in-place settings") + } + + tests := map[string]struct { + mutateGoal func(*aksmachine.GoalState) + mutateState func(*State) + mutateNode func(*corev1.Node) + }{ + "Kubernetes version changed": { + mutateGoal: func(goal *aksmachine.GoalState) { goal.KubernetesVersion = "1.35.0" }, + }, + "max pods changed": { + mutateGoal: func(goal *aksmachine.GoalState) { goal.MaxPods = 50 }, + }, + "kubelet config changed": { + mutateGoal: func(goal *aksmachine.GoalState) { goal.KubeletConfig.ImageGCHighThreshold = 70 }, + }, + "applied goal unavailable": { + mutateState: func(state *State) { state.AppliedGoal = nil }, + }, + "desired label missing": { + mutateNode: func(node *corev1.Node) { delete(node.Labels, "workload") }, + }, + "desired empty label missing": { + mutateGoal: func(goal *aksmachine.GoalState) { goal.NodeLabels = map[string]string{"empty": ""} }, + mutateNode: func(node *corev1.Node) { delete(node.Labels, "empty") }, + }, + "removed label remains": { + mutateNode: func(node *corev1.Node) { node.Labels["removed"] = "true" }, + }, + "desired taint missing": { + mutateNode: func(node *corev1.Node) { node.Spec.Taints = node.Spec.Taints[1:] }, + }, + "removed taint remains": { + mutateNode: func(node *corev1.Node) { + node.Spec.Taints = append(node.Spec.Taints, corev1.Taint{Key: "removed", Value: "true", Effect: corev1.TaintEffectNoExecute}) + }, + }, + } + + for name, tt := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + goal := desiredGoal + stateCopy := *state + appliedGoalCopy := appliedGoal + stateCopy.AppliedGoal = &appliedGoalCopy + node := matchingNode.DeepCopy() + if tt.mutateGoal != nil { + tt.mutateGoal(&goal) + } + if tt.mutateState != nil { + tt.mutateState(&stateCopy) + } + if tt.mutateNode != nil { + tt.mutateNode(node) + } + if goalReflectedOnNode(goal, &stateCopy, node) { + t.Fatal("goalReflectedOnNode returned true") + } + }) + } +} + func TestHasDeletionSignal(t *testing.T) { t.Parallel() diff --git a/pkg/daemon/repave_reconciler.go b/pkg/daemon/repave_reconciler.go index 8d4f99c5..9859ffa3 100644 --- a/pkg/daemon/repave_reconciler.go +++ b/pkg/daemon/repave_reconciler.go @@ -7,6 +7,7 @@ import ( "fmt" "log/slog" "math/big" + "reflect" "strings" "time" @@ -42,6 +43,7 @@ type decisionKind string const ( decisionNoop decisionKind = "Noop" decisionApplyGoalState decisionKind = "ApplyGoalState" + decisionAcknowledgeGoalState decisionKind = "AcknowledgeGoalState" decisionResetDelete decisionKind = "ResetDelete" decisionWaitForMachineDelete decisionKind = "WaitForMachineDelete" decisionWaitForNodeSignal decisionKind = "WaitForNodeSignal" @@ -182,6 +184,8 @@ func (r *repaveReconciler) reconcileOnce(ctx context.Context) error { return r.patchStatus(ctx, aksmachine.ProvisioningStateSucceeded, decision.Goal.SettingsVersion, decision.Reason) case decisionApplyGoalState: return r.applyGoalState(ctx, state, decision.Goal) + case decisionAcknowledgeGoalState: + return r.acknowledgeGoalState(ctx, state, decision.Goal) case decisionResetDelete: return r.resetDelete(ctx) default: @@ -198,6 +202,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 } @@ -220,7 +227,16 @@ func (r *repaveReconciler) applyGoalState(ctx context.Context, state *State, goa _ = r.patchStatus(ctx, aksmachine.ProvisioningStateFailed, stateObservedVersion(state), err.Error()) return err } - return r.patchStatus(ctx, aksmachine.ProvisioningStateSucceeded, newState.AppliedSettingsVersion, "machine goal state applied") + return r.patchStatus(ctx, aksmachine.ProvisioningStateSucceeded, stateObservedVersion(newState), "machine goal state applied") +} + +func (r *repaveReconciler) acknowledgeGoalState(ctx context.Context, state *State, goal aksmachine.GoalState) error { + newState, err := r.operator.AcknowledgeGoalState(ctx, goal) + if err != nil { + _ = r.patchStatus(ctx, aksmachine.ProvisioningStateFailed, stateObservedVersion(state), err.Error()) + return err + } + return r.patchStatus(ctx, aksmachine.ProvisioningStateSucceeded, stateObservedVersion(newState), "in-place machine goal state observed") } func (r *repaveReconciler) resetDelete(ctx context.Context) error { @@ -270,15 +286,84 @@ func decide(machine machineSnapshot, node nodeSnapshot, state *State) decision { if goalApplied(goal, state) { return decision{Kind: decisionReportSucceeded, Goal: goal, Reason: "goal state is applied"} } + if goalReflectedOnNode(goal, state, node.node) { + return decision{Kind: decisionAcknowledgeGoalState, Goal: goal, Reason: "goal update is already reflected on the node"} + } return decision{Kind: decisionWaitForNodeSignal, Goal: goal, Reason: "goal state differs but node deletion trigger is absent"} } +// goalApplied reports whether local daemon state has already acknowledged the +// exact desired settings version; no further verification or action is needed. func goalApplied(goal aksmachine.GoalState, state *State) bool { - if state == nil { + return goal.SettingsVersion != "" && stateObservedVersion(state) == goal.SettingsVersion +} + +// goalReflectedOnNode handles an unacknowledged settings version: it recognizes +// goal updates already applied externally by AKS RP so the agent can acknowledge +// them without mutating or repaving the host. +func goalReflectedOnNode(goal aksmachine.GoalState, state *State, node *corev1.Node) bool { + if goal.SettingsVersion == "" || state == nil || state.AppliedGoal == nil || node == nil { + return false + } + + appliedGoal := state.AppliedGoal + if !goalMatchesOutsideNodeState(goal, *appliedGoal) { return false } - return goal.SettingsVersion != "" && state.AppliedSettingsVersion == goal.SettingsVersion + + return nodeLabelsMatchGoal(node.Labels, appliedGoal.NodeLabels, goal.NodeLabels) && + nodeTaintsMatchGoal(node.Spec.Taints, appliedGoal.NodeTaints, goal.NodeTaints) +} + +func goalMatchesOutsideNodeState(goal, applied aksmachine.GoalState) bool { + // Normalize the version and Node-observable fields verified below. Comparing + // the remaining complete value makes future GoalState fields fail closed + // until their application contract is explicitly implemented here. + applied.SettingsVersion = goal.SettingsVersion + applied.NodeLabels = goal.NodeLabels + applied.NodeTaints = goal.NodeTaints + return reflect.DeepEqual(applied, goal) +} + +func nodeLabelsMatchGoal(nodeLabels, appliedLabels, desiredLabels map[string]string) bool { + for key, desiredValue := range desiredLabels { + if currentValue, present := nodeLabels[key]; !present || currentValue != desiredValue { + return false + } + } + for key := range appliedLabels { + if _, stillDesired := desiredLabels[key]; stillDesired { + continue + } + if _, stillPresent := nodeLabels[key]; stillPresent { + return false + } + } + return true +} + +func nodeTaintsMatchGoal(nodeTaints []corev1.Taint, appliedTaints, desiredTaints []string) bool { + nodeTaintSet := make(map[string]struct{}, len(nodeTaints)) + for i := range nodeTaints { + nodeTaintSet[nodeTaints[i].ToString()] = struct{}{} + } + desiredTaintSet := make(map[string]struct{}, len(desiredTaints)) + for _, taint := range desiredTaints { + desiredTaintSet[taint] = struct{}{} + if _, present := nodeTaintSet[taint]; !present { + return false + } + } + for _, taint := range appliedTaints { + if _, stillDesired := desiredTaintSet[taint]; stillDesired { + continue + } + if _, stillPresent := nodeTaintSet[taint]; stillPresent { + return false + } + } + return true } func hasDeletionSignal(taints []corev1.Taint) bool { @@ -291,10 +376,10 @@ func hasDeletionSignal(taints []corev1.Taint) bool { } func stateObservedVersion(state *State) string { - if state == nil { + if state == nil || state.AppliedGoal == nil { return "" } - return state.AppliedSettingsVersion + return state.AppliedGoal.SettingsVersion } func machineReconcileJitter(interval time.Duration) time.Duration { diff --git a/pkg/daemon/repave_reconciler_test.go b/pkg/daemon/repave_reconciler_test.go index 633b2eac..d0217e41 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" @@ -18,7 +19,17 @@ func TestRepaveReconcilerApplyGoalState(t *testing.T) { t.Parallel() machines := &fakeMachineClient{machine: &aksmachine.Machine{Goal: aksmachine.GoalState{KubernetesVersion: "1.34.0", SettingsVersion: "42"}}} - operator := &fakeNodeOperator{state: &State{AppliedSettingsVersion: "41", AppliedKubernetesVersion: "1.33.0", ActiveMachine: "kube1"}, newState: &State{AppliedSettingsVersion: "42", AppliedKubernetesVersion: "1.34.0", PreviousSettingsVersion: "41", PreviousKubernetesVersion: "1.33.0", ActiveMachine: "kube2"}} + operator := &fakeNodeOperator{ + state: &State{ + AppliedGoal: &aksmachine.GoalState{KubernetesVersion: "1.33.0", SettingsVersion: "41"}, + ActiveMachine: "kube1", + }, + newState: &State{ + AppliedGoal: &aksmachine.GoalState{KubernetesVersion: "1.34.0", SettingsVersion: "42"}, + PreviousAppliedGoal: &aksmachine.GoalState{KubernetesVersion: "1.33.0", SettingsVersion: "41"}, + ActiveMachine: "kube2", + }, + } repaves := newTestRepaveReconciler(t, machines, fakeClient(), operator) if err := repaves.reconcileOnce(context.Background()); err != nil { @@ -27,9 +38,61 @@ func TestRepaveReconcilerApplyGoalState(t *testing.T) { if !operator.applied { t.Fatal("ApplyGoalState was not called") } - if operator.state.AppliedSettingsVersion != "42" || operator.state.PreviousSettingsVersion != "41" || operator.state.ActiveMachine != "kube2" { + if stateObservedVersion(operator.state) != "42" || operator.state.ActiveMachine != "kube2" || + operator.state.PreviousAppliedGoal == nil || operator.state.PreviousAppliedGoal.SettingsVersion != "41" { + t.Fatalf("state = %#v", operator.state) + } + if got := machines.status.ProvisioningState; got != aksmachine.ProvisioningStateSucceeded { + t.Fatalf("status = %s", got) + } +} + +func TestRepaveReconcilerAcknowledgesInPlaceGoalState(t *testing.T) { + t.Parallel() + + appliedGoal := aksmachine.GoalState{ + KubernetesVersion: "1.34.0", + SettingsVersion: "41", + NodeLabels: map[string]string{"workload": "old"}, + NodeTaints: []string{"dedicated=old:NoSchedule"}, + } + desiredGoal := aksmachine.GoalState{ + KubernetesVersion: "1.34.0", + SettingsVersion: "42", + NodeLabels: map[string]string{"workload": "new"}, + NodeTaints: []string{"dedicated=new:NoSchedule"}, + } + machines := &fakeMachineClient{machine: &aksmachine.Machine{Goal: desiredGoal}} + operator := &fakeNodeOperator{state: &State{ + AppliedGoal: &appliedGoal, + ActiveMachine: "kube1", + }} + node := &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: "node1", Labels: map[string]string{"workload": "new"}}, + Spec: corev1.NodeSpec{Taints: []corev1.Taint{ + {Key: "dedicated", Value: "new", Effect: corev1.TaintEffectNoSchedule}, + }}, + } + repaves := newTestRepaveReconciler(t, machines, fakeClient(node), operator) + + if err := repaves.reconcileOnce(context.Background()); err != nil { + t.Fatalf("Reconcile: %v", err) + } + if !operator.acknowledged { + t.Fatal("AcknowledgeGoalState was not called") + } + if operator.applied { + t.Fatal("ApplyGoalState was called for an in-place update") + } + if stateObservedVersion(operator.state) != "42" || operator.state.AppliedGoal == nil || operator.state.AppliedGoal.NodeLabels["workload"] != "new" { t.Fatalf("state = %#v", operator.state) } + if operator.state.PreviousAppliedGoal == nil || operator.state.PreviousAppliedGoal.SettingsVersion != "41" { + t.Fatalf("PreviousAppliedGoal = %#v, want settings version 41", operator.state.PreviousAppliedGoal) + } + if got := machines.status.ObservedSettingsVersion; got != "42" { + t.Fatalf("observed settings version = %q, want 42", got) + } if got := machines.status.ProvisioningState; got != aksmachine.ProvisioningStateSucceeded { t.Fatalf("status = %s", got) } @@ -73,6 +136,25 @@ func TestRepaveReconcilerStateLoadFailurePatchesFailed(t *testing.T) { } } +func TestRepaveReconcilerRejectsInvalidMachineGoal(t *testing.T) { + t.Parallel() + + machines := &fakeMachineClient{machine: &aksmachine.Machine{}} + operator := &fakeNodeOperator{state: &State{ + AppliedGoal: &aksmachine.GoalState{KubernetesVersion: "1.34.0", SettingsVersion: "41"}, + ActiveMachine: "kube1", + }} + repaves := newTestRepaveReconciler(t, machines, fakeClient(), operator) + + err := repaves.reconcileOnce(context.Background()) + 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{ @@ -115,16 +197,17 @@ func (f *fakeMachineClient) PatchStatus(_ context.Context, status aksmachine.Sta } type fakeNodeOperator struct { - state *State - newState *State - err error - restartErr error - resetErr error - stopErr error - applied bool - restarted bool - reset bool - stopped bool + state *State + newState *State + err error + restartErr error + resetErr error + stopErr error + applied bool + acknowledged bool + restarted bool + reset bool + stopped bool } func (f *fakeNodeOperator) LoadState(context.Context) (*State, error) { @@ -140,6 +223,23 @@ func (f *fakeNodeOperator) ApplyGoalState(context.Context, *slog.Logger, aksmach return f.state, nil } +func (f *fakeNodeOperator) AcknowledgeGoalState(_ context.Context, goal aksmachine.GoalState) (*State, error) { + f.acknowledged = true + if f.err != nil { + return nil, f.err + } + var previousGoal *aksmachine.GoalState + if f.state != nil && f.state.AppliedGoal != nil { + previousGoal = cloneGoalState(*f.state.AppliedGoal) + } + f.state = &State{ + AppliedGoal: cloneGoalState(goal), + PreviousAppliedGoal: previousGoal, + ActiveMachine: f.state.ActiveMachine, + } + return f.state, nil +} + func (f *fakeNodeOperator) RestartNode(context.Context, *slog.Logger) error { f.restarted = true return f.restartErr diff --git a/pkg/daemon/state.go b/pkg/daemon/state.go index 3c16a4fc..3bef968d 100644 --- a/pkg/daemon/state.go +++ b/pkg/daemon/state.go @@ -7,8 +7,10 @@ import ( "encoding/json" "errors" "fmt" + "maps" "os" "path/filepath" + "slices" "strings" "github.com/Azure/AKSFlexNode/pkg/aksmachine" @@ -23,14 +25,24 @@ const ( stateFileName = "daemon-state.json" ) -// State records the last safely applied AKS machine goal and the previous -// known-good goal needed for rollback-oriented reconciliation. +// State records the current and previous safely applied AKS machine goals and +// the active nspawn machine. type State struct { - AppliedSettingsVersion string `json:"appliedSettingsVersion,omitempty"` - AppliedKubernetesVersion string `json:"appliedKubernetesVersion,omitempty"` - PreviousSettingsVersion string `json:"previousSettingsVersion,omitempty"` - PreviousKubernetesVersion string `json:"previousKubernetesVersion,omitempty"` - ActiveMachine string `json:"activeMachine,omitempty"` + AppliedGoal *aksmachine.GoalState `json:"appliedGoal,omitempty"` + PreviousAppliedGoal *aksmachine.GoalState `json:"previousAppliedGoal,omitempty"` + ActiveMachine string `json:"activeMachine,omitempty"` +} + +func (s *State) validate() error { + if s == nil { + return fmt.Errorf("daemon state is nil") + } + // SettingsVersion can be empty when best-effort Machine registration fails, + // but the applied goal itself is still required to restart safely. + if s.AppliedGoal == nil { + return fmt.Errorf("daemon state applied goal is missing") + } + return nil } type saveStateTask struct { @@ -59,12 +71,18 @@ func (t *saveStateTask) Do(ctx context.Context) error { func SeededState(goal aksmachine.GoalState) *State { return &State{ - AppliedSettingsVersion: goal.SettingsVersion, - AppliedKubernetesVersion: goal.KubernetesVersion, - ActiveMachine: goalstates.NSpawnMachineKube1, + AppliedGoal: cloneGoalState(goal), + ActiveMachine: goalstates.NSpawnMachineKube1, } } +func cloneGoalState(goal aksmachine.GoalState) *aksmachine.GoalState { + cloned := goal + cloned.NodeLabels = maps.Clone(goal.NodeLabels) + cloned.NodeTaints = slices.Clone(goal.NodeTaints) + return &cloned +} + func validActiveMachine(machine string) bool { return machine == goalstates.NSpawnMachineKube1 || machine == goalstates.NSpawnMachineKube2 } @@ -77,6 +95,9 @@ func activeMachineFromStore(ctx context.Context, store stateStore) (*activeMachi if state == nil { return nil, fmt.Errorf("daemon state is missing active machine") } + if err := state.validate(); err != nil { + return nil, err + } if !validActiveMachine(state.ActiveMachine) { return nil, fmt.Errorf("daemon state active machine %q is invalid", state.ActiveMachine) } @@ -125,12 +146,15 @@ func (s *fileStateStore) Load(context.Context) (*State, error) { if err := json.Unmarshal(data, &state); err != nil { return nil, fmt.Errorf("decode daemon state %s: %w", s.path, err) } + if err := state.validate(); err != nil { + return nil, fmt.Errorf("validate daemon state %s: %w", s.path, err) + } return &state, nil } func (s *fileStateStore) Save(_ context.Context, state *State) error { - if state == nil { - return fmt.Errorf("daemon state is nil") + if err := state.validate(); err != nil { + return err } data, err := json.MarshalIndent(state, "", " ") if err != nil { diff --git a/pkg/daemon/state_test.go b/pkg/daemon/state_test.go index 620b221a..7315e8f5 100644 --- a/pkg/daemon/state_test.go +++ b/pkg/daemon/state_test.go @@ -18,11 +18,17 @@ func TestFileStateStoreSaveLoad(t *testing.T) { t.Fatalf("newFileStateStore: %v", err) } want := &State{ - AppliedSettingsVersion: "42", - AppliedKubernetesVersion: "1.34.0", - PreviousSettingsVersion: "41", - PreviousKubernetesVersion: "1.33.0", - ActiveMachine: "kube2", + AppliedGoal: &aksmachine.GoalState{ + KubernetesVersion: "1.34.0", + SettingsVersion: "42", + NodeLabels: map[string]string{"workload": "flex"}, + NodeTaints: []string{"dedicated=flex:NoSchedule"}, + }, + PreviousAppliedGoal: &aksmachine.GoalState{ + KubernetesVersion: "1.33.0", + SettingsVersion: "41", + }, + ActiveMachine: "kube2", } if err := store.Save(context.Background(), want); err != nil { @@ -32,7 +38,9 @@ func TestFileStateStoreSaveLoad(t *testing.T) { if err != nil { t.Fatalf("Load: %v", err) } - if got.AppliedSettingsVersion != want.AppliedSettingsVersion || got.ActiveMachine != want.ActiveMachine { + if got.ActiveMachine != want.ActiveMachine || + got.AppliedGoal == nil || got.AppliedGoal.NodeLabels["workload"] != "flex" || len(got.AppliedGoal.NodeTaints) != 1 || + got.PreviousAppliedGoal == nil || got.PreviousAppliedGoal.SettingsVersion != "41" { t.Fatalf("state = %#v, want %#v", got, want) } } @@ -53,6 +61,64 @@ func TestFileStateStoreLoadMissing(t *testing.T) { } } +func TestFileStateStoreRejectsOldStateWithoutAppliedGoal(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "state.json") + store, err := newFileStateStore(path) + if err != nil { + t.Fatalf("newFileStateStore: %v", err) + } + data := []byte(`{"appliedSettingsVersion":"42","appliedKubernetesVersion":"1.34.0","activeMachine":"kube1"}`) + if err := os.WriteFile(path, data, 0o600); err != nil { + t.Fatalf("WriteFile: %v", err) + } + if err := os.WriteFile(path+".sha256", []byte(checksum(data)+"\n"), 0o600); err != nil { + t.Fatalf("WriteFile checksum: %v", err) + } + + _, err = store.Load(t.Context()) + if err == nil || !strings.Contains(err.Error(), "daemon state applied goal is missing") { + t.Fatalf("Load error = %v, want missing applied goal", err) + } +} + +func TestFileStateStoreAllowsAppliedGoalWithoutSettingsVersion(t *testing.T) { + t.Parallel() + + store, err := newFileStateStore(filepath.Join(t.TempDir(), "state.json")) + if err != nil { + t.Fatalf("newFileStateStore: %v", err) + } + want := &State{ + AppliedGoal: &aksmachine.GoalState{KubernetesVersion: "1.34.0"}, + ActiveMachine: "kube1", + } + if err := store.Save(t.Context(), want); err != nil { + t.Fatalf("Save: %v", err) + } + got, err := store.Load(t.Context()) + if err != nil { + t.Fatalf("Load: %v", err) + } + if got.AppliedGoal == nil || got.AppliedGoal.KubernetesVersion != "1.34.0" || got.AppliedGoal.SettingsVersion != "" { + t.Fatalf("AppliedGoal = %#v, want goal with empty settings version", got.AppliedGoal) + } +} + +func TestFileStateStoreRejectsSaveWithoutAppliedGoal(t *testing.T) { + t.Parallel() + + store, err := newFileStateStore(filepath.Join(t.TempDir(), "state.json")) + if err != nil { + t.Fatalf("newFileStateStore: %v", err) + } + err = store.Save(t.Context(), &State{ActiveMachine: "kube1"}) + if err == nil || !strings.Contains(err.Error(), "daemon state applied goal is missing") { + t.Fatalf("Save error = %v, want missing applied goal", err) + } +} + func TestFileStateStoreChecksumMismatch(t *testing.T) { t.Parallel() @@ -61,10 +127,10 @@ func TestFileStateStoreChecksumMismatch(t *testing.T) { if err != nil { t.Fatalf("newFileStateStore: %v", err) } - if err := store.Save(context.Background(), &State{AppliedSettingsVersion: "42"}); err != nil { + if err := store.Save(context.Background(), &State{AppliedGoal: &aksmachine.GoalState{KubernetesVersion: "1.34.0", SettingsVersion: "42"}}); err != nil { t.Fatalf("Save: %v", err) } - if err := os.WriteFile(path, []byte(`{"appliedSettingsVersion":"43"}`), 0o600); err != nil { + if err := os.WriteFile(path, []byte(`{"appliedGoal":{"settingsVersion":"43"}}`), 0o600); err != nil { t.Fatalf("WriteFile: %v", err) } _, err = store.Load(context.Background()) @@ -81,7 +147,7 @@ func TestFileStateStoreCorruptJSON(t *testing.T) { if err != nil { t.Fatalf("newFileStateStore: %v", err) } - data := []byte(`{"appliedSettingsVersion":`) + data := []byte(`{"appliedGoal":`) if err := os.WriteFile(path, data, 0o600); err != nil { t.Fatalf("WriteFile: %v", err) } @@ -102,7 +168,7 @@ func TestFileStateStoreDelete(t *testing.T) { if err != nil { t.Fatalf("newFileStateStore: %v", err) } - if err := store.Save(context.Background(), &State{AppliedSettingsVersion: "42"}); err != nil { + if err := store.Save(context.Background(), &State{AppliedGoal: &aksmachine.GoalState{KubernetesVersion: "1.34.0", SettingsVersion: "42"}}); err != nil { t.Fatalf("Save: %v", err) } if err := store.Delete(context.Background()); err != nil { @@ -119,18 +185,24 @@ func TestFileStateStoreDelete(t *testing.T) { func TestSeededState(t *testing.T) { t.Parallel() - state := SeededState(aksmachine.GoalState{KubernetesVersion: "1.34.0", SettingsVersion: "42"}) - if state.AppliedSettingsVersion != "42" { - t.Fatalf("AppliedSettingsVersion = %q, want 42", state.AppliedSettingsVersion) + goal := aksmachine.GoalState{ + KubernetesVersion: "1.34.0", + SettingsVersion: "42", + NodeLabels: map[string]string{"workload": "flex"}, + NodeTaints: []string{"dedicated=flex:NoSchedule"}, } - if state.AppliedKubernetesVersion != "1.34.0" { - t.Fatalf("AppliedKubernetesVersion = %q, want 1.34.0", state.AppliedKubernetesVersion) + state := SeededState(goal) + if state.AppliedGoal == nil || state.AppliedGoal.SettingsVersion != "42" || state.AppliedGoal.KubernetesVersion != "1.34.0" { + t.Fatalf("AppliedGoal = %#v", state.AppliedGoal) } if state.ActiveMachine != "kube1" { t.Fatalf("ActiveMachine = %q, want kube1", state.ActiveMachine) } - if state.PreviousSettingsVersion != "" || state.PreviousKubernetesVersion != "" { - t.Fatalf("previous state = %#v, want empty", state) + if state.PreviousAppliedGoal != nil { + t.Fatalf("PreviousAppliedGoal = %#v, want nil", state.PreviousAppliedGoal) + } + if state.AppliedGoal == nil || state.AppliedGoal.NodeLabels["workload"] != "flex" || len(state.AppliedGoal.NodeTaints) != 1 { + t.Fatalf("AppliedGoal = %#v, want complete goal", state.AppliedGoal) } } @@ -161,18 +233,22 @@ func TestActiveMachineFromStore(t *testing.T) { wantErr bool }{ "kube1": { - state: &State{ActiveMachine: "kube1"}, + state: &State{AppliedGoal: &aksmachine.GoalState{KubernetesVersion: "1.34.0"}, ActiveMachine: "kube1"}, want: "kube1", }, "kube2": { - state: &State{ActiveMachine: "kube2"}, + state: &State{AppliedGoal: &aksmachine.GoalState{KubernetesVersion: "1.34.0"}, ActiveMachine: "kube2"}, want: "kube2", }, "missing state": { wantErr: true, }, + "missing applied goal": { + state: &State{ActiveMachine: "kube1"}, + wantErr: true, + }, "invalid active machine": { - state: &State{ActiveMachine: "kube3"}, + state: &State{AppliedGoal: &aksmachine.GoalState{KubernetesVersion: "1.34.0"}, ActiveMachine: "kube3"}, wantErr: true, }, } From f652f36e12304d84e4e1379346c9db402f6897e0 Mon Sep 17 00:00:00 2001 From: wenxuanW Date: Wed, 29 Jul 2026 14:30:59 -0700 Subject: [PATCH 3/4] Preserve legacy daemon state fields --- pkg/daemon/state.go | 47 +++++++++++++++- pkg/daemon/state_test.go | 115 +++++++++++++++++++++++++++++++++------ 2 files changed, 143 insertions(+), 19 deletions(-) diff --git a/pkg/daemon/state.go b/pkg/daemon/state.go index 3bef968d..c28f278f 100644 --- a/pkg/daemon/state.go +++ b/pkg/daemon/state.go @@ -30,7 +30,15 @@ const ( type State struct { AppliedGoal *aksmachine.GoalState `json:"appliedGoal,omitempty"` PreviousAppliedGoal *aksmachine.GoalState `json:"previousAppliedGoal,omitempty"` - ActiveMachine string `json:"activeMachine,omitempty"` + + // Deprecated: these projections keep state readable by older agent binaries. + // AppliedGoal and PreviousAppliedGoal remain authoritative. + AppliedSettingsVersion string `json:"appliedSettingsVersion,omitempty"` + AppliedKubernetesVersion string `json:"appliedKubernetesVersion,omitempty"` + PreviousSettingsVersion string `json:"previousSettingsVersion,omitempty"` + PreviousKubernetesVersion string `json:"previousKubernetesVersion,omitempty"` + + ActiveMachine string `json:"activeMachine,omitempty"` } func (s *State) validate() error { @@ -83,6 +91,37 @@ func cloneGoalState(goal aksmachine.GoalState) *aksmachine.GoalState { return &cloned } +func (s *State) migrateLegacyGoals() { + if s.AppliedGoal == nil && (s.AppliedSettingsVersion != "" || s.AppliedKubernetesVersion != "") { + s.AppliedGoal = &aksmachine.GoalState{ + SettingsVersion: s.AppliedSettingsVersion, + KubernetesVersion: s.AppliedKubernetesVersion, + } + } + if s.PreviousAppliedGoal == nil && (s.PreviousSettingsVersion != "" || s.PreviousKubernetesVersion != "") { + s.PreviousAppliedGoal = &aksmachine.GoalState{ + SettingsVersion: s.PreviousSettingsVersion, + KubernetesVersion: s.PreviousKubernetesVersion, + } + } +} + +func (s *State) populateLegacyFields() { + s.AppliedSettingsVersion = "" + s.AppliedKubernetesVersion = "" + if s.AppliedGoal != nil { + s.AppliedSettingsVersion = s.AppliedGoal.SettingsVersion + s.AppliedKubernetesVersion = s.AppliedGoal.KubernetesVersion + } + + s.PreviousSettingsVersion = "" + s.PreviousKubernetesVersion = "" + if s.PreviousAppliedGoal != nil { + s.PreviousSettingsVersion = s.PreviousAppliedGoal.SettingsVersion + s.PreviousKubernetesVersion = s.PreviousAppliedGoal.KubernetesVersion + } +} + func validActiveMachine(machine string) bool { return machine == goalstates.NSpawnMachineKube1 || machine == goalstates.NSpawnMachineKube2 } @@ -146,9 +185,11 @@ func (s *fileStateStore) Load(context.Context) (*State, error) { if err := json.Unmarshal(data, &state); err != nil { return nil, fmt.Errorf("decode daemon state %s: %w", s.path, err) } + state.migrateLegacyGoals() if err := state.validate(); err != nil { return nil, fmt.Errorf("validate daemon state %s: %w", s.path, err) } + state.populateLegacyFields() return &state, nil } @@ -156,7 +197,9 @@ func (s *fileStateStore) Save(_ context.Context, state *State) error { if err := state.validate(); err != nil { return err } - data, err := json.MarshalIndent(state, "", " ") + stateForPersistence := *state + stateForPersistence.populateLegacyFields() + data, err := json.MarshalIndent(&stateForPersistence, "", " ") if err != nil { return fmt.Errorf("marshal daemon state: %w", err) } diff --git a/pkg/daemon/state_test.go b/pkg/daemon/state_test.go index 7315e8f5..4e43831d 100644 --- a/pkg/daemon/state_test.go +++ b/pkg/daemon/state_test.go @@ -2,6 +2,7 @@ package daemon import ( "context" + "encoding/json" "os" "path/filepath" "strings" @@ -28,19 +29,36 @@ func TestFileStateStoreSaveLoad(t *testing.T) { KubernetesVersion: "1.33.0", SettingsVersion: "41", }, - ActiveMachine: "kube2", + AppliedSettingsVersion: "stale-applied", + AppliedKubernetesVersion: "stale-applied", + PreviousSettingsVersion: "stale-previous", + PreviousKubernetesVersion: "stale-previous", + ActiveMachine: "kube2", } if err := store.Save(context.Background(), want); err != nil { t.Fatalf("Save: %v", err) } + persistedData, err := os.ReadFile(store.path) + if err != nil { + t.Fatalf("ReadFile: %v", err) + } + var persisted State + if err := json.Unmarshal(persistedData, &persisted); err != nil { + t.Fatalf("Unmarshal persisted state: %v", err) + } + if persisted.AppliedSettingsVersion != "42" || persisted.AppliedKubernetesVersion != "1.34.0" || + persisted.PreviousSettingsVersion != "41" || persisted.PreviousKubernetesVersion != "1.33.0" { + t.Fatalf("legacy projections = %#v", persisted) + } got, err := store.Load(context.Background()) if err != nil { t.Fatalf("Load: %v", err) } if got.ActiveMachine != want.ActiveMachine || got.AppliedGoal == nil || got.AppliedGoal.NodeLabels["workload"] != "flex" || len(got.AppliedGoal.NodeTaints) != 1 || - got.PreviousAppliedGoal == nil || got.PreviousAppliedGoal.SettingsVersion != "41" { + got.PreviousAppliedGoal == nil || got.PreviousAppliedGoal.SettingsVersion != "41" || + got.AppliedSettingsVersion != "42" || got.PreviousSettingsVersion != "41" { t.Fatalf("state = %#v, want %#v", got, want) } } @@ -61,25 +79,88 @@ func TestFileStateStoreLoadMissing(t *testing.T) { } } -func TestFileStateStoreRejectsOldStateWithoutAppliedGoal(t *testing.T) { +func TestFileStateStoreLoadCompatibility(t *testing.T) { t.Parallel() - path := filepath.Join(t.TempDir(), "state.json") - store, err := newFileStateStore(path) - if err != nil { - t.Fatalf("newFileStateStore: %v", err) - } - data := []byte(`{"appliedSettingsVersion":"42","appliedKubernetesVersion":"1.34.0","activeMachine":"kube1"}`) - if err := os.WriteFile(path, data, 0o600); err != nil { - t.Fatalf("WriteFile: %v", err) - } - if err := os.WriteFile(path+".sha256", []byte(checksum(data)+"\n"), 0o600); err != nil { - t.Fatalf("WriteFile checksum: %v", err) + tests := map[string]struct { + data string + wantErr string + check func(*testing.T, *State) + }{ + "legacy state is migrated": { + data: `{ + "appliedSettingsVersion":"42", + "appliedKubernetesVersion":"1.34.0", + "previousSettingsVersion":"41", + "previousKubernetesVersion":"1.33.0", + "activeMachine":"kube1" + }`, + check: func(t *testing.T, state *State) { + t.Helper() + if state.AppliedGoal == nil || state.AppliedGoal.SettingsVersion != "42" || state.AppliedGoal.KubernetesVersion != "1.34.0" { + t.Fatalf("AppliedGoal = %#v, want migrated legacy goal", state.AppliedGoal) + } + if state.PreviousAppliedGoal == nil || state.PreviousAppliedGoal.SettingsVersion != "41" || state.PreviousAppliedGoal.KubernetesVersion != "1.33.0" { + t.Fatalf("PreviousAppliedGoal = %#v, want migrated legacy goal", state.PreviousAppliedGoal) + } + }, + }, + "full goals override stale legacy projections": { + data: `{ + "appliedGoal":{"kubernetesVersion":"1.34.0","settingsVersion":"42","nodeLabels":{"workload":"flex"}}, + "previousAppliedGoal":{"kubernetesVersion":"1.33.0","settingsVersion":"41"}, + "appliedSettingsVersion":"99", + "appliedKubernetesVersion":"1.99.0", + "previousSettingsVersion":"98", + "previousKubernetesVersion":"1.98.0", + "activeMachine":"kube2" + }`, + check: func(t *testing.T, state *State) { + t.Helper() + if state.AppliedGoal == nil || state.AppliedGoal.SettingsVersion != "42" || state.AppliedGoal.NodeLabels["workload"] != "flex" { + t.Fatalf("AppliedGoal = %#v, want complete goal", state.AppliedGoal) + } + if state.AppliedSettingsVersion != "42" || state.AppliedKubernetesVersion != "1.34.0" || + state.PreviousSettingsVersion != "41" || state.PreviousKubernetesVersion != "1.33.0" { + t.Fatalf("legacy projections = %#v, want values derived from full goals", state) + } + }, + }, + "state without either format is rejected": { + data: `{"activeMachine":"kube1"}`, + wantErr: "daemon state applied goal is missing", + }, } - _, err = store.Load(t.Context()) - if err == nil || !strings.Contains(err.Error(), "daemon state applied goal is missing") { - t.Fatalf("Load error = %v, want missing applied goal", err) + for name, tt := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "state.json") + store, err := newFileStateStore(path) + if err != nil { + t.Fatalf("newFileStateStore: %v", err) + } + data := []byte(tt.data) + if err := os.WriteFile(path, data, 0o600); err != nil { + t.Fatalf("WriteFile: %v", err) + } + if err := os.WriteFile(path+".sha256", []byte(checksum(data)+"\n"), 0o600); err != nil { + t.Fatalf("WriteFile checksum: %v", err) + } + + got, err := store.Load(t.Context()) + if tt.wantErr != "" { + if err == nil || !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("Load error = %v, want %q", err, tt.wantErr) + } + return + } + if err != nil { + t.Fatalf("Load: %v", err) + } + tt.check(t, got) + }) } } From 8ce89afb6d820a9a840547b3a9bc464f80c68cfc Mon Sep 17 00:00:00 2001 From: wenxuanW Date: Thu, 30 Jul 2026 10:46:28 -0700 Subject: [PATCH 4/4] Use AKS Machine goal during preflight --- docs/usages/operations.md | 2 +- pkg/cmd/preflight/machine_goal.go | 125 ++++++++++++++++ pkg/cmd/preflight/preflight.go | 9 +- pkg/cmd/preflight/preflight_test.go | 220 ++++++++++++++++++++++++++++ 4 files changed, 354 insertions(+), 2 deletions(-) create mode 100644 pkg/cmd/preflight/machine_goal.go diff --git a/docs/usages/operations.md b/docs/usages/operations.md index bd2bbc9b..17a99509 100644 --- a/docs/usages/operations.md +++ b/docs/usages/operations.md @@ -4,7 +4,7 @@ This guide summarizes common host and cluster operations for AKS Flex Node. ## Preflight -Run preflight before mutating the host. The command validates the config, resolves the nspawn goal state, and checks host prerequisites, API server reachability, rootfs image reachability, and bootstrap artifact sources. +Run preflight before mutating the host. The command validates the config, reads any existing AKS Machine, resolves the effective nspawn goal state, and checks host prerequisites, API server reachability, rootfs image reachability, and bootstrap artifact sources. An existing Machine is authoritative. When its goal differs from local configuration, preflight reports an error and validates bootstrap inputs derived from the Machine goal. If the Machine does not exist, preflight keeps the original config-only behavior. ```bash aks-flex-node preflight --config /etc/aks-flex-node/config.json diff --git a/pkg/cmd/preflight/machine_goal.go b/pkg/cmd/preflight/machine_goal.go new file mode 100644 index 00000000..7af5e6b5 --- /dev/null +++ b/pkg/cmd/preflight/machine_goal.go @@ -0,0 +1,125 @@ +package preflight + +import ( + "context" + "errors" + "fmt" + "log/slog" + "reflect" + "slices" + + "github.com/Azure/AKSFlexNode/pkg/aksmachine" + "github.com/Azure/AKSFlexNode/pkg/config" + "github.com/Azure/unbounded/pkg/agent/preflight" +) + +const machineGoalCheckName = "AKSMachineGoal" + +type machineGoalCheck struct { + remoteGoal *aksmachine.GoalState + result preflight.Result + log *slog.Logger + err error +} + +func (c machineGoalCheck) Name() string { return c.result.Name } + +func (c machineGoalCheck) Check(context.Context) []preflight.Result { + if c.err != nil { + c.log.Warn("preflight could not use AKS Machine goal", "error", c.err) + } + return preflight.Results(c.result) +} + +// newMachineGoalCheck resolves the authoritative goal before the other preflight +// checks are created so they validate the same inputs that bootstrap will use. +func newMachineGoalCheck( + ctx context.Context, + log *slog.Logger, + cfg *config.Config, +) (*machineGoalCheck, error) { + localGoal, err := aksmachine.GoalStateFromConfig(cfg) + if err != nil { + return nil, fmt.Errorf("build goal state from config: %w", err) + } + machines, err := aksmachine.NewMachineClient(cfg, log, aksmachine.MachineClientOptions{}) + if err != nil { + return nil, fmt.Errorf("create AKS machine client: %w", err) + } + + requireMachineRegistration := cfg.Agent.RequireMachineRegistration != nil && *cfg.Agent.RequireMachineRegistration + remoteGoal, result, err := resolveMachineGoal(ctx, machines, localGoal, requireMachineRegistration) + return &machineGoalCheck{remoteGoal: remoteGoal, result: result, log: log, err: err}, nil +} + +// resolveMachineGoal adopts an existing Machine read-only so the remaining +// checks validate the bootstrap inputs derived from the same remote goal as start. +func resolveMachineGoal( + ctx context.Context, + machines aksmachine.MachineClient, + localGoal aksmachine.GoalState, + requireMachineRegistration bool, +) (*aksmachine.GoalState, preflight.Result, error) { + machine, err := machines.Get(ctx) + if err != nil { + var notFound *aksmachine.NotFoundError + if errors.As(err, ¬Found) { + // A nil remote goal preserves the original config-only resolution path. + return nil, preflight.OK( + machineGoalCheckName, + "AKS Machine goal", + "AKS Machine does not exist; validating the local bootstrap goal", + ), nil + } + return machineGoalFailure(requireMachineRegistration, "read AKS Machine", err) + } + if err := machine.Validate(); err != nil { + return machineGoalFailure(requireMachineRegistration, "validate AKS Machine", err) + } + + if !goalsMatch(machine.Goal, localGoal) { + return &machine.Goal, preflight.Error( + machineGoalCheckName, + "AKS Machine goal", + "existing AKS Machine goal is authoritative and differs from local config; validating bootstrap inputs derived from the AKS Machine goal", + ), nil + } + + return &machine.Goal, preflight.OK( + machineGoalCheckName, + "AKS Machine goal", + "existing AKS Machine goal is authoritative and matches local config; validating bootstrap inputs derived from the AKS Machine goal", + ), nil +} + +func goalsMatch(remote, local aksmachine.GoalState) bool { + normalize := func(goal aksmachine.GoalState) aksmachine.GoalState { + // SettingsVersion is the remote ETag and has no corresponding local setting. + goal.SettingsVersion = "" + if len(goal.NodeLabels) == 0 { + goal.NodeLabels = nil + } + if len(goal.NodeTaints) == 0 { + goal.NodeTaints = nil + } else { + goal.NodeTaints = slices.Clone(goal.NodeTaints) + slices.Sort(goal.NodeTaints) + } + return goal + } + remote = normalize(remote) + local = normalize(local) + return reflect.DeepEqual(remote, local) +} + +func machineGoalFailure( + requireMachineRegistration bool, + operation string, + err error, +) (*aksmachine.GoalState, preflight.Result, error) { + result := preflight.Warning(machineGoalCheckName, "AKS Machine goal", "AKS Machine goal is unavailable; validating the local bootstrap goal") + if requireMachineRegistration { + result = preflight.Error(machineGoalCheckName, "AKS Machine goal", "AKS Machine goal is unavailable but machine registration is required") + } + return nil, result, fmt.Errorf("%s: %w", operation, err) +} diff --git a/pkg/cmd/preflight/preflight.go b/pkg/cmd/preflight/preflight.go index 7046bac9..9f06b356 100644 --- a/pkg/cmd/preflight/preflight.go +++ b/pkg/cmd/preflight/preflight.go @@ -12,6 +12,7 @@ import ( "github.com/spf13/cobra" "github.com/Azure/AKSFlexNode/pkg/config" + "github.com/Azure/AKSFlexNode/pkg/daemon" "github.com/Azure/AKSFlexNode/pkg/logger" "github.com/Azure/AKSFlexNode/pkg/npd" "github.com/Azure/unbounded/pkg/agent/goalstates" @@ -68,12 +69,18 @@ func (h *handler) execute(ctx context.Context) error { } log := createPreflightLogger(cfg.Agent.LogLevel) - agentCfg, gs, _, err := config.ResolveMachineGoalState(ctx, log, cfg, goalstates.NSpawnMachineKube1) + machineGoalCheck, err := newMachineGoalCheck(ctx, log, cfg) + if err != nil { + return err + } + + agentCfg, gs, _, err := daemon.ResolveMachineGoalState(ctx, log, cfg, goalstates.NSpawnMachineKube1, machineGoalCheck.remoteGoal) if err != nil { return fmt.Errorf("preflight failed to resolve goal state: %w", err) } checks := preflight.Flatten( + []preflight.Checker{machineGoalCheck}, host.Preflight(log, *agentCfg, gs), nodestart.Preflight(log, *agentCfg, gs), rootfs.Preflight(log, *agentCfg, gs), diff --git a/pkg/cmd/preflight/preflight_test.go b/pkg/cmd/preflight/preflight_test.go index 0dbedc57..495e7f70 100644 --- a/pkg/cmd/preflight/preflight_test.go +++ b/pkg/cmd/preflight/preflight_test.go @@ -2,9 +2,15 @@ package preflight import ( "bytes" + "context" + "errors" + "maps" + "reflect" + "slices" "strings" "testing" + "github.com/Azure/AKSFlexNode/pkg/aksmachine" "github.com/Azure/unbounded/pkg/agent/preflight" ) @@ -89,3 +95,217 @@ func TestWriteText(t *testing.T) { } } } + +func TestResolveMachineGoal(t *testing.T) { + t.Parallel() + + localGoal := aksmachine.GoalState{ + KubernetesVersion: "1.34.0", + MaxPods: 30, + NodeLabels: map[string]string{"source": "local"}, + NodeTaints: []string{"dedicated=local:NoSchedule"}, + KubeletConfig: aksmachine.KubeletConfig{ + ImageGCHighThreshold: 85, + ImageGCLowThreshold: 80, + }, + } + newRemoteGoal := func() aksmachine.GoalState { + goal := localGoal + goal.SettingsVersion = "etag-42" + goal.NodeLabels = maps.Clone(localGoal.NodeLabels) + goal.NodeTaints = slices.Clone(localGoal.NodeTaints) + return goal + } + + matchingGoal := newRemoteGoal() + versionMismatch := newRemoteGoal() + versionMismatch.KubernetesVersion = "1.35.0" + maxPodsMismatch := newRemoteGoal() + maxPodsMismatch.MaxPods = 110 + labelsMismatch := newRemoteGoal() + labelsMismatch.NodeLabels["source"] = "remote" + taintsMismatch := newRemoteGoal() + taintsMismatch.NodeTaints = []string{"dedicated=remote:NoExecute"} + imageGCHighMismatch := newRemoteGoal() + imageGCHighMismatch.KubeletConfig.ImageGCHighThreshold = 90 + imageGCLowMismatch := newRemoteGoal() + imageGCLowMismatch.KubeletConfig.ImageGCLowThreshold = 75 + + tests := map[string]struct { + machine *aksmachine.Machine + getErr error + require bool + wantRemote *aksmachine.GoalState + severity preflight.Severity + message string + wantErr string + }{ + "matching remote goal is used": { + machine: &aksmachine.Machine{Goal: matchingGoal}, + wantRemote: &matchingGoal, + severity: preflight.SeverityOK, + message: "matches local config", + }, + "Kubernetes version mismatch fails and uses remote": { + machine: &aksmachine.Machine{Goal: versionMismatch}, + wantRemote: &versionMismatch, + severity: preflight.SeverityError, + message: "is authoritative and differs from local config", + }, + "max pods mismatch fails and uses remote": { + machine: &aksmachine.Machine{Goal: maxPodsMismatch}, + wantRemote: &maxPodsMismatch, + severity: preflight.SeverityError, + message: "is authoritative and differs from local config", + }, + "labels mismatch fails and uses remote": { + machine: &aksmachine.Machine{Goal: labelsMismatch}, + wantRemote: &labelsMismatch, + severity: preflight.SeverityError, + message: "is authoritative and differs from local config", + }, + "taints mismatch fails and uses remote": { + machine: &aksmachine.Machine{Goal: taintsMismatch}, + wantRemote: &taintsMismatch, + severity: preflight.SeverityError, + message: "is authoritative and differs from local config", + }, + "image GC high threshold mismatch fails and uses remote": { + machine: &aksmachine.Machine{Goal: imageGCHighMismatch}, + wantRemote: &imageGCHighMismatch, + severity: preflight.SeverityError, + message: "is authoritative and differs from local config", + }, + "image GC low threshold mismatch fails and uses remote": { + machine: &aksmachine.Machine{Goal: imageGCLowMismatch}, + wantRemote: &imageGCLowMismatch, + severity: preflight.SeverityError, + message: "is authoritative and differs from local config", + }, + "missing machine preserves config-only resolution": { + getErr: &aksmachine.NotFoundError{Resource: "machine"}, + severity: preflight.SeverityOK, + message: "does not exist", + }, + "optional read failure warns and preserves config-only resolution": { + getErr: errors.New("boom"), + severity: preflight.SeverityWarning, + message: "validating the local bootstrap goal", + wantErr: "read AKS Machine: boom", + }, + "required read failure fails and preserves config-only resolution": { + getErr: errors.New("boom"), + require: true, + severity: preflight.SeverityError, + message: "machine registration is required", + wantErr: "read AKS Machine: boom", + }, + "optional invalid machine warns and preserves config-only resolution": { + machine: &aksmachine.Machine{Goal: aksmachine.GoalState{KubernetesVersion: "1.35.0"}}, + severity: preflight.SeverityWarning, + message: "validating the local bootstrap goal", + wantErr: "validate AKS Machine: goal settings version is empty", + }, + "required nil machine fails and preserves config-only resolution": { + require: true, + severity: preflight.SeverityError, + message: "machine registration is required", + wantErr: "validate AKS Machine: machine is nil", + }, + } + + for name, tt := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + + client := &preflightMachineClient{machine: tt.machine, getErr: tt.getErr} + remoteGoal, result, err := resolveMachineGoal(t.Context(), client, localGoal, tt.require) + + if tt.wantErr == "" { + if err != nil { + t.Fatalf("resolveMachineGoal() error = %v", err) + } + } else if err == nil || !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("resolveMachineGoal() error = %v, want containing %q", err, tt.wantErr) + } + if !reflect.DeepEqual(remoteGoal, tt.wantRemote) { + t.Fatalf("resolveMachineGoal() remote goal = %#v, want %#v", remoteGoal, tt.wantRemote) + } + if result.Name != machineGoalCheckName || result.Severity != tt.severity || !strings.Contains(result.Message, tt.message) { + t.Fatalf("resolveMachineGoal() result = %#v, want severity %q and message containing %q", result, tt.severity, tt.message) + } + if client.createCalls != 0 { + t.Fatalf("Create() calls = %d, want 0", client.createCalls) + } + }) + } +} + +func TestGoalsMatch(t *testing.T) { + t.Parallel() + + tests := map[string]struct { + remote aksmachine.GoalState + local aksmachine.GoalState + want bool + }{ + "settings version is ignored": { + remote: aksmachine.GoalState{KubernetesVersion: "1.34.0", SettingsVersion: "etag-42"}, + local: aksmachine.GoalState{KubernetesVersion: "1.34.0"}, + want: true, + }, + "nil and empty collections are equivalent": { + remote: aksmachine.GoalState{KubernetesVersion: "1.34.0", SettingsVersion: "etag-42"}, + local: aksmachine.GoalState{ + KubernetesVersion: "1.34.0", + NodeLabels: map[string]string{}, + NodeTaints: []string{}, + }, + want: true, + }, + "taint order is ignored": { + remote: aksmachine.GoalState{ + KubernetesVersion: "1.34.0", + SettingsVersion: "etag-42", + NodeTaints: []string{"b=true:NoExecute", "a=true:NoSchedule"}, + }, + local: aksmachine.GoalState{ + KubernetesVersion: "1.34.0", + NodeTaints: []string{"a=true:NoSchedule", "b=true:NoExecute"}, + }, + want: true, + }, + "different labels do not match": { + remote: aksmachine.GoalState{KubernetesVersion: "1.34.0", NodeLabels: map[string]string{"source": "remote"}}, + local: aksmachine.GoalState{KubernetesVersion: "1.34.0", NodeLabels: map[string]string{"source": "local"}}, + }, + } + + for name, tt := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + if got := goalsMatch(tt.remote, tt.local); got != tt.want { + t.Fatalf("goalsMatch() = %t, want %t", got, tt.want) + } + }) + } +} + +type preflightMachineClient struct { + machine *aksmachine.Machine + getErr error + createCalls int +} + +func (c *preflightMachineClient) Create(context.Context, aksmachine.GoalState) (*aksmachine.Machine, error) { + c.createCalls++ + return nil, errors.New("unexpected Create call") +} + +func (c *preflightMachineClient) Get(context.Context) (*aksmachine.Machine, error) { + return c.machine, c.getErr +} + +func (*preflightMachineClient) PatchStatus(context.Context, aksmachine.Status) error { + return errors.New("unexpected PatchStatus call") +}