diff --git a/cmd/aks-flex-node/main.go b/cmd/aks-flex-node/main.go index e355f51e..a2f26019 100644 --- a/cmd/aks-flex-node/main.go +++ b/cmd/aks-flex-node/main.go @@ -47,7 +47,7 @@ func newRootCommand() *cobra.Command { rootCmd.AddCommand(start.NewCommand()) rootCmd.AddCommand(bootstrapdata.NewCommand()) rootCmd.AddCommand(preflight.NewCommand()) - rootCmd.AddCommand(daemon.NewCommand()) + rootCmd.AddCommand(daemon.NewCommands()...) rootCmd.AddCommand(nspawnlifecycle.NewCommand()) rootCmd.AddCommand(reset.NewCommand()) rootCmd.AddCommand(version.NewCommand()) diff --git a/docs/usages/operations.md b/docs/usages/operations.md index fae735a4..bd2bbc9b 100644 --- a/docs/usages/operations.md +++ b/docs/usages/operations.md @@ -47,6 +47,42 @@ systemctl is-active aks-flex-node-agent journalctl -u aks-flex-node-agent -f ``` +## Managed Agent Upgrade + +When the Unbounded `MachineOperation` API is installed, submit an `AgentUpgrade` with an HTTP or HTTPS release archive and, when available, the SHA-256 of the compressed archive: + +```yaml +apiVersion: unbounded-cloud.io/v1alpha3 +kind: MachineOperation +metadata: + name: upgrade-agent-worker-01 +spec: + machineRef: worker-01 + operationKind: AgentUpgrade + parameters: + downloadURL: https://example.com/aks-flex-node-linux-amd64.tar.gz + sha256: 0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef +``` + +The archive must contain exactly the architecture-specific release member used by AKS Flex Node (`aks-flex-node-linux-amd64` or `aks-flex-node-linux-arm64`). The `sha256` parameter is optional; when supplied, the daemon verifies the compressed archive digest. Prefer HTTPS and a digest for production downloads. Plain HTTP is intended for explicitly trusted networks such as a VM-local loopback server; omit the digest only when both the archive source and transport path are trusted. The daemon always verifies the candidate `version` command before switching its blue/green binary links. It also atomically updates the binary in the active nspawn rootfs so kubelet exec authentication uses the same version. + +The restarted daemon marks the operation `Complete`. If the candidate cannot remain running, systemd restores the last-known-good host and nspawn binaries and marks the operation `Failed`. URL query strings, which may contain SAS credentials, are omitted from logs and operation status. + +MachineOperations are cluster-scoped. The daemon group requires cluster-wide read access to MachineOperations and Nodes, plus MachineOperation status update access, so restrict who can create operations and treat parameter values as sensitive API data. Prefer short-lived, read-only download credentials. + +```bash +kubectl get machineoperation upgrade-agent-worker-01 -w +``` + +A host provisioning system that has already authenticated and staged a candidate can activate it directly without creating an Unbounded `MachineOperation`: + +```bash +sudo /var/tmp/aks-flex-node-candidate agent-upgrade --preflight +sudo /var/tmp/aks-flex-node-candidate agent-upgrade +``` + +The candidate must be staged separately from the installed binary. Direct activation and `MachineOperation` activation share one host lock and refuse to overlap with a pending operation signal. Both paths verify the candidate, switch the same blue/green layout, and restore last-good on activation failure. If `aks-flex-node-agent.service` is active, direct activation restarts it, verifies the running executable, and synchronizes the active nspawn exec-credential binary. If the service is already inactive during reset/rejoin provisioning, activation preserves that stopped state; the subsequent bootstrap starts the service and worker. + ## Nspawn Worker Inspect the local nspawn-backed worker: diff --git a/docs/usages/operator-first-boot.md b/docs/usages/operator-first-boot.md index a461db35..822a1d90 100644 --- a/docs/usages/operator-first-boot.md +++ b/docs/usages/operator-first-boot.md @@ -249,7 +249,7 @@ label. ## 3. Install temporary AKS Flex daemon RBAC > [!IMPORTANT] -> **Temporary preview requirement:** When the Machina MachineOperation CRD is +> **Temporary preview requirement:** When the Unbounded MachineOperation CRD is > installed, AKS Flex Node discovers it and enables its MachineOperation > reconciler. A future AKS RP release will install and manage the required > ClusterRole and ClusterRoleBinding automatically as part of FlexNodes pool @@ -287,7 +287,17 @@ rules: resources: - machineoperations/status verbs: + - get + - patch - update +- apiGroups: + - "" + resources: + - nodes + verbs: + - get + - list + - watch --- apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding diff --git a/hack/controller-deployment/rbac.yaml b/hack/controller-deployment/rbac.yaml index bea3b066..acf0ed58 100644 --- a/hack/controller-deployment/rbac.yaml +++ b/hack/controller-deployment/rbac.yaml @@ -103,3 +103,33 @@ subjects: - apiGroup: rbac.authorization.k8s.io kind: Group name: aks-flex-node-daemons +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: aks-flex-node-daemon-machineoperations +rules: + - apiGroups: ["unbounded-cloud.io"] + resources: ["machineoperations"] + verbs: ["get", "list", "watch"] + - apiGroups: ["unbounded-cloud.io"] + resources: ["machineoperations/status"] + verbs: ["get", "patch", "update"] + # The shared MachineOperation selector implementation evaluates labels on the + # local Node when an operation uses machineSelector. + - apiGroups: [""] + resources: ["nodes"] + verbs: ["get", "list", "watch"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRoleBinding +metadata: + name: aks-flex-node-daemon-machineoperations +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: ClusterRole + name: aks-flex-node-daemon-machineoperations +subjects: + - apiGroup: rbac.authorization.k8s.io + kind: Group + name: aks-flex-node-daemons diff --git a/hack/e2e/README.md b/hack/e2e/README.md index 4c056317..964224c8 100644 --- a/hack/e2e/README.md +++ b/hack/e2e/README.md @@ -54,8 +54,9 @@ The default `all` command runs: 7. Unjoin all Flex Nodes and verify they are absent, including reset cleanup of host network artifacts. 8. Rejoin all Flex Nodes and validate again. 9. Validate the installed nspawn lifecycle helper and generated systemd hooks, then reconcile a running node through the helper. -10. Run controller-machine-driven repave validation. -11. Collect logs and clean up Azure resources. +10. Validate managed agent upgrade, forced rollback, retry, nspawn synchronization, and kubelet authentication. +11. Run controller-machine-driven repave validation after the agent upgrade. +12. Collect logs and clean up Azure resources. ## Commands @@ -63,7 +64,7 @@ The default `all` command runs: | Command | Description | |---------|-------------| -| `all` | Full flow: build, infra, join, validate, unjoin, validate absent, rejoin, validate, repave, logs, cleanup. | +| `all` | Full flow: build, infra, join, validate, unjoin, validate absent, rejoin, validate, lifecycle, agent upgrade, repave, logs, cleanup. | | `infra` | Deploy AKS cluster, four VMs, Unbounded-Net CNI, the local registry, and the in-cluster controller. | | `join` | Join all Flex Node VMs. | | `join-msi` | Join only the managed-identity node. | @@ -79,6 +80,7 @@ The default `all` command runs: | `validate-absent` | Verify Flex Node objects are absent after unjoin. | | `smoke` | Run smoke workloads only. | | `nspawn-lifecycle` | Validate lifecycle helper installation and generated hooks on all nodes, then regenerate config and restart the token node through lifecycle reconciliation. | +| `agent-upgrade` | Validate managed agent upgrade, forced rollback, retry, direct host activation, and nspawn synchronization. | | `upgrade-drift` | Validate controller-machine-driven repave to the alternate nspawn side. | | `logs` | Collect logs from VMs. | | `cleanup` | Collect logs and delete Azure resources. | @@ -123,6 +125,7 @@ Additional environment variables: | `E2E_SSH_WAIT_TIMEOUT` | `300` | Timeout in seconds while waiting for SSH. | | `E2E_NODE_JOIN_TIMEOUT` | `300` | Timeout in seconds while waiting for node bootstrap. | | `E2E_POD_READY_TIMEOUT` | `120` | Timeout in seconds while waiting for smoke pods. | +| `E2E_AGENT_UPGRADE_TIMEOUT` | `300` | Timeout in seconds while waiting for an AgentUpgrade result. | | `E2E_DRIFT_UPGRADE_TIMEOUT` | `900` | Timeout in seconds while waiting for repave. | | `AZURE_SUBSCRIPTION_ID` | auto-detected | Azure subscription. | | `AZURE_TENANT_ID` | auto-detected | Azure tenant. | @@ -143,7 +146,26 @@ its Azure resource name remains lowercase. This verifies that an omitted `agent.nodeName` is derived from the normalized hostname and still joins the cluster under the lowercase VM name. -Each join path uploads the locally built binary, renders a config file, installs the binary through `scripts/install.sh` with `AKS_FLEX_NODE_LOCAL_BINARY`, and starts the node through a transient systemd unit. The installed agent service is then validated with systemd checks. +Each join path uploads the locally built binary and renders a config file. Fresh hosts install it through `scripts/install.sh` with `AKS_FLEX_NODE_LOCAL_BINARY`; rejoin hosts with an existing managed layout invoke the uploaded candidate's `agent-upgrade` command before bootstrap. The node starts through a transient systemd unit, and the installed agent service is then validated with systemd checks. + +## Agent Upgrade Validation + +The `agent-upgrade` command uses the bootstrap-token VM to exercise the complete managed binary lifecycle: + +1. Serve architecture-specific release archives over VM-local loopback HTTP to validate HTTP transport support. +2. Submit an `AgentUpgrade` with an archive SHA-256 and a query credential. +3. Verify successful daemon restart, operation completion, binary replacement, and host/nspawn binary equality. +4. Restart kubelet to exercise the synchronized nspawn exec-credential binary and require the Node to remain Ready. +5. Upgrade to a candidate that passes `version` but fails daemon startup, then verify automatic rollback and a failed operation. +6. Confirm status does not expose the sensitive URL query and retry successfully into the inactive slot without the optional archive digest. +7. Stage a distinct candidate and validate direct host activation preflight, inactive-slot switch, service health, shared layout, and active-nspawn synchronization without creating a `MachineOperation` signal. +8. Restart kubelet through the directly activated nspawn credential binary, require Lease renewal and Node readiness, then run a workload before the subsequent repave test. + +Run it against an already joined environment: + +```bash +./hack/e2e/run.sh agent-upgrade +``` ## Nspawn Lifecycle Validation @@ -234,6 +256,7 @@ hack/e2e/ node-join-offline.sh Offline artifacts join/unjoin. node-join-kubeadm.sh Kubeadm-style bootstrap-token join/unjoin. nspawn-lifecycle.sh Lifecycle helper and managed restart validation. + agent-upgrade.sh Managed agent upgrade and rollback validation. upgrade-drift.sh Controller machine goal repave validation. validate.sh Node readiness and smoke tests. cleanup.sh Log collection and Azure resource cleanup. diff --git a/hack/e2e/lib/agent-upgrade.sh b/hack/e2e/lib/agent-upgrade.sh new file mode 100644 index 00000000..63a2d704 --- /dev/null +++ b/hack/e2e/lib/agent-upgrade.sh @@ -0,0 +1,284 @@ +#!/usr/bin/env bash +# ============================================================================= +# AgentUpgrade blue/green, rollback, retry, and nspawn synchronization E2E test. +# ============================================================================= +set -euo pipefail + +[[ -n "${_E2E_AGENT_UPGRADE_LOADED:-}" ]] && return 0 +readonly _E2E_AGENT_UPGRADE_LOADED=1 + +# shellcheck disable=SC1091 +source "$(dirname "${BASH_SOURCE[0]}")/common.sh" + +_agent_upgrade_ensure_api() { + local unbounded_dir + unbounded_dir="$(cd "${REPO_ROOT}" && go list -m -f '{{.Dir}}' github.com/Azure/unbounded)" + kubectl apply -f "${unbounded_dir}/deploy/machina/crd/unbounded-cloud.io_machineoperations.yaml" + kubectl wait --for=condition=Established \ + customresourcedefinition/machineoperations.unbounded-cloud.io --timeout=60s +} + +_agent_upgrade_prepare_server() { + local vm_ip="$1" + local upload_path="/tmp/aks-flex-node-e2e-upgrade-binary" + remote_copy "${E2E_BINARY}" "${vm_ip}" "${upload_path}" + + remote_exec "${vm_ip}" 'bash -s' <<'REMOTE' +set -euo pipefail +work=/opt/aks-flex-node-e2e-upgrade +sudo rm -rf "${work}" +sudo install -d -m 0755 "${work}" +sudo install -m 0755 /tmp/aks-flex-node-e2e-upgrade-binary "${work}/aks-flex-node-linux-amd64" +sudo tar -C "${work}" -czf "${work}/success.tar.gz" aks-flex-node-linux-amd64 + +cat >/tmp/aks-flex-node-e2e-broken <<'BROKEN' +#!/bin/sh +if [ "${1:-}" = "version" ]; then + echo "e2e-forced-daemon-failure" + exit 0 +fi +exit 42 +BROKEN +sudo install -m 0755 /tmp/aks-flex-node-e2e-broken "${work}/aks-flex-node-linux-amd64" +sudo tar -C "${work}" -czf "${work}/failure.tar.gz" aks-flex-node-linux-amd64 +check_dir="$(mktemp -d)" +sudo tar -C "${check_dir}" -xzf "${work}/failure.tar.gz" +sudo "${check_dir}/aks-flex-node-linux-amd64" version >/dev/null +if sudo "${check_dir}/aks-flex-node-linux-amd64" agent >/dev/null 2>&1; then + echo 'forced-failure archive unexpectedly starts the daemon command' >&2 + exit 1 +fi +rm -rf "${check_dir}" +sudo install -m 0755 /tmp/aks-flex-node-e2e-upgrade-binary "${work}/aks-flex-node-linux-amd64" + +sudo systemctl stop aks-flex-node-e2e-upgrade-server.service 2>/dev/null || true +sudo systemd-run --unit=aks-flex-node-e2e-upgrade-server.service \ + --property=WorkingDirectory="${work}" \ + /usr/bin/python3 -m http.server 18080 --bind 127.0.0.1 >/dev/null +for _ in $(seq 1 30); do + if curl --silent --fail http://127.0.0.1:18080/success.tar.gz >/dev/null; then + exit 0 + fi + sleep 1 +done +echo 'upgrade archive server did not become ready' >&2 +exit 1 +REMOTE +} + +_agent_upgrade_digest() { + local vm_ip="$1" archive="$2" + remote_exec "${vm_ip}" "sha256sum /opt/aks-flex-node-e2e-upgrade/${archive} | awk '{print \$1}'" +} + +_agent_upgrade_apply() { + local operation="$1" vm_name="$2" archive="$3" digest="$4" token="$5" digest_parameter="" + if [[ -n "${digest}" ]]; then + digest_parameter=" sha256: ${digest}" + fi + cat </dev/null || true)" + ready="$(kubectl get node "${vm_name}" -o jsonpath='{.status.conditions[?(@.type=="Ready")].status}' 2>/dev/null || true)" + if [[ -n "${renew}" && "${renew}" != "${before_renew}" && "${ready}" == "True" ]]; then + log_success "Kubelet renewed its lease after restart using the synchronized exec credential binary" + return 0 + fi + sleep 5 + elapsed=$((elapsed + 5)) + done + log_error "Kubelet did not renew its lease after AgentUpgrade" + return 1 +} + +_agent_upgrade_direct_activation() { + local vm_name="$1" vm_ip="$2" before_snapshot before_slot before_digest after_snapshot after_slot after_digest + before_snapshot="$(_agent_upgrade_snapshot "${vm_ip}")" + IFS='|' read -r before_slot _ before_digest _ <<<"${before_snapshot}" + + remote_exec "${vm_ip}" 'bash -s' <<'REMOTE' +set -euo pipefail +work=/opt/aks-flex-node-e2e-upgrade +candidate="${work}/aks-flex-node-direct-candidate" +current_link=/usr/local/lib/aks-flex-node/aks-flex-node-current +last_good_link=/usr/local/lib/aks-flex-node/aks-flex-node-last-good +service=/etc/systemd/system/aks-flex-node-agent.service + +sudo cp "${work}/aks-flex-node-linux-amd64" "${candidate}" +printf '\nAKS-FLEX-DIRECT-ACTIVATION-E2E\n' | sudo tee -a "${candidate}" >/dev/null +sudo chmod 0755 "${candidate}" +current_before="$(sudo readlink -f "${current_link}")" +last_good_before="$(sudo readlink -f "${last_good_link}")" +unit_before="$(sudo sha256sum "${service}" | awk '{print $1}')" + +sudo "${candidate}" agent-upgrade --preflight | tee /tmp/direct-agent-upgrade-preflight.log +[[ "$(sudo readlink -f "${current_link}")" == "${current_before}" ]] +[[ "$(sudo readlink -f "${last_good_link}")" == "${last_good_before}" ]] +[[ "$(sudo sha256sum "${service}" | awk '{print $1}')" == "${unit_before}" ]] +sudo "${candidate}" agent-upgrade | tee /tmp/direct-agent-upgrade.log +current_after="$(sudo readlink -f "${current_link}")" +[[ "${current_after}" != "${current_before}" ]] +[[ "$(sudo readlink -f "${last_good_link}")" == "${current_before}" ]] +[[ "$(sudo readlink -f /usr/local/bin/aks-flex-node)" == "${current_after}" ]] +[[ "$(sudo sha256sum "${candidate}" | awk '{print $1}')" == "$(sudo sha256sum "${current_after}" | awk '{print $1}')" ]] +sudo grep -Fq "ExecStart=${current_link} agent" "${service}" +sudo systemctl is-active --quiet aks-flex-node-agent.service +pid="$(sudo systemctl show --property MainPID --value aks-flex-node-agent.service)" +[[ "$(sudo readlink -f "/proc/${pid}/exe")" == "${current_after}" ]] +[[ ! -e /etc/aks-flex-node/agent-upgrade-signal.json ]] +REMOTE + + after_snapshot="$(_agent_upgrade_snapshot "${vm_ip}")" + IFS='|' read -r after_slot _ after_digest _ <<<"${after_snapshot}" + if [[ -z "${after_slot}" || "${after_slot}" == "${before_slot}" || "${after_digest}" == "${before_digest}" ]]; then + log_error "Direct activation did not install a distinct inactive-slot candidate: before=${before_snapshot} after=${after_snapshot}" + return 1 + fi + _agent_upgrade_assert_synchronized "${vm_ip}" + _agent_upgrade_validate_kubelet_auth "${vm_name}" "${vm_ip}" + validate_node_joined "${vm_name}" + log_success "Direct host activation preflight, switch, service health, and nspawn synchronization passed" +} + +agent_upgrade_e2e() { + log_section "Managed and Direct AgentUpgrade E2E" + local vm_name vm_ip suffix success_digest failure_digest before before_slot success_snapshot success_slot success_binary_digest rollback_snapshot rollback_binary_digest retry_snapshot retry_binary_digest + vm_name="$(state_get token_vm_name)" + vm_ip="$(state_get token_vm_ip)" + suffix="$(date +%s)" + + validate_node_joined "${vm_name}" + _agent_upgrade_ensure_api + # MachineOperation discovery occurs during daemon startup. Restart after the + # CRD is established so this focused command also works on previously joined + # environments where the API was absent. + remote_exec "${vm_ip}" 'sudo systemctl restart aks-flex-node-agent.service' + validate_node_joined "${vm_name}" + _agent_upgrade_prepare_server "${vm_ip}" + success_digest="$(_agent_upgrade_digest "${vm_ip}" success.tar.gz)" + failure_digest="$(_agent_upgrade_digest "${vm_ip}" failure.tar.gz)" + before="$(_agent_upgrade_snapshot "${vm_ip}")" + before_slot="$(cut -d'|' -f1 <<<"${before}")" + log_info "Pre-upgrade agent snapshot: ${before}" + + local success_op="agent-upgrade-success-${suffix}" + _agent_upgrade_apply "${success_op}" "${vm_name}" success.tar.gz "${success_digest}" "success-${suffix}" + _agent_upgrade_wait_phase "${success_op}" Complete + validate_node_joined "${vm_name}" + _agent_upgrade_assert_synchronized "${vm_ip}" + success_snapshot="$(_agent_upgrade_snapshot "${vm_ip}")" + IFS='|' read -r success_slot _ success_binary_digest _ <<<"${success_snapshot}" + if [[ -z "${success_slot}" || "${success_slot}" == "${before_slot}" ]]; then + log_error "Successful AgentUpgrade did not switch the active binary slot: before=${before} after=${success_snapshot}" + return 1 + fi + + # Restart kubelet and require a fresh Lease renewal so this proves the + # synchronized nspawn exec-credential binary can still authenticate. + _agent_upgrade_validate_kubelet_auth "${vm_name}" "${vm_ip}" + + local failure_op="agent-upgrade-rollback-${suffix}" + _agent_upgrade_apply "${failure_op}" "${vm_name}" failure.tar.gz "${failure_digest}" "failure-${suffix}" + _agent_upgrade_wait_phase "${failure_op}" Failed + rollback_snapshot="$(_agent_upgrade_snapshot "${vm_ip}")" + IFS='|' read -r _ _ rollback_binary_digest _ <<<"${rollback_snapshot}" + if [[ "${rollback_binary_digest}" != "${success_binary_digest}" ]]; then + log_error "Rollback did not restore the successful candidate: success=${success_snapshot} rollback=${rollback_snapshot}" + return 1 + fi + if kubectl get machineoperation "${failure_op}" -o jsonpath='{.status.message}' | grep -q "failure-${suffix}"; then + log_error "AgentUpgrade status leaked sensitive URL query data" + return 1 + fi + remote_exec "${vm_ip}" 'sudo systemctl is-active --quiet aks-flex-node-agent.service' + _agent_upgrade_assert_synchronized "${vm_ip}" + validate_node_joined "${vm_name}" + + local retry_op="agent-upgrade-retry-${suffix}" + # The digest is optional when the VM-local source and loopback transport + # provide the trust boundary. + _agent_upgrade_apply "${retry_op}" "${vm_name}" success.tar.gz "" "retry-${suffix}" + _agent_upgrade_wait_phase "${retry_op}" Complete + retry_snapshot="$(_agent_upgrade_snapshot "${vm_ip}")" + IFS='|' read -r _ _ retry_binary_digest _ <<<"${retry_snapshot}" + if [[ "${retry_binary_digest}" != "${success_binary_digest}" ]]; then + log_error "Retry did not install the expected successful candidate: ${retry_snapshot}" + return 1 + fi + _agent_upgrade_assert_synchronized "${vm_ip}" + validate_node_joined "${vm_name}" + + _agent_upgrade_direct_activation "${vm_name}" "${vm_ip}" + smoke_test "${vm_name}" "agent-upgrade" + + log_success "Managed AgentUpgrade success/rollback/retry and direct host activation E2E passed" +} diff --git a/hack/e2e/lib/controller.sh b/hack/e2e/lib/controller.sh index 2f2046ac..37c990e1 100644 --- a/hack/e2e/lib/controller.sh +++ b/hack/e2e/lib/controller.sh @@ -159,9 +159,6 @@ _build_controller_image() { log_section "Building AKS Flex Controller Image" log_info "Building controller image ${local_image} and pushing to in-cluster local registry" - pf_pid="" - _start_registry_port_forward pf_pid "${local_port}" || return 1 - if ! ( cd "${REPO_ROOT}" DOCKER_BUILDKIT=1 docker build \ @@ -172,13 +169,26 @@ _build_controller_image() { --build-arg "BUILD_TIME=${build_time}" \ --tag "${local_image}" \ . - docker push "${local_image}" ); then + return 1 + fi + + local pushed=0 attempt + for attempt in 1 2 3; do + pf_pid="" + if _start_registry_port_forward pf_pid "${local_port}" && docker push "${local_image}"; then + pushed=1 + _stop_registry_port_forward "${pf_pid}" + break + fi _stop_registry_port_forward "${pf_pid}" + log_warn "Controller image push attempt ${attempt} failed; reopening registry tunnel" + sleep 2 + done + if [[ "${pushed}" != "1" ]]; then return 1 fi - _stop_registry_port_forward "${pf_pid}" docker image rm "${local_image}" >/dev/null 2>&1 || true out_image="${cluster_image}" } @@ -285,7 +295,14 @@ _wait_for_controller_ready() { } _ensure_flex_controller_unlocked() { - local image + local image unbounded_dir + # Install the optional API before any Flex daemon starts so its startup-time + # discovery enables MachineOperation watches without requiring a restart. + unbounded_dir="$(cd "${REPO_ROOT}" && go list -m -f '{{.Dir}}' github.com/Azure/unbounded)" + kubectl apply -f "${unbounded_dir}/deploy/machina/crd/unbounded-cloud.io_machineoperations.yaml" || return 1 + kubectl wait --for=condition=Established \ + customresourcedefinition/machineoperations.unbounded-cloud.io --timeout=60s || return 1 + image="$(_controller_image_from_state_or_env)" if [[ -n "${E2E_CONTROLLER_IMAGE:-}" ]]; then diff --git a/hack/e2e/lib/node-join.sh b/hack/e2e/lib/node-join.sh index 63db3c5c..b2948d55 100755 --- a/hack/e2e/lib/node-join.sh +++ b/hack/e2e/lib/node-join.sh @@ -39,12 +39,19 @@ _deploy_and_start_agent() { remote_exec "${vm_ip}" "UNIT_NAME=${unit_name} E2E_NODE_JOIN_TIMEOUT=${E2E_NODE_JOIN_TIMEOUT} E2E_KUBERNETES_VERSION=${E2E_KUBERNETES_VERSION} bash -s" <<'REMOTE' set -euo pipefail -sudo AKS_FLEX_NODE_LOCAL_BINARY=/tmp/aks-flex-node-binary \ - AKS_FLEX_NODE_VERSION=e2e-local \ - SKIP_AZCLI=true \ - bash /tmp/aks-flex-node-install.sh --yes +managed_current=/usr/local/lib/aks-flex-node/aks-flex-node-current +if [[ -e "${managed_current}" || -L "${managed_current}" ]]; then + echo "Existing managed layout found; activating the separately staged E2E candidate..." + sudo chmod 0755 /tmp/aks-flex-node-binary + sudo /tmp/aks-flex-node-binary agent-upgrade +else + sudo AKS_FLEX_NODE_LOCAL_BINARY=/tmp/aks-flex-node-binary \ + AKS_FLEX_NODE_VERSION=e2e-local \ + SKIP_AZCLI=true \ + bash /tmp/aks-flex-node-install.sh --yes +fi -aks-flex-node version +sudo /usr/local/bin/aks-flex-node version sudo cp /tmp/config.json /etc/aks-flex-node/ diff --git a/hack/e2e/run.sh b/hack/e2e/run.sh index 5032fc5f..04cfdb59 100755 --- a/hack/e2e/run.sh +++ b/hack/e2e/run.sh @@ -23,6 +23,7 @@ # validate-absent Verify all flex nodes are gone after unjoin # smoke Run smoke tests only (pods on flex nodes) # nspawn-lifecycle Validate generated lifecycle hooks and restart reconciliation +# agent-upgrade Validate managed binary upgrade, rollback, and retry # upgrade-drift Run controller-machine Kubernetes version drift repave test # logs Collect logs from VMs # cleanup Tear down Azure resources @@ -107,6 +108,8 @@ source "${SCRIPT_DIR}/lib/nspawn-lifecycle.sh" # shellcheck disable=SC1091 source "${SCRIPT_DIR}/lib/upgrade-drift.sh" # shellcheck disable=SC1091 +source "${SCRIPT_DIR}/lib/agent-upgrade.sh" +# shellcheck disable=SC1091 source "${SCRIPT_DIR}/lib/cleanup.sh" # shellcheck disable=SC1091 source "${SCRIPT_DIR}/lib/runner.sh" @@ -131,7 +134,7 @@ usage() { parse_args() { while [[ $# -gt 0 ]]; do case "$1" in - all|infra|join|join-msi|join-token|join-offline|join-kubeadm|unjoin|unjoin-msi|unjoin-token|unjoin-offline|unjoin-kubeadm|validate|validate-absent|smoke|nspawn-lifecycle|upgrade-drift|logs|cleanup|runner-cleanup|status) + all|infra|join|join-msi|join-token|join-offline|join-kubeadm|unjoin|unjoin-msi|unjoin-token|unjoin-offline|unjoin-kubeadm|validate|validate-absent|smoke|nspawn-lifecycle|agent-upgrade|upgrade-drift|logs|cleanup|runner-cleanup|status) COMMAND="$1"; shift ;; -g|--resource-group) export E2E_RESOURCE_GROUP="$2"; shift 2 ;; -l|--location) export E2E_LOCATION="$2"; shift 2 ;; @@ -203,7 +206,10 @@ cmd_all() { # ── Host nspawn lifecycle restart ────────────────────────────────────── nspawn_lifecycle_all - # ── Controller-backed machine repave ─────────────────────────────────── + # ── Managed host agent binary upgrade ───────────────────────────────── + agent_upgrade_e2e + + # ── Controller-backed machine repave after agent upgrade ─────────────── upgrade_drift_all # Collect logs (always, even if tests fail) @@ -319,6 +325,11 @@ main() { nspawn-lifecycle) nspawn_lifecycle_all ;; + agent-upgrade) + ensure_binary + ensure_cluster_dependencies + agent_upgrade_e2e + ;; upgrade-drift) ensure_binary ensure_cluster_dependencies diff --git a/pkg/cmd/daemon/agent_upgrade_recovery.go b/pkg/cmd/daemon/agent_upgrade_recovery.go new file mode 100644 index 00000000..40287004 --- /dev/null +++ b/pkg/cmd/daemon/agent_upgrade_recovery.go @@ -0,0 +1,29 @@ +package daemon + +import ( + "fmt" + + "github.com/spf13/cobra" + + hostdaemon "github.com/Azure/AKSFlexNode/pkg/daemon" +) + +// newAgentUpgradeRecoveryCommand remains callable only as a local root +// operation by the systemd recovery unit. +func newAgentUpgradeRecoveryCommand() *cobra.Command { + var message string + cmd := &cobra.Command{ + Use: "recover-agent-upgrade", + Short: "Restore the last-known-good agent after a failed upgrade", + Hidden: true, + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + if err := hostdaemon.RecoverAgentUpgrade(cmd.Context(), message); err != nil { + return fmt.Errorf("recover AgentUpgrade: %w", err) + } + return nil + }, + } + cmd.Flags().StringVar(&message, "message", "", "failure message to publish") + return cmd +} diff --git a/pkg/cmd/daemon/daemon.go b/pkg/cmd/daemon/daemon.go index 45453d6b..af9ae234 100644 --- a/pkg/cmd/daemon/daemon.go +++ b/pkg/cmd/daemon/daemon.go @@ -10,7 +10,16 @@ import ( "github.com/Azure/AKSFlexNode/pkg/logger" ) -func NewCommand() *cobra.Command { +// NewCommands returns all daemon runtime and internal lifecycle commands. +func NewCommands() []*cobra.Command { + return []*cobra.Command{ + newCommand(), + newHostAgentUpgradeCommand(), + newAgentUpgradeRecoveryCommand(), + } +} + +func newCommand() *cobra.Command { var configPath string cmd := &cobra.Command{ Use: "daemon", diff --git a/pkg/cmd/daemon/daemon_test.go b/pkg/cmd/daemon/daemon_test.go new file mode 100644 index 00000000..b1ed6882 --- /dev/null +++ b/pkg/cmd/daemon/daemon_test.go @@ -0,0 +1,31 @@ +package daemon + +import "testing" + +func TestNewCommands(t *testing.T) { + t.Parallel() + + commands := NewCommands() + if len(commands) != 3 { + t.Fatalf("len(NewCommands()) = %d, want 3", len(commands)) + } + want := map[string]bool{ + "daemon": false, + "agent-upgrade": false, + "recover-agent-upgrade": false, + } + for _, command := range commands { + if _, ok := want[command.Name()]; !ok { + t.Fatalf("unexpected command %q", command.Name()) + } + if want[command.Name()] { + t.Fatalf("duplicate command %q", command.Name()) + } + want[command.Name()] = true + } + for name, found := range want { + if !found { + t.Errorf("command %q is missing", name) + } + } +} diff --git a/pkg/cmd/daemon/host_agent_upgrade.go b/pkg/cmd/daemon/host_agent_upgrade.go new file mode 100644 index 00000000..ea497d58 --- /dev/null +++ b/pkg/cmd/daemon/host_agent_upgrade.go @@ -0,0 +1,65 @@ +package daemon + +import ( + "context" + "fmt" + "io" + "log/slog" + "os" + "path/filepath" + + "github.com/spf13/cobra" + + hostdaemon "github.com/Azure/AKSFlexNode/pkg/daemon" + "github.com/Azure/AKSFlexNode/pkg/logger" +) + +func newHostAgentUpgradeCommand() *cobra.Command { + var preflight bool + cmd := &cobra.Command{ + Use: "agent-upgrade", + Short: "Activate this executable as the host agent daemon", + Hidden: true, + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + candidate, err := os.Executable() + if err != nil { + return fmt.Errorf("resolve candidate executable: %w", err) + } + candidate, err = filepath.Abs(candidate) + if err != nil { + return fmt.Errorf("resolve absolute candidate executable path: %w", err) + } + log := logger.CreateLogger("info", "") + candidate = filepath.Clean(candidate) + if preflight { + return runHostAgentUpgradePreflight(cmd.Context(), cmd.OutOrStdout(), log, candidate) + } + result, err := hostdaemon.ActivateHostAgent(cmd.Context(), log, candidate) + if err != nil { + return err + } + _, err = fmt.Fprintf(cmd.OutOrStdout(), "activated host agent daemon: %s -> %s\n", result.PreviousPath, result.CurrentPath) + return err + }, + } + cmd.Flags().BoolVar(&preflight, "preflight", false, "Show and validate the host activation plan without applying it") + return cmd +} + +func runHostAgentUpgradePreflight(ctx context.Context, output io.Writer, log *slog.Logger, candidate string) error { + plan, err := hostdaemon.PreflightHostAgentActivation(ctx, log, candidate) + if err != nil { + return err + } + if _, err := fmt.Fprintf(output, "Candidate: %s\nActive binary: %s\nInstall target: %s\n", plan.CandidatePath, plan.ActivePath, plan.TargetPath); err != nil { + return err + } + for _, action := range plan.Actions { + if _, err := fmt.Fprintf(output, "- %s\n", action); err != nil { + return err + } + } + _, err = fmt.Fprintln(output, "Preflight: no changes applied") + return err +} diff --git a/pkg/config/artifacts.go b/pkg/config/artifacts.go new file mode 100644 index 00000000..ee88867a --- /dev/null +++ b/pkg/config/artifacts.go @@ -0,0 +1,20 @@ +package config + +import "fmt" + +// AgentBinaryBaseName is the installed command and release artifact base name. +const AgentBinaryBaseName = "aks-flex-node" + +// AgentBinaryArchiveMember returns the binary member name used by a release +// archive for the requested platform. +func AgentBinaryArchiveMember(goos, goarch string) (string, error) { + if goos != "linux" { + return "", fmt.Errorf("unsupported agent release operating system %q", goos) + } + switch goarch { + case "amd64", "arm64": + return AgentBinaryBaseName + "-" + goos + "-" + goarch, nil + default: + return "", fmt.Errorf("unsupported agent release architecture %q", goarch) + } +} diff --git a/pkg/config/artifacts_test.go b/pkg/config/artifacts_test.go new file mode 100644 index 00000000..e7c8b673 --- /dev/null +++ b/pkg/config/artifacts_test.go @@ -0,0 +1,47 @@ +package config + +import "testing" + +func TestAgentBinaryArchiveMember(t *testing.T) { + t.Parallel() + + tests := map[string]struct { + goos string + goarch string + want string + wantErr bool + }{ + "amd64": { + goos: "linux", + goarch: "amd64", + want: "aks-flex-node-linux-amd64", + }, + "arm64": { + goos: "linux", + goarch: "arm64", + want: "aks-flex-node-linux-arm64", + }, + "unsupported operating system": { + goos: "windows", + goarch: "amd64", + wantErr: true, + }, + "unsupported architecture": { + goos: "linux", + goarch: "riscv64", + wantErr: true, + }, + } + for name, tt := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + got, err := AgentBinaryArchiveMember(tt.goos, tt.goarch) + if (err != nil) != tt.wantErr { + t.Fatalf("AgentBinaryArchiveMember() error = %v, wantErr %v", err, tt.wantErr) + } + if got != tt.want { + t.Fatalf("AgentBinaryArchiveMember() = %q, want %q", got, tt.want) + } + }) + } +} diff --git a/pkg/daemon/agent_upgrade.go b/pkg/daemon/agent_upgrade.go new file mode 100644 index 00000000..a18d7617 --- /dev/null +++ b/pkg/daemon/agent_upgrade.go @@ -0,0 +1,609 @@ +package daemon + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "os" + "path/filepath" + "strings" + "time" + + "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/Azure/AKSFlexNode/pkg/utils/utilexec" + "github.com/Azure/AKSFlexNode/pkg/utils/utilio" + machinav1alpha3 "github.com/Azure/unbounded/api/machina/v1alpha3" + "github.com/Azure/unbounded/pkg/agent/agentbinary" + agentdaemon "github.com/Azure/unbounded/pkg/agent/daemon" + "github.com/Azure/unbounded/pkg/agent/goalstates" +) + +const ( + agentUpgradeDownloadURLParameter = "downloadURL" + agentUpgradeSHA256Parameter = "sha256" +) + +var errAgentUpgradeAlreadyPending = errors.New("AgentUpgrade operation is already pending") + +func defaultAgentUpgradePaths() agentUpgradePaths { + const binaryDir = "/usr/local/lib/aks-flex-node" + return agentUpgradePaths{ + BinaryPath: "/usr/local/bin/aks-flex-node", + BluePath: filepath.Join(binaryDir, "aks-flex-node-blue"), + GreenPath: filepath.Join(binaryDir, "aks-flex-node-green"), + CurrentPath: filepath.Join(binaryDir, "aks-flex-node-current"), + LastGoodPath: filepath.Join(binaryDir, "aks-flex-node-last-good"), + SignalPath: "/etc/aks-flex-node/agent-upgrade-signal.json", + } +} + +type agentUpgradeRequest struct { + downloadURL string + sha256 string +} + +func parseAgentUpgradeRequest(parameters map[string]string) (agentUpgradeRequest, error) { + request := agentUpgradeRequest{ + downloadURL: strings.TrimSpace(parameters[agentUpgradeDownloadURLParameter]), + sha256: strings.TrimSpace(parameters[agentUpgradeSHA256Parameter]), + } + if request.downloadURL == "" { + return agentUpgradeRequest{}, fmt.Errorf("missing required parameter %q", agentUpgradeDownloadURLParameter) + } + if _, err := secureAgentInstallOptions(request.downloadURL, request.sha256); err != nil { + return agentUpgradeRequest{}, err + } + return request, nil +} + +type agentUpgradeSignal struct { + OperationName string `json:"operationName"` + ActiveMachine string `json:"activeMachine,omitempty"` + CandidatePath string `json:"candidatePath,omitempty"` + InitiatingDaemonInstance string `json:"initiatingDaemonInstance,omitempty"` + SwitchCommitted bool `json:"switchCommitted,omitempty"` + RecoveryRequired bool `json:"recoveryRequired,omitempty"` + Failure string `json:"failure,omitempty"` +} + +type agentUpgradeSignalStore struct { + path string +} + +func (s agentUpgradeSignalStore) recordPending(operationName, activeMachine, daemonInstance string) error { + return s.write(agentUpgradeSignal{ + OperationName: operationName, + ActiveMachine: activeMachine, + InitiatingDaemonInstance: daemonInstance, + }) +} + +func (s agentUpgradeSignalStore) recordCandidate(candidatePath string) error { + signal, err := s.read() + if err != nil { + return err + } + if signal == nil { + return fmt.Errorf("no pending AgentUpgrade signal") + } + signal.CandidatePath = candidatePath + return s.write(*signal) +} + +func (s agentUpgradeSignalStore) recordSwitchCommitted() error { + signal, err := s.read() + if err != nil { + return err + } + if signal == nil { + return fmt.Errorf("no pending AgentUpgrade signal") + } + signal.SwitchCommitted = true + return s.write(*signal) +} + +func (s agentUpgradeSignalStore) recordFailure(message string) error { + signal, err := s.read() + if err != nil { + return err + } + if signal == nil { + return nil + } + message = strings.TrimSpace(message) + if message == "" { + message = "upgraded daemon failed to start; restored last-good binary" + } + signal.Failure = message + signal.RecoveryRequired = true + return s.write(*signal) +} + +func (s agentUpgradeSignalStore) read() (*agentUpgradeSignal, error) { + data, err := os.ReadFile(s.path) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil, nil + } + return nil, fmt.Errorf("read AgentUpgrade signal: %w", err) + } + var signal agentUpgradeSignal + if err := json.Unmarshal(data, &signal); err != nil { + return nil, fmt.Errorf("decode AgentUpgrade signal: %w", err) + } + signal.OperationName = strings.TrimSpace(signal.OperationName) + signal.ActiveMachine = strings.TrimSpace(signal.ActiveMachine) + signal.CandidatePath = strings.TrimSpace(signal.CandidatePath) + signal.Failure = strings.TrimSpace(signal.Failure) + if signal.OperationName == "" { + return nil, fmt.Errorf("AgentUpgrade signal has no operation name") + } + if signal.ActiveMachine != "" && !validNspawnMachine(signal.ActiveMachine) { + return nil, fmt.Errorf("AgentUpgrade signal has invalid active machine %q", signal.ActiveMachine) + } + return &signal, nil +} + +func (s agentUpgradeSignalStore) write(signal agentUpgradeSignal) error { + data, err := json.Marshal(signal) + if err != nil { + return fmt.Errorf("encode AgentUpgrade signal: %w", err) + } + if err := utilio.WriteFile(s.path, append(data, '\n'), 0o600); err != nil { + return fmt.Errorf("write AgentUpgrade signal: %w", err) + } + return nil +} + +func (s agentUpgradeSignalStore) clear() error { + if err := os.Remove(s.path); err != nil && !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("remove AgentUpgrade signal: %w", err) + } + return nil +} + +type agentUpgradeExecutor interface { + Acquire() (io.Closer, error) + RecordPending(context.Context, string) error + RetryRecovery(context.Context) error + RecordFailure(string) error + Stage(context.Context, agentUpgradeRequest) error + Abort(context.Context) error + Restart(context.Context) error +} + +type agentUpgradeStateLoader interface { + LoadState(context.Context) (*State, error) +} + +type hostAgentUpgradeExecutor struct { + log *slog.Logger + paths agentUpgradePaths + state agentUpgradeStateLoader + signals agentUpgradeSignalStore + runSystemdRun func(context.Context, ...string) error + finishMachineOperation func(context.Context, client.Client, agentdaemon.MachineOperation, agentdaemon.MachineOperationResult[int64]) error + runningExecutable func() (string, error) + nspawnBinaryPath func(string) string + instanceID string +} + +func newHostAgentUpgradeExecutor(log *slog.Logger, state agentUpgradeStateLoader) (*hostAgentUpgradeExecutor, error) { + paths := defaultAgentUpgradePaths() + instanceID, err := newDaemonInstanceID() + if err != nil { + return nil, fmt.Errorf("create daemon instance ID: %w", err) + } + return &hostAgentUpgradeExecutor{ + log: log, + paths: paths, + state: state, + signals: agentUpgradeSignalStore{path: paths.SignalPath}, + runSystemdRun: func(ctx context.Context, args ...string) error { + return utilexec.RunCmd(ctx, log, utilexec.SystemdRun(), args...) + }, + finishMachineOperation: agentdaemon.FinishMachineOperation, + runningExecutable: runningAgentExecutable, + nspawnBinaryPath: activeNspawnAgentBinaryPath, + instanceID: instanceID, + }, nil +} + +func newDaemonInstanceID() (string, error) { + var value [16]byte + if _, err := rand.Read(value[:]); err != nil { + return "", err + } + return hex.EncodeToString(value[:]), nil +} + +func (e *hostAgentUpgradeExecutor) Acquire() (io.Closer, error) { + return agentbinary.AcquireHostActivationLock(agentUpgradeLockPath) +} + +func (e *hostAgentUpgradeExecutor) RecordPending(ctx context.Context, operationName string) error { + existing, err := e.signals.read() + if err != nil { + return err + } + if existing != nil { + if existing.OperationName == operationName { + return errAgentUpgradeAlreadyPending + } + return fmt.Errorf("another AgentUpgrade operation %q is pending", existing.OperationName) + } + state, err := e.state.LoadState(ctx) + if err != nil { + return fmt.Errorf("load daemon state for AgentUpgrade: %w", err) + } + if state == nil || !validNspawnMachine(state.ActiveMachine) { + return fmt.Errorf("no valid active nspawn machine for AgentUpgrade") + } + if err := e.signals.recordPending(operationName, state.ActiveMachine, e.instanceID); err != nil { + return err + } + return nil +} + +func (e *hostAgentUpgradeExecutor) RetryRecovery(ctx context.Context) error { + signal, err := e.signals.read() + if err != nil { + return err + } + if signal == nil || !signal.RecoveryRequired { + return nil + } + cleanupCtx, cancel := agentUpgradeCleanupContext(ctx) + defer cancel() + if err := e.Restart(cleanupCtx); err != nil { + return fmt.Errorf("retry AgentUpgrade recovery restart: %w", err) + } + return nil +} + +func (e *hostAgentUpgradeExecutor) RecordFailure(message string) error { + return e.signals.recordFailure(message) +} + +func (e *hostAgentUpgradeExecutor) Stage(ctx context.Context, request agentUpgradeRequest) error { + if err := ensureAgentUpgradeLayout(ctx, e.log, e.paths); err != nil { + return fmt.Errorf("initialize agent binary layout: %w", err) + } + current, err := resolvedExecutable(e.paths.CurrentPath) + if err != nil { + return fmt.Errorf("resolve current agent binary: %w", err) + } + candidate := e.paths.BluePath + if current == e.paths.BluePath { + candidate = e.paths.GreenPath + } + if err := e.signals.recordCandidate(candidate); err != nil { + return err + } + if err := installAndSwitchAgentBinary(ctx, e.log, request.downloadURL, request.sha256, e.paths); err != nil { + return err + } + if err := e.signals.recordSwitchCommitted(); err != nil { + return e.rollbackAfterStage(ctx, fmt.Errorf("record committed AgentUpgrade switch: %w", err)) + } + + signal, err := e.signals.read() + if err != nil { + return e.rollbackAfterStage(ctx, err) + } + if signal == nil || !validNspawnMachine(signal.ActiveMachine) { + return e.rollbackAfterStage(ctx, fmt.Errorf("pending AgentUpgrade signal has no valid active machine")) + } + current, err = resolvedExecutable(e.paths.CurrentPath) + if err != nil { + return e.rollbackAfterStage(ctx, fmt.Errorf("resolve staged agent binary: %w", err)) + } + if current != candidate { + return e.rollbackAfterStage(ctx, fmt.Errorf("staged agent binary resolved to unexpected slot")) + } + if err := synchronizeNspawnAgentBinary(current, signal.ActiveMachine); err != nil { + return e.rollbackAfterStage(ctx, fmt.Errorf("synchronize active nspawn agent binary: %w", err)) + } + return nil +} + +func (e *hostAgentUpgradeExecutor) rollbackAfterStage(ctx context.Context, stageErr error) error { + cleanupCtx, cancel := agentUpgradeCleanupContext(ctx) + defer cancel() + if rollbackErr := e.rollback(cleanupCtx); rollbackErr != nil { + return fmt.Errorf("%w; rollback failed: %v", stageErr, rollbackErr) + } + return stageErr +} + +func (e *hostAgentUpgradeExecutor) Abort(ctx context.Context) error { + cleanupCtx, cancel := agentUpgradeCleanupContext(ctx) + defer cancel() + if err := e.rollback(cleanupCtx); err != nil { + // Preserve the signal so startup recovery can retry the rollback. + return err + } + return e.signals.clear() +} + +func agentUpgradeCleanupContext(ctx context.Context) (context.Context, context.CancelFunc) { + return context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) +} + +func (e *hostAgentUpgradeExecutor) rollback(ctx context.Context) error { + if err := ctx.Err(); err != nil { + return err + } + signal, err := e.signals.read() + if err != nil { + return err + } + return rollbackAgentUpgradeFiles(e.paths, signal) +} + +func (e *hostAgentUpgradeExecutor) Restart(ctx context.Context) error { + if err := ctx.Err(); err != nil { + return err + } + // A direct self-restart can terminate the systemctl child before the + // handler records a successful handoff. Schedule the restart in a separate + // transient unit so this process returns while its durable signal is intact. + unit := fmt.Sprintf("aks-flex-node-agent-upgrade-restart-%d", time.Now().UnixNano()) + return e.runSystemdRun( + ctx, + "--quiet", + "--collect", + "--unit="+unit, + "--on-active=1s", + "/usr/bin/systemctl", + "restart", + ServiceUnitName, + ) +} + +func validNspawnMachine(machine string) bool { + return machine == goalstates.NSpawnMachineKube1 || machine == goalstates.NSpawnMachineKube2 +} + +func synchronizeNspawnAgentBinary(sourcePath, machine string) error { + if !validNspawnMachine(machine) { + return fmt.Errorf("invalid nspawn machine %q", machine) + } + destination := filepath.Join("/var/lib/machines", machine, "usr", "local", "bin", "aks-flex-node") + if err := copyExecutable(sourcePath, destination); err != nil { + return fmt.Errorf("copy agent binary to %s: %w", machine, err) + } + return nil +} + +// RecoverAgentUpgrade records failure and restores both host and active nspawn +// binaries. It is invoked by the systemd recovery unit through last-good. +func RecoverAgentUpgrade(ctx context.Context, message string) error { + paths := defaultAgentUpgradePaths() + signals := agentUpgradeSignalStore{path: paths.SignalPath} + if err := signals.recordFailure(message); err != nil { + return err + } + signal, err := signals.read() + if err != nil { + return err + } + if err := rollbackAgentUpgradeFiles(paths, signal); err != nil { + return err + } + return ctx.Err() +} + +func publishAndClearAgentUpgradeSignal(ctx context.Context, log *slog.Logger, c client.Client, executor *hostAgentUpgradeExecutor) error { + paths := executor.paths + signals := executor.signals + signal, err := signals.read() + if err != nil { + return err + } + if signal == nil { + return nil + } + if signal.Failure == "" && signal.InitiatingDaemonInstance == executor.instanceID { + // The initiating daemon must not consume its own signal while staging. + return nil + } + + if signal.Failure == "" { + nspawnBinaryPath := executor.nspawnBinaryPath + if nspawnBinaryPath == nil { + nspawnBinaryPath = activeNspawnAgentBinaryPath + } + if validationErr := validateStartedAgentUpgrade(paths, signal, nspawnBinaryPath); validationErr != nil { + signal.Failure = validationErr.Error() + candidateActive, activeErr := agentUpgradeCandidateIsActive(paths, signal) + if activeErr != nil { + signal.Failure = errors.Join(validationErr, activeErr).Error() + signal.RecoveryRequired = true + } else { + signal.RecoveryRequired = candidateActive + } + if err := signals.write(*signal); err != nil { + return err + } + } + } + + result := agentdaemon.MachineOperationResult[int64]{ + Phase: machinav1alpha3.OperationPhaseComplete, + Reason: "Succeeded", + Message: "AgentUpgrade completed", + } + if signal.Failure != "" { + result.Phase = machinav1alpha3.OperationPhaseFailed + result.Reason = "DaemonFailed" + result.Message = signal.Failure + + if err := rollbackAgentUpgradeFiles(paths, signal); err != nil { + return fmt.Errorf("roll back failed AgentUpgrade: %w", err) + } + } + + if signal.RecoveryRequired { + lastGood, err := resolvedExecutable(paths.LastGoodPath) + if err != nil { + return fmt.Errorf("resolve last-good agent for recovery restart: %w", err) + } + runningExecutable := executor.runningExecutable + if runningExecutable == nil { + runningExecutable = runningAgentExecutable + } + running, err := runningExecutable() + if err != nil { + return err + } + if running != lastGood { + cleanupCtx, cancel := agentUpgradeCleanupContext(ctx) + restartErr := executor.Restart(cleanupCtx) + cancel() + if restartErr != nil { + return fmt.Errorf("restart last-good agent: %w", restartErr) + } + // Keep the signal and terminal status unpublished until the last-good + // process confirms that it is running. + return nil + } + } + finishOperation := executor.finishMachineOperation + if finishOperation == nil { + finishOperation = agentdaemon.FinishMachineOperation + } + finishErr := finishOperation(ctx, c, agentdaemon.MachineOperation{Name: signal.OperationName}, result) + if finishErr != nil { + return fmt.Errorf("publish AgentUpgrade result: %w", finishErr) + } + if err := signals.clear(); err != nil { + return err + } + log.Info("published AgentUpgrade result", "operation", signal.OperationName, "phase", result.Phase) + return nil +} + +func runningAgentExecutable() (string, error) { + path, err := os.Executable() + if err != nil { + return "", fmt.Errorf("resolve running agent executable: %w", err) + } + resolved, err := filepath.EvalSymlinks(path) + if err != nil { + return "", fmt.Errorf("resolve running agent executable symlinks: %w", err) + } + return resolved, nil +} + +func wrapOptionalError(context string, err error) error { + if err == nil { + return nil + } + return fmt.Errorf("%s: %w", context, err) +} + +func validateStartedAgentUpgrade(paths agentUpgradePaths, signal *agentUpgradeSignal, nspawnBinaryPath func(string) string) error { + if signal.CandidatePath != paths.BluePath && signal.CandidatePath != paths.GreenPath { + return fmt.Errorf("AgentUpgrade was interrupted before selecting a candidate slot") + } + current, err := resolvedExecutable(paths.CurrentPath) + if err != nil { + return fmt.Errorf("resolve current upgraded agent binary: %w", err) + } + if current != signal.CandidatePath { + return fmt.Errorf("AgentUpgrade was interrupted before switching the candidate binary") + } + if !validNspawnMachine(signal.ActiveMachine) { + return fmt.Errorf("AgentUpgrade has no valid active nspawn machine") + } + nspawnPath := nspawnBinaryPath(signal.ActiveMachine) + equal, err := filesHaveEqualSHA256(current, nspawnPath) + if err != nil { + return fmt.Errorf("verify synchronized nspawn agent binary: %w", err) + } + if !equal { + return fmt.Errorf("upgraded host and nspawn agent binaries do not match") + } + return nil +} + +func agentUpgradeCandidateIsActive(paths agentUpgradePaths, signal *agentUpgradeSignal) (bool, error) { + if signal == nil || signal.CandidatePath != paths.BluePath && signal.CandidatePath != paths.GreenPath { + return false, nil + } + current, err := resolvedExecutable(paths.CurrentPath) + if err != nil { + return false, fmt.Errorf("resolve current agent binary for rollback: %w", err) + } + return current == signal.CandidatePath, nil +} + +func activeNspawnAgentBinaryPath(machine string) string { + return filepath.Join("/var/lib/machines", machine, "usr", "local", "bin", "aks-flex-node") +} + +func rollbackAgentUpgradeFiles(paths agentUpgradePaths, signal *agentUpgradeSignal) error { + if signal == nil { + return nil + } + candidateActive, err := agentUpgradeCandidateIsActive(paths, signal) + if err != nil { + return err + } + if !signal.SwitchCommitted && !signal.RecoveryRequired && !candidateActive { + return nil + } + lastGood, err := resolvedExecutable(paths.LastGoodPath) + if err != nil { + return fmt.Errorf("resolve last-good agent binary: %w", err) + } + if err := replaceSymlink(paths.CurrentPath, lastGood); err != nil { + return fmt.Errorf("restore last-good agent binary: %w", err) + } + if validNspawnMachine(signal.ActiveMachine) { + if err := synchronizeNspawnAgentBinary(lastGood, signal.ActiveMachine); err != nil { + return err + } + } + return nil +} + +func filesHaveEqualSHA256(firstPath, secondPath string) (bool, error) { + first, err := fileSHA256(firstPath) + if err != nil { + return false, err + } + second, err := fileSHA256(secondPath) + if err != nil { + return false, err + } + return first == second, nil +} + +func fileSHA256(path string) ([sha256.Size]byte, error) { + var digest [sha256.Size]byte + file, err := os.Open(path) //nolint:gosec // fixed root-owned agent paths + if err != nil { + return digest, err + } + defer file.Close() //nolint:errcheck // read result is authoritative + hasher := sha256.New() + limited := io.LimitReader(file, agentUpgradeMaxBinaryBytes+1) + n, err := io.Copy(hasher, limited) + if err != nil { + return digest, err + } + if n > agentUpgradeMaxBinaryBytes { + return digest, fmt.Errorf("agent binary exceeds %d-byte limit", agentUpgradeMaxBinaryBytes) + } + copy(digest[:], hasher.Sum(nil)) + return digest, nil +} diff --git a/pkg/daemon/agent_upgrade_binary.go b/pkg/daemon/agent_upgrade_binary.go new file mode 100644 index 00000000..41cbca66 --- /dev/null +++ b/pkg/daemon/agent_upgrade_binary.go @@ -0,0 +1,229 @@ +package daemon + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "log/slog" + "net/http" + "net/url" + "os" + "path/filepath" + "runtime" + "strings" + "syscall" + "time" + + "github.com/Azure/AKSFlexNode/pkg/config" + "github.com/Azure/AKSFlexNode/pkg/utils/utilio" + "github.com/Azure/unbounded/pkg/agent/agentbinary" + "github.com/Azure/unbounded/pkg/agent/goalstates" +) + +const ( + agentUpgradeBinaryMode = 0o755 + agentUpgradeMaxArchiveBytes = 256 << 20 + agentUpgradeMaxBinaryBytes = 256 << 20 + agentUpgradeLockPath = "/run/aks-flex-node-agent-upgrade.lock" +) + +type agentUpgradePaths struct { + BinaryPath string + BluePath string + GreenPath string + CurrentPath string + LastGoodPath string + SignalPath string +} + +func (p agentUpgradePaths) layout() agentbinary.Layout { + return agentbinary.Layout{ + BinaryPath: p.BinaryPath, + BluePath: p.BluePath, + GreenPath: p.GreenPath, + CurrentPath: p.CurrentPath, + LastGoodPath: p.LastGoodPath, + } +} + +func (p agentUpgradePaths) sharedPaths() (goalstates.AgentUpgradePaths, error) { + paths := goalstates.AgentUpgradePaths{ + BinaryPath: p.BinaryPath, + BluePath: p.BluePath, + GreenPath: p.GreenPath, + CurrentPath: p.CurrentPath, + LastGoodPath: p.LastGoodPath, + SignalPath: p.SignalPath, + } + target, err := filepath.EvalSymlinks(p.CurrentPath) + if err == nil { + paths.CurrentTargetPath = target + return paths, nil + } + if errors.Is(err, os.ErrNotExist) { + paths.CurrentTargetPath = p.BinaryPath + return paths, nil + } + return goalstates.AgentUpgradePaths{}, fmt.Errorf("resolve current agent binary: %w", err) +} + +// ensureAgentUpgradeLayout adds Flex-specific ownership validation around the +// shared idempotent migration and link initialization implementation. +func ensureAgentUpgradeLayout(ctx context.Context, log *slog.Logger, paths agentUpgradePaths) error { + if log == nil { + return fmt.Errorf("logger is nil") + } + if err := validateAgentUpgradePaths(paths); err != nil { + return err + } + productionPaths := paths == defaultAgentUpgradePaths() + if productionPaths && os.Geteuid() != 0 { + return fmt.Errorf("agent binary layout must be initialized as root") + } + sharedPaths, err := paths.sharedPaths() + if err != nil { + return err + } + if err := agentbinary.EnsureDaemonBinaryLinks(ctx, log, sharedPaths); err != nil { + return err + } + if productionPaths { + return validateRootOwnedAgentUpgradePaths(paths) + } + return nil +} + +func validateRootOwnedAgentUpgradePaths(paths agentUpgradePaths) error { + for _, path := range []string{ + filepath.Dir(paths.BluePath), + paths.BinaryPath, + paths.BluePath, + paths.GreenPath, + paths.CurrentPath, + paths.LastGoodPath, + paths.SignalPath, + } { + info, err := os.Lstat(path) + if errors.Is(err, os.ErrNotExist) { + continue + } + if err != nil { + return fmt.Errorf("inspect ownership of %s: %w", path, err) + } + stat, ok := info.Sys().(*syscall.Stat_t) + if !ok || stat.Uid != 0 { + return fmt.Errorf("agent upgrade path %s is not root-owned", path) + } + } + return nil +} + +func validateAgentUpgradePaths(paths agentUpgradePaths) error { + values := []string{paths.BinaryPath, paths.BluePath, paths.GreenPath, paths.CurrentPath, paths.LastGoodPath, paths.SignalPath} + seen := make(map[string]struct{}, len(values)) + for _, value := range values { + if value == "" || !filepath.IsAbs(value) || filepath.Clean(value) != value { + return fmt.Errorf("invalid agent upgrade path %q", value) + } + if _, ok := seen[value]; ok { + return fmt.Errorf("duplicate agent upgrade path %q", value) + } + seen[value] = struct{}{} + } + return nil +} + +func resolvedExecutable(path string) (string, error) { + resolved, err := filepath.EvalSymlinks(path) + if err != nil { + return "", err + } + info, err := os.Stat(resolved) + if err != nil { + return "", err + } + if !info.Mode().IsRegular() || info.Mode().Perm()&0o111 == 0 { + return "", fmt.Errorf("%s is not a regular executable file", path) + } + return resolved, nil +} + +func copyExecutable(sourcePath, targetPath string) (err error) { + source, err := os.Open(sourcePath) //nolint:gosec // paths are fixed daemon configuration + if err != nil { + return err + } + defer func() { + if closeErr := source.Close(); closeErr != nil && err == nil { + err = closeErr + } + }() + return utilio.InstallFileWithLimitedSize(targetPath, source, agentUpgradeBinaryMode, agentUpgradeMaxBinaryBytes) +} + +func replaceSymlink(linkPath, targetPath string) error { + if err := os.MkdirAll(filepath.Dir(linkPath), 0o750); err != nil { + return err + } + temp, err := os.CreateTemp(filepath.Dir(linkPath), ".aks-flex-node-link-*") + if err != nil { + return err + } + tempPath := temp.Name() + if err := temp.Close(); err != nil { + _ = os.Remove(tempPath) + return err + } + if err := os.Remove(tempPath); err != nil { + return err + } + defer os.Remove(tempPath) //nolint:errcheck // best-effort cleanup before/after rename + if err := os.Symlink(targetPath, tempPath); err != nil { + return err + } + return os.Rename(tempPath, linkPath) +} + +// secureAgentInstallOptions intentionally follows the merged Unbounded +// MachineOperation contract: HTTP transport and an omitted digest are allowed +// when the control plane trusts the archive source and transport path. Flex +// still bounds the archive, requires one exact member, and verifies the +// candidate executable. Production callers should supply HTTPS and SHA-256. +func secureAgentInstallOptions(rawURL, expectedDigest string) (agentbinary.InstallOptions, error) { + parsedURL, err := url.ParseRequestURI(strings.TrimSpace(rawURL)) + if err != nil || parsedURL.Scheme != "http" && parsedURL.Scheme != "https" || parsedURL.Host == "" || parsedURL.User != nil || parsedURL.Fragment != "" { + return agentbinary.InstallOptions{}, fmt.Errorf("download URL must use HTTP or HTTPS, include a host, omit user information, and omit fragments") + } + digest := strings.TrimPrefix(strings.TrimSpace(expectedDigest), "sha256:") + if digest != "" { + decodedDigest, decodeErr := hex.DecodeString(digest) + if decodeErr != nil || len(decodedDigest) != sha256.Size { + return agentbinary.InstallOptions{}, fmt.Errorf("expected SHA-256 must be exactly 64 hexadecimal characters") + } + } + member, err := config.AgentBinaryArchiveMember(runtime.GOOS, runtime.GOARCH) + if err != nil { + return agentbinary.InstallOptions{}, err + } + return agentbinary.InstallOptions{ + DownloadURL: rawURL, + ExpectedSHA256: expectedDigest, + ExpectedMember: member, + Mode: agentUpgradeBinaryMode, + MaxArchiveBytes: agentUpgradeMaxArchiveBytes, + MaxExtractedBytes: agentUpgradeMaxBinaryBytes, + ExactMember: true, + HTTPClient: &http.Client{Timeout: 10 * time.Minute}, + }, nil +} + +func installAndSwitchAgentBinary(ctx context.Context, log *slog.Logger, rawURL, expectedDigest string, paths agentUpgradePaths) error { + opts, err := secureAgentInstallOptions(rawURL, expectedDigest) + if err != nil { + return err + } + _, err = agentbinary.InstallAndSwitchFromTarGz(ctx, log, paths.layout(), opts) + return err +} diff --git a/pkg/daemon/agent_upgrade_binary_test.go b/pkg/daemon/agent_upgrade_binary_test.go new file mode 100644 index 00000000..02f625bb --- /dev/null +++ b/pkg/daemon/agent_upgrade_binary_test.go @@ -0,0 +1,174 @@ +package daemon + +import ( + "log/slog" + "os" + "path/filepath" + "runtime" + "strings" + "testing" + + "github.com/Azure/AKSFlexNode/pkg/config" +) + +func TestEnsureAgentUpgradeLayoutMigratesLegacyBinaryIdempotently(t *testing.T) { + t.Parallel() + + paths := testAgentUpgradePaths(t) + if err := os.MkdirAll(filepath.Dir(paths.BinaryPath), 0o755); err != nil { + t.Fatalf("MkdirAll: %v", err) + } + if err := os.WriteFile(paths.BinaryPath, []byte("legacy"), 0o755); err != nil { + t.Fatalf("WriteFile: %v", err) + } + + for range 2 { + if err := ensureAgentUpgradeLayout(t.Context(), slog.Default(), paths); err != nil { + t.Fatalf("ensureAgentUpgradeLayout: %v", err) + } + } + + assertResolvedPath(t, paths.CurrentPath, paths.BluePath) + assertResolvedPath(t, paths.LastGoodPath, paths.BluePath) + assertResolvedPath(t, paths.BinaryPath, paths.BluePath) + data, err := os.ReadFile(paths.BluePath) + if err != nil { + t.Fatalf("ReadFile: %v", err) + } + if string(data) != "legacy" { + t.Fatalf("blue slot = %q, want legacy", data) + } +} + +func TestEnsureAgentUpgradeLayoutRecoversInterruptedLastGoodInitialization(t *testing.T) { + t.Parallel() + + paths := testAgentUpgradePaths(t) + if err := os.MkdirAll(filepath.Dir(paths.BluePath), 0o755); err != nil { + t.Fatalf("MkdirAll: %v", err) + } + if err := os.WriteFile(paths.BluePath, []byte("active"), 0o755); err != nil { + t.Fatalf("WriteFile: %v", err) + } + if err := os.Symlink(paths.BluePath, paths.CurrentPath); err != nil { + t.Fatalf("Symlink current: %v", err) + } + + if err := ensureAgentUpgradeLayout(t.Context(), slog.Default(), paths); err != nil { + t.Fatalf("ensureAgentUpgradeLayout: %v", err) + } + + assertResolvedPath(t, paths.CurrentPath, paths.BluePath) + assertResolvedPath(t, paths.LastGoodPath, paths.BluePath) + assertResolvedPath(t, paths.BinaryPath, paths.BluePath) +} + +func TestSecureAgentInstallOptions(t *testing.T) { + t.Parallel() + + digest := strings.Repeat("a", 64) + opts, err := secureAgentInstallOptions("https://example.com/agent.tar.gz?sig=secret", digest) + if err != nil { + t.Fatalf("secureAgentInstallOptions: %v", err) + } + wantMember, err := config.AgentBinaryArchiveMember(runtime.GOOS, runtime.GOARCH) + if err != nil { + t.Fatalf("AgentBinaryArchiveMember: %v", err) + } + if opts.ExpectedMember != wantMember { + t.Fatalf("ExpectedMember = %q, want %q", opts.ExpectedMember, wantMember) + } + if opts.MaxArchiveBytes != agentUpgradeMaxArchiveBytes || opts.MaxExtractedBytes != agentUpgradeMaxBinaryBytes { + t.Fatalf("size limits = %d, %d", opts.MaxArchiveBytes, opts.MaxExtractedBytes) + } + if !opts.ExactMember { + t.Fatal("ExactMember = false") + } + withoutDigest, err := secureAgentInstallOptions("https://example.com/agent.tar.gz", "") + if err != nil { + t.Fatalf("secureAgentInstallOptions without digest: %v", err) + } + if withoutDigest.ExpectedSHA256 != "" { + t.Fatalf("ExpectedSHA256 = %q, want empty", withoutDigest.ExpectedSHA256) + } + if _, err := secureAgentInstallOptions("http://127.0.0.1/agent.tar.gz", digest); err != nil { + t.Fatalf("secureAgentInstallOptions with HTTP: %v", err) + } +} + +func TestSecureAgentInstallOptionsRejectsInvalidInputs(t *testing.T) { + t.Parallel() + + tests := map[string]struct { + url string + digest string + }{ + "unsupported scheme": { + url: "ftp://example.com/agent.tar.gz", + digest: strings.Repeat("a", 64), + }, + "invalid digest": { + url: "https://example.com/agent.tar.gz", + digest: "bad", + }, + } + for name, tt := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + if _, err := secureAgentInstallOptions(tt.url, tt.digest); err == nil { + t.Fatal("secureAgentInstallOptions error = nil") + } + }) + } +} + +func TestInstallAndSwitchAgentBinaryRejectsInvalidInputsWithoutSwitching(t *testing.T) { + t.Parallel() + + paths := testAgentUpgradePaths(t) + if err := os.MkdirAll(filepath.Dir(paths.BluePath), 0o755); err != nil { + t.Fatalf("MkdirAll: %v", err) + } + if err := os.WriteFile(paths.BluePath, []byte("old"), 0o755); err != nil { + t.Fatalf("WriteFile: %v", err) + } + if err := os.Symlink(paths.BluePath, paths.CurrentPath); err != nil { + t.Fatalf("Symlink: %v", err) + } + + err := installAndSwitchAgentBinary( + t.Context(), + slog.Default(), + "ftp://example.com/agent.tar.gz", + strings.Repeat("0", 64), + paths, + ) + if err == nil { + t.Fatal("installAndSwitchAgentBinary error = nil") + } + assertResolvedPath(t, paths.CurrentPath, paths.BluePath) +} + +func testAgentUpgradePaths(t *testing.T) agentUpgradePaths { + t.Helper() + dir := t.TempDir() + return agentUpgradePaths{ + BinaryPath: filepath.Join(dir, "bin", "aks-flex-node"), + BluePath: filepath.Join(dir, "lib", "aks-flex-node-blue"), + GreenPath: filepath.Join(dir, "lib", "aks-flex-node-green"), + CurrentPath: filepath.Join(dir, "lib", "aks-flex-node-current"), + LastGoodPath: filepath.Join(dir, "lib", "aks-flex-node-last-good"), + SignalPath: filepath.Join(dir, "etc", "agent-upgrade-signal.json"), + } +} + +func assertResolvedPath(t *testing.T, path, want string) { + t.Helper() + got, err := filepath.EvalSymlinks(path) + if err != nil { + t.Fatalf("EvalSymlinks(%s): %v", path, err) + } + if got != want { + t.Fatalf("resolved %s = %s, want %s", path, got, want) + } +} diff --git a/pkg/daemon/agent_upgrade_test.go b/pkg/daemon/agent_upgrade_test.go new file mode 100644 index 00000000..37a0f679 --- /dev/null +++ b/pkg/daemon/agent_upgrade_test.go @@ -0,0 +1,501 @@ +package daemon + +import ( + "context" + "errors" + "log/slog" + "os" + "path/filepath" + "strings" + "testing" + + "sigs.k8s.io/controller-runtime/pkg/client" + + machinav1alpha3 "github.com/Azure/unbounded/api/machina/v1alpha3" + agentdaemon "github.com/Azure/unbounded/pkg/agent/daemon" +) + +func TestNewDaemonInstanceID(t *testing.T) { + t.Parallel() + + first, err := newDaemonInstanceID() + if err != nil { + t.Fatalf("newDaemonInstanceID: %v", err) + } + second, err := newDaemonInstanceID() + if err != nil { + t.Fatalf("newDaemonInstanceID: %v", err) + } + if first == "" || second == "" || first == second { + t.Fatalf("instance IDs = %q, %q", first, second) + } +} + +func TestParseAgentUpgradeRequest(t *testing.T) { + t.Parallel() + + tests := map[string]struct { + parameters map[string]string + wantErr string + }{ + "valid": { + parameters: map[string]string{ + agentUpgradeDownloadURLParameter: "https://example.com/agent.tar.gz?sig=secret", + agentUpgradeSHA256Parameter: strings.Repeat("a", 64), + }, + }, + "missing URL": { + parameters: map[string]string{agentUpgradeSHA256Parameter: strings.Repeat("a", 64)}, + wantErr: agentUpgradeDownloadURLParameter, + }, + "HTTP URL": { + parameters: map[string]string{ + agentUpgradeDownloadURLParameter: "http://example.com/agent.tar.gz", + agentUpgradeSHA256Parameter: strings.Repeat("a", 64), + }, + }, + "optional digest omitted": { + parameters: map[string]string{agentUpgradeDownloadURLParameter: "https://example.com/agent.tar.gz"}, + }, + "invalid digest": { + parameters: map[string]string{ + agentUpgradeDownloadURLParameter: "https://example.com/agent.tar.gz", + agentUpgradeSHA256Parameter: "bad", + }, + wantErr: "64 hexadecimal", + }, + } + for name, tt := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + _, err := parseAgentUpgradeRequest(tt.parameters) + if tt.wantErr == "" { + if err != nil { + t.Fatalf("parseAgentUpgradeRequest: %v", err) + } + return + } + if err == nil || !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("error = %v, want substring %q", err, tt.wantErr) + } + }) + } +} + +func TestHostAgentUpgradeExecutorRecordPendingIsIdempotent(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "agent-upgrade.json") + executor := &hostAgentUpgradeExecutor{ + state: &fakeNodeOperator{state: &State{ActiveMachine: "kube1"}}, + signals: agentUpgradeSignalStore{path: path}, + } + if err := executor.RecordPending(t.Context(), "operation-1"); err != nil { + t.Fatalf("first RecordPending: %v", err) + } + if err := executor.RecordPending(t.Context(), "operation-1"); !errors.Is(err, errAgentUpgradeAlreadyPending) { + t.Fatalf("second RecordPending error = %v, want errAgentUpgradeAlreadyPending", err) + } + if err := executor.RecordPending(t.Context(), "operation-2"); err == nil || !strings.Contains(err.Error(), "operation-1") { + t.Fatalf("competing RecordPending error = %v", err) + } +} + +func TestAgentUpgradeSignalStoreLifecycle(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "signals", "agent-upgrade.json") + store := agentUpgradeSignalStore{path: path} + if err := store.recordPending("operation-1", "kube1", "instance-1"); err != nil { + t.Fatalf("recordPending: %v", err) + } + if err := store.recordCandidate("/slots/green"); err != nil { + t.Fatalf("recordCandidate: %v", err) + } + if err := store.recordSwitchCommitted(); err != nil { + t.Fatalf("recordSwitchCommitted: %v", err) + } + if err := store.recordFailure("rolled back"); err != nil { + t.Fatalf("recordFailure: %v", err) + } + signal, err := store.read() + if err != nil { + t.Fatalf("read: %v", err) + } + if signal == nil || signal.OperationName != "operation-1" || signal.ActiveMachine != "kube1" || signal.CandidatePath != "/slots/green" || signal.InitiatingDaemonInstance != "instance-1" || !signal.SwitchCommitted || signal.Failure != "rolled back" || !signal.RecoveryRequired { + t.Fatalf("signal = %#v", signal) + } + info, err := os.Stat(path) + if err != nil { + t.Fatalf("Stat: %v", err) + } + if info.Mode().Perm() != 0o600 { + t.Fatalf("signal mode = %o, want 600", info.Mode().Perm()) + } + if err := store.clear(); err != nil { + t.Fatalf("clear: %v", err) + } + if _, err := os.Stat(path); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("signal still exists: %v", err) + } +} + +func TestAgentUpgradeSignalStoreRejectsInvalidMachine(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "signal.json") + if err := os.WriteFile(path, []byte(`{"operationName":"operation-1","activeMachine":"../../etc"}`), 0o600); err != nil { + t.Fatalf("WriteFile: %v", err) + } + _, err := (agentUpgradeSignalStore{path: path}).read() + if err == nil || !strings.Contains(err.Error(), "invalid active machine") { + t.Fatalf("error = %v", err) + } +} + +func TestHostAgentUpgradeExecutorSchedulesRestartOutsideService(t *testing.T) { + t.Parallel() + + executor := &hostAgentUpgradeExecutor{ + runSystemdRun: func(_ context.Context, args ...string) error { + joined := strings.Join(args, " ") + for _, expected := range []string{ + "--collect", + "--on-active=1s", + "--unit=aks-flex-node-agent-upgrade-restart-", + "/usr/bin/systemctl restart " + ServiceUnitName, + } { + if !strings.Contains(joined, expected) { + t.Fatalf("systemd-run args %q do not contain %q", joined, expected) + } + } + return nil + }, + } + if err := executor.Restart(t.Context()); err != nil { + t.Fatalf("Restart: %v", err) + } +} + +func TestHostAgentUpgradeExecutorRestartRejectsPreCanceledContext(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + called := false + executor := &hostAgentUpgradeExecutor{runSystemdRun: func(context.Context, ...string) error { + called = true + return nil + }} + if err := executor.Restart(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("Restart error = %v, want context.Canceled", err) + } + if called { + t.Fatal("systemctl called with an already canceled context") + } +} + +func TestHostAgentUpgradeExecutorAbortUsesCleanupContext(t *testing.T) { + t.Parallel() + + paths := testAgentUpgradePaths(t) + if err := os.MkdirAll(filepath.Dir(paths.BluePath), 0o750); err != nil { + t.Fatalf("MkdirAll: %v", err) + } + for path, content := range map[string]string{paths.BluePath: "blue", paths.GreenPath: "green"} { + if err := os.WriteFile(path, []byte(content), 0o755); err != nil { + t.Fatalf("WriteFile(%s): %v", path, err) + } + } + if err := os.Symlink(paths.GreenPath, paths.CurrentPath); err != nil { + t.Fatalf("Symlink current: %v", err) + } + if err := os.Symlink(paths.BluePath, paths.LastGoodPath); err != nil { + t.Fatalf("Symlink last-good: %v", err) + } + signals := agentUpgradeSignalStore{path: paths.SignalPath} + if err := signals.recordPending("operation-1", "", "instance-1"); err != nil { + t.Fatalf("recordPending: %v", err) + } + if err := signals.recordCandidate(paths.GreenPath); err != nil { + t.Fatalf("recordCandidate: %v", err) + } + executor := &hostAgentUpgradeExecutor{paths: paths, signals: signals, instanceID: "instance-1"} + ctx, cancel := context.WithCancel(t.Context()) + cancel() + if err := executor.Abort(ctx); err != nil { + t.Fatalf("Abort: %v", err) + } + assertResolvedPath(t, paths.CurrentPath, paths.BluePath) + if signal, err := signals.read(); err != nil || signal != nil { + t.Fatalf("signal after Abort = %#v, %v", signal, err) + } +} + +func TestHostAgentUpgradeExecutorAbortPreservesSignalOnRollbackFailure(t *testing.T) { + t.Parallel() + + paths := testAgentUpgradePaths(t) + signals := agentUpgradeSignalStore{path: paths.SignalPath} + if err := signals.recordPending("operation-1", "", "instance-1"); err != nil { + t.Fatalf("recordPending: %v", err) + } + if err := signals.recordCandidate(paths.BluePath); err != nil { + t.Fatalf("recordCandidate: %v", err) + } + executor := &hostAgentUpgradeExecutor{paths: paths, signals: signals, instanceID: "instance-1"} + if err := executor.Abort(t.Context()); err == nil { + t.Fatal("Abort error = nil") + } + if signal, err := signals.read(); err != nil || signal == nil { + t.Fatalf("signal after failed Abort = %#v, %v", signal, err) + } +} + +func TestPublishAgentUpgradeSignalPreservesSignalWhenRollbackFails(t *testing.T) { + t.Parallel() + + paths := testAgentUpgradePaths(t) + if err := os.MkdirAll(filepath.Dir(paths.BluePath), 0o750); err != nil { + t.Fatalf("MkdirAll: %v", err) + } + if err := os.WriteFile(paths.BluePath, []byte("candidate"), 0o755); err != nil { + t.Fatalf("WriteFile: %v", err) + } + if err := os.Symlink(paths.BluePath, paths.CurrentPath); err != nil { + t.Fatalf("Symlink: %v", err) + } + signals := agentUpgradeSignalStore{path: paths.SignalPath} + if err := signals.write(agentUpgradeSignal{ + OperationName: "operation-1", + ActiveMachine: "kube1", + CandidatePath: paths.BluePath, + InitiatingDaemonInstance: "other-instance", + }); err != nil { + t.Fatalf("write signal: %v", err) + } + executor := &hostAgentUpgradeExecutor{paths: paths, signals: signals, instanceID: "current-instance"} + if err := publishAndClearAgentUpgradeSignal(t.Context(), slog.Default(), nil, executor); err == nil { + t.Fatal("publishAndClearAgentUpgradeSignal error = nil") + } + if signal, err := signals.read(); err != nil || signal == nil { + t.Fatalf("signal after failed rollback = %#v, %v", signal, err) + } +} + +func TestPublishAgentUpgradeFailureRestartsIntoLastGoodBeforeClearingSignal(t *testing.T) { + t.Parallel() + + paths := testAgentUpgradePaths(t) + if err := os.MkdirAll(filepath.Dir(paths.BluePath), 0o750); err != nil { + t.Fatalf("MkdirAll: %v", err) + } + for path, content := range map[string]string{paths.BluePath: "candidate", paths.GreenPath: "last-good"} { + if err := os.WriteFile(path, []byte(content), 0o755); err != nil { + t.Fatalf("WriteFile(%s): %v", path, err) + } + } + if err := os.Symlink(paths.BluePath, paths.CurrentPath); err != nil { + t.Fatalf("Symlink current: %v", err) + } + if err := os.Symlink(paths.GreenPath, paths.LastGoodPath); err != nil { + t.Fatalf("Symlink last-good: %v", err) + } + signals := agentUpgradeSignalStore{path: paths.SignalPath} + if err := signals.write(agentUpgradeSignal{ + OperationName: "operation-1", + CandidatePath: paths.BluePath, + InitiatingDaemonInstance: "initiator", + RecoveryRequired: true, + Failure: "candidate failed", + }); err != nil { + t.Fatalf("write signal: %v", err) + } + restarts := 0 + finished := 0 + executor := &hostAgentUpgradeExecutor{ + paths: paths, + signals: signals, + instanceID: "candidate-instance", + runSystemdRun: func(context.Context, ...string) error { + restarts++ + return nil + }, + finishMachineOperation: func(_ context.Context, _ client.Client, _ agentdaemon.MachineOperation, result agentdaemon.MachineOperationResult[int64]) error { + finished++ + if result.Phase != machinav1alpha3.OperationPhaseFailed { + t.Fatalf("phase = %s, want Failed", result.Phase) + } + return nil + }, + runningExecutable: func() (string, error) { return paths.BluePath, nil }, + } + if err := publishAndClearAgentUpgradeSignal(t.Context(), slog.Default(), nil, executor); err != nil { + t.Fatalf("publish candidate recovery: %v", err) + } + assertResolvedPath(t, paths.CurrentPath, paths.GreenPath) + if signal, err := signals.read(); err != nil || signal == nil { + t.Fatalf("signal cleared before last-good startup: %#v, %v", signal, err) + } + if restarts != 1 || finished != 0 { + t.Fatalf("restarts = %d, finished = %d; terminal status must wait for last-good", restarts, finished) + } + + executor.instanceID = "last-good-instance" + executor.runningExecutable = func() (string, error) { return paths.GreenPath, nil } + if err := publishAndClearAgentUpgradeSignal(t.Context(), slog.Default(), nil, executor); err != nil { + t.Fatalf("publish last-good recovery: %v", err) + } + if signal, err := signals.read(); err != nil || signal != nil { + t.Fatalf("signal after last-good startup = %#v, %v", signal, err) + } + if restarts != 1 || finished != 1 { + t.Fatalf("restarts = %d, finished = %d", restarts, finished) + } +} + +func TestPublishAgentUpgradeSuccessCompletesAndClearsSignal(t *testing.T) { + t.Parallel() + + paths := testAgentUpgradePaths(t) + if err := os.MkdirAll(filepath.Dir(paths.BluePath), 0o750); err != nil { + t.Fatalf("MkdirAll: %v", err) + } + if err := os.WriteFile(paths.BluePath, []byte("candidate"), 0o755); err != nil { + t.Fatalf("WriteFile candidate: %v", err) + } + if err := os.Symlink(paths.BluePath, paths.CurrentPath); err != nil { + t.Fatalf("Symlink current: %v", err) + } + nspawnBinary := filepath.Join(t.TempDir(), "nspawn-agent") + if err := os.WriteFile(nspawnBinary, []byte("candidate"), 0o755); err != nil { + t.Fatalf("WriteFile nspawn: %v", err) + } + signals := agentUpgradeSignalStore{path: paths.SignalPath} + if err := signals.write(agentUpgradeSignal{ + OperationName: "operation-1", + ActiveMachine: "kube1", + CandidatePath: paths.BluePath, + InitiatingDaemonInstance: "previous-instance", + SwitchCommitted: true, + }); err != nil { + t.Fatalf("write signal: %v", err) + } + finished := 0 + executor := &hostAgentUpgradeExecutor{ + paths: paths, + signals: signals, + instanceID: "restarted-instance", + nspawnBinaryPath: func(string) string { + return nspawnBinary + }, + finishMachineOperation: func(_ context.Context, _ client.Client, _ agentdaemon.MachineOperation, result agentdaemon.MachineOperationResult[int64]) error { + finished++ + if result.Phase != machinav1alpha3.OperationPhaseComplete { + t.Fatalf("phase = %s, want Complete", result.Phase) + } + return nil + }, + } + if err := publishAndClearAgentUpgradeSignal(t.Context(), slog.Default(), nil, executor); err != nil { + t.Fatalf("publishAndClearAgentUpgradeSignal: %v", err) + } + if finished != 1 { + t.Fatalf("finish calls = %d, want 1", finished) + } + if signal, err := signals.read(); err != nil || signal != nil { + t.Fatalf("signal after success = %#v, %v", signal, err) + } +} + +func TestPublishAgentUpgradeSignalIgnoresInitiatingProcess(t *testing.T) { + t.Parallel() + + paths := testAgentUpgradePaths(t) + signals := agentUpgradeSignalStore{path: paths.SignalPath} + if err := signals.recordPending("operation-1", "kube1", "current-instance"); err != nil { + t.Fatalf("recordPending: %v", err) + } + executor := &hostAgentUpgradeExecutor{paths: paths, signals: signals, instanceID: "current-instance"} + if err := publishAndClearAgentUpgradeSignal(t.Context(), slog.Default(), nil, executor); err != nil { + t.Fatalf("publishAndClearAgentUpgradeSignal: %v", err) + } + if signal, err := signals.read(); err != nil || signal == nil { + t.Fatalf("initiating process consumed signal: %#v, %v", signal, err) + } +} + +func TestRollbackAgentUpgradeFilesDoesNotDowngradeBeforeSwitch(t *testing.T) { + t.Parallel() + + paths := testAgentUpgradePaths(t) + if err := os.MkdirAll(filepath.Dir(paths.BluePath), 0o750); err != nil { + t.Fatalf("MkdirAll: %v", err) + } + for path, content := range map[string]string{paths.BluePath: "last-good", paths.GreenPath: "active"} { + if err := os.WriteFile(path, []byte(content), 0o755); err != nil { + t.Fatalf("WriteFile(%s): %v", path, err) + } + } + if err := os.Symlink(paths.GreenPath, paths.CurrentPath); err != nil { + t.Fatalf("Symlink current: %v", err) + } + if err := os.Symlink(paths.BluePath, paths.LastGoodPath); err != nil { + t.Fatalf("Symlink last-good: %v", err) + } + + if err := rollbackAgentUpgradeFiles(paths, &agentUpgradeSignal{CandidatePath: paths.BluePath}); err != nil { + t.Fatalf("rollbackAgentUpgradeFiles: %v", err) + } + assertResolvedPath(t, paths.CurrentPath, paths.GreenPath) +} + +func TestRollbackAgentUpgradeFilesRetriesAfterHostLinkWasRestored(t *testing.T) { + t.Parallel() + + paths := testAgentUpgradePaths(t) + if err := os.MkdirAll(filepath.Dir(paths.BluePath), 0o750); err != nil { + t.Fatalf("MkdirAll: %v", err) + } + for _, path := range []string{paths.BluePath, paths.GreenPath} { + if err := os.WriteFile(path, []byte(path), 0o755); err != nil { + t.Fatalf("WriteFile(%s): %v", path, err) + } + } + if err := os.Symlink(paths.GreenPath, paths.CurrentPath); err != nil { + t.Fatalf("Symlink current: %v", err) + } + + err := rollbackAgentUpgradeFiles(paths, &agentUpgradeSignal{ + CandidatePath: paths.BluePath, + RecoveryRequired: true, + }) + if err == nil || !strings.Contains(err.Error(), "resolve last-good") { + t.Fatalf("rollbackAgentUpgradeFiles error = %v, want retry to resolve last-good", err) + } +} + +func TestFilesHaveEqualSHA256(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + first := filepath.Join(dir, "first") + second := filepath.Join(dir, "second") + if err := os.WriteFile(first, []byte("same"), 0o600); err != nil { + t.Fatalf("WriteFile first: %v", err) + } + if err := os.WriteFile(second, []byte("same"), 0o600); err != nil { + t.Fatalf("WriteFile second: %v", err) + } + equal, err := filesHaveEqualSHA256(first, second) + if err != nil || !equal { + t.Fatalf("filesHaveEqualSHA256 = %v, %v", equal, err) + } + if err := os.WriteFile(second, []byte("different"), 0o600); err != nil { + t.Fatalf("WriteFile second: %v", err) + } + equal, err = filesHaveEqualSHA256(first, second) + if err != nil || equal { + t.Fatalf("filesHaveEqualSHA256 = %v, %v", equal, err) + } +} diff --git a/pkg/daemon/assets/aks-flex-node-agent-recovery.service b/pkg/daemon/assets/aks-flex-node-agent-recovery.service new file mode 100644 index 00000000..d579d5ac --- /dev/null +++ b/pkg/daemon/assets/aks-flex-node-agent-recovery.service @@ -0,0 +1,6 @@ +[Unit] +Description=Recover AKS Flex Node Agent to last-known-good binary + +[Service] +Type=oneshot +ExecStart=/usr/local/lib/aks-flex-node/aks-flex-node-recovery.sh diff --git a/pkg/daemon/assets/aks-flex-node-agent.service b/pkg/daemon/assets/aks-flex-node-agent.service index 628a77ce..37e09a62 100644 --- a/pkg/daemon/assets/aks-flex-node-agent.service +++ b/pkg/daemon/assets/aks-flex-node-agent.service @@ -2,7 +2,8 @@ Description=AKS Flex Node Agent After=network-online.target Wants=network-online.target -# Restart on failure to enable auto-recovery +OnFailure=aks-flex-node-agent-recovery.service +# Trigger last-good recovery after the upgraded daemon repeatedly fails. StartLimitIntervalSec=300 StartLimitBurst=5 diff --git a/pkg/daemon/assets/aks-flex-node-recovery.sh b/pkg/daemon/assets/aks-flex-node-recovery.sh new file mode 100644 index 00000000..5cd2432d --- /dev/null +++ b/pkg/daemon/assets/aks-flex-node-recovery.sh @@ -0,0 +1,21 @@ +#!/bin/bash +set -euo pipefail + +signal=/etc/aks-flex-node/agent-upgrade-signal.json +last_good="$(readlink -f /usr/local/lib/aks-flex-node/aks-flex-node-last-good || true)" + +# An ordinary daemon failure must not change the selected binary. +if [[ ! -f "${signal}" ]]; then + exit 0 +fi +if [[ -z "${last_good}" || ! -x "${last_good}" ]]; then + echo "no executable last-known-good AKS Flex Node agent binary" >&2 + exit 1 +fi + +status=0 +"${last_good}" recover-agent-upgrade \ + --message "upgraded daemon failed repeatedly; restored last-good binary" || status=$? +systemctl reset-failed aks-flex-node-agent.service || status=$? +systemctl --no-block restart aks-flex-node-agent.service || status=$? +exit "${status}" diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index 80e0e53f..74798ae6 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -26,8 +26,10 @@ import ( ) const ( - daemonCredentialDir = "daemon-credentials" //nolint:gosec // Directory name, not a credential. - daemonCredentialGroup = "aks-flex-node-daemons" //nolint:gosec // Kubernetes group name, not a credential. + agentUpgradeStartupStability = 3 * time.Second + agentUpgradePublishRetry = 10 * time.Second + daemonCredentialDir = "daemon-credentials" //nolint:gosec // Directory name, not a credential. + daemonCredentialGroup = "aks-flex-node-daemons" //nolint:gosec // Kubernetes group name, not a credential. ) // Run starts the machine-driven daemon loop. @@ -36,6 +38,11 @@ func Run(ctx context.Context, cfg *config.Config, log *slog.Logger) error { // logs unless the process explicitly configures its global logger. ctrl.SetLogger(logr.FromSlogHandler(log.Handler())) + // Existing direct-file installations may predate the recovery units. Keep + // the binary layout and systemd rollback assets converged on every startup. + if err := ensureAgentUpgradeServiceAssets(ctx, log); err != nil { + return err + } restCfg, stopCredentials, err := daemonRESTConfig(ctx, cfg) if err != nil { return err @@ -74,6 +81,14 @@ func Run(ctx context.Context, cfg *config.Config, log *slog.Logger) error { if err != nil { return err } + upgrades, err := newHostAgentUpgradeExecutor(log, operator) + if err != nil { + return err + } + directClient, err := client.New(restCfg, client.Options{Scheme: newScheme()}) + if err != nil { + return fmt.Errorf("create direct Kubernetes client: %w", err) + } repaves, err := newRepaveReconciler(repaveReconcilerOptions{ Log: log, Machines: machines, @@ -92,14 +107,57 @@ func Run(ctx context.Context, cfg *config.Config, log *slog.Logger) error { AKSMachineName: aksMachineName, MachineOperationMode: cfg.Agent.MachineOperationMode, Operator: repaves.operator, + AgentUpgrade: upgrades, }) if err != nil { return err } - if err := daemon.SetupController("aks-flex-node-daemon", mgr, machineOperations, repaves); err != nil { + gate := newStartupGate() + if err := daemon.SetupController( + "aks-flex-node-daemon", + mgr, + gatedMachineOperationReconciler{delegate: machineOperations, gate: gate}, + gatedRepaveReconciler{delegate: repaves, gate: gate}, + ); err != nil { return fmt.Errorf("setup daemon controller: %w", err) } + go func() { + // Keep host-mutating reconciliation gated until the candidate has reached + // cache readiness and remained alive for a bounded stability interval. + if !mgr.GetCache().WaitForCacheSync(ctx) { + return + } + select { + case <-ctx.Done(): + return + case <-time.After(agentUpgradeStartupStability): + } + for { + publishErr := publishAndClearAgentUpgradeSignal(ctx, log, directClient, upgrades) + pending, readErr := upgrades.signals.read() + switch { + case publishErr != nil: + log.Warn("failed to publish AgentUpgrade startup result; will retry", "error", publishErr) + case readErr != nil: + log.Warn("failed to confirm AgentUpgrade startup signal; will retry", "error", readErr) + case pending != nil: + // Recovery scheduled a restart. Keep reconciliation gated until the + // last-good process starts and consumes the retained signal. + return + default: + gate.open() + return + } + + select { + case <-ctx.Done(): + return + case <-time.After(agentUpgradePublishRetry): + } + } + }() + err = mgr.Start(ctx) repaves.log.Info("daemon shutting down") return err diff --git a/pkg/daemon/host_agent_activation.go b/pkg/daemon/host_agent_activation.go new file mode 100644 index 00000000..db3f3568 --- /dev/null +++ b/pkg/daemon/host_agent_activation.go @@ -0,0 +1,235 @@ +package daemon + +import ( + "bytes" + "context" + "errors" + "fmt" + "log/slog" + "os" + "path/filepath" + "strconv" + "strings" + "time" + + "github.com/Azure/AKSFlexNode/pkg/utils/utilexec" + "github.com/Azure/unbounded/pkg/agent/agentbinary" +) + +const ( + hostAgentHealthTimeout = 30 * time.Second + hostAgentStableDuration = 3 * time.Second + hostAgentHealthPoll = 250 * time.Millisecond +) + +// PreflightHostAgentActivation validates a directly staged Flex agent binary +// and returns the shared activation plan without changing host state. +func PreflightHostAgentActivation(ctx context.Context, log *slog.Logger, candidatePath string) (agentbinary.ActivationPlan, error) { + service, paths, err := newFlexDaemonActivationService(log) + if err != nil { + return agentbinary.ActivationPlan{}, err + } + return agentbinary.PreflightHostDaemonActivation(ctx, hostAgentActivationOptions(paths, candidatePath), service) +} + +// ActivateHostAgent activates a directly staged Flex agent binary. It uses the +// same lock and binary layout as MachineOperation upgrades. +func ActivateHostAgent(ctx context.Context, log *slog.Logger, candidatePath string) (agentbinary.ActivationResult, error) { + if os.Geteuid() != 0 { + return agentbinary.ActivationResult{}, fmt.Errorf("host agent upgrade requires root privileges") + } + service, paths, err := newFlexDaemonActivationService(log) + if err != nil { + return agentbinary.ActivationResult{}, err + } + return agentbinary.ActivateHostDaemon(ctx, log, hostAgentActivationOptions(paths, candidatePath), service) +} + +func hostAgentActivationOptions(paths agentUpgradePaths, candidatePath string) agentbinary.ActivationOptions { + return agentbinary.ActivationOptions{ + Layout: paths.layout(), + CandidatePath: candidatePath, + BinaryMode: agentUpgradeBinaryMode, + LockPath: agentUpgradeLockPath, + } +} + +type flexDaemonActivationService struct { + log *slog.Logger + paths agentUpgradePaths + state stateStore + systemdDir string + recoveryScript string + inspectService func(context.Context, *slog.Logger, string) (bool, error) + serviceWasActive bool +} + +func newFlexDaemonActivationService(log *slog.Logger) (*flexDaemonActivationService, agentUpgradePaths, error) { + if log == nil { + log = slog.Default() + } + state, err := NewFileStateStore() + if err != nil { + return nil, agentUpgradePaths{}, err + } + paths := defaultAgentUpgradePaths() + return &flexDaemonActivationService{ + log: log, + paths: paths, + state: state, + systemdDir: systemdSystemDir, + recoveryScript: recoveryScriptPath, + inspectService: inspectAgentServiceActive, + }, paths, nil +} + +func (s *flexDaemonActivationService) Preflight(ctx context.Context, currentBinaryPath string) (agentbinary.ServicePlan, error) { + inspectService := s.inspectService + if inspectService == nil { + inspectService = inspectAgentServiceActive + } + serviceWasActive, err := inspectService(ctx, s.log, ServiceUnitName) + if err != nil { + return agentbinary.ServicePlan{}, fmt.Errorf("inspect agent service state: %w", err) + } + s.serviceWasActive = serviceWasActive + if _, err := os.Stat(s.paths.SignalPath); err == nil { + return agentbinary.ServicePlan{}, fmt.Errorf("AgentUpgrade MachineOperation signal exists at %s", s.paths.SignalPath) + } else if !errors.Is(err, os.ErrNotExist) { + return agentbinary.ServicePlan{}, fmt.Errorf("inspect AgentUpgrade MachineOperation signal: %w", err) + } + for _, asset := range desiredAgentServiceAssets(s.paths, s.systemdDir, s.recoveryScript, currentBinaryPath) { + actual, err := os.ReadFile(asset.path) + if errors.Is(err, os.ErrNotExist) || err == nil && !bytes.Equal(actual, asset.content) { + return agentbinary.ServicePlan{ + UpdateRequired: true, + Description: "install or update AKS Flex Node agent systemd assets", + }, nil + } + if err != nil { + return agentbinary.ServicePlan{}, fmt.Errorf("read daemon asset %s: %w", asset.path, err) + } + } + return agentbinary.ServicePlan{Description: "AKS Flex Node agent systemd assets are current"}, nil +} + +func inspectAgentServiceActive(ctx context.Context, log *slog.Logger, service string) (bool, error) { + loadState, err := utilexec.OutputCmdAt(ctx, log, slog.LevelDebug, "systemctl", "show", "--property=LoadState", "--value", service) + if err != nil { + return false, err + } + if strings.TrimSpace(loadState) == "not-found" { + // Reset removes the unit while intentionally retaining the managed binary + // layout for a later rejoin. + return false, nil + } + state, err := utilexec.OutputCmdAt(ctx, log, slog.LevelDebug, "systemctl", "show", "--property=ActiveState", "--value", service) + if err != nil { + return false, err + } + switch strings.TrimSpace(state) { + case "inactive": + return false, nil + case "active", "activating", "reloading", "deactivating", "failed": + return true, nil + default: + return false, fmt.Errorf("unexpected ActiveState %q for %s", state, service) + } +} + +func (s *flexDaemonActivationService) Prepare(_ context.Context, currentBinaryPath string) error { + return writeAgentServiceAssets(s.paths, s.systemdDir, s.recoveryScript, currentBinaryPath) +} + +func (s *flexDaemonActivationService) Reload(ctx context.Context) error { + if err := utilexec.ReloadSystemd(ctx, s.log); err != nil { + return fmt.Errorf("systemctl daemon-reload: %w", err) + } + return nil +} + +func (s *flexDaemonActivationService) Restart(ctx context.Context) error { + if !s.serviceWasActive { + s.log.Info("leaving inactive host agent service stopped after activation") + return nil + } + if err := utilexec.RunCmd(ctx, s.log, utilexec.Systemctl(), "restart", ServiceUnitName); err != nil { + return fmt.Errorf("systemctl restart %s: %w", ServiceUnitName, err) + } + return nil +} + +// WaitHealthy preserves an inactive service during reset/reinstall. Otherwise, +// it synchronizes the active nspawn exec credential only after systemd is +// stably running the expected host binary. Shared rollback calls this again +// with last-good. +func (s *flexDaemonActivationService) WaitHealthy(ctx context.Context, expectedBinaryPath string) error { + healthCtx, cancel := context.WithTimeout(ctx, hostAgentHealthTimeout) + defer cancel() + expected, err := filepath.EvalSymlinks(expectedBinaryPath) + if err != nil { + return fmt.Errorf("resolve expected daemon binary: %w", err) + } + if !s.serviceWasActive { + return s.synchronizeActiveNspawn(healthCtx, expected) + } + var healthySince time.Time + ticker := time.NewTicker(hostAgentHealthPoll) + defer ticker.Stop() + for { + healthy, checkErr := s.isExpectedDaemonActive(healthCtx, expected) + if checkErr == nil && healthy { + if healthySince.IsZero() { + healthySince = time.Now() + } else if time.Since(healthySince) >= hostAgentStableDuration { + return s.synchronizeActiveNspawn(healthCtx, expected) + } + } else { + healthySince = time.Time{} + } + select { + case <-healthCtx.Done(): + if checkErr != nil { + return fmt.Errorf("daemon did not become healthy: %w", checkErr) + } + return fmt.Errorf("daemon did not execute expected binary %s: %w", expected, healthCtx.Err()) + case <-ticker.C: + } + } +} + +func (s *flexDaemonActivationService) synchronizeActiveNspawn(ctx context.Context, expected string) error { + state, err := s.state.Load(ctx) + if err != nil { + return fmt.Errorf("load active nspawn state: %w", err) + } + if state == nil { + s.log.Info("activated host agent without active nspawn synchronization") + return nil + } + if !validNspawnMachine(state.ActiveMachine) { + return fmt.Errorf("no valid active nspawn machine for agent activation") + } + if err := synchronizeNspawnAgentBinary(expected, state.ActiveMachine); err != nil { + return fmt.Errorf("synchronize active nspawn agent binary: %w", err) + } + return nil +} + +func (s *flexDaemonActivationService) isExpectedDaemonActive(ctx context.Context, expected string) (bool, error) { + output, err := utilexec.OutputCmd(ctx, s.log, "systemctl", "show", "--property", "MainPID", "--value", ServiceUnitName) + if err != nil { + return false, err + } + pid, err := strconv.Atoi(strings.TrimSpace(output)) + if err != nil || pid <= 0 { + return false, fmt.Errorf("invalid daemon MainPID %q", output) + } + running, err := filepath.EvalSymlinks(fmt.Sprintf("/proc/%d/exe", pid)) + if err != nil { + return false, err + } + return running == expected, nil +} + +var _ agentbinary.DaemonService = (*flexDaemonActivationService)(nil) diff --git a/pkg/daemon/host_agent_activation_test.go b/pkg/daemon/host_agent_activation_test.go new file mode 100644 index 00000000..aaf694d7 --- /dev/null +++ b/pkg/daemon/host_agent_activation_test.go @@ -0,0 +1,96 @@ +package daemon + +import ( + "context" + "errors" + "log/slog" + "os" + "path/filepath" + "testing" +) + +func TestFlexDaemonActivationPreflightUsesFlexAssetsWithoutMutation(t *testing.T) { + t.Parallel() + + paths := testAgentUpgradePaths(t) + systemdDir := filepath.Join(t.TempDir(), "systemd") + recoveryScript := filepath.Join(t.TempDir(), "recovery.sh") + service := &flexDaemonActivationService{ + log: slog.Default(), + paths: paths, + systemdDir: systemdDir, + recoveryScript: recoveryScript, + inspectService: func(context.Context, *slog.Logger, string) (bool, error) { + return false, nil + }, + } + plan, err := service.Preflight(t.Context(), paths.CurrentPath) + if err != nil { + t.Fatalf("Preflight: %v", err) + } + if !plan.UpdateRequired { + t.Fatal("UpdateRequired = false for missing Flex service assets") + } + for _, path := range []string{systemdDir, recoveryScript, paths.CurrentPath} { + if _, err := os.Lstat(path); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("preflight mutated %s: %v", path, err) + } + } +} + +func TestFlexDaemonActivationLeavesInactiveResetHostStopped(t *testing.T) { + t.Parallel() + + binary := filepath.Join(t.TempDir(), "candidate") + if err := os.WriteFile(binary, []byte("candidate"), 0o755); err != nil { + t.Fatalf("WriteFile: %v", err) + } + service := &flexDaemonActivationService{ + log: slog.Default(), + state: &testStateStore{}, + } + if err := service.Restart(t.Context()); err != nil { + t.Fatalf("Restart: %v", err) + } + if err := service.WaitHealthy(t.Context(), binary); err != nil { + t.Fatalf("WaitHealthy: %v", err) + } +} + +func TestFlexDaemonActivationPreflightRejectsUnknownServiceState(t *testing.T) { + t.Parallel() + + service := &flexDaemonActivationService{ + log: slog.Default(), + inspectService: func(context.Context, *slog.Logger, string) (bool, error) { + return false, errors.New("systemd unavailable") + }, + } + if _, err := service.Preflight(t.Context(), "/unused/current"); err == nil { + t.Fatal("Preflight accepted an unknown service state") + } +} + +func TestFlexDaemonActivationPreflightRejectsMachineOperationSignal(t *testing.T) { + t.Parallel() + + paths := testAgentUpgradePaths(t) + if err := os.MkdirAll(filepath.Dir(paths.SignalPath), 0o750); err != nil { + t.Fatalf("MkdirAll: %v", err) + } + if err := os.WriteFile(paths.SignalPath, []byte("{}"), 0o600); err != nil { + t.Fatalf("WriteFile: %v", err) + } + service := &flexDaemonActivationService{ + log: slog.Default(), + paths: paths, + systemdDir: t.TempDir(), + recoveryScript: filepath.Join(t.TempDir(), "recovery.sh"), + inspectService: func(context.Context, *slog.Logger, string) (bool, error) { + return false, nil + }, + } + if _, err := service.Preflight(t.Context(), paths.CurrentPath); err == nil { + t.Fatal("Preflight accepted a pending MachineOperation signal") + } +} diff --git a/pkg/daemon/lifecycle.go b/pkg/daemon/lifecycle.go index 822d01b1..1a35ba52 100644 --- a/pkg/daemon/lifecycle.go +++ b/pkg/daemon/lifecycle.go @@ -1,6 +1,7 @@ package daemon import ( + "bytes" "context" _ "embed" "fmt" @@ -14,13 +15,21 @@ import ( ) const ( - ServiceUnitName = "aks-flex-node-agent.service" - systemdSystemDir = "/etc/systemd/system" + ServiceUnitName = "aks-flex-node-agent.service" + recoveryServiceUnitName = "aks-flex-node-agent-recovery.service" + recoveryScriptPath = "/usr/local/lib/aks-flex-node/aks-flex-node-recovery.sh" + systemdSystemDir = "/etc/systemd/system" ) //go:embed assets/aks-flex-node-agent.service var serviceUnitContent []byte +//go:embed assets/aks-flex-node-agent-recovery.service +var recoveryServiceUnitContent []byte + +//go:embed assets/aks-flex-node-recovery.sh +var recoveryScriptContent []byte + type installServiceTask struct { log *slog.Logger } @@ -33,13 +42,8 @@ func InstallService(log *slog.Logger) phases.Task { func (t *installServiceTask) Name() string { return "install-service" } func (t *installServiceTask) Do(ctx context.Context) error { - unitPath := filepath.Join(systemdSystemDir, ServiceUnitName) - if err := utilio.WriteFile(unitPath, serviceUnitContent, 0o644); err != nil { //nolint:gosec // service files must be world-readable - return fmt.Errorf("write %s: %w", unitPath, err) - } - - if err := utilexec.ReloadSystemd(ctx, t.log); err != nil { - return fmt.Errorf("systemctl daemon-reload: %w", err) + if err := ensureAgentUpgradeServiceAssets(ctx, t.log); err != nil { + return err } if err := utilexec.RunCmd(ctx, t.log, utilexec.Systemctl(), "enable", ServiceUnitName); err != nil { return fmt.Errorf("systemctl enable %s: %w", ServiceUnitName, err) @@ -52,6 +56,70 @@ func (t *installServiceTask) Do(ctx context.Context) error { return nil } +func ensureAgentUpgradeServiceAssets(ctx context.Context, log *slog.Logger) error { + return ensureAgentUpgradeServiceAssetsAt( + ctx, + log, + defaultAgentUpgradePaths(), + systemdSystemDir, + recoveryScriptPath, + utilexec.ReloadSystemd, + ) +} + +func ensureAgentUpgradeServiceAssetsAt( + ctx context.Context, + log *slog.Logger, + binaryPaths agentUpgradePaths, + systemdDir, recoveryScript string, + reload func(context.Context, *slog.Logger) error, +) error { + if err := ensureAgentUpgradeLayout(ctx, log, binaryPaths); err != nil { + return fmt.Errorf("initialize agent binary layout: %w", err) + } + if err := writeAgentServiceAssets(binaryPaths, systemdDir, recoveryScript, binaryPaths.CurrentPath); err != nil { + return err + } + if err := reload(ctx, log); err != nil { + return fmt.Errorf("systemctl daemon-reload: %w", err) + } + return nil +} + +type agentServiceAsset struct { + path string + content []byte + mode os.FileMode +} + +func desiredAgentServiceAssets(binaryPaths agentUpgradePaths, systemdDir, recoveryScript, currentBinaryPath string) []agentServiceAsset { + serviceContent := bytes.ReplaceAll(serviceUnitContent, []byte(defaultAgentUpgradePaths().BinaryPath), []byte(currentBinaryPath)) + recoveryServiceContent := bytes.ReplaceAll(recoveryServiceUnitContent, []byte(recoveryScriptPath), []byte(recoveryScript)) + recoveryContent := recoveryScriptContent + for oldPath, newPath := range map[string]string{ + defaultAgentUpgradePaths().LastGoodPath: binaryPaths.LastGoodPath, + defaultAgentUpgradePaths().SignalPath: binaryPaths.SignalPath, + } { + recoveryContent = bytes.ReplaceAll(recoveryContent, []byte(oldPath), []byte(newPath)) + } + // Publish dependencies before the main unit that references OnFailure, so an + // interrupted update never leaves systemd pointing at missing recovery assets. + return []agentServiceAsset{ + {path: recoveryScript, content: recoveryContent, mode: 0o750}, + {path: filepath.Join(systemdDir, recoveryServiceUnitName), content: recoveryServiceContent, mode: 0o644}, + {path: filepath.Join(systemdDir, ServiceUnitName), content: serviceContent, mode: 0o644}, + } +} + +func writeAgentServiceAssets(binaryPaths agentUpgradePaths, systemdDir, recoveryScript, currentBinaryPath string) error { + for _, asset := range desiredAgentServiceAssets(binaryPaths, systemdDir, recoveryScript, currentBinaryPath) { + if err := utilio.WriteFile(asset.path, asset.content, asset.mode); err != nil { + return fmt.Errorf("write %s: %w", asset.path, err) + } + } + return nil +} + type uninstallServiceTask struct { log *slog.Logger } @@ -71,9 +139,15 @@ func (t *uninstallServiceTask) Do(ctx context.Context) error { t.log.Warn("failed to disable service (may not be enabled)", "unit", ServiceUnitName, "error", err) } - unitPath := filepath.Join(systemdSystemDir, ServiceUnitName) - if err := os.Remove(unitPath); err != nil && !os.IsNotExist(err) { - return fmt.Errorf("remove %s: %w", unitPath, err) + for _, path := range []string{ + filepath.Join(systemdSystemDir, ServiceUnitName), + filepath.Join(systemdSystemDir, recoveryServiceUnitName), + recoveryScriptPath, + defaultAgentUpgradePaths().SignalPath, + } { + if err := os.Remove(path); err != nil && !os.IsNotExist(err) { + return fmt.Errorf("remove %s: %w", path, err) + } } if err := utilexec.ReloadSystemd(ctx, t.log); err != nil { diff --git a/pkg/daemon/lifecycle_test.go b/pkg/daemon/lifecycle_test.go new file mode 100644 index 00000000..d0d87e4d --- /dev/null +++ b/pkg/daemon/lifecycle_test.go @@ -0,0 +1,101 @@ +package daemon + +import ( + "context" + "log/slog" + "os" + "path/filepath" + "strings" + "testing" +) + +func TestEnsureAgentUpgradeServiceAssetsMigratesExistingInstallation(t *testing.T) { + t.Parallel() + + paths := testAgentUpgradePaths(t) + if err := os.MkdirAll(filepath.Dir(paths.BinaryPath), 0o750); err != nil { + t.Fatalf("MkdirAll: %v", err) + } + if err := os.WriteFile(paths.BinaryPath, []byte("legacy"), 0o755); err != nil { + t.Fatalf("write legacy binary: %v", err) + } + systemdDir := filepath.Join(t.TempDir(), "systemd") + if err := os.MkdirAll(systemdDir, 0o750); err != nil { + t.Fatalf("MkdirAll systemd: %v", err) + } + unitPath := filepath.Join(systemdDir, ServiceUnitName) + if err := os.WriteFile(unitPath, []byte("[Service]\nExecStart=/usr/local/bin/aks-flex-node agent\n"), 0o644); err != nil { + t.Fatalf("write legacy unit: %v", err) + } + recoveryPath := filepath.Join(t.TempDir(), "aks-flex-node-recovery.sh") + reloaded := false + if err := ensureAgentUpgradeServiceAssetsAt( + t.Context(), + slog.Default(), + paths, + systemdDir, + recoveryPath, + func(context.Context, *slog.Logger) error { + reloaded = true + return nil + }, + ); err != nil { + t.Fatalf("ensureAgentUpgradeServiceAssetsAt: %v", err) + } + if !reloaded { + t.Fatal("systemd reload was not requested") + } + assertResolvedPath(t, paths.BinaryPath, paths.BluePath) + assertResolvedPath(t, paths.CurrentPath, paths.BluePath) + assertResolvedPath(t, paths.LastGoodPath, paths.BluePath) + unit, err := os.ReadFile(unitPath) + if err != nil { + t.Fatalf("read service unit: %v", err) + } + if !strings.Contains(string(unit), "OnFailure="+recoveryServiceUnitName) { + t.Fatalf("updated unit does not include recovery: %s", unit) + } + if !strings.Contains(string(unit), "ExecStart="+paths.CurrentPath+" agent") { + t.Fatalf("updated unit does not execute the managed current link: %s", unit) + } + recoveryService, err := os.ReadFile(filepath.Join(systemdDir, recoveryServiceUnitName)) + if err != nil { + t.Fatalf("recovery service was not installed: %v", err) + } + if !strings.Contains(string(recoveryService), "ExecStart="+recoveryPath) { + t.Fatalf("recovery service does not use installed script: %s", recoveryService) + } + info, err := os.Stat(recoveryPath) + if err != nil { + t.Fatalf("recovery script was not installed: %v", err) + } + if info.Mode().Perm() != 0o750 { + t.Fatalf("recovery script mode = %o, want 750", info.Mode().Perm()) + } +} + +func TestAgentServiceIncludesUpgradeRecovery(t *testing.T) { + t.Parallel() + + service := string(serviceUnitContent) + if !strings.Contains(service, "OnFailure="+recoveryServiceUnitName) { + t.Fatalf("service does not activate %s on failure", recoveryServiceUnitName) + } + if !strings.Contains(string(recoveryServiceUnitContent), "ExecStart="+recoveryScriptPath) { + t.Fatalf("recovery service does not execute %s", recoveryScriptPath) + } + script := string(recoveryScriptContent) + for _, expected := range []string{ + "recover-agent-upgrade", + "aks-flex-node-agent.service", + "aks-flex-node-last-good", + "agent-upgrade-signal.json", + "systemctl --no-block restart", + "|| status=$?", + "exit \"${status}\"", + } { + if !strings.Contains(script, expected) { + t.Fatalf("recovery script does not contain %q", expected) + } + } +} diff --git a/pkg/daemon/machineoperation_reconciler.go b/pkg/daemon/machineoperation_reconciler.go index 94f7fbf6..564c63c8 100644 --- a/pkg/daemon/machineoperation_reconciler.go +++ b/pkg/daemon/machineoperation_reconciler.go @@ -2,8 +2,10 @@ package daemon import ( "context" + "errors" "fmt" "log/slog" + "time" apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/meta" @@ -12,6 +14,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" machinav1alpha3 "github.com/Azure/unbounded/api/machina/v1alpha3" + "github.com/Azure/unbounded/pkg/agent/agentbinary" "github.com/Azure/unbounded/pkg/agent/daemon" ) @@ -25,11 +28,13 @@ type machineOperationReconcilerOptions struct { AKSMachineName string MachineOperationMode string Operator nodeOperator + AgentUpgrade agentUpgradeExecutor } type machineOperationHandlers struct { - log *slog.Logger - operator nodeOperator + log *slog.Logger + operator nodeOperator + agentUpgrade agentUpgradeExecutor } // machineOperationReconciler runs MachineOperations when the Machina CRD is available. @@ -46,6 +51,9 @@ func machineOperationReconciler( if opts.Operator == nil { return nil, fmt.Errorf("node operator is nil") } + if opts.AgentUpgrade == nil { + return nil, fmt.Errorf("agent upgrade executor is nil") + } if opts.MachineOperationMode == "" { opts.MachineOperationMode = machineOperationModeAuto } @@ -73,14 +81,18 @@ func machineOperationReconciler( return nil, fmt.Errorf("AKS machine name is empty") } - handlers := &machineOperationHandlers{log: opts.Log, operator: opts.Operator} + handlers := &machineOperationHandlers{ + log: opts.Log, + operator: opts.Operator, + agentUpgrade: opts.AgentUpgrade, + } reconciler, err := daemon.NewMachinaMachineOperationReconciler( opts.Client, opts.NodeName, opts.AKSMachineName, daemon.MachineOperationHandlers{ machinav1alpha3.OperationNodeReboot: handlers.reconcileNodeReboot, - machinav1alpha3.OperationAgentUpgrade: handlers.unsupportedOperation, + machinav1alpha3.OperationAgentUpgrade: handlers.reconcileAgentUpgrade, machinav1alpha3.OperationAgentReset: handlers.reconcileAgentReset, }, ) @@ -136,6 +148,94 @@ func (h *machineOperationHandlers) reconcileNodeReboot( return ctrl.Result{}, nil } +func (h *machineOperationHandlers) reconcileAgentUpgrade( + ctx context.Context, + store daemon.MachineOperationStore[int64], + op daemon.MachineOperation, +) (ctrl.Result, error) { + request, err := parseAgentUpgradeRequest(op.Parameters) + if err != nil { + return h.finishFailedMachineOperation(ctx, store, op, "InvalidParameters", err.Error()) + } + activationLock, err := h.agentUpgrade.Acquire() + if errors.Is(err, agentbinary.ErrActivationInProgress) { + return ctrl.Result{RequeueAfter: 5 * time.Second}, nil + } + if err != nil { + return ctrl.Result{}, fmt.Errorf("acquire agent activation lock: %w", err) + } + defer func() { + if closeErr := activationLock.Close(); closeErr != nil { + h.log.Warn("failed to release agent activation lock", "error", closeErr) + } + }() + // Persist the recovery signal before InProgress. The shared reconciler does + // not enqueue InProgress operations after a process crash, so the signal + // must exist before the status can become non-reconcilable. + if err := h.agentUpgrade.RecordPending(ctx, op.Name); err != nil { + if errors.Is(err, errAgentUpgradeAlreadyPending) { + // Retry a previously failed recovery handoff. An ordinary duplicate + // remains a no-op while the delayed daemon restart is pending. + return ctrl.Result{}, h.agentUpgrade.RetryRecovery(ctx) + } + return h.finishFailedMachineOperation(ctx, store, op, "ExecutionFailed", err.Error()) + } + if err := store.MarkInProgress(ctx, op, "staging upgraded AKS Flex Node agent binary"); err != nil { + cleanupCtx, cancel := agentUpgradeCleanupContext(ctx) + abortErr := h.agentUpgrade.Abort(cleanupCtx) + cancel() + return ctrl.Result{}, errors.Join( + fmt.Errorf("mark AgentUpgrade MachineOperation in progress: %w", err), + wrapOptionalError("clear pending AgentUpgrade signal", abortErr), + ) + } + if err := h.agentUpgrade.Stage(ctx, request); err != nil { + if abortErr := h.agentUpgrade.Abort(ctx); abortErr != nil { + return h.beginAgentUpgradeRecovery(ctx, op, err, abortErr) + } + return h.finishFailedMachineOperation(ctx, store, op, "ExecutionFailed", err.Error()) + } + if err := h.agentUpgrade.Restart(ctx); err != nil { + if abortErr := h.agentUpgrade.Abort(ctx); abortErr != nil { + return h.beginAgentUpgradeRecovery(ctx, op, err, abortErr) + } + return h.finishFailedMachineOperation(ctx, store, op, "ExecutionFailed", "failed to restart upgraded agent daemon") + } + // Restart scheduling is the old daemon's final responsibility. Keep the + // operation InProgress and let the restarted or recovery daemon publish the + // only terminal result. + return ctrl.Result{}, nil +} + +func (h *machineOperationHandlers) beginAgentUpgradeRecovery( + ctx context.Context, + op daemon.MachineOperation, + executionErr, rollbackErr error, +) (ctrl.Result, error) { + message := fmt.Sprintf("AgentUpgrade execution failed and requires recovery: %v", executionErr) + recordErr := h.agentUpgrade.RecordFailure(message) + if recordErr != nil { + h.log.Error("failed to annotate durable AgentUpgrade recovery signal", "operation", op.Name, "error", recordErr) + } + h.log.Error("AgentUpgrade rollback failed; restarting daemon for durable recovery", + "operation", op.Name, + "error", rollbackErr, + ) + cleanupCtx, cancel := agentUpgradeCleanupContext(ctx) + defer cancel() + restartErr := h.agentUpgrade.Restart(cleanupCtx) + if restartErr != nil { + h.log.Error("failed to restart daemon for AgentUpgrade recovery", "operation", op.Name, "error", restartErr) + } + if recordErr != nil || restartErr != nil { + return ctrl.Result{}, errors.Join( + wrapOptionalError("record AgentUpgrade recovery failure", recordErr), + wrapOptionalError("restart daemon for AgentUpgrade recovery", restartErr), + ) + } + return ctrl.Result{}, nil +} + func (h *machineOperationHandlers) reconcileAgentReset( ctx context.Context, store daemon.MachineOperationStore[int64], diff --git a/pkg/daemon/machineoperation_reconciler_test.go b/pkg/daemon/machineoperation_reconciler_test.go index dffa321a..d682b1bc 100644 --- a/pkg/daemon/machineoperation_reconciler_test.go +++ b/pkg/daemon/machineoperation_reconciler_test.go @@ -3,6 +3,7 @@ package daemon import ( "context" "errors" + "io" "log/slog" "strings" "testing" @@ -12,6 +13,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client/fake" machinav1alpha3 "github.com/Azure/unbounded/api/machina/v1alpha3" + "github.com/Azure/unbounded/pkg/agent/agentbinary" "github.com/Azure/unbounded/pkg/agent/daemon" ) @@ -67,6 +69,7 @@ func TestMachineOperationReconcilerDisableModeSkipsDiscovery(t *testing.T) { Log: slog.Default(), MachineOperationMode: machineOperationModeDisable, Operator: &fakeNodeOperator{}, + AgentUpgrade: &fakeAgentUpgradeExecutor{}, }) if err != nil { t.Fatalf("machineOperationReconciler: %v", err) @@ -99,11 +102,20 @@ func TestMachineOperationReconcilerRequiresDependencies(t *testing.T) { }, "missing operator": { opts: machineOperationReconcilerOptions{ - Client: fake.NewClientBuilder().Build(), - Log: slog.Default(), + Client: fake.NewClientBuilder().Build(), + Log: slog.Default(), + AgentUpgrade: &fakeAgentUpgradeExecutor{}, }, wantErr: "node operator is nil", }, + "missing agent upgrade executor": { + opts: machineOperationReconcilerOptions{ + Client: fake.NewClientBuilder().Build(), + Log: slog.Default(), + Operator: &fakeNodeOperator{}, + }, + wantErr: "agent upgrade executor is nil", + }, } for name, tt := range tests { @@ -151,6 +163,7 @@ func TestMachineOperationReconcilerEnabledRequiresNames(t *testing.T) { NodeName: tt.nodeName, AKSMachineName: tt.aksMachineName, Operator: &fakeNodeOperator{}, + AgentUpgrade: &fakeAgentUpgradeExecutor{}, }) if err == nil { t.Fatal("machineOperationReconciler error = nil, want error") @@ -278,6 +291,193 @@ func TestMachineOperationHandlersUnsupportedOperation(t *testing.T) { } } +func TestMachineOperationHandlersAgentUpgrade(t *testing.T) { + t.Parallel() + + upgrader := &fakeAgentUpgradeExecutor{} + store := &fakeMachineOperationStore{} + target := &machineOperationHandlers{log: slog.Default(), operator: &fakeNodeOperator{}, agentUpgrade: upgrader} + digest := strings.Repeat("a", 64) + op := daemon.MachineOperation{ + Name: "upgrade-1", + Kind: machinav1alpha3.OperationAgentUpgrade, + Parameters: map[string]string{ + agentUpgradeDownloadURLParameter: "https://example.com/agent.tar.gz?sig=secret", + agentUpgradeSHA256Parameter: digest, + }, + } + + if _, err := target.reconcileAgentUpgrade(t.Context(), store, op); err != nil { + t.Fatalf("reconcileAgentUpgrade: %v", err) + } + if !store.inProgress || !upgrader.pending || !upgrader.staged || !upgrader.restarted { + t.Fatalf("upgrade calls = inProgress:%v pending:%v staged:%v restarted:%v", store.inProgress, upgrader.pending, upgrader.staged, upgrader.restarted) + } + if store.result.Phase != "" { + t.Fatalf("phase = %s, want non-terminal until daemon restart", store.result.Phase) + } +} + +func TestMachineOperationHandlersAgentUpgradeInvalidParameters(t *testing.T) { + t.Parallel() + + upgrader := &fakeAgentUpgradeExecutor{} + store := &fakeMachineOperationStore{} + target := &machineOperationHandlers{log: slog.Default(), operator: &fakeNodeOperator{}, agentUpgrade: upgrader} + + if _, err := target.reconcileAgentUpgrade(t.Context(), store, daemon.MachineOperation{Name: "upgrade-1", Kind: machinav1alpha3.OperationAgentUpgrade}); err != nil { + t.Fatalf("reconcileAgentUpgrade: %v", err) + } + if store.result.Phase != machinav1alpha3.OperationPhaseFailed || store.result.Reason != "InvalidParameters" { + t.Fatalf("result = %#v, want InvalidParameters failure", store.result) + } + if upgrader.pending || upgrader.staged || upgrader.restarted { + t.Fatal("executor called for invalid parameters") + } +} + +func TestMachineOperationHandlersAgentUpgradeDuplicateReconcileIsNoop(t *testing.T) { + t.Parallel() + + upgrader := &fakeAgentUpgradeExecutor{pendingErr: errAgentUpgradeAlreadyPending} + store := &fakeMachineOperationStore{} + target := &machineOperationHandlers{log: slog.Default(), operator: &fakeNodeOperator{}, agentUpgrade: upgrader} + op := daemon.MachineOperation{Name: "upgrade-1", Parameters: map[string]string{ + agentUpgradeDownloadURLParameter: "https://example.com/agent.tar.gz", + agentUpgradeSHA256Parameter: strings.Repeat("a", 64), + }} + + if _, err := target.reconcileAgentUpgrade(t.Context(), store, op); err != nil { + t.Fatalf("reconcileAgentUpgrade: %v", err) + } + if upgrader.staged || upgrader.restarted || upgrader.aborted { + t.Fatal("duplicate reconciliation executed upgrade actions") + } + if store.result.Phase != "" { + t.Fatalf("duplicate reconciliation produced terminal phase %s", store.result.Phase) + } +} + +func TestMachineOperationHandlersAgentUpgradeRequeuesWhenDirectActivationHoldsLock(t *testing.T) { + t.Parallel() + + upgrader := &fakeAgentUpgradeExecutor{acquireErr: agentbinary.ErrActivationInProgress} + store := &fakeMachineOperationStore{} + target := &machineOperationHandlers{log: slog.Default(), operator: &fakeNodeOperator{}, agentUpgrade: upgrader} + op := daemon.MachineOperation{Name: "upgrade-1", Parameters: map[string]string{ + agentUpgradeDownloadURLParameter: "https://example.com/agent.tar.gz", + agentUpgradeSHA256Parameter: strings.Repeat("a", 64), + }} + result, err := target.reconcileAgentUpgrade(t.Context(), store, op) + if err != nil { + t.Fatalf("reconcileAgentUpgrade: %v", err) + } + if result.RequeueAfter <= 0 { + t.Fatalf("RequeueAfter = %v, want positive duration", result.RequeueAfter) + } + if store.inProgress || upgrader.pending || upgrader.staged || upgrader.restarted { + t.Fatal("lock contention mutated the operation or staged an upgrade") + } +} + +func TestMachineOperationHandlersAgentUpgradeMarkFailureClearsPendingSignal(t *testing.T) { + t.Parallel() + + upgrader := &fakeAgentUpgradeExecutor{} + store := &fakeMachineOperationStore{markErr: errors.New("status update failed")} + target := &machineOperationHandlers{log: slog.Default(), operator: &fakeNodeOperator{}, agentUpgrade: upgrader} + op := daemon.MachineOperation{Name: "upgrade-1", Parameters: map[string]string{ + agentUpgradeDownloadURLParameter: "https://example.com/agent.tar.gz", + }} + + _, err := target.reconcileAgentUpgrade(t.Context(), store, op) + if err == nil || !strings.Contains(err.Error(), "mark AgentUpgrade MachineOperation in progress") { + t.Fatalf("reconcileAgentUpgrade error = %v", err) + } + if !upgrader.pending || !upgrader.aborted { + t.Fatalf("pending = %v, aborted = %v; want durable signal followed by cleanup", upgrader.pending, upgrader.aborted) + } + if upgrader.staged || upgrader.restarted { + t.Fatal("status failure staged or restarted the agent") + } +} + +func TestMachineOperationHandlersAgentUpgradeStageFailureRollsBack(t *testing.T) { + t.Parallel() + + upgrader := &fakeAgentUpgradeExecutor{stageErr: errors.New("bad archive")} + store := &fakeMachineOperationStore{} + target := &machineOperationHandlers{log: slog.Default(), operator: &fakeNodeOperator{}, agentUpgrade: upgrader} + op := daemon.MachineOperation{Name: "upgrade-1", Parameters: map[string]string{ + agentUpgradeDownloadURLParameter: "https://example.com/agent.tar.gz", + agentUpgradeSHA256Parameter: strings.Repeat("a", 64), + }} + + if _, err := target.reconcileAgentUpgrade(t.Context(), store, op); err != nil { + t.Fatalf("reconcileAgentUpgrade: %v", err) + } + if !upgrader.aborted { + t.Fatal("Abort was not called") + } + if store.result.Phase != machinav1alpha3.OperationPhaseFailed || store.result.Message != "bad archive" { + t.Fatalf("result = %#v", store.result) + } +} + +func TestMachineOperationHandlersAgentUpgradeAbortFailureStartsRecovery(t *testing.T) { + t.Parallel() + + upgrader := &fakeAgentUpgradeExecutor{ + stageErr: errors.New("stage failed"), + abortErr: errors.New("rollback failed"), + } + store := &fakeMachineOperationStore{} + target := &machineOperationHandlers{log: slog.Default(), operator: &fakeNodeOperator{}, agentUpgrade: upgrader} + op := daemon.MachineOperation{Name: "upgrade-1", Parameters: map[string]string{ + agentUpgradeDownloadURLParameter: "https://example.com/agent.tar.gz", + agentUpgradeSHA256Parameter: strings.Repeat("a", 64), + }} + + if _, err := target.reconcileAgentUpgrade(t.Context(), store, op); err != nil { + t.Fatalf("reconcileAgentUpgrade: %v", err) + } + if upgrader.failure == "" || !upgrader.restarted { + t.Fatalf("recovery failure = %q, restarted = %v", upgrader.failure, upgrader.restarted) + } + if store.result.Phase != "" { + t.Fatalf("phase = %s, want recovery to publish terminal status", store.result.Phase) + } +} + +func TestMachineOperationHandlersAgentUpgradeRecoveryRestartFailureRequeues(t *testing.T) { + t.Parallel() + + upgrader := &fakeAgentUpgradeExecutor{ + stageErr: errors.New("stage failed"), + abortErr: errors.New("rollback failed"), + restartErr: errors.New("restart failed"), + } + target := &machineOperationHandlers{log: slog.Default(), operator: &fakeNodeOperator{}, agentUpgrade: upgrader} + op := daemon.MachineOperation{Name: "upgrade-1", Parameters: map[string]string{ + agentUpgradeDownloadURLParameter: "https://example.com/agent.tar.gz", + }} + + _, err := target.reconcileAgentUpgrade(t.Context(), &fakeMachineOperationStore{}, op) + if err == nil || !strings.Contains(err.Error(), "restart daemon for AgentUpgrade recovery") { + t.Fatalf("reconcileAgentUpgrade error = %v, want recovery restart error", err) + } + + upgrader.pendingErr = errAgentUpgradeAlreadyPending + upgrader.restartErr = nil + upgrader.restarted = false + if _, err := target.reconcileAgentUpgrade(t.Context(), &fakeMachineOperationStore{}, op); err != nil { + t.Fatalf("reconcileAgentUpgrade retry: %v", err) + } + if !upgrader.restarted { + t.Fatal("requeued recovery did not retry the daemon restart") + } +} + func TestMachineOperationHandlersAgentReset(t *testing.T) { t.Parallel() @@ -342,6 +542,63 @@ func TestMachineOperationHandlersAgentResetStopFailure(t *testing.T) { } } +type fakeAgentUpgradeExecutor struct { + pending bool + staged bool + aborted bool + restarted bool + failure string + recovering bool + acquireErr error + pendingErr error + stageErr error + abortErr error + restartErr error +} + +func (f *fakeAgentUpgradeExecutor) Acquire() (io.Closer, error) { + if f.acquireErr != nil { + return nil, f.acquireErr + } + return io.NopCloser(strings.NewReader("")), nil +} + +func (f *fakeAgentUpgradeExecutor) RecordPending(context.Context, string) error { + f.pending = true + return f.pendingErr +} + +func (f *fakeAgentUpgradeExecutor) RetryRecovery(context.Context) error { + if !f.recovering { + return nil + } + f.restarted = true + return f.restartErr +} + +func (f *fakeAgentUpgradeExecutor) RecordFailure(message string) error { + f.failure = message + f.recovering = true + return nil +} + +func (f *fakeAgentUpgradeExecutor) Stage(context.Context, agentUpgradeRequest) error { + f.staged = true + return f.stageErr +} + +func (f *fakeAgentUpgradeExecutor) Abort(context.Context) error { + f.aborted = true + return f.abortErr +} + +func (f *fakeAgentUpgradeExecutor) Restart(context.Context) error { + f.restarted = true + return f.restartErr +} + +var _ agentUpgradeExecutor = (*fakeAgentUpgradeExecutor)(nil) + type fakeMachineOperationStore struct { inProgress bool operation daemon.MachineOperation diff --git a/pkg/daemon/startup_gate.go b/pkg/daemon/startup_gate.go new file mode 100644 index 00000000..20b084e5 --- /dev/null +++ b/pkg/daemon/startup_gate.go @@ -0,0 +1,65 @@ +package daemon + +import ( + "context" + "sync" + + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/builder" + + agentdaemon "github.com/Azure/unbounded/pkg/agent/daemon" +) + +type startupGate struct { + ready chan struct{} + once sync.Once +} + +func newStartupGate() *startupGate { + return &startupGate{ready: make(chan struct{})} +} + +func (g *startupGate) open() { + g.once.Do(func() { close(g.ready) }) +} + +func (g *startupGate) wait(ctx context.Context) error { + select { + case <-g.ready: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +type gatedMachineOperationReconciler struct { + delegate agentdaemon.MachineOperationRequestReconciler + gate *startupGate +} + +func (r gatedMachineOperationReconciler) SetupController(b *builder.TypedBuilder[agentdaemon.Request]) *builder.TypedBuilder[agentdaemon.Request] { + return r.delegate.SetupController(b) +} + +func (r gatedMachineOperationReconciler) ReconcileMachineOperation(ctx context.Context, name string) (ctrl.Result, error) { + if err := r.gate.wait(ctx); err != nil { + return ctrl.Result{}, err + } + return r.delegate.ReconcileMachineOperation(ctx, name) +} + +type gatedRepaveReconciler struct { + delegate agentdaemon.RepaveReconciler + gate *startupGate +} + +func (r gatedRepaveReconciler) SetupController(b *builder.TypedBuilder[agentdaemon.Request]) *builder.TypedBuilder[agentdaemon.Request] { + return r.delegate.SetupController(b) +} + +func (r gatedRepaveReconciler) ReconcileRepave(ctx context.Context, source string) (ctrl.Result, error) { + if err := r.gate.wait(ctx); err != nil { + return ctrl.Result{}, err + } + return r.delegate.ReconcileRepave(ctx, source) +} diff --git a/pkg/daemon/startup_gate_test.go b/pkg/daemon/startup_gate_test.go new file mode 100644 index 00000000..72e226e0 --- /dev/null +++ b/pkg/daemon/startup_gate_test.go @@ -0,0 +1,41 @@ +package daemon + +import ( + "context" + "errors" + "testing" +) + +func TestStartupGate(t *testing.T) { + t.Parallel() + + t.Run("blocks until opened", func(t *testing.T) { + t.Parallel() + gate := newStartupGate() + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan error, 1) + go func() { done <- gate.wait(ctx) }() + + select { + case err := <-done: + t.Fatalf("wait returned before open: %v", err) + default: + } + gate.open() + if err := <-done; err != nil { + t.Fatalf("wait after open: %v", err) + } + gate.open() // idempotent + }) + + t.Run("honors cancellation", func(t *testing.T) { + t.Parallel() + gate := newStartupGate() + ctx, cancel := context.WithCancel(t.Context()) + cancel() + if err := gate.wait(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("wait error = %v, want context.Canceled", err) + } + }) +} diff --git a/pkg/utils/utilexec/exec.go b/pkg/utils/utilexec/exec.go index 88f25478..a98d199f 100644 --- a/pkg/utils/utilexec/exec.go +++ b/pkg/utils/utilexec/exec.go @@ -140,6 +140,13 @@ func Systemctl() func(context.Context) *exec.Cmd { } } +// SystemdRun returns a command factory for systemd-run. +func SystemdRun() func(context.Context) *exec.Cmd { + return func(ctx context.Context) *exec.Cmd { + return exec.CommandContext(ctx, "systemd-run") // #nosec G204 -- fixed binary + } +} + // Azcmagent returns a command factory for azcmagent. func Azcmagent() func(context.Context) *exec.Cmd { return func(ctx context.Context) *exec.Cmd {