Detect offline clusters

.github/scripts/deploy-fleet.sh

@@ -63,6 +63,10 @@ eventually helm upgrade --install fleet charts/fleet \
   --set agentImage.tag="$agentTag" \
   --set agentImage.imagePullPolicy=IfNotPresent \
   --set bootstrap.agentNamespace=cattle-fleet-local-system \
+  --set agentCheckinInterval=5s \
+  --set clusterMonitor.enabled=true \
+  --set clusterMonitor.threshold=20s \
+  --set clusterMonitor.interval=10s \
   --set apiServerCA="$ca" \
   --set apiServerURL="$server" \
   $shards_settings \

.github/workflows/e2e-multicluster-ci.yml

@@ -82,7 +82,7 @@ jobs:
             -p "443:443@agent:0:direct" \
             --api-port 6443 \
             --agents 1 \
-            --network "nw01"
+            --network "fleet"
       -
         name: Provision k3d Downstream Cluster for agent-initiated registration
         run: |
@@ -91,7 +91,7 @@ jobs:
             -p "444:443@agent:0:direct" \
             --api-port 6644 \
             --agents 1 \
-            --network "nw01"
+            --network "fleet"
       -
         name: Provision k3d Downstream Cluster for manager-initiated registration
         run: |
@@ -100,7 +100,7 @@ jobs:
             -p "445:443@agent:0:direct" \
             --api-port 6645 \
             --agents 1 \
-            --network "nw01"
+            --network "fleet"
       -
         name: Import Images Into k3d
         run: |
@@ -213,6 +213,7 @@ jobs:
         env:
           FLEET_E2E_NS: fleet-local
           FLEET_E2E_NS_DOWNSTREAM: fleet-default
+          FLEET_E2E_CLUSTER_DOWNSTREAM: k3d-downstream
         run: |
           # Force use of non-managed downstream cluster for portability
           export CI_REGISTERED_CLUSTER=$(kubectl get clusters.fleet.cattle.io -n $FLEET_E2E_NS_DOWNSTREAM -o jsonpath='{range .items[*]}{.metadata.name}{"\n"}' | grep -v second)

charts/fleet/templates/configmap.yaml

@@ -16,6 +16,11 @@ data:
             "bundledeployment": "{{.Values.agent.reconciler.workers.bundledeployment}}",
             "drift": "{{.Values.agent.reconciler.workers.drift}}"
       },
+      "clusterMonitor": {
+            "enabled": {{.Values.clusterMonitor.enabled}},
+            "interval": "{{.Values.clusterMonitor.interval}}",
+            "threshold": "{{.Values.clusterMonitor.threshold}}"
+      },
       {{ if .Values.garbageCollectionInterval }}
       "garbageCollectionInterval": "{{.Values.garbageCollectionInterval}}",
       {{ end }}

charts/fleet/values.yaml

@@ -70,6 +70,17 @@ bootstrap:
   branch: master
   paths: ""
 
+## Cluster monitor for offline cluster detection
+clusterMonitor:
+  enabled: false
+  # Determines how long must have elapsed since a downstream cluster's Fleet agent last reported its status to the
+  # management cluster, before that downstream cluster is considered offline.
+  # If this configured value is shorter than three times the agent check-in interval, then that check-in
+  # interval-based value will be used instead to prevent false positives.
+  threshold: "45m"
+  # Determines how often the cluster monitor will check for offline downstream clusters.
+  interval: "10m"
+
 global:
   cattle:
     systemDefaultRegistry: ""

dev/setup-fleet

@@ -44,6 +44,10 @@ helm -n cattle-fleet-system upgrade --install --create-namespace --wait --reset-
   --set agent.leaderElection.leaseDuration=10s \
   --set agent.leaderElection.retryPeriod=1s \
   --set agent.leaderElection.renewDeadline=5s \
+  --set agentCheckinInterval=5s \
+  --set clusterMonitor.enabled=true \
+  --set clusterMonitor.threshold=20s \
+  --set clusterMonitor.interval=10s \
   --set garbageCollectionInterval=1s \
   --set insecureSkipHostKeyChecks=false \
   --set imagescan.enabled=true \

e2e/multi-cluster/installation/agent_test.go

