Merge pull request #7 from weaveworks/wait-for-controlplane

Wait for controlplane to be ready before creating the jobs in the origin cluster.

Author: bigkevmcd

api/v1alpha1/clusterbootstrapconfig_types.go

@@ -17,10 +17,14 @@ limitations under the License.
 package v1alpha1
 
 import (
+	"time"
+
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+const defaultWaitDuration = time.Second * 60
+
 const BootstrappedAnnotation = "capi.weave.works/bootstrapped"
 
 // JobTemplate describes a job to create
@@ -40,7 +44,16 @@ type JobTemplate struct {
 // ClusterBootstrapConfigSpec defines the desired state of ClusterBootstrapConfig
 type ClusterBootstrapConfigSpec struct {
 	ClusterSelector metav1.LabelSelector `json:"clusterSelector"`
-	Template        JobTemplate          `json:"jobTemplate,omitempty"`
+	Template        JobTemplate          `json:"jobTemplate"`
+
+	// Wait for the remote cluster to be "ready" before creating the jobs.
+	// Defaults to false.
+	//+kubebuilder:default:false
+	RequireClusterReady bool `json:"requireClusterReady"`
+	// When checking for readiness, this is the time to wait before
+	// checking again.
+	//+kubebuilder:default:60s
+	ClusterReadinessBackoff *metav1.Duration `json:"clusterReadinessBackoff,omitempty"`
 }
 
 // ClusterBootstrapConfigStatus defines the observed state of ClusterBootstrapConfig
@@ -59,6 +72,15 @@ type ClusterBootstrapConfig struct {
 	Status ClusterBootstrapConfigStatus `json:"status,omitempty"`
 }
 
+// ClusterReadinessRequeue returns the configured ClusterReadinessBackoff or a default
+// value if not configured.
+func (c ClusterBootstrapConfig) ClusterReadinessRequeue() time.Duration {
+	if v := c.Spec.ClusterReadinessBackoff; v != nil {
+		return v.Duration
+	}
+	return defaultWaitDuration
+}
+
 //+kubebuilder:object:root=true
 
 // ClusterBootstrapConfigList contains a list of ClusterBootstrapConfig

api/v1alpha1/clusterbootstrapconfig_types_test.go

@@ -0,0 +1,38 @@
+/*
+Copyright 2022.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1alpha1
+
+import (
+	"testing"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestControl(t *testing.T) {
+	cfg := ClusterBootstrapConfig{}
+
+	if v := cfg.ClusterReadinessRequeue(); v != defaultWaitDuration {
+		t.Fatalf("ClusterReadinessRequeue() got %v, want %v", v, defaultWaitDuration)
+	}
+
+	want := time.Second * 20
+	cfg.Spec.ClusterReadinessBackoff = &metav1.Duration{Duration: want}
+	if v := cfg.ClusterReadinessRequeue(); v != want {
+		t.Fatalf("ClusterReadinessRequeue() got %v, want %v", v, want)
+	}
+}

api/v1alpha1/zz_generated.deepcopy.go

@@ -37,6 +37,10 @@ spec:
           spec:
             description: ClusterBootstrapConfigSpec defines the desired state of ClusterBootstrapConfig
             properties:
+              clusterReadinessBackoff:
+                description: When checking for readiness, this is the time to wait
+                  before checking again.
+                type: string
               clusterSelector:
                 description: A label selector is a label query over a set of resources.
                   The result of matchLabels and matchExpressions are ANDed. An empty
@@ -6889,8 +6893,14 @@ spec:
                 - generateName
                 - spec
                 type: object
+              requireClusterReady:
+                description: Wait for the remote cluster to be "ready" before creating
+                  the jobs. Defaults to false.
+                type: boolean
             required:
             - clusterSelector
+            - jobTemplate
+            - requireClusterReady
             type: object
           status:
             description: ClusterBootstrapConfigStatus defines the observed state of

config/crd/bases/capi.weave.works_clusterbootstrapconfigs.yaml

@@ -1,5 +1,5 @@
 # Adds namespace to all resources.
-namespace: cluster-bootstrap-controller-system
+namespace: flux-system
 
 # Value of this field is prepended to the
 # names of all resources, e.g. a deployment named

config/default/kustomization.yaml

@@ -6,6 +6,14 @@ metadata:
   creationTimestamp: null
   name: manager-role
 rules:
+- apiGroups:
+  - ""
+  resources:
+  - secrets
+  verbs:
+  - get
+  - list
+  - watch
 - apiGroups:
   - batch
   resources:

config/rbac/role.yaml

@@ -11,6 +11,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	ptrutils "k8s.io/utils/pointer"
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -181,12 +182,18 @@ func makeTestClusterBootstrapConfig(opts ...func(*capiv1alpha1.ClusterBootstrapC
 }
 
 func makeTestClient(t *testing.T, objs ...runtime.Object) client.Client {
+	_, client := makeTestClientAndScheme(t, objs...)
+	return client
+}
+
+func makeTestClientAndScheme(t *testing.T, objs ...runtime.Object) (*runtime.Scheme, client.Client) {
 	t.Helper()
 	s := runtime.NewScheme()
+	test.AssertNoError(t, clientgoscheme.AddToScheme(s))
 	test.AssertNoError(t, capiv1alpha1.AddToScheme(s))
 	test.AssertNoError(t, clusterv1.AddToScheme(s))
 	test.AssertNoError(t, batchv1.AddToScheme(s))
-	return fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(objs...).Build()
+	return s, fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(objs...).Build()
 }
 
 func makeTestVolume(name, secretName string) corev1.Volume {

controllers/bootstrap_test.go

@@ -0,0 +1,70 @@
+package controllers
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/go-logr/logr"
+	corev1 "k8s.io/api/core/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+)
+
+const (
+	deprecatedControlPlaneLabel = "node-role.kubernetes.io/master"
+	controlPlaneLabel           = "node-role.kubernetes.io/control-plane"
+)
+
+// IsControlPlaneReady takes a client connected to a cluster and reports whether or
+// not the control-plane for the cluster is "ready".
+func IsControlPlaneReady(ctx context.Context, cl client.Client) (bool, error) {
+	logger := log.FromContext(ctx)
+	readiness := []bool{}
+	readyNodes, err := listReadyNodesWithLabel(ctx, logger, cl, controlPlaneLabel)
+	if err != nil {
+		return false, err
+	}
+	readiness = append(readiness, readyNodes...)
+
+	if len(readyNodes) == 0 {
+		readyNodes, err := listReadyNodesWithLabel(ctx, logger, cl, deprecatedControlPlaneLabel)
+		if err != nil {
+			return false, err
+		}
+		readiness = append(readiness, readyNodes...)
+	}
+
+	isReady := func(bools []bool) bool {
+		for _, v := range bools {
+			if !v {
+				return false
+			}
+		}
+		return true
+	}
+	logger.Info("readiness", "len", len(readiness), "is-ready", isReady(readiness))
+
+	// If we have no statuses, then we really don't know if we're ready or not.
+	return (len(readiness) > 0 && isReady(readiness)), nil
+}
+
+func listReadyNodesWithLabel(ctx context.Context, logger logr.Logger, cl client.Client, label string) ([]bool, error) {
+	nodes := &corev1.NodeList{}
+	// https://github.com/kubernetes/enhancements/blob/master/keps/sig-cluster-lifecycle/kubeadm/2067-rename-master-label-taint/README.md#design-details
+	err := cl.List(ctx, nodes, client.HasLabels([]string{label}))
+	if err != nil {
+		return nil, fmt.Errorf("failed to query cluster node list: %w", err)
+	}
+	logger.Info("listed nodes with control plane label", "label", label, "count", len(nodes.Items))
+
+	readiness := []bool{}
+	for _, node := range nodes.Items {
+		for _, c := range node.Status.Conditions {
+			switch c.Type {
+			case corev1.NodeReady:
+				readiness = append(readiness, c.Status == corev1.ConditionTrue)
+			}
+		}
+	}
+	return readiness, nil
+}

controllers/cluster.go

@@ -0,0 +1,97 @@
+package controllers
+
+import (
+	"context"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+)
+
+func TestIsControlPlaneReady(t *testing.T) {
+	controlPlaneLabels := map[string]string{
+		"node-role.kubernetes.io/master":        "",
+		"node-role.kubernetes.io/control-plane": "",
+		"beta.kubernetes.io/arch":               "amd64",
+		"beta.kubernetes.io/os":                 "linux",
+		"kubernetes.io/arch":                    "amd64",
+		"kubernetes.io/hostname":                "kind-control-plane",
+		"kubernetes.io/os":                      "linux",
+	}
+
+	nodeTests := []struct {
+		name       string
+		labels     map[string]string
+		conditions []corev1.NodeCondition
+		wantReady  bool
+	}{
+		{
+			name:   "control plane not ready",
+			labels: controlPlaneLabels,
+			conditions: makeConditions(
+				corev1.NodeCondition{Type: corev1.NodeReady, Status: corev1.ConditionFalse, LastHeartbeatTime: metav1.Now(), LastTransitionTime: metav1.Now(), Reason: "KubeletNotReady", Message: "container runtime network not ready: NetworkReady=false reason:NetworkPluginNotReady message:Network plugin returns error: cni plugin not initialized"},
+			),
+		},
+		{
+			name:   "control plane ready",
+			labels: controlPlaneLabels,
+			conditions: makeConditions(
+				corev1.NodeCondition{Type: "NetworkUnavailable", Status: "False", LastHeartbeatTime: metav1.Now(), LastTransitionTime: metav1.Now(), Reason: "CalicoIsUp", Message: "Calico is running on this node"},
+				corev1.NodeCondition{Type: "Ready", Status: "True", LastHeartbeatTime: metav1.Now(), LastTransitionTime: metav1.Now(), Reason: "KubeletReady", Message: "kubelet is posting ready status"},
+			),
+			wantReady: true,
+		},
+		{
+			name:   "no control plane",
+			labels: map[string]string{},
+			conditions: makeConditions(
+				corev1.NodeCondition{Type: corev1.NodeReady, Status: corev1.ConditionFalse, LastHeartbeatTime: metav1.Now(), LastTransitionTime: metav1.Now(), Reason: "KubeletNotReady", Message: "container runtime network not ready: NetworkReady=false reason:NetworkPluginNotReady message:Network plugin returns error: cni plugin not initialized"},
+			),
+		},
+	}
+	for _, tt := range nodeTests {
+		t.Run(tt.name, func(t *testing.T) {
+			cl := makeClient(makeNode(tt.labels, tt.conditions...))
+
+			ready, err := IsControlPlaneReady(context.TODO(), cl)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			if ready != tt.wantReady {
+				t.Fatalf("IsControlPlaneReady() got %v, want %v", ready, tt.wantReady)
+			}
+		})
+	}
+}
+
+func makeNode(labels map[string]string, conds ...corev1.NodeCondition) *corev1.Node {
+	return &corev1.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   "test-control-plane",
+			Labels: labels,
+			UID:    types.UID("f046e20e-df55-40d0-ab3a-76ff56617575"),
+		},
+		Spec: corev1.NodeSpec{},
+		Status: corev1.NodeStatus{
+			Conditions: conds,
+		},
+	}
+}
+
+func makeClient(objs ...runtime.Object) client.Client {
+	return fake.NewClientBuilder().WithRuntimeObjects(objs...).Build()
+}
+
+func makeConditions(conds ...corev1.NodeCondition) []corev1.NodeCondition {
+	base := []corev1.NodeCondition{
+		corev1.NodeCondition{Type: corev1.NodeMemoryPressure, Status: corev1.ConditionFalse, LastHeartbeatTime: metav1.Now(), LastTransitionTime: metav1.Now(), Reason: "KubeletHasSufficientMemory", Message: "kubelet has sufficient memory available"},
+		corev1.NodeCondition{Type: corev1.NodeDiskPressure, Status: corev1.ConditionFalse, LastHeartbeatTime: metav1.Now(), LastTransitionTime: metav1.Now(), Reason: "KubeletHasNoDiskPressure", Message: "kubelet has no disk pressure"},
+		corev1.NodeCondition{Type: corev1.NodePIDPressure, Status: corev1.ConditionFalse, LastHeartbeatTime: metav1.Now(), LastTransitionTime: metav1.Now(), Reason: "KubeletHasSufficientPID", Message: "kubelet has sufficient PID available"},
+	}
+	return append(conds, base...)
+}