feat(ha): node local loadbalancing (#2012)

* feat(ha): node local loadbalancing

* f

* f

* f

* f

* f

* f

* f

* f

* f

* test with new kots binary

* no-op

* fix kots version

* no-op

* f

* no-op

* keep noop

* f

* f

* f

* f

* f

* f

* f

* f

* f

---------

Co-authored-by: Salah Aldeen Al Saleh <[email protected]>

Author: emosbaugh

.github/workflows/ci.yaml

@@ -113,6 +113,8 @@ jobs:
     - name: Checkout
       uses: actions/checkout@v4
     - name: Make manifests
+      env:
+        GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
       run: make -C operator manifests
     - name: Check CRDs
       run: |

cmd/buildtools/k0s.go

@@ -76,6 +76,12 @@ var k0sImageComponents = map[string]addonComponent{
 			return fmt.Sprintf("registry.k8s.io/pause:%s", opts.upstreamVersion.Original()), nil
 		},
 	},
+	"quay.io/k0sproject/envoy-distroless": {
+		name: "envoy-distroless",
+		getWolfiPackageName: func(opts addonComponentOptions) string {
+			return fmt.Sprintf("envoy-%d.%d", opts.upstreamVersion.Major(), opts.upstreamVersion.Minor())
+		},
+	},
 }
 
 var updateK0sImagesCommand = &cli.Command{

cmd/buildtools/utils.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"bufio"
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -145,10 +146,12 @@ func ResolveApkoPackageVersion(componentName, packageName, packageVersion string
 		fmt.Sprintf("PACKAGE_NAME=%s", packageName),
 		fmt.Sprintf("PACKAGE_VERSION=%s", packageVersion),
 	}
+	var errBuf bytes.Buffer
 	cmd := exec.Command("make", args...)
+	cmd.Stderr = &errBuf
 	out, err := cmd.Output()
 	if err != nil {
-		return "", fmt.Errorf("run command: %w: %s", err, string(out))
+		return "", fmt.Errorf("run command: %w: %s", err, errBuf.String())
 	}
 	return strings.TrimSpace(string(out)), nil
 }

deploy/images/envoy-distroless/apko.tmpl.yaml

@@ -0,0 +1,31 @@
+# source: https://github.com/chainguard-images/images/blob/bea234042585fd6db129bc2c836aad4937f55799/images/envoy/config/main.tf
+contents:
+  repositories:
+    - https://packages.wolfi.dev/os
+  keyring:
+    - https://packages.wolfi.dev/os/wolfi-signing.rsa.pub
+  packages:
+    - envoy~${PACKAGE_VERSION}
+    - envoy-config~${PACKAGE_VERSION}
+    - envoy-oci-entrypoint~${PACKAGE_VERSION}
+    - su-exec
+
+accounts:
+  groups:
+    - groupname: nonroot
+      gid: 65532
+  users:
+    - username: nonroot
+      uid: 65532
+      gid: 65532
+  run-as: 65532
+
+paths:
+  - path: /etc/envoy
+    type: directory
+    uid: 65532
+    gid: 65532
+    permissions: 0o755
+
+entrypoint:
+  command: /var/lib/envoy/init/envoy-entrypoint.sh

e2e/cluster/docker/cluster.go