@@ -36,8 +36,9 @@ var _ = Describe("Fleet installation with TLS agent modes", func() {
 			"cattle-fleet-system",
 			"--type=merge",
 			"-p",
+			// Agent check-in interval cannot be 0. Any other value will do here.
 			fmt.Sprintf(
-				`{"data":{"config":"{\"apiServerURL\": \"https://google.com\", \"apiServerCA\": \"\", \"agentTLSMode\": \"%s\"}"}}`,
+				`{"data":{"config":"{\"apiServerURL\": \"https://google.com\", \"apiServerCA\": \"\", \"agentTLSMode\": \"%s\", \"agentCheckinInterval\": \"1m\"}"}}`,
 				agentMode,
 			),
 		)

e2e/multi-cluster/offline_cluster_test.go

@@ -0,0 +1,167 @@
+package multicluster_test
+
+import (
+	"encoding/json"
+	"os/exec"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	corev1 "k8s.io/api/core/v1"
+
+	"github.com/rancher/fleet/e2e/testenv"
+	"github.com/rancher/fleet/e2e/testenv/kubectl"
+	"github.com/rancher/wrangler/v3/pkg/genericcondition"
+)
+
+// This test uses two clusters to demonstrate offline cluster handling.
+var _ = Describe("Offline cluster detection", func() {
+	var (
+		k     kubectl.Command
+		asset string
+		name  string
+	)
+
+	BeforeEach(func() {
+		k = env.Kubectl.Context(env.Upstream)
+	})
+
+	Context("cluster with deployed workload becomes offline", func() {
+		BeforeEach(func() {
+			asset = "multi-cluster/helmop.yaml"
+			name = "test-offline-cluster"
+
+			err := testenv.ApplyTemplate(k.Namespace(env.ClusterRegistrationNamespace), testenv.AssetPath(asset), struct {
+				Name                  string
+				Namespace             string
+				Repo                  string
+				Chart                 string
+				Version               string
+				PollingInterval       time.Duration
+				HelmSecretName        string
+				InsecureSkipTLSVerify bool
+			}{
+				name,
+				env.ClusterRegistrationNamespace,
+				"",
+				"https://github.com/rancher/fleet/raw/refs/heads/main/integrationtests/cli/assets/helmrepository/config-chart-0.1.0.tgz",
+				"",
+				0,
+				"",
+				false,
+			})
+			Expect(err).ToNot(HaveOccurred())
+
+			DeferCleanup(func() {
+				connectCmd := exec.Command("docker", "network", "connect", "fleet", "k3d-"+k3dDownstreamCluster+"-server-0")
+				_, _ = connectCmd.CombinedOutput() // will fail if the node is already connected.
+
+				_, _ = k.Namespace(env.ClusterRegistrationNamespace).Delete("helmop", name)
+			})
+		})
+		It("marks any offline cluster as such, along with its bundle deployments", func() {
+			By("checking the initial online state of the cluster")
+			// Cluster should be ready
+			Eventually(func(g Gomega) {
+				out, err := k.Get(
+					"cluster", "-n", env.ClusterRegistrationNamespace,
+					"-o", `jsonpath={.items[0].status.conditions}`,
+				)
+				g.Expect(err).ToNot(HaveOccurred(), out)
+
+				checkReadyCondition(g, out, "", "True")
+			}).To(Succeed())
+
+			// Bundle deployment should be ready
+			Eventually(func(g Gomega) {
+				out, err := k.Get(
+					"bundledeployments", "-A",
+					"-l", "fleet.cattle.io/bundle-name="+name,
+					"-l", "fleet.cattle.io/bundle-namespace="+env.ClusterRegistrationNamespace,
+					"-o", `jsonpath={.items[0].status.conditions}`,
+				)
+				g.Expect(err).ToNot(HaveOccurred(), out)
+
+				checkReadyCondition(g, out, "", "True")
+			}).To(Succeed())
+
+			By("taking the cluster offline")
+			disconnectCmd := exec.Command("docker", "network", "disconnect", "fleet", "k3d-"+k3dDownstreamCluster+"-server-0")
+			out, err := disconnectCmd.CombinedOutput()
+			Expect(err).ToNot(HaveOccurred(), string(out))
+
+			By("checking that the bundle deployment and the cluster appear offline")
+			// Cluster should be offline
+			Eventually(func(g Gomega) {
+				out, err := k.Get(
+					"cluster", "-n", env.ClusterRegistrationNamespace,
+					"-o", `jsonpath={.items[0].status.conditions}`,
+				)
+				g.Expect(err).ToNot(HaveOccurred(), out)
+
+				checkReadyCondition(g, out, "cluster is offline", "Unknown")
+			}).To(Succeed())
+
+			// Bundle deployment should be offline
+			Eventually(func(g Gomega) {
+				out, err := k.Get(
+					"bundledeployments", "-A",
+					"-l", "fleet.cattle.io/bundle-name="+name,
+					"-l", "fleet.cattle.io/bundle-namespace="+env.ClusterRegistrationNamespace,
+					"-o", `jsonpath={.items[0].status.conditions}`,
+				)
+				g.Expect(err).ToNot(HaveOccurred(), out)
+
+				checkReadyCondition(g, out, "cluster is offline", "Unknown")
+			}).To(Succeed())
+
+			By("taking the cluster back online")
+			connectCmd := exec.Command("docker", "network", "connect", "fleet", "k3d-"+k3dDownstreamCluster+"-server-0")
+			out, err = connectCmd.CombinedOutput()
+			Expect(err).ToNot(HaveOccurred(), string(out))
+
+			By("checking the new online state of the cluster")
+			// Cluster should be ready again
+			Eventually(func(g Gomega) {
+				out, err := k.Get(
+					"cluster", "-n", env.ClusterRegistrationNamespace,
+					"-o", `jsonpath={.items[0].status.conditions}`,
+				)
+				g.Expect(err).ToNot(HaveOccurred(), out)
+
+				checkReadyCondition(g, out, "", "True")
+			}).To(Succeed())
+
+			// Bundle deployment should be ready again
+			Eventually(func(g Gomega) {
+				out, err := k.Get(
+					"bundledeployments", "-A",
+					"-l", "fleet.cattle.io/bundle-name="+name,
+					"-l", "fleet.cattle.io/bundle-namespace="+env.ClusterRegistrationNamespace,
+					"-o", `jsonpath={.items[0].status.conditions}`,
+				)
+				g.Expect(err).ToNot(HaveOccurred(), out)
+
+				checkReadyCondition(g, out, "", "True")
+			}).To(Succeed())
+		})
+	})
+})
+
+func checkReadyCondition(g Gomega, out, msg, status string) {
+	conds := []genericcondition.GenericCondition{}
+	err := json.Unmarshal([]byte(out), &conds)
+	g.Expect(err).ToNot(HaveOccurred())
+
+	var readyCond *genericcondition.GenericCondition
+	for i, c := range conds {
+		if c.Type == "Ready" {
+			readyCond = &conds[i]
+		}
+	}
+
+	g.Expect(readyCond).NotTo(BeNil())
+	g.Expect(readyCond.Message).To(ContainSubstring(msg))
+	g.Expect(readyCond.Status).To(Equal(corev1.ConditionStatus(status)))
+}

e2e/multi-cluster/suite_test.go

@@ -3,6 +3,7 @@ package multicluster_test
 
 import (
 	"os"
+	"strings"
 	"testing"
 
 	"github.com/rancher/fleet/e2e/testenv"
@@ -17,8 +18,9 @@ func TestE2E(t *testing.T) {
 }
 
 var (
-	env       *testenv.Env
-	dsCluster = "second"
+	env                  *testenv.Env
+	dsCluster            = "second" // name of the Fleet Cluster resource
+	k3dDownstreamCluster = "downstream"
 )
 
 var _ = BeforeSuite(func() {
@@ -30,4 +32,8 @@ var _ = BeforeSuite(func() {
 	if dsClusterEnvVar := os.Getenv("CI_REGISTERED_CLUSTER"); dsClusterEnvVar != "" {
 		dsCluster = dsClusterEnvVar
 	}
+
+	if k3dDSClusterEnvVar := os.Getenv("FLEET_E2E_CLUSTER_DOWNSTREAM"); k3dDSClusterEnvVar != "" {
+		k3dDownstreamCluster = strings.TrimPrefix(k3dDSClusterEnvVar, "k3d-")
+	}
 })