Merge pull request #2169 from chuckha/failure-domain

✨ implement failure domain picking

Author: k8s-ci-robot

controlplane/kubeadm/controllers/kubeadm_control_plane_controller.go

@@ -33,6 +33,7 @@ import (
 	kerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apiserver/pkg/storage/names"
 	"k8s.io/client-go/tools/record"
+	"sigs.k8s.io/cluster-api/controlplane/kubeadm/internal"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
@@ -226,6 +227,7 @@ func (r *KubeadmControlPlaneReconciler) reconcile(ctx context.Context, cluster *
 	// Once we start handling upgrade, we'll need to filter this list and act appropriately
 	numMachines := len(ownedMachines.Items)
 	desiredReplicas := int(*kcp.Spec.Replicas)
+
 	switch {
 	// We are creating the first replica
 	case numMachines < desiredReplicas && numMachines == 0:
@@ -423,7 +425,26 @@ func (r *KubeadmControlPlaneReconciler) generateKubeadmConfig(ctx context.Contex
 	return bootstrapRef, nil
 }
 
+func (r *KubeadmControlPlaneReconciler) failureDomainForScaleUp(ctx context.Context, kcp *controlplanev1.KubeadmControlPlane, cluster *clusterv1.Cluster) (*string, error) {
+	// Don't do anything if there are no failure domains defined on the cluster.
+	if len(cluster.Status.FailureDomains) == 0 {
+		return nil, nil
+	}
+	machineList, err := r.getMachines(ctx, types.NamespacedName{Namespace: cluster.GetNamespace(), Name: cluster.GetName()})
+	if err != nil {
+		return nil, err
+	}
+	machineList = r.filterOwnedMachines(kcp, machineList)
+	picker := internal.FailureDomainPicker{Log: r.Log}
+	failureDomain := picker.PickFewest(cluster.Status.FailureDomains.FilterControlPlane(), machineList.Items)
+	return &failureDomain, nil
+}
+
 func (r *KubeadmControlPlaneReconciler) generateMachine(ctx context.Context, kcp *controlplanev1.KubeadmControlPlane, cluster *clusterv1.Cluster, infraRef, bootstrapRef *corev1.ObjectReference) error {
+	fd, err := r.failureDomainForScaleUp(ctx, kcp, cluster)
+	if err != nil {
+		return err
+	}
 	machine := &clusterv1.Machine{
 		ObjectMeta: metav1.ObjectMeta{
 			Labels:    generateKubeadmControlPlaneLabels(cluster.Name),
@@ -439,6 +460,7 @@ func (r *KubeadmControlPlaneReconciler) generateMachine(ctx context.Context, kcp
 			},
 		},
 	}
+	machine.Spec.FailureDomain = fd
 
 	owner := metav1.NewControllerRef(kcp, controlplanev1.GroupVersion.WithKind("KubeadmControlPlane"))
 	if owner != nil {

controlplane/kubeadm/internal/failure_domain.go

@@ -0,0 +1,111 @@
+/*
+Copyright 2020 The Kubernetes Authors.
+
+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 internal
+
+import (
+	"sort"
+
+	clusterv1 "sigs.k8s.io/cluster-api/api/v1alpha3"
+)
+
+type failureDomainAggregation struct {
+	id    string
+	count int
+}
+type failureDomainAggregations []failureDomainAggregation
+
+// Len is the number of elements in the collection.
+func (f failureDomainAggregations) Len() int {
+	return len(f)
+}
+
+// Less reports whether the element with
+// index i should sort before the element with index j.
+func (f failureDomainAggregations) Less(i, j int) bool {
+	return f[i].count < f[j].count
+}
+
+// Swap swaps the elements with indexes i and j.
+func (f failureDomainAggregations) Swap(i, j int) {
+	f[i], f[j] = f[j], f[i]
+}
+
+// Scope down logr.Logger
+type logger interface {
+	Info(msg string, keysAndValues ...interface{})
+}
+
+// FailureDomainPicker picks a failure domain given a list of failure domains and a list of machines.
+type FailureDomainPicker struct {
+	Log logger
+}
+
+// PickMost returns the failure domain with the most number of machines.
+func (f *FailureDomainPicker) PickMost(failureDomains clusterv1.FailureDomains, machines []clusterv1.Machine) string {
+	aggregations := f.pick(failureDomains, machines)
+	if len(aggregations) == 0 {
+		return ""
+	}
+	sort.Sort(sort.Reverse(aggregations))
+	return aggregations[0].id
+
+}
+
+// PickFewest returns the failure domain with the fewest number of machines.
+func (f *FailureDomainPicker) PickFewest(failureDomains clusterv1.FailureDomains, machines []clusterv1.Machine) string {
+	aggregations := f.pick(failureDomains, machines)
+	if len(aggregations) == 0 {
+		return ""
+	}
+	sort.Sort(aggregations)
+	return aggregations[0].id
+}
+
+func (f *FailureDomainPicker) pick(failureDomains clusterv1.FailureDomains, machines []clusterv1.Machine) failureDomainAggregations {
+	if len(failureDomains) == 0 {
+		return failureDomainAggregations{}
+	}
+
+	counters := map[string]int{}
+
+	// Initialize the known failure domain keys to find out if an existing machine is in an unsupported failure domain.
+	for fd := range failureDomains {
+		counters[fd] = 0
+	}
+
+	// Count how many machines are in each failure domain.
+	for _, m := range machines {
+		if m.Spec.FailureDomain == nil {
+			continue
+		}
+		id := *m.Spec.FailureDomain
+		if _, ok := failureDomains[id]; !ok {
+			f.Log.Info("unknown failure domain", "machine-name", m.GetName(), "failure-domain-id", id, "known-failure-domains", failureDomains)
+			continue
+		}
+		counters[id]++
+	}
+
+	aggregations := make(failureDomainAggregations, 0)
+
+	// Gather up tuples of failure domains ids and counts
+	for fd, count := range counters {
+		aggregations = append(aggregations, failureDomainAggregation{id: fd, count: count})
+	}
+
+	return aggregations
+}

controlplane/kubeadm/internal/failure_domain_test.go

@@ -0,0 +1,108 @@
+/*
+Copyright 2020 The Kubernetes Authors.
+
+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 internal
+
+import (
+	"testing"
+
+	"k8s.io/klog/klogr"
+	clusterv1 "sigs.k8s.io/cluster-api/api/v1alpha3"
+)
+
+func TestNewFailureDomainPicker(t *testing.T) {
+	a := "us-west-1a"
+	b := "us-west-1b"
+
+	fds := clusterv1.FailureDomains{
+		a: clusterv1.FailureDomainSpec{},
+		b: clusterv1.FailureDomainSpec{},
+	}
+	machinea := clusterv1.Machine{Spec: clusterv1.MachineSpec{FailureDomain: &a}}
+	machineb := clusterv1.Machine{Spec: clusterv1.MachineSpec{FailureDomain: &b}}
+	machinenil := clusterv1.Machine{Spec: clusterv1.MachineSpec{FailureDomain: nil}}
+
+	testcases := []struct {
+		name     string
+		logger   logger
+		fds      clusterv1.FailureDomains
+		machines []clusterv1.Machine
+		expected []string
+	}{
+		{
+			name:     "simple",
+			expected: []string{""},
+		},
+		{
+			name: "no machines",
+			fds: clusterv1.FailureDomains{
+				a: clusterv1.FailureDomainSpec{},
+			},
+			expected: []string{a},
+		},
+		{
+			name: "one machine in a failure domain",
+			fds:  fds,
+			machines: []clusterv1.Machine{
+				machinea,
+			},
+			expected: []string{b},
+		},
+		{
+			name: "no failure domain specified on machine",
+			fds: clusterv1.FailureDomains{
+				a: clusterv1.FailureDomainSpec{},
+			},
+			machines: []clusterv1.Machine{
+				machinenil,
+			},
+			expected: []string{a, b},
+		},
+		{
+			name:   "mismatched failure domain on machine",
+			logger: klogr.New(),
+			fds: clusterv1.FailureDomains{
+				a: clusterv1.FailureDomainSpec{},
+			},
+			machines: []clusterv1.Machine{
+				machineb,
+			},
+			expected: []string{a},
+		},
+		{
+			name:     "failure domains and no machines should return a valid failure domain",
+			fds:      fds,
+			machines: []clusterv1.Machine{},
+			expected: []string{a, b},
+		},
+	}
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			picker := FailureDomainPicker{Log: tc.logger}
+			fd := picker.PickFewest(tc.fds, tc.machines)
+
+			found := false
+			for _, expectation := range tc.expected {
+				if fd == expectation {
+					found = true
+				}
+			}
+			if !found {
+				t.Fatal("did not find expected value")
+			}
+		})
+	}
+}

test/infrastructure/docker/api/v1alpha3/dockercluster_types.go

@@ -18,6 +18,7 @@ package v1alpha3
 
 import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	clusterv1 "sigs.k8s.io/cluster-api/api/v1alpha3"
 )
 
 const (
@@ -34,12 +35,23 @@ type DockerClusterSpec struct {
 	// ControlPlaneEndpoint represents the endpoint used to communicate with the control plane.
 	// +optional
 	ControlPlaneEndpoint APIEndpoint `json:"controlPlaneEndpoint"`
+
+	// FailureDomains are not usulaly defined on the spec.
+	// The docker provider is special since failure domains don't mean anything in a local docker environment.
+	// Instead, the docker cluster controller will simply copy these into the Status and allow the Cluster API
+	// controllers to do what they will with the defined failure domains.
+	// +optional
+	FailureDomains clusterv1.FailureDomains `json:"failureDomains,omitempty"`
 }
 
 // DockerClusterStatus defines the observed state of DockerCluster.
 type DockerClusterStatus struct {
 	// Ready denotes that the docker cluster (infrastructure) is ready.
 	Ready bool `json:"ready"`
+
+	// FailureDomains don't mean much in CAPD since it's all local, but we can see how the rest of cluster API
+	// will use this if we populate it.
+	FailureDomains clusterv1.FailureDomains `json:"failureDomains,omitempty"`
 }
 
 // APIEndpoint represents a reachable Kubernetes API endpoint.