@@ -13,21 +13,23 @@ import (
 type Cluster struct {
 	Nodes []*Container
 
-	t *testing.T
+	t                      *testing.T
+	supportBundleNodeIndex int
 }
 
 type ClusterInput struct {
-	T                    *testing.T
-	Nodes                int
-	Distro               string
-	LicensePath          string
-	ECBinaryPath         string
-	ECReleaseBuilderPath string
-	K0sDir               string
+	T                      *testing.T
+	Nodes                  int
+	Distro                 string
+	LicensePath            string
+	ECBinaryPath           string
+	ECReleaseBuilderPath   string
+	K0sDir                 string
+	SupportBundleNodeIndex int
 }
 
 func NewCluster(in *ClusterInput) *Cluster {
-	c := &Cluster{t: in.T}
+	c := &Cluster{t: in.T, supportBundleNodeIndex: in.SupportBundleNodeIndex}
 
 	c.Nodes = make([]*Container, in.Nodes)
 
@@ -170,16 +172,17 @@ func (c *Cluster) generateSupportBundle(envs ...map[string]string) {
 		}(i, &wg)
 	}
 
-	c.t.Logf("%s: generating cluster support bundle from node 0", time.Now().Format(time.RFC3339))
-	if stdout, stderr, err := c.RunCommandOnNode(0, []string{"collect-support-bundle-cluster.sh"}, envs...); err != nil {
+	node := c.Nodes[c.supportBundleNodeIndex]
+	c.t.Logf("%s: generating cluster support bundle from node %d", time.Now().Format(time.RFC3339), c.supportBundleNodeIndex)
+	if stdout, stderr, err := c.RunCommandOnNode(c.supportBundleNodeIndex, []string{"collect-support-bundle-cluster.sh"}, envs...); err != nil {
 		c.t.Logf("stdout: %s", stdout)
 		c.t.Logf("stderr: %s", stderr)
-		c.t.Logf("fail to generate cluster support from node %d bundle: %v", 0, err)
+		c.t.Logf("fail to generate cluster support from node %d bundle: %v", c.supportBundleNodeIndex, err)
 	} else {
-		c.t.Logf("%s: copying cluster support bundle from node 0 to local machine", time.Now().Format(time.RFC3339))
-		src := fmt.Sprintf("%s:cluster.tar.gz", c.Nodes[0].GetName())
+		c.t.Logf("%s: copying cluster support bundle from node %d to local machine", time.Now().Format(time.RFC3339), c.supportBundleNodeIndex)
+		src := fmt.Sprintf("%s:cluster.tar.gz", node.GetName())
 		dst := "support-bundle-cluster.tar.gz"
-		if stdout, stderr, err := c.Nodes[0].CopyFile(src, dst); err != nil {
+		if stdout, stderr, err := node.CopyFile(src, dst); err != nil {
 			c.t.Logf("stdout: %s", stdout)
 			c.t.Logf("stderr: %s", stderr)
 			c.t.Logf("fail to generate cluster support bundle from node 0: %v", err)

e2e/cluster/lxd/cluster.go

@@ -70,6 +70,7 @@ type ClusterInput struct {
 	WithProxy                         bool
 	id                                string
 	AdditionalFiles                   []File
+	SupportBundleNodeIndex            int
 }
 
 // File holds information about a file that must be uploaded to a node.
@@ -88,12 +89,13 @@ type Dir struct {
 // Cluster is returned when a cluster is created. Contain a list of all node
 // names and the cluster id.
 type Cluster struct {
-	Nodes   []string
-	IPs     []string
-	network string
-	id      string
-	T       *testing.T
-	Proxy   string
+	Nodes                  []string
+	IPs                    []string
+	network                string
+	id                     string
+	T                      *testing.T
+	Proxy                  string
+	supportBundleNodeIndex int
 }
 
 // Destroy destroys a cluster pointed by the id property.
@@ -210,9 +212,10 @@ func NewCluster(in *ClusterInput) *Cluster {
 	in.network = <-networkaddr
 
 	out := &Cluster{
-		T:       in.T,
-		network: in.network,
-		id:      in.id,
+		T:                      in.T,
+		network:                in.network,
+		id:                     in.id,
+		supportBundleNodeIndex: in.SupportBundleNodeIndex,
 	}
 	out.T.Cleanup(out.Destroy)
 
@@ -1174,10 +1177,10 @@ func (c *Cluster) generateSupportBundle(envs ...map[string]string) {
 		}(i, &wg)
 	}
 
-	node := c.Nodes[0]
+	node := c.Nodes[c.supportBundleNodeIndex]
 	c.T.Logf("%s: generating cluster support bundle from node %s", time.Now().Format(time.RFC3339), node)
 	line := []string{"collect-support-bundle-cluster.sh"}
-	if stdout, stderr, err := c.RunCommandOnNode(0, line, envs...); err != nil {
+	if stdout, stderr, err := c.RunCommandOnNode(c.supportBundleNodeIndex, line, envs...); err != nil {
 		c.T.Logf("stdout: %s", stdout)
 		c.T.Logf("stderr: %s", stderr)
 		c.T.Logf("fail to generate cluster support from node %s bundle: %v", node, err)

e2e/install_test.go

@@ -1851,11 +1851,12 @@ func TestMultiNodeAirgapUpgradePreviousStable(t *testing.T) {
 // for them to report ready. Runs additional high availability validations afterwards.
 func TestMultiNodeHAInstallation(t *testing.T) {
 	tc := docker.NewCluster(&docker.ClusterInput{
-		T:            t,
-		Nodes:        4,
-		Distro:       "debian-bookworm",
-		LicensePath:  "license.yaml",
-		ECBinaryPath: "../output/bin/embedded-cluster",
+		T:                      t,
+		Nodes:                  4,
+		Distro:                 "debian-bookworm",
+		LicensePath:            "license.yaml",
+		ECBinaryPath:           "../output/bin/embedded-cluster",
+		SupportBundleNodeIndex: 2,
 	})
 	defer tc.Cleanup()
 
@@ -1943,24 +1944,30 @@ func TestMultiNodeHAInstallation(t *testing.T) {
 	}
 
 	bin := strings.Split(command, " ")[0]
-	t.Logf("%s: resetting controller node 2", time.Now().Format(time.RFC3339))
-	stdout, stderr, err = tc.RunCommandOnNode(2, []string{bin, "reset", "--yes"})
+	t.Logf("%s: resetting controller node 0", time.Now().Format(time.RFC3339))
+	stdout, stderr, err = tc.RunCommandOnNode(0, []string{bin, "reset", "--yes"})
 	if err != nil {
-		t.Fatalf("fail to remove controller node 2: %v: %s: %s", err, stdout, stderr)
+		t.Fatalf("fail to remove controller node 0: %v: %s: %s", err, stdout, stderr)
 	}
 	if !strings.Contains(stdout, "High-availability clusters must maintain at least three controller nodes") {
 		t.Errorf("reset output does not contain the ha warning")
 		t.Logf("stdout: %s\nstderr: %s", stdout, stderr)
 	}
 
-	stdout, stderr, err = tc.RunCommandOnNode(0, []string{"check-nodes-removed.sh", "3"})
+	stdout, stderr, err = tc.RunCommandOnNode(2, []string{"check-nodes-removed.sh", "3"})
 	if err != nil {
-		t.Fatalf("fail to remove worker node 0: %v: %s: %s", err, stdout, stderr)
+		t.Fatalf("fail to check nodes removed: %v: %s: %s", err, stdout, stderr)
+	}
+
+	t.Logf("%s: checking nllb", time.Now().Format(time.RFC3339))
+	line = []string{"check-nllb.sh"}
+	if stdout, stderr, err := tc.RunCommandOnNode(2, line); err != nil {
+		t.Fatalf("fail to check nllb: %v: %s: %s", err, stdout, stderr)
 	}
 
 	t.Logf("%s: checking installation state after upgrade", time.Now().Format(time.RFC3339))
 	line = []string{"check-postupgrade-state.sh", k8sVersion(), ecUpgradeTargetVersion()}
-	if stdout, stderr, err := tc.RunCommandOnNode(0, line); err != nil {
+	if stdout, stderr, err := tc.RunCommandOnNode(2, line); err != nil {
 		t.Fatalf("fail to check postupgrade state: %v: %s: %s", err, stdout, stderr)
 	}
 
@@ -1993,6 +2000,7 @@ func TestMultiNodeAirgapHAInstallation(t *testing.T) {
 		WithProxy:               true,
 		AirgapInstallBundlePath: airgapInstallBundlePath,
 		AirgapUpgradeBundlePath: airgapUpgradeBundlePath,
+		SupportBundleNodeIndex:  2,
 	})
 	defer tc.Cleanup()
 
@@ -2023,10 +2031,7 @@ func TestMultiNodeAirgapHAInstallation(t *testing.T) {
 	if _, _, err := tc.RunCommandOnNode(0, line); err != nil {
 		t.Fatalf("fail to remove airgap bundle on node %s: %v", tc.Nodes[0], err)
 	}
-	line = []string{"rm", "/usr/local/bin/embedded-cluster"}
-	if _, _, err := tc.RunCommandOnNode(0, line); err != nil {
-		t.Fatalf("fail to remove embedded-cluster binary on node %s: %v", tc.Nodes[0], err)
-	}
+	// do not remove the embedded-cluster binary as it is used for reset
 
 	if _, _, err := tc.SetupPlaywrightAndRunTest("deploy-app"); err != nil {
 		t.Fatalf("fail to run playwright test deploy-app: %v", err)
@@ -2054,8 +2059,8 @@ func TestMultiNodeAirgapHAInstallation(t *testing.T) {
 		t.Fatalf("fail to prepare airgap files on node 1: %v", err)
 	}
 	t.Logf("%s: joining node 1 to the cluster as a worker", time.Now().Format(time.RFC3339))
-	if _, _, err := tc.RunCommandOnNode(1, strings.Split(command, " ")); err != nil {
-		t.Fatalf("fail to join node 1 to the cluster as a worker: %v", err)
+	if stdout, stderr, err := tc.RunCommandOnNode(1, strings.Split(command, " ")); err != nil {
+		t.Fatalf("fail to join node 1 to the cluster as a worker: %v: %s: %s", err, stdout, stderr)
 	}
 	// remove the airgap bundle and binary after joining
 	line = []string{"rm", "/assets/release.airgap"}
@@ -2091,7 +2096,10 @@ func TestMultiNodeAirgapHAInstallation(t *testing.T) {
 	if _, _, err := tc.RunCommandOnNode(2, line); err != nil {
 		t.Fatalf("fail to remove airgap bundle on node 2: %v", err)
 	}
-	// don't remove the embedded-cluster binary as it is used for reset
+	line = []string{"rm", "/usr/local/bin/embedded-cluster"}
+	if _, _, err := tc.RunCommandOnNode(2, line); err != nil {
+		t.Fatalf("fail to remove embedded-cluster binary on node 2: %v", err)
+	}
 
 	// join another controller in HA mode
 	stdout, stderr, err = tc.RunPlaywrightTest("get-join-controller-command")
@@ -2163,27 +2171,32 @@ func TestMultiNodeAirgapHAInstallation(t *testing.T) {
 	}
 
 	bin := strings.Split(command, " ")[0]
-	t.Logf("%s: resetting controller node 2 with bin %q", time.Now().Format(time.RFC3339), bin)
-	stdout, stderr, err = tc.RunCommandOnNode(2, []string{bin, "reset", "--yes"})
+	t.Logf("%s: resetting controller node 0 with bin %q", time.Now().Format(time.RFC3339), bin)
+	stdout, stderr, err = tc.RunCommandOnNode(0, []string{bin, "reset", "--yes"})
 	if err != nil {
 		t.Logf("stdout: %s\nstderr: %s", stdout, stderr)
-		t.Fatalf("fail to remove controller node %s:", err)
+		t.Fatalf("fail to remove controller node 0 %s:", err)
 	}
 	if !strings.Contains(stdout, "High-availability clusters must maintain at least three controller nodes") {
 		t.Errorf("reset output does not contain the ha warning")
 		t.Logf("stdout: %s\nstderr: %s", stdout, stderr)
 	}
 
-	stdout, _, err = tc.RunCommandOnNode(0, []string{"check-nodes-removed.sh", "3"})
+	stdout, _, err = tc.RunCommandOnNode(2, []string{"check-nodes-removed.sh", "3"})
 	if err != nil {
-		t.Log(stdout)
-		t.Fatalf("fail to remove worker node %s:", err)
+		t.Fatalf("fail to check nodes removed: %v: %s: %s", err, stdout, stderr)
+	}
+
+	t.Logf("%s: checking nllb", time.Now().Format(time.RFC3339))
+	line = []string{"check-nllb.sh"}
+	if stdout, stderr, err := tc.RunCommandOnNode(2, line); err != nil {
+		t.Fatalf("fail to check nllb: %v: %s: %s", err, stdout, stderr)
 	}
 
 	t.Logf("%s: checking installation state after upgrade", time.Now().Format(time.RFC3339))
 	line = []string{"check-postupgrade-state.sh", k8sVersion(), ecUpgradeTargetVersion()}
-	if _, _, err := tc.RunCommandOnNode(0, line); err != nil {
-		t.Fatalf("fail to check postupgrade state: %v", err)
+	if stdout, stderr, err := tc.RunCommandOnNode(2, line); err != nil {
+		t.Fatalf("fail to check postupgrade state: %v: %s: %s", err, stdout, stderr)
 	}
 
 	t.Logf("%s: test complete", time.Now().Format(time.RFC3339))

e2e/reset_test.go

@@ -109,7 +109,7 @@ func TestMultiNodeReset(t *testing.T) {
 
 	stdout, stderr, err = tc.RunCommandOnNode(0, []string{"check-nodes-removed.sh", "2"})
 	if err != nil {
-		t.Fatalf("fail to remove worker node 0: %v: %s: %s", err, stdout, stderr)
+		t.Fatalf("fail to check nodes removed: %v: %s: %s", err, stdout, stderr)
 	}
 
 	t.Logf("%s: checking installation state", time.Now().Format(time.RFC3339))

e2e/scripts/check-nllb.sh

@@ -0,0 +1,40 @@
+#!/bin/bash
+
+# This script is meant to be run after deleting the first control plane node to ensure the cluster
+# is still functional.
+
+set -euxo pipefail
+
+DIR=/usr/local/bin
+. $DIR/common.sh
+
+function main() {
+    local worker_node=
+    worker_node=$(kubectl get nodes -l node-role.kubernetes.io/control-plane!=true -oname | awk -F/ 'NR==1{print $2}')
+
+    if [ -z "$worker_node" ]; then
+        echo "No worker node found"
+        exit 1
+    fi
+
+    local kotsadm_image=
+    kotsadm_image=$(kubectl -n kotsadm get deploy kotsadm  -o jsonpath='{.spec.template.spec.containers[0].image}')
+
+    if [ -z "$kotsadm_image" ]; then
+        echo "No kotsadm image found"
+        exit 1
+    fi
+
+    # run the pod on a worker node
+    kubectl run test-nllb --image "$kotsadm_image" \
+        --overrides='{"spec": { "nodeSelector": {"kubernetes.io/hostname": "'"$worker_node"'"}}}' --command -- sleep infinity
+
+    # wait for the pod to be running
+    if ! kubectl wait --for=condition=ready pod/test-nllb --timeout=1m; then
+        echo "Pod test-nllb did not become ready"
+        kubectl describe pod test-nllb
+        exit 1
+    fi
+}
+
+main "$@"

e2e/scripts/restore-installation-airgap.exp

@@ -92,7 +92,7 @@ expect {
 }
 
 expect {
-    -timeout 210 "Velero is ready!" {}
+    -timeout 300 "Velero is ready!" {}
     timeout {
       puts "\n\nFailed to wait for Velero to be ready."
       exit 1