Add coordinator fault tolerance check (#1827)

* Add coordinator fault tolerance check

* Fix mock client for DNS setup

Author: johscheuer

api/v1beta2/foundationdb_process_address.go

@@ -256,7 +256,6 @@ func (address ProcessAddress) String() string {
 	sb.WriteString(net.JoinHostPort(address.MachineAddress(), strconv.Itoa(address.Port)))
 
 	flags := address.SortedFlags()
-
 	if len(flags) > 0 {
 		sb.WriteString(":" + strings.Join(flags, ":"))
 	}

controllers/cluster_controller_test.go

@@ -331,8 +331,7 @@ var _ = Describe("cluster_controller", func() {
 
 			It("should update the pods", func() {
 				pods := &corev1.PodList{}
-				err = k8sClient.List(context.TODO(), pods, getListOptions(cluster)...)
-				Expect(err).NotTo(HaveOccurred())
+				Expect(k8sClient.List(context.TODO(), pods, getListOptions(cluster)...)).NotTo(HaveOccurred())
 				for _, pod := range pods.Items {
 					env := make(map[string]string)
 					for _, envVar := range pod.Spec.InitContainers[0].Env {
@@ -342,6 +341,7 @@ var _ = Describe("cluster_controller", func() {
 				}
 			})
 		})
+
 		Context("when buggifying a pod to make it crash loop", func() {
 			BeforeEach(func() {
 				cluster.Spec.Buggify.CrashLoop = []fdbv1beta2.ProcessGroupID{"storage-1"}

controllers/remove_process_groups_test.go

@@ -43,8 +43,7 @@ var _ = Describe("remove_process_groups", func() {
 	Context("validating process removal", func() {
 		BeforeEach(func() {
 			cluster = internal.CreateDefaultCluster()
-			err := k8sClient.Create(context.TODO(), cluster)
-			Expect(err).NotTo(HaveOccurred())
+			Expect(k8sClient.Create(context.TODO(), cluster)).NotTo(HaveOccurred())
 
 			result, err := reconcileCluster(cluster)
 			Expect(err).NotTo(HaveOccurred())
@@ -193,15 +192,15 @@ var _ = Describe("remove_process_groups", func() {
 				})
 			})
 
-			When("Removing multiple process groups", func() {
+			When("removing multiple process groups", func() {
 				var initialCnt int
 				var secondRemovedProcessGroup *fdbv1beta2.ProcessGroupStatus
 
 				When("the removal mode is the default zone", func() {
 					BeforeEach(func() {
 						Expect(cluster.Spec.AutomationOptions.RemovalMode).To(BeEmpty())
 						initialCnt = len(cluster.Status.ProcessGroups)
-						secondRemovedProcessGroup = cluster.Status.ProcessGroups[1]
+						secondRemovedProcessGroup = cluster.Status.ProcessGroups[6]
 						marked, processGroup := fdbv1beta2.MarkProcessGroupForRemoval(cluster.Status.ProcessGroups, secondRemovedProcessGroup.ProcessGroupID, secondRemovedProcessGroup.ProcessClass, removedProcessGroup.Addresses[0])
 						Expect(marked).To(BeTrue())
 						Expect(processGroup).To(BeNil())
@@ -303,7 +302,7 @@ var _ = Describe("remove_process_groups", func() {
 						Expect(err).NotTo(HaveOccurred())
 
 						initialCnt = len(cluster.Status.ProcessGroups)
-						secondRemovedProcessGroup = cluster.Status.ProcessGroups[1]
+						secondRemovedProcessGroup = cluster.Status.ProcessGroups[6]
 						marked, processGroup := fdbv1beta2.MarkProcessGroupForRemoval(cluster.Status.ProcessGroups, secondRemovedProcessGroup.ProcessGroupID, secondRemovedProcessGroup.ProcessClass, removedProcessGroup.Addresses[0])
 						Expect(marked).To(BeTrue())
 						Expect(processGroup).To(BeNil())

pkg/fdbadminclient/mock/admin_client_mock.go

@@ -228,6 +228,13 @@ func (client *AdminClient) GetStatus() (*fdbv1beta2.FoundationDBStatus, error) {
 					for _, envVar := range container.Env {
 						if envVar.Name == "FDB_DNS_NAME" {
 							locality[fdbv1beta2.FDBLocalityDNSNameKey] = envVar.Value
+
+							if client.Cluster.UseDNSInClusterFile() {
+								fullAddress.StringAddress = envVar.Value
+							}
+							// TODO (johscheuer): This should be set to true. This will add additional information to the
+							// return address and needs some additional refactoring in the mock code.
+							// fullAddress.FromHostname = true
 						}
 					}
 				}
@@ -350,6 +357,7 @@ func (client *AdminClient) GetStatus() (*fdbv1beta2.FoundationDBStatus, error) {
 		status.Cluster.Clients.SupportedVersions = supportedVersions
 	}
 
+	var countReachableCoordinators int
 	for address, reachable := range coordinators {
 		pAddr, err := fdbv1beta2.ParseProcessAddress(address)
 		if err != nil {
@@ -360,8 +368,14 @@ func (client *AdminClient) GetStatus() (*fdbv1beta2.FoundationDBStatus, error) {
 			Address:   pAddr,
 			Reachable: reachable,
 		})
+
+		if reachable {
+			countReachableCoordinators++
+		}
 	}
 
+	minReachableCoordinators := (client.Cluster.DesiredCoordinatorCount() + 1) / 2
+	status.Client.Coordinators.QuorumReachable = countReachableCoordinators >= minReachableCoordinators
 	status.Client.DatabaseStatus.Available = true
 	status.Client.DatabaseStatus.Healthy = true
 

pkg/fdbstatus/status_checks.go

@@ -390,9 +390,26 @@ func DoLogServerFaultDomainCheckOnStatus(status *fdbv1beta2.FoundationDBStatus)
 }
 
 // DoCoordinatorFaultDomainCheckOnStatus does a coordinator related fault domain check over the given status object.
-// @note an empty function for now. We will revisit this later.
-func DoCoordinatorFaultDomainCheckOnStatus(_ *fdbv1beta2.FoundationDBStatus) error {
-	// TODO: decide if we need to do coordinator related check.
+func DoCoordinatorFaultDomainCheckOnStatus(status *fdbv1beta2.FoundationDBStatus) error {
+	notReachable := make([]string, 0, len(status.Client.Coordinators.Coordinators))
+	for _, coordinator := range status.Client.Coordinators.Coordinators {
+		if coordinator.Reachable {
+			continue
+		}
+
+		notReachable = append(notReachable, coordinator.Address.String())
+	}
+
+	if len(notReachable) > 0 {
+		return fmt.Errorf("not all coordinators are reachable, unreachable coordinators: %s", strings.Join(notReachable, ","))
+	}
+
+	// If this is the case the statements above should already catch the unreachable coordinators and printout a more
+	// detailed message.
+	if !status.Client.Coordinators.QuorumReachable {
+		return fmt.Errorf("quorum of coordinators is not reachable")
+	}
+
 	return nil
 }
 
@@ -430,7 +447,6 @@ func HasDesiredFaultToleranceFromStatus(log logr.Logger, status *fdbv1beta2.Foun
 		return false
 	}
 
-	// TODO (johscheuer): Should those checks be specific to the Kubernetes cluster (DC) we are requesting?
 	// Should we also add a method to check the different process classes? Currently the degraded log fault tolerance
 	// will block the removal of a storage process.
 	err := DoStorageServerFaultDomainCheckOnStatus(status)
@@ -445,5 +461,11 @@ func HasDesiredFaultToleranceFromStatus(log logr.Logger, status *fdbv1beta2.Foun
 		return false
 	}
 
+	err = DoCoordinatorFaultDomainCheckOnStatus(status)
+	if err != nil {
+		log.Info("Fault domain check for coordinator servers failed", "error", err)
+		return false
+	}
+
 	return true
 }