From 9e4b26093f853ffd2d52867c8dad93f13ce3066c Mon Sep 17 00:00:00 2001 From: "Johannes M. Scheuermann" Date: Fri, 21 Aug 2026 09:36:16 +0100 Subject: [PATCH 1/4] Implement single cluster recovery command --- docs/manual/operations.md | 4 +- e2e/fixtures/fdb_cluster.go | 68 ++- e2e/fixtures/ha_fdb_cluster.go | 2 +- .../operator_backup_test.go | 2 +- .../operator_creation_velocity_test.go | 2 +- .../operator_ha_failure_test.go | 4 +- .../operator_migate_image_type_test.go | 6 +- .../operator_plugin_test.go | 177 ++++++- .../operator_stress_test.go | 4 +- kubectl-fdb/cmd/recover.go | 383 +++++++++++++++ .../cmd/recover_multi_region_cluster.go | 317 +----------- kubectl-fdb/cmd/recover_single_dc_cluster.go | 461 ++++++++++++++++++ kubectl-fdb/cmd/root.go | 2 +- 13 files changed, 1085 insertions(+), 347 deletions(-) create mode 100644 kubectl-fdb/cmd/recover.go create mode 100644 kubectl-fdb/cmd/recover_single_dc_cluster.go diff --git a/docs/manual/operations.md b/docs/manual/operations.md index 655446aea..c25acc6a7 100644 --- a/docs/manual/operations.md +++ b/docs/manual/operations.md @@ -272,8 +272,8 @@ If the cluster is a multi-region cluster, perform this step for all running regi - Now you can set `spec.Skip = false` to let the operator take over again. - Depending on the state of the multi-region cluster, you probably want to change the desired database configuration to drop ha. -The [kubectl-fdb plugin](../../kubectl-fdb/Readme.md) provides a `recover-multi-region-cluster` command that can be used to automatically recover a cluster with the above steps. -The command has some additional safety checks, to ensure the steps are only performed on a cluster that is unhealthy and the majority of coordinators are unreachable. +The [kubectl-fdb plugin](../../kubectl-fdb/Readme.md) provides `recover multi-region` and `recover single-dc` commands that can be used to automatically recover a cluster with the above steps. +The commands have some additional safety checks, to ensure the steps are only performed on a cluster that is unhealthy and the majority of coordinators are unreachable. ## Next diff --git a/e2e/fixtures/fdb_cluster.go b/e2e/fixtures/fdb_cluster.go index 073463a43..635b92c77 100644 --- a/e2e/fixtures/fdb_cluster.go +++ b/e2e/fixtures/fdb_cluster.go @@ -42,7 +42,7 @@ import ( "github.com/onsi/gomega" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/equality" - kubeErrors "k8s.io/apimachinery/pkg/api/errors" + k8serrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/resource" "k8s.io/apimachinery/pkg/util/wait" "sigs.k8s.io/controller-runtime/pkg/client" @@ -829,6 +829,58 @@ func (fdbCluster *FdbCluster) SetPodAsUnschedulable(ctx context.Context, pod cor }).WithTimeout(5*time.Minute).WithPolling(2*time.Second).MustPassRepeatedly(5).Should(gomega.BeEmpty(), "Not able to set pod as unschedulable") } +// SetPodsAsUnschedulable sets the provided slice of Pods on the NoSchedule list of the current FoundationDBCluster. This will make +// sure that the Pods are stuck in Pending. +func (fdbCluster *FdbCluster) SetPodsAsUnschedulable(ctx context.Context, pods []corev1.Pod) { + unschedulableProcessGroups := make([]fdbv1beta2.ProcessGroupID, 0, len(pods)) + + for _, pod := range pods { + unschedulableProcessGroups = append(unschedulableProcessGroups, GetProcessGroupID(pod)) + } + + fdbCluster.SetProcessGroupsAsUnschedulable( + ctx, + unschedulableProcessGroups, + ) + + // context aware. + time.Sleep(5 * time.Second) + + for _, pod := range pods { + fetchedPod := &corev1.Pod{} + err := fdbCluster.getClient(). + Get(ctx, client.ObjectKeyFromObject(&pod), fetchedPod) + if err != nil { + continue + } + // check if error is absent -> case not created, just deleted + gomega.Expect(err).NotTo(gomega.HaveOccurred()) + + // Try deleting the Pod as a workaround until the operator handles all cases. + if fetchedPod.Spec.NodeName != "" && fetchedPod.DeletionTimestamp.IsZero() { + gomega.Expect(fdbCluster.getClient().Delete(ctx, &pod)). + NotTo(gomega.HaveOccurred()) + } + } + + gomega.Eventually(func(g gomega.Gomega) { + for _, pod := range pods { + fetchedPod := &corev1.Pod{} + err := fdbCluster.getClient(). + Get(ctx, client.ObjectKeyFromObject(&pod), fetchedPod) + g.Expect(err).NotTo(gomega.HaveOccurred()) + + // Try deleting the Pod as a workaround until the operator handles all cases. + if fetchedPod.Spec.NodeName != "" && fetchedPod.DeletionTimestamp.IsZero() { + g.Expect(fdbCluster.getClient().Delete(ctx, &pod)). + NotTo(gomega.HaveOccurred()) + } + + g.Expect(fetchedPod.Spec.NodeName).To(gomega.BeEmpty()) + } + }).WithTimeout(5*time.Minute).WithPolling(2*time.Second).MustPassRepeatedly(5).Should(gomega.Succeed(), "Not able to set pods as unschedulable") +} + // SetProcessGroupsAsUnschedulable sets the provided process groups on the NoSchedule list of the current FoundationDBCluster. This will make // sure that the Pod is stuck in Pending. func (fdbCluster *FdbCluster) SetProcessGroupsAsUnschedulable( @@ -980,7 +1032,7 @@ func (fdbCluster *FdbCluster) WaitForPodRemoval(ctx context.Context, pod *corev1 gomega.Eventually(func() bool { err := fdbCluster.getClient(). Get(ctx, client.ObjectKeyFromObject(pod), fetchedPod) - if err != nil && kubeErrors.IsNotFound(err) { + if err != nil && k8serrors.IsNotFound(err) { return true } @@ -1216,7 +1268,7 @@ func (fdbCluster *FdbCluster) CheckPodIsDeleted(ctx context.Context, podName str Get(ctx, client.ObjectKey{Namespace: fdbCluster.Namespace(), Name: podName}, pod) if err != nil { - if kubeErrors.IsNotFound(err) { + if k8serrors.IsNotFound(err) { return true } } @@ -1260,13 +1312,13 @@ func (fdbCluster *FdbCluster) SetUseDNSInClusterFile( return fdbCluster.WaitForReconciliation(ctx) } -// Destroy will remove the underlying cluster. -func (fdbCluster *FdbCluster) Destroy(ctx context.Context) error { - return fdbCluster.DestroyWithWaitForTearDown(ctx, false) +// Delete will remove the underlying cluster. +func (fdbCluster *FdbCluster) Delete(ctx context.Context) error { + return fdbCluster.DeleteWithWaitForTearDown(ctx, false) } -// DestroyWithWaitForTearDown will remove the underlying cluster and wait for the resources to be removed. -func (fdbCluster *FdbCluster) DestroyWithWaitForTearDown( +// DeleteWithWaitForTearDown will remove the underlying cluster and wait for the resources to be removed. +func (fdbCluster *FdbCluster) DeleteWithWaitForTearDown( ctx context.Context, waitForTearDown bool, ) error { diff --git a/e2e/fixtures/ha_fdb_cluster.go b/e2e/fixtures/ha_fdb_cluster.go index 04458317f..5347252a7 100644 --- a/e2e/fixtures/ha_fdb_cluster.go +++ b/e2e/fixtures/ha_fdb_cluster.go @@ -219,7 +219,7 @@ func (factory *Factory) createHaFdbClusterSpec( // Delete removes all Clusters associated FoundationDBClusters. func (haFDBCluster *HaFdbCluster) Delete(ctx context.Context) { for _, cluster := range haFDBCluster.GetAllClusters() { - gomega.Expect(cluster.Destroy(ctx)).NotTo(gomega.HaveOccurred()) + gomega.Expect(cluster.Delete(ctx)).NotTo(gomega.HaveOccurred()) } } diff --git a/e2e/test_operator_backups/operator_backup_test.go b/e2e/test_operator_backups/operator_backup_test.go index 7f9607ed8..0dd35fe52 100644 --- a/e2e/test_operator_backups/operator_backup_test.go +++ b/e2e/test_operator_backups/operator_backup_test.go @@ -103,7 +103,7 @@ var _ = Describe("Operator Backup", Label("e2e", "pr", "foundationdb-pr"), func( namespace := fdbCluster.Namespace() // Delete the FDB cluster to have a clean start. - Expect(fdbCluster.DestroyWithWaitForTearDown(ctx, true)).To(Succeed()) + Expect(fdbCluster.DeleteWithWaitForTearDown(ctx, true)).To(Succeed()) // Restart the operator pods. factory.RecreateOperatorPods(ctx, namespace) }) diff --git a/e2e/test_operator_creation_velocity/operator_creation_velocity_test.go b/e2e/test_operator_creation_velocity/operator_creation_velocity_test.go index 3f51bfe93..a0987384f 100644 --- a/e2e/test_operator_creation_velocity/operator_creation_velocity_test.go +++ b/e2e/test_operator_creation_velocity/operator_creation_velocity_test.go @@ -85,7 +85,7 @@ var _ = Describe("Test Operator Velocity", Label("e2e"), func() { runTime := time.Since(startTime) log.Println("Single-DC cluster creation took: ", runTime.String()) - Expect(fdbCluster.Destroy(ctx)).ToNot(HaveOccurred()) + Expect(fdbCluster.Delete(ctx)).ToNot(HaveOccurred()) }) }) diff --git a/e2e/test_operator_ha_failure/operator_ha_failure_test.go b/e2e/test_operator_ha_failure/operator_ha_failure_test.go index 93307ba07..69608de30 100644 --- a/e2e/test_operator_ha_failure/operator_ha_failure_test.go +++ b/e2e/test_operator_ha_failure/operator_ha_failure_test.go @@ -134,7 +134,7 @@ var _ = Describe("Operator HA Failure tests", Label("e2e"), func() { keyValues = primary.GenerateRandomValues(10, prefix) primary.WriteKeyValuesWithTimeout(ctx, keyValues, 120) - // Destroy primary and primary satellite (should have mutations that are not present in the remote side). + // Delete primary and primary satellite (should have mutations that are not present in the remote side). primary.SetSkipReconciliation(ctx, true) primarySatellite.SetSkipReconciliation(ctx, true) // We also destroy the remote satellite, it shouldn't matter in this case as the remote satellite @@ -199,7 +199,7 @@ var _ = Describe("Operator HA Failure tests", Label("e2e"), func() { &operatorPod, "manager", fmt.Sprintf( - "kubectl-fdb -n %s recover-multi-region-cluster --version-check=false --wait=false %s", + "kubectl-fdb -n %s recover multi-region --version-check=false --wait=false %s", remote.Namespace(), remote.Name(), ), diff --git a/e2e/test_operator_migrate_image_type/operator_migate_image_type_test.go b/e2e/test_operator_migrate_image_type/operator_migate_image_type_test.go index 5578d8fc5..fe2afbde2 100644 --- a/e2e/test_operator_migrate_image_type/operator_migate_image_type_test.go +++ b/e2e/test_operator_migrate_image_type/operator_migate_image_type_test.go @@ -82,7 +82,7 @@ var _ = PDescribe("Operator Migrate Image Type", Label("e2e"), func() { }) AfterEach(func(ctx SpecContext) { - Expect(fdbCluster.Destroy(ctx)).NotTo(HaveOccurred()) + Expect(fdbCluster.Delete(ctx)).NotTo(HaveOccurred()) }) It("should convert the cluster", func(ctx SpecContext) { @@ -136,7 +136,7 @@ var _ = PDescribe("Operator Migrate Image Type", Label("e2e"), func() { }) AfterEach(func(ctx SpecContext) { - Expect(fdbCluster.Destroy(ctx)).NotTo(HaveOccurred()) + Expect(fdbCluster.Delete(ctx)).NotTo(HaveOccurred()) }) It("should convert the cluster", func(ctx SpecContext) { @@ -184,7 +184,7 @@ var _ = PDescribe("Operator Migrate Image Type", Label("e2e"), func() { }) AfterEach(func(ctx SpecContext) { - Expect(fdbCluster.Destroy(ctx)).NotTo(HaveOccurred()) + Expect(fdbCluster.Delete(ctx)).NotTo(HaveOccurred()) }) It("should convert the cluster", func(ctx SpecContext) { diff --git a/e2e/test_operator_plugin/operator_plugin_test.go b/e2e/test_operator_plugin/operator_plugin_test.go index d235611d1..e127ad8ca 100644 --- a/e2e/test_operator_plugin/operator_plugin_test.go +++ b/e2e/test_operator_plugin/operator_plugin_test.go @@ -32,6 +32,7 @@ import ( fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/v2/api/v1beta2" corev1 "k8s.io/api/core/v1" + "k8s.io/utils/ptr" ctrlClient "sigs.k8s.io/controller-runtime/pkg/client" "github.com/FoundationDB/fdb-kubernetes-operator/v2/e2e/fixtures" @@ -42,7 +43,6 @@ import ( var ( factory *fixtures.Factory - fdbCluster *fixtures.HaFdbCluster testOptions *fixtures.FactoryOptions clusterConfig *fixtures.ClusterConfig ) @@ -51,10 +51,8 @@ func init() { testOptions = fixtures.InitFlags() } -var _ = BeforeSuite(func(ctx SpecContext) { +var _ = BeforeSuite(func() { factory = fixtures.CreateFactory(testOptions) - clusterConfig = fixtures.DefaultClusterConfigWithHaMode(fixtures.HaFourZoneSingleSat, false) - fdbCluster = factory.CreateFdbHaCluster(ctx, clusterConfig) }) var _ = AfterSuite(func(ctx SpecContext) { @@ -66,11 +64,23 @@ var _ = AfterSuite(func(ctx SpecContext) { var _ = Describe("Operator Plugin", Label("e2e", "pr"), func() { When("getting the plugin version from the operator pod", func() { + var fdbCluster *fixtures.FdbCluster + + BeforeEach(func(ctx SpecContext) { + clusterConfig = fixtures.DefaultClusterConfig(false) + fdbCluster = factory.CreateFdbCluster(ctx, clusterConfig) + }) + + AfterEach(func(ctx SpecContext) { + // Delete the cluster. + Expect(fdbCluster.Delete(ctx)).NotTo(HaveOccurred()) + }) + It("should print the version", func(ctx SpecContext) { // Pick one operator pod and execute the kubectl version command to ensure that kubectl-fdb is present // and can be executed. operatorPod := factory.RandomPickOnePod( - factory.GetOperatorPods(ctx, fdbCluster.GetPrimary().Namespace()).Items, + factory.GetOperatorPods(ctx, fdbCluster.Namespace()).Items, ) log.Println("operatorPod:", operatorPod.Name) Eventually(func(g Gomega) string { @@ -80,7 +90,7 @@ var _ = Describe("Operator Plugin", Label("e2e", "pr"), func() { "manager", fmt.Sprintf( "kubectl-fdb -n %s --version-check=false version", - fdbCluster.GetPrimary().Namespace(), + fdbCluster.Namespace(), ), false, ) @@ -92,8 +102,20 @@ var _ = Describe("Operator Plugin", Label("e2e", "pr"), func() { When("all Pods in the primary and satellites are down", func() { var useDNS bool + var fdbCluster *fixtures.HaFdbCluster + + AfterEach(func(ctx SpecContext) { + // Delete the cluster. + fdbCluster.Delete(ctx) + }) JustBeforeEach(func(ctx SpecContext) { + clusterConfig = fixtures.DefaultClusterConfigWithHaMode( + fixtures.HaFourZoneSingleSat, + false, + ) + fdbCluster = factory.CreateFdbHaCluster(ctx, clusterConfig) + var errGroup errgroup.Group // Enable DNS names in the cluster file for the whole cluster. for _, cluster := range fdbCluster.GetAllClusters() { @@ -191,7 +213,7 @@ var _ = Describe("Operator Plugin", Label("e2e", "pr"), func() { &operatorPod, "manager", fmt.Sprintf( - "kubectl-fdb -n %s recover-multi-region-cluster --version-check=false --wait=false %s", + "kubectl-fdb -n %s recover multi-region --version-check=false --wait=false %s", remote.Namespace(), remote.Name(), ), @@ -245,7 +267,7 @@ var _ = Describe("Operator Plugin", Label("e2e", "pr"), func() { &operatorPod, "manager", fmt.Sprintf( - "kubectl-fdb -n %s recover-multi-region-cluster --version-check=false --wait=false %s", + "kubectl-fdb -n %s recover multi-region --version-check=false --wait=false %s", remote.Namespace(), remote.Name(), ), @@ -277,14 +299,141 @@ var _ = Describe("Operator Plugin", Label("e2e", "pr"), func() { } }) }) + }) + + When("a majority of coordinators are down in a single dc cluster", func() { + var useDNS bool + var fdbCluster *fixtures.FdbCluster AfterEach(func(ctx SpecContext) { - log.Println("Recreate cluster") - // Delete the broken cluster. - factory.Shutdown(ctx) - // Recreate the cluster to make sure the next tests can proceed - factory = fixtures.CreateFactory(testOptions) - fdbCluster = factory.CreateFdbHaCluster(ctx, clusterConfig) + // Delete the cluster. + Expect(fdbCluster.Delete(ctx)).NotTo(HaveOccurred()) + }) + + JustBeforeEach(func(ctx SpecContext) { + clusterConfig = fixtures.DefaultClusterConfig(false) + clusterConfig.UseDNS = ptr.To(useDNS) + fdbCluster = factory.CreateFdbCluster(ctx, clusterConfig) + coordinators := fdbCluster.GetCoordinators(ctx) + minimumFaultDomains := fdbCluster.GetCluster(ctx).MinimumFaultDomains() + downCoordinators := make([]corev1.Pod, 0, minimumFaultDomains) + for _, coordinator := range coordinators { + if len(downCoordinators) >= minimumFaultDomains { + break + } + + downCoordinators = append(downCoordinators, coordinator) + } + + // Set those coordinators as unschedulable to simulate that those coordinators are down. Another option + // would be to create a network partition. + fdbCluster.SetPodsAsUnschedulable(ctx, downCoordinators) + }) + + // Default case is to run with DNS enabled. The test case with IPs enabled can run into issues when + // the underlying Kubernetes cluster deletes pods. + // Because of the above issues the test case is currently disabled (marked as pending) and can be used + // to run the test manually if needed. + PWhen("DNS is disabled", func() { + BeforeEach(func(_ SpecContext) { + useDNS = false + }) + + It("should recover the coordinators", func(ctx SpecContext) { + // Pick one operator pod and execute the recovery command + operatorPod := factory.RandomPickOnePod( + factory.GetOperatorPods(ctx, fdbCluster.Namespace()).Items, + ) + log.Println("operatorPod:", operatorPod.Name) + stdout, stderr, err := factory.ExecuteCmdOnPod( + ctx, + &operatorPod, + "manager", + fmt.Sprintf( + "kubectl-fdb -n %s recover single-dc --version-check=false --wait=false %s", + fdbCluster.Namespace(), + fdbCluster.Name(), + ), + false, + ) + log.Println("stdout:", stdout, "stderr:", stderr) + Expect(err).NotTo(HaveOccurred()) + + // Ensure the cluster is available again. + Eventually(func() bool { + return fdbCluster.GetStatus(ctx).Client.DatabaseStatus.Available + }).WithTimeout(2 * time.Minute).WithPolling(1 * time.Second).Should(BeTrue()) + + fdbCluster.SetSkipReconciliation(ctx, false) + // Recreate the operator pods to ensure they get the new connection string. + factory.RecreateOperatorPods(ctx, fdbCluster.Namespace()) + // Ensure that the cluster is able to reconcile + Expect(fdbCluster.WaitForReconciliation(ctx)).To(Succeed()) + + log.Println( + "new connection string:", + fdbCluster.GetCluster(ctx).Status.ConnectionString, + ) + connectionString, err := fdbv1beta2.ParseConnectionString( + fdbCluster.GetCluster(ctx).Status.ConnectionString, + ) + Expect(err).NotTo(HaveOccurred()) + + for _, coordinator := range connectionString.Coordinators { + address, err := fdbv1beta2.ParseProcessAddress(coordinator) + Expect(err).NotTo(HaveOccurred()) + Expect(address.StringAddress).To(BeEmpty()) + } + }) + }) + + When("DNS names in the cluster file are used", func() { + BeforeEach(func(_ SpecContext) { + useDNS = true + }) + + It("should recover the coordinators", func(ctx SpecContext) { + // Pick one operator pod and execute the recovery command + operatorPod := factory.RandomPickOnePod( + factory.GetOperatorPods(ctx, fdbCluster.Namespace()).Items, + ) + log.Println("operatorPod:", operatorPod.Name) + stdout, stderr, err := factory.ExecuteCmdOnPod( + ctx, + &operatorPod, + "manager", + fmt.Sprintf( + "kubectl-fdb -n %s recover single-dc --version-check=false --wait=false %s", + fdbCluster.Namespace(), + fdbCluster.Name(), + ), + false, + ) + log.Println("stdout:", stdout, "stderr:", stderr) + if strings.Contains(stderr, "Error determining public address") { + Skip( + "plugin was not able to determine public address, this means that all coordinators are probably gone", + ) + } + Expect(err).NotTo(HaveOccurred()) + + // Ensure the cluster is available again. + Eventually(func() bool { + return fdbCluster.GetStatus(ctx).Client.DatabaseStatus.Available + }).WithTimeout(2 * time.Minute).WithPolling(1 * time.Second).Should(BeTrue()) + + currentConnectionString := fdbCluster.GetStatus(ctx).Cluster.ConnectionString + log.Println("new connection string:", currentConnectionString) + connectionString, err := fdbv1beta2.ParseConnectionString(currentConnectionString) + Expect(err).NotTo(HaveOccurred()) + + for _, coordinator := range connectionString.Coordinators { + address, err := fdbv1beta2.ParseProcessAddress(coordinator) + log.Println("address", address) + Expect(err).NotTo(HaveOccurred()) + Expect(address.StringAddress).NotTo(BeEmpty()) + } + }) }) }) }) diff --git a/e2e/test_operator_stress/operator_stress_test.go b/e2e/test_operator_stress/operator_stress_test.go index 260a075ff..eb2703288 100644 --- a/e2e/test_operator_stress/operator_stress_test.go +++ b/e2e/test_operator_stress/operator_stress_test.go @@ -58,7 +58,7 @@ var _ = Describe("Operator Stress", Label("e2e"), func() { fixtures.DefaultClusterConfig(false), ) Expect(fdbCluster.IsAvailable(ctx)).To(BeTrue()) - Expect(fdbCluster.Destroy(ctx)).NotTo(HaveOccurred()) + Expect(fdbCluster.Delete(ctx)).NotTo(HaveOccurred()) } }) }) @@ -74,7 +74,7 @@ var _ = Describe("Operator Stress", Label("e2e"), func() { }) AfterEach(func(ctx SpecContext) { - Expect(fdbCluster.Destroy(ctx)).NotTo(HaveOccurred()) + Expect(fdbCluster.Delete(ctx)).NotTo(HaveOccurred()) }) It("should replace the targeted Pod", func(ctx SpecContext) { diff --git a/kubectl-fdb/cmd/recover.go b/kubectl-fdb/cmd/recover.go new file mode 100644 index 000000000..90d8365e2 --- /dev/null +++ b/kubectl-fdb/cmd/recover.go @@ -0,0 +1,383 @@ +/* + * recover_multi_region_cluster.go + * + * This source file is part of the FoundationDB open source project + * + * Copyright 2018-2026 Apple Inc. and the FoundationDB project 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 cmd + +import ( + "context" + "fmt" + "io" + "log" + "os" + "path" + "strings" + "time" + + fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/v2/api/v1beta2" + "github.com/FoundationDB/fdb-kubernetes-operator/v2/internal" + kubeHelper "github.com/FoundationDB/fdb-kubernetes-operator/v2/internal/kubernetes" + "github.com/spf13/cobra" + corev1 "k8s.io/api/core/v1" + k8serrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/cli-runtime/pkg/genericiooptions" + "k8s.io/client-go/rest" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// RecoveryOpts struct to pass down all args to the actual runner. +type RecoveryOpts struct { + // Client is the client.Client to interact with the Kubernetes API. + Client client.Client + // Config is the rest.Config to interact with the Kubernetes API + Config *rest.Config + // ClusterName represents the cluster name of the targeted cluster. + ClusterName string + // Namespace represents the namespace of the targeted cluster. + Namespace string + // Stdout to print commands stdout output. + Stdout io.Writer + // Stderr to print commands stderr output. + Stderr io.Writer + // excludedCoordinators defines the coordinators that should be skipped during the recovery effort. + excludedCoordinators []string +} + +func newRecoverCmd(streams genericiooptions.IOStreams) *cobra.Command { + o := newFDBOptions(streams) + + cmd := &cobra.Command{ + Use: "recover", + Short: "Subcommand to recover a cluster if a majority of coordinators is lost permanently", + Long: "Subcommand to recover a cluster if a majority of coordinators is lost permanently", + RunE: func(c *cobra.Command, _ []string) error { + return c.Help() + }, + Example: ` +# Recover the multi-region cluster "sample-cluster-1" in the current Namespace +kubectl fdb recover multi-region sample-cluster-1 + +# Recover the multi-region cluster "sample-cluster-1" in the "testing" Namespace +kubectl fdb recover multi-region -n testing sample-cluster-1 + +# Recover the single-dc cluster "sample-cluster-1" in the current Namespace +kubectl fdb recover single-dc sample-cluster-1 + +# Recover the single-dc cluster "sample-cluster-1" in the "testing" Namespace +kubectl fdb recover single-dc -n testing sample-cluster-1 +`, + } + cmd.SetOut(o.Out) + cmd.SetErr(o.ErrOut) + cmd.SetIn(o.In) + + cmd.AddCommand(newRecoverMultiRegionClusterCmd(streams)) + cmd.AddCommand(newRecoverSingleDCClusterCmd(streams)) + o.configFlags.AddFlags(cmd.Flags()) + + return cmd +} + +// getDataDir will return the target data directory to upload the coordinator files to. The directory can be different, depending +// on the used image type and if more than one process should be running inside the Pod. +func getDataDir(dataDir string, pod *corev1.Pod, cluster *fdbv1beta2.FoundationDBCluster) string { + baseDir := dataDir + // If the dataDir has a suffix for the process we remove it. + if dataDir != "/var/fdb/data" { + baseDir = path.Dir(dataDir) + } + + // If the unified image is used we can simply return /var/fdb/data/1, as the unified image will always add the process + // directory, even if only a single process is running inside the Pod. + if cluster.UseUnifiedImage() { + return path.Join(baseDir, "/1") + } + + // In this path we use the split image, so the process directory is only added if more than one process should be running + processClass := internal.GetProcessClassFromMeta(cluster, pod.ObjectMeta) + + if processClass.IsLogProcess() && cluster.GetLogServersPerPod() > 1 { + return path.Join(baseDir, "/1") + } + + if processClass == fdbv1beta2.ProcessClassStorage && cluster.GetStorageServersPerPod() > 1 { + return path.Join(baseDir, "/1") + } + + // This is the default case if we are running one process per Pod for this storage class and using the split image. + return baseDir +} + +func downloadCoordinatorFile( + ctx context.Context, + kubeClient client.Client, + config *rest.Config, + pod *corev1.Pod, + src string, + dst string, +) error { + tmpCoordinatorFile, err := os.OpenFile(dst, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0600) + if err != nil { + return err + } + + defer func() { + _ = tmpCoordinatorFile.Close() + }() + + log.Println( + "Download files, target:", + dst, + "source", + src, + "pod", + pod.Name, + "Namespace", + pod.Namespace, + ) + err = kubeHelper.DownloadFile( + ctx, + kubeClient, + config, + pod, + fdbv1beta2.MainContainerName, + src, + tmpCoordinatorFile, + ) + if err != nil { + return err + } + + fileInfo, err := os.Stat(tmpCoordinatorFile.Name()) + if err != nil { + return err + } + + if fileInfo.Size() <= 0 { + return fmt.Errorf("file %s is empty", tmpCoordinatorFile.Name()) + } + + return nil +} + +func uploadCoordinatorFile( + ctx context.Context, + kubeClient client.Client, + config *rest.Config, + pod *corev1.Pod, + src string, + dst string, +) error { + tmpCoordinatorFile, err := os.OpenFile(src, os.O_RDONLY, 0600) + if err != nil { + return err + } + + defer func() { + _ = tmpCoordinatorFile.Close() + }() + + log.Println( + "Upload files, target:", + dst, + "source", + src, + "pod", + pod.Name, + "Namespace", + pod.Namespace, + ) + + return kubeHelper.UploadFile( + ctx, + kubeClient, + config, + pod, + fdbv1beta2.MainContainerName, + tmpCoordinatorFile, + dst, + ) +} + +// restartFdbserverInCluster will try to restart all fdbserver processes inside all the pods of the cluster. If the restart fails, it will be retried again two more times. +func restartFdbserverInCluster( + ctx context.Context, + kubeClient client.Client, + config *rest.Config, + cluster *fdbv1beta2.FoundationDBCluster, +) error { + pods, err := getRunningPodsForCluster(ctx, kubeClient, cluster) + if err != nil { + return err + } + + // Now all Pods must be restarted and the previous local cluster file must be deleted to make sure the fdbserver is picking the connection string from the seed cluster file (`/var/dynamic-conf/fdb.cluster`). + retryRestart := make([]corev1.Pod, 0, len(pods.Items)) + for _, pod := range pods.Items { + var stderr string + _, _, err = kubeHelper.ExecuteCommand( + context.Background(), + kubeClient, + config, + pod.Namespace, + pod.Name, + fdbv1beta2.MainContainerName, + "pkill fdbserver && rm -f /var/fdb/data/fdb.cluster && pkill fdbserver || true", + false, + ) + if err != nil { + // If the pod doesn't exist anymore ignore the error. The pod will have the new configuration when recreated again. + if k8serrors.IsNotFound(err) { + continue + } + + time.Sleep(1 * time.Second) + log.Println( + "error restarting process in pod", + pod.Name, + "got error", + err.Error(), + "will be directly retried, stderr:", + stderr, + ) + _, stderr, err = kubeHelper.ExecuteCommand( + context.Background(), + kubeClient, + config, + pod.Namespace, + pod.Name, + fdbv1beta2.MainContainerName, + "pkill fdbserver && rm -f /var/fdb/data/fdb.cluster && pkill fdbserver || true", + false, + ) + if err != nil { + log.Println( + "error restarting process in pod", + pod.Name, + "got error", + err.Error(), + "will be retried later, stderr:", + stderr, + ) + retryRestart = append(retryRestart, pod) + } + } + } + + if len(retryRestart) == 0 { + return nil + } + + // If we have more than one pod where we failed to restart the fdbserver processes, wait ten seconds before trying again. + time.Sleep(10 * time.Second) + log.Println( + "Failed to restart the fdbserver processes in", + len(retryRestart), + "pods, will be retried now.", + ) + + for _, pod := range retryRestart { + // Pod is marked for deletion, so we can skip it here. + if !pod.DeletionTimestamp.IsZero() { + continue + } + + _, _, err = kubeHelper.ExecuteCommand( + context.Background(), + kubeClient, + config, + pod.Namespace, + pod.Name, + fdbv1beta2.MainContainerName, + "pkill fdbserver && rm -f /var/fdb/data/fdb.cluster && pkill fdbserver || true", + false, + ) + if err != nil { + // If the pod doesn't exist anymore ignore the error. The pod will have the new configuration when recreated again. + if k8serrors.IsNotFound(err) { + continue + } + + return err + } + } + + return nil +} + +func checkIfClusterIsUnavailableAndMajorityOfCoordinatorsAreUnreachable( + ctx context.Context, + kubeClient client.Client, + config *rest.Config, + cluster *fdbv1beta2.FoundationDBCluster, +) error { + pods, err := getRunningPodsForCluster(ctx, kubeClient, cluster) + if err != nil { + return err + } + + clientPod, err := kubeHelper.PickRandomPod(pods) + if err != nil { + return err + } + + log.Println("Getting the status from:", clientPod.Name) + for range 5 { + err = getStatusAndCheckIfClusterShouldBeRecovered(ctx, kubeClient, config, clientPod) + if err == nil { + break + } + + time.Sleep(5 * time.Second) + } + + // If DNS is used for the cluster file, we could hit cases where no DNS entry can be resolved, in this case we could + // assume that the cluster is also down. The error from the client side is the following: + // Error: error getting status: Error determining public address. + // ERROR: Unable to bind to network (1512) + if err != nil && strings.Contains(err.Error(), "Error determining public address") { + return err + } + + return err +} + +func getStatusAndCheckIfClusterShouldBeRecovered(ctx context.Context, + kubeClient client.Client, + config *rest.Config, + clientPod *corev1.Pod) error { + status, err := getStatus(ctx, kubeClient, config, clientPod) + if err != nil { + return err + } + + if status.Client.DatabaseStatus.Available { + return fmt.Errorf("cluster is available, will abort any further actions") + } + + if status.Client.DatabaseStatus.Healthy { + return fmt.Errorf("cluster is healthy, will abort any further actions") + } + + if status.Client.Coordinators.QuorumReachable { + return fmt.Errorf("quorum of coordinators are reachable, will abort any further actions") + } + + return nil +} diff --git a/kubectl-fdb/cmd/recover_multi_region_cluster.go b/kubectl-fdb/cmd/recover_multi_region_cluster.go index 9f7eb3797..76fb42272 100644 --- a/kubectl-fdb/cmd/recover_multi_region_cluster.go +++ b/kubectl-fdb/cmd/recover_multi_region_cluster.go @@ -23,7 +23,6 @@ package cmd import ( "context" "fmt" - "io" "log" "os" "path" @@ -36,33 +35,15 @@ import ( "github.com/go-logr/logr" "github.com/spf13/cobra" corev1 "k8s.io/api/core/v1" - k8serrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/cli-runtime/pkg/genericiooptions" - "k8s.io/client-go/rest" "sigs.k8s.io/controller-runtime/pkg/client" ) -// RecoverMultiRegionClusterOpts struct to pass down all args to the actual runner. -type RecoverMultiRegionClusterOpts struct { - // Client is the client.Client to interact with the Kubernetes API. - Client client.Client - // Config is the rest.Config to interact with the Kubernetes API - Config *rest.Config - // ClusterName represents the cluster name of the targeted cluster. - ClusterName string - // Namespace represents the namespace of the targeted cluster. - Namespace string - // Stdout to print commands stdout output. - Stdout io.Writer - // Stderr to print commands stderr output. - Stderr io.Writer -} - func newRecoverMultiRegionClusterCmd(streams genericiooptions.IOStreams) *cobra.Command { o := newFDBOptions(streams) cmd := &cobra.Command{ - Use: "recover-multi-region-cluster", + Use: "multi-region", Short: "Recover a multi-region cluster if a majority of coordinators is lost permanently", Long: "Recover a multi-region cluster if a majority of coordinators is lost permanently", RunE: func(cmd *cobra.Command, args []string) error { @@ -116,7 +97,7 @@ func newRecoverMultiRegionClusterCmd(streams genericiooptions.IOStreams) *cobra. } return RecoverMultiRegionCluster(cmd.Context(), - RecoverMultiRegionClusterOpts{ + RecoveryOpts{ Client: kubeClient, Config: config, ClusterName: clusterName, @@ -127,10 +108,10 @@ func newRecoverMultiRegionClusterCmd(streams genericiooptions.IOStreams) *cobra. }, Example: ` # Recover the multi-region cluster "sample-cluster-1" in the current Namespace -kubectl fdb recover-multi-region-cluster sample-cluster-1 +kubectl fdb recover multi-region sample-cluster-1 # Recover the multi-region cluster "sample-cluster-1" in the "testing" Namespace -kubectl fdb recover-multi-region-cluster -n testing sample-cluster-1 +kubectl fdb recover multi-region -n testing sample-cluster-1 `, } cmd.SetOut(o.Out) @@ -144,7 +125,7 @@ kubectl fdb recover-multi-region-cluster -n testing sample-cluster-1 // RecoverMultiRegionCluster will forcefully recover a multi-region cluster if a majority of coordinators are lost. // Performing this action can result in data loss. -func RecoverMultiRegionCluster(ctx context.Context, opts RecoverMultiRegionClusterOpts) error { +func RecoverMultiRegionCluster(ctx context.Context, opts RecoveryOpts) error { cluster := &fdbv1beta2.FoundationDBCluster{} err := opts.Client.Get( ctx, @@ -493,291 +474,3 @@ func RecoverMultiRegionCluster(ctx context.Context, opts RecoverMultiRegionClust return nil } - -// getDataDir will return the target data directory to upload the coordinator files to. The directory can be different, depending -// on the used image type and if more than one process should be running inside the Pod. -func getDataDir(dataDir string, pod *corev1.Pod, cluster *fdbv1beta2.FoundationDBCluster) string { - baseDir := dataDir - // If the dataDir has a suffix for the process we remove it. - if dataDir != "/var/fdb/data" { - baseDir = path.Dir(dataDir) - } - - // If the unified image is used we can simply return /var/fdb/data/1, as the unified image will always add the process - // directory, even if only a single process is running inside the Pod. - if cluster.UseUnifiedImage() { - return path.Join(baseDir, "/1") - } - - // In this path we use the split image, so the process directory is only added if more than one process should be running - processClass := internal.GetProcessClassFromMeta(cluster, pod.ObjectMeta) - - if processClass.IsLogProcess() && cluster.GetLogServersPerPod() > 1 { - return path.Join(baseDir, "/1") - } - - if processClass == fdbv1beta2.ProcessClassStorage && cluster.GetStorageServersPerPod() > 1 { - return path.Join(baseDir, "/1") - } - - // This is the default case if we are running one process per Pod for this storage class and using the split image. - return baseDir -} - -func downloadCoordinatorFile( - ctx context.Context, - kubeClient client.Client, - config *rest.Config, - pod *corev1.Pod, - src string, - dst string, -) error { - tmpCoordinatorFile, err := os.OpenFile(dst, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0600) - if err != nil { - return err - } - - defer func() { - _ = tmpCoordinatorFile.Close() - }() - - log.Println( - "Download files, target:", - dst, - "source", - src, - "pod", - pod.Name, - "Namespace", - pod.Namespace, - ) - err = kubeHelper.DownloadFile( - ctx, - kubeClient, - config, - pod, - fdbv1beta2.MainContainerName, - src, - tmpCoordinatorFile, - ) - if err != nil { - return err - } - - fileInfo, err := os.Stat(tmpCoordinatorFile.Name()) - if err != nil { - return err - } - - if fileInfo.Size() <= 0 { - return fmt.Errorf("file %s is empty", tmpCoordinatorFile.Name()) - } - - return nil -} - -func uploadCoordinatorFile( - ctx context.Context, - kubeClient client.Client, - config *rest.Config, - pod *corev1.Pod, - src string, - dst string, -) error { - tmpCoordinatorFile, err := os.OpenFile(src, os.O_RDONLY, 0600) - if err != nil { - return err - } - - defer func() { - _ = tmpCoordinatorFile.Close() - }() - - log.Println( - "Upload files, target:", - dst, - "source", - src, - "pod", - pod.Name, - "Namespace", - pod.Namespace, - ) - - return kubeHelper.UploadFile( - ctx, - kubeClient, - config, - pod, - fdbv1beta2.MainContainerName, - tmpCoordinatorFile, - dst, - ) -} - -// restartFdbserverInCluster will try to restart all fdbserver processes inside all the pods of the cluster. If the restart fails, it will be retried again two more times. -func restartFdbserverInCluster( - ctx context.Context, - kubeClient client.Client, - config *rest.Config, - cluster *fdbv1beta2.FoundationDBCluster, -) error { - pods, err := getRunningPodsForCluster(ctx, kubeClient, cluster) - if err != nil { - return err - } - - // Now all Pods must be restarted and the previous local cluster file must be deleted to make sure the fdbserver is picking the connection string from the seed cluster file (`/var/dynamic-conf/fdb.cluster`). - retryRestart := make([]corev1.Pod, 0, len(pods.Items)) - for _, pod := range pods.Items { - var stderr string - _, _, err = kubeHelper.ExecuteCommand( - context.Background(), - kubeClient, - config, - pod.Namespace, - pod.Name, - fdbv1beta2.MainContainerName, - "pkill fdbserver && rm -f /var/fdb/data/fdb.cluster && pkill fdbserver || true", - false, - ) - if err != nil { - // If the pod doesn't exist anymore ignore the error. The pod will have the new configuration when recreated again. - if k8serrors.IsNotFound(err) { - continue - } - - time.Sleep(1 * time.Second) - log.Println( - "error restarting process in pod", - pod.Name, - "got error", - err.Error(), - "will be directly retried, stderr:", - stderr, - ) - _, stderr, err = kubeHelper.ExecuteCommand( - context.Background(), - kubeClient, - config, - pod.Namespace, - pod.Name, - fdbv1beta2.MainContainerName, - "pkill fdbserver && rm -f /var/fdb/data/fdb.cluster && pkill fdbserver || true", - false, - ) - if err != nil { - log.Println( - "error restarting process in pod", - pod.Name, - "got error", - err.Error(), - "will be retried later, stderr:", - stderr, - ) - retryRestart = append(retryRestart, pod) - } - } - } - - if len(retryRestart) == 0 { - return nil - } - - // If we have more than one pod where we failed to restart the fdbserver processes, wait ten seconds before trying again. - time.Sleep(10 * time.Second) - log.Println( - "Failed to restart the fdbserver processes in", - len(retryRestart), - "pods, will be retried now.", - ) - - for _, pod := range retryRestart { - // Pod is marked for deletion, so we can skip it here. - if !pod.DeletionTimestamp.IsZero() { - continue - } - - _, _, err = kubeHelper.ExecuteCommand( - context.Background(), - kubeClient, - config, - pod.Namespace, - pod.Name, - fdbv1beta2.MainContainerName, - "pkill fdbserver && rm -f /var/fdb/data/fdb.cluster && pkill fdbserver || true", - false, - ) - if err != nil { - // If the pod doesn't exist anymore ignore the error. The pod will have the new configuration when recreated again. - if k8serrors.IsNotFound(err) { - continue - } - - return err - } - } - - return nil -} - -func checkIfClusterIsUnavailableAndMajorityOfCoordinatorsAreUnreachable( - ctx context.Context, - kubeClient client.Client, - config *rest.Config, - cluster *fdbv1beta2.FoundationDBCluster, -) error { - pods, err := getRunningPodsForCluster(ctx, kubeClient, cluster) - if err != nil { - return err - } - - clientPod, err := kubeHelper.PickRandomPod(pods) - if err != nil { - return err - } - - log.Println("Getting the status from:", clientPod.Name) - for range 5 { - err = getStatusAndCheckIfClusterShouldBeRecovered(ctx, kubeClient, config, clientPod) - if err == nil { - break - } - - time.Sleep(5 * time.Second) - } - - // If DNS is used for the cluster file, we could hit cases where no DNS entry can be resolved, in this case we could - // assume that the cluster is also down. The error from the client side is the following: - // Error: error getting status: Error determining public address. - // ERROR: Unable to bind to network (1512) - if err != nil && strings.Contains(err.Error(), "Error determining public address") { - return err - } - - return err -} - -func getStatusAndCheckIfClusterShouldBeRecovered(ctx context.Context, - kubeClient client.Client, - config *rest.Config, - clientPod *corev1.Pod) error { - status, err := getStatus(ctx, kubeClient, config, clientPod) - if err != nil { - return err - } - - if status.Client.DatabaseStatus.Available { - return fmt.Errorf("cluster is available, will abort any further actions") - } - - if status.Client.DatabaseStatus.Healthy { - return fmt.Errorf("cluster is healthy, will abort any further actions") - } - - if status.Client.Coordinators.QuorumReachable { - return fmt.Errorf("quorum of coordinators are reachable, will abort any further actions") - } - - return nil -} diff --git a/kubectl-fdb/cmd/recover_single_dc_cluster.go b/kubectl-fdb/cmd/recover_single_dc_cluster.go new file mode 100644 index 000000000..da8b539d1 --- /dev/null +++ b/kubectl-fdb/cmd/recover_single_dc_cluster.go @@ -0,0 +1,461 @@ +/* + * recover_multi_region_cluster.go + * + * This source file is part of the FoundationDB open source project + * + * Copyright 2018-2026 Apple Inc. and the FoundationDB project 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 cmd + +import ( + "context" + "fmt" + "log" + "os" + "path" + "strings" + "time" + + fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/v2/api/v1beta2" + "github.com/FoundationDB/fdb-kubernetes-operator/v2/internal" + kubeHelper "github.com/FoundationDB/fdb-kubernetes-operator/v2/internal/kubernetes" + "github.com/go-logr/logr" + "github.com/spf13/cobra" + corev1 "k8s.io/api/core/v1" + "k8s.io/cli-runtime/pkg/genericiooptions" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +func newRecoverSingleDCClusterCmd(streams genericiooptions.IOStreams) *cobra.Command { + o := newFDBOptions(streams) + + cmd := &cobra.Command{ + Use: "single-dc", + Short: "Recover a single dc cluster if a majority of coordinators is lost permanently", + Long: "Recover a single dc cluster if a majority of coordinators is lost permanently", + RunE: func(cmd *cobra.Command, args []string) error { + wait, err := cmd.Root().Flags().GetBool("wait") + if err != nil { + return err + } + + if len(args) != 1 { + return fmt.Errorf( + "exactly one cluster name must be specified, provided args: %v", + args, + ) + } + + clusterName := args[0] + + kubeClient, err := getKubeClient(cmd.Context(), o) + if err != nil { + return err + } + + namespace, err := getNamespace(*o.configFlags.Namespace) + if err != nil { + return err + } + + config, err := o.configFlags.ToRESTConfig() + if err != nil { + return err + } + + excludedCoordinators, err := cmd.Flags().GetStringArray("exclude-coordinator") + if err != nil { + return err + } + + if wait { + confirmed := confirmAction( + fmt.Sprintf( + "WARNING:\nThe cluster: %s/%s will be force recovered.\nOnly perform those steps if you are unable to recover the coordinator pods.\nPerforming this action could lead to data loss.\n At leas one coordinator must be active and running to copy the coordinator state.\n", + namespace, + clusterName, + ), + ) + if !confirmed { + return fmt.Errorf("aborted recover single-dc aciton") + } + } + + return RecoverSingleDCCluster(cmd.Context(), + RecoveryOpts{ + Client: kubeClient, + Config: config, + ClusterName: clusterName, + Namespace: namespace, + Stdout: cmd.OutOrStdout(), + Stderr: cmd.OutOrStderr(), + excludedCoordinators: excludedCoordinators, + }) + }, + Example: ` +# Recover the single dc cluster "sample-cluster-1" in the current Namespace +kubectl fdb recover single-dc sample-cluster-1 + +# Recover the single-dc cluster "sample-cluster-1" in the "testing" Namespace +kubectl fdb recover single-dc -n testing sample-cluster-1 + +# Recover the single-dc cluster "sample-cluster-1" in the "testing" Namespace and excluding the coordinator on pod +# sample-cluster-1-storage-42 +kubectl fdb recover single-dc -n testing sample-cluster-1 --exclude-coordinator sample-cluster-1-storage-42 +`, + } + cmd.SetOut(o.Out) + cmd.SetErr(o.ErrOut) + cmd.SetIn(o.In) + cmd.Flags(). + StringArray("exclude-coordinator", []string{}, "Exclude a coordinator from the recovery process, e.g. because the coordinator pod is running but not able to reach the rest of the cluster. The provided name must match the pod name of the coordinator.") + + o.configFlags.AddFlags(cmd.Flags()) + + return cmd +} + +// RecoverSingleDCCluster will forcefully recover a single-dc cluster if a majority of coordinators are lost. +// Performing this action can result in data loss. +func RecoverSingleDCCluster(ctx context.Context, opts RecoveryOpts) error { + cluster := &fdbv1beta2.FoundationDBCluster{} + err := opts.Client.Get( + ctx, + client.ObjectKey{Name: opts.ClusterName, Namespace: opts.Namespace}, + cluster, + ) + if err != nil { + return err + } + + err = checkIfClusterIsUnavailableAndMajorityOfCoordinatorsAreUnreachable( + ctx, + opts.Client, + opts.Config, + cluster, + ) + if err != nil { + return err + } + + // Skip the cluster, make sure the operator is not taking any action on the cluster. + err = setSkipReconciliation(ctx, opts.Client, cluster, true) + if err != nil { + return err + } + + // Fetch the last connection string from the `FoundationDBCluster` status, e.g. `kubectl get fdb ${cluster} -o jsonpath='{ .status.connectionString }'`. + lastConnectionString := cluster.Status.ConnectionString + lastConnectionStringParts := strings.Split(lastConnectionString, "@") + addresses := strings.Split(lastConnectionStringParts[1], ",") + usesDNSInClusterFile := cluster.UseDNSInClusterFile() + + log.Println( + "current connection string", + lastConnectionString, + "cluster uses DNS:", + usesDNSInClusterFile, + ) + var useTLS bool + coordinators := map[string]fdbv1beta2.ProcessAddress{} + for _, addr := range addresses { + parsed, parseErr := fdbv1beta2.ParseProcessAddress(addr) + if parseErr != nil { + return parseErr + } + + log.Println("found coordinator", parsed.String()) + coordinators[parsed.MachineAddress()] = parsed + // If the tls flag is present we assume that the coordinators should make use of TLS. + _, useTLS = parsed.Flags["tls"] + } + + log.Println("Current coordinators", coordinators, "useTLS", useTLS) + // Fetch all Pods and coordinators for the remote and remote satellite. + runningCoordinators := map[string]fdbv1beta2.None{} + newCoordinators := make([]fdbv1beta2.ProcessAddress, 0, cluster.DesiredCoordinatorCount()) + processCounts, err := cluster.GetProcessCountsWithDefaults() + if err != nil { + return err + } + candidates := make([]*corev1.Pod, 0, processCounts.Total()) + + pods, err := getRunningPodsForCluster(ctx, opts.Client, cluster) + if err != nil { + return err + } + + excludedCoordinators := map[string]fdbv1beta2.None{} + for _, excludedCoordinator := range opts.excludedCoordinators { + excludedCoordinators[excludedCoordinator] = fdbv1beta2.None{} + } + + // Find a running coordinator to copy the coordinator files from. Note: Running doesn't necessarily mean that the + // coordinator is healthy from a cluster perspective, as the coordinator hosting pods could be up and running but have + // networking issues like a network partition. + var runningCoordinator *corev1.Pod + for _, pod := range pods.Items { + if _, ok := excludedCoordinators[pod.Name]; ok { + log.Println("Skipping pod as excluded from recovery coordinator set:", pod.Name) + continue + } + + if pod.Status.Phase != corev1.PodRunning { + log.Println( + "Skipping pod as pod's phase is not running, current phase:", + pod.Status.Phase, + ) + continue + } + + var addr fdbv1beta2.ProcessAddress + if usesDNSInClusterFile { + addr = fdbv1beta2.ProcessAddress{ + StringAddress: internal.GetPodDNSName(cluster, pod.GetName()), + } + } else { + currentPod := pod + publicIPs := internal.GetPublicIPsForPod(¤tPod, logr.Discard()) + if len(publicIPs) == 0 { + log.Println("Found no public IPs for pod:", pod.Name) + continue + } + + var parseErr error + addr, parseErr = fdbv1beta2.ParseProcessAddress(publicIPs[0]) + if parseErr != nil { + return parseErr + } + } + + log.Println("Checking pod", pod.Name, "address", addr.MachineAddress()) + loopPod := pod + if coordinatorAddr, ok := coordinators[addr.MachineAddress()]; ok { + log.Println("Found coordinator for cluster", pod.Name, "address", addr.MachineAddress()) + runningCoordinators[addr.MachineAddress()] = fdbv1beta2.None{} + newCoordinators = append(newCoordinators, coordinatorAddr) + + runningCoordinator = &loopPod + continue + } + + if !internal.GetProcessClassFromMeta(cluster, pod.ObjectMeta).IsStateful() { + continue + } + + candidates = append(candidates, &loopPod) + } + + if runningCoordinator == nil { + return fmt.Errorf("could not find any running coordinator for this cluster") + } + + // Pick new coordinators. + needsUpload := make([]*corev1.Pod, 0, cluster.DesiredCoordinatorCount()) + candidateIdx := 0 + for len(newCoordinators) < cluster.DesiredCoordinatorCount() { + if candidateIdx >= len(candidates) { + return fmt.Errorf( + "not enough coordinator candidates: need %d more, have %d running and %d candidates", + cluster.DesiredCoordinatorCount()-len(newCoordinators), len(newCoordinators), len(candidates), + ) + } + log.Println("Current coordinators:", len(newCoordinators)) + candidate := candidates[candidateIdx] + candidateIdx++ + + var addr fdbv1beta2.ProcessAddress + if usesDNSInClusterFile { + dnsName := internal.GetPodDNSName(cluster, candidate.GetName()) + addr = fdbv1beta2.ProcessAddress{StringAddress: dnsName} + } else { + var parseErr error + addr, parseErr = fdbv1beta2.ParseProcessAddress(candidate.Status.PodIP) + if parseErr != nil { + return parseErr + } + } + + if useTLS { + addr.Port = 4500 + addr.Flags = map[string]bool{"tls": true} + } else { + addr.Port = 4501 + } + + log.Println("Adding new coordinator:", addr.String()) + newCoordinators = append(newCoordinators, addr) + needsUpload = append(needsUpload, candidate) + } + + // If at least one coordinator needs to get the files uploaded, we perform the download and upload for the coordinators. + if len(needsUpload) > 0 { + // Copy the coordinator state from one of the running coordinators to your local machine: + coordinatorFiles := []string{"coordination-0.fdq", "coordination-1.fdq"} + tmpCoordinatorFiles := make([]string, 2) + tmpDir := os.TempDir() + for idx, coordinatorFile := range coordinatorFiles { + tmpCoordinatorFiles[idx] = path.Join(tmpDir, coordinatorFile) + } + + log.Println( + "tmpCoordinatorFiles", + tmpCoordinatorFiles, + "checking the location of the coordination-0.fdq in Pod", + runningCoordinator.Name, + ) + stdout, stderr, err := kubeHelper.ExecuteCommandOnPod( + context.Background(), + opts.Client, + opts.Config, + runningCoordinator, + fdbv1beta2.MainContainerName, + "find /var/fdb/data/ -type f -name 'coordination-0.fdq' -print -quit | head -n 1", + false, + ) + if err != nil { + log.Println(stderr) + return err + } + + trimmedStdout := strings.TrimSpace(stdout) + if trimmedStdout == "" { + return fmt.Errorf("no coordination file found in %s", runningCoordinator.Name) + } + + lines := strings.Split(trimmedStdout, "\n") + dataDir := path.Dir(strings.TrimSpace(lines[0])) + log.Println("dataDir:", dataDir) + for idx, coordinatorFile := range coordinatorFiles { + err = downloadCoordinatorFile( + ctx, + opts.Client, + opts.Config, + runningCoordinator, + path.Join(dataDir, coordinatorFile), + tmpCoordinatorFiles[idx], + ) + if err != nil { + return err + } + } + + for _, target := range needsUpload { + targetDataDir := getDataDir(dataDir, target, cluster) + + for idx, coordinatorFile := range coordinatorFiles { + err = uploadCoordinatorFile( + ctx, + opts.Client, + opts.Config, + target, + tmpCoordinatorFiles[idx], + path.Join(targetDataDir, coordinatorFile), + ) + if err != nil { + return err + } + } + } + } + + // Update the `ConfigMap` to contain the new connection string, the new connection string must contain the still existing coordinators and the new coordinators. The old entries must be removed. + var newConnectionString strings.Builder + newConnectionString.WriteString(lastConnectionStringParts[0]) + newConnectionString.WriteString("@") + for idx, coordinator := range newCoordinators { + newConnectionString.WriteString(coordinator.String()) + if idx == len(newCoordinators)-1 { + break + } + + newConnectionString.WriteString(",") + } + + newCS := newConnectionString.String() + log.Println("new connection string:", newCS) + err = updateConnectionString(ctx, opts.Client, cluster, newCS) + if err != nil { + return err + } + + // Wait ~1 min until the `ConfigMap` is synced to all Pods, you can check the `/var/dynamic-conf/fdb.cluster` inside a Pod if you are unsure. + time.Sleep(2 * time.Minute) + + // If the split image is used we have to update the copied files by making a POST request against the sidecar API. + // In the unified image, this step is not required as the dynamic files are directly mounted in the main container. + // We are not deleting the Pods as the operator is set to skip the reconciliation and therefore the deleted Pods + // would not be recreated. + if !cluster.UseUnifiedImage() { + log.Println("The cluster uses the split image, the plugin will update the copied files") + for _, pod := range pods.Items { + loopPod := pod + + command := []string{"/bin/bash", "-c"} + + var curlStr strings.Builder + curlStr.WriteString("curl -X POST") + if internal.PodHasSidecarTLS(&loopPod) { + curlStr.WriteString( + " --cacert ${FDB_TLS_CA_FILE} --cert ${FDB_TLS_CERTIFICATE_FILE} --key ${FDB_TLS_KEY_FILE} -k https://", + ) + } else { + curlStr.WriteString(" http://") + } + + curlStr.WriteString(loopPod.Status.PodIP) + curlStr.WriteString(":8080/copy_files > /dev/null") + + command = append(command, curlStr.String()) + + err = kubeHelper.ExecuteCommandRaw( + ctx, + opts.Client, + opts.Config, + runningCoordinator.Namespace, + runningCoordinator.Name, + fdbv1beta2.MainContainerName, + command, + nil, + opts.Stdout, + opts.Stderr, + false, + ) + if err != nil { + return err + } + } + } + + log.Println("Killing fdbserver processes") + // Now all Pods must be restarted and the previous local cluster file must be deleted to make sure the fdbserver is picking the connection string from the seed cluster file (`/var/dynamic-conf/fdb.cluster`). + err = restartFdbserverInCluster(ctx, opts.Client, opts.Config, cluster) + if err != nil { + return err + } + + // Wait until all fdbservers have started again. + time.Sleep(1 * time.Minute) + + // Now you can set `spec.Skip = false` to let the operator take over again. + // Skip the cluster, make sure the operator is not taking any action on the cluster. + err = setSkipReconciliation(ctx, opts.Client, cluster, false) + if err != nil { + return err + } + + return nil +} diff --git a/kubectl-fdb/cmd/root.go b/kubectl-fdb/cmd/root.go index c6e071ed4..802143eb5 100644 --- a/kubectl-fdb/cmd/root.go +++ b/kubectl-fdb/cmd/root.go @@ -100,7 +100,7 @@ func NewRootCmd( newFixCoordinatorIPsCmd(streams), newGetCmd(streams), newBuggifyCmd(streams), - newRecoverMultiRegionClusterCmd(streams), + newRecoverCmd(streams), newUpdateCmd(streams), ) From fabb99abc19d97b23c3738d4d45c90b5346d1d4c Mon Sep 17 00:00:00 2001 From: "Johannes M. Scheuermann" Date: Thu, 27 Aug 2026 21:18:06 +0200 Subject: [PATCH 2/4] Fix format --- kubectl-fdb/cmd/recover_single_dc_cluster.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/kubectl-fdb/cmd/recover_single_dc_cluster.go b/kubectl-fdb/cmd/recover_single_dc_cluster.go index da8b539d1..ffe16a556 100644 --- a/kubectl-fdb/cmd/recover_single_dc_cluster.go +++ b/kubectl-fdb/cmd/recover_single_dc_cluster.go @@ -270,7 +270,11 @@ func RecoverSingleDCCluster(ctx context.Context, opts RecoveryOpts) error { if candidateIdx >= len(candidates) { return fmt.Errorf( "not enough coordinator candidates: need %d more, have %d running and %d candidates", - cluster.DesiredCoordinatorCount()-len(newCoordinators), len(newCoordinators), len(candidates), + cluster.DesiredCoordinatorCount()-len( + newCoordinators, + ), + len(newCoordinators), + len(candidates), ) } log.Println("Current coordinators:", len(newCoordinators)) From a8f192c468458bc999a6e8be94201c0649d5a515 Mon Sep 17 00:00:00 2001 From: "Johannes M. Scheuermann" Date: Tue, 1 Sep 2026 19:59:28 +0200 Subject: [PATCH 3/4] Fix header --- kubectl-fdb/cmd/recover.go | 2 +- kubectl-fdb/cmd/recover_single_dc_cluster.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/kubectl-fdb/cmd/recover.go b/kubectl-fdb/cmd/recover.go index 90d8365e2..d77d83a3e 100644 --- a/kubectl-fdb/cmd/recover.go +++ b/kubectl-fdb/cmd/recover.go @@ -1,5 +1,5 @@ /* - * recover_multi_region_cluster.go + * recover.go * * This source file is part of the FoundationDB open source project * diff --git a/kubectl-fdb/cmd/recover_single_dc_cluster.go b/kubectl-fdb/cmd/recover_single_dc_cluster.go index ffe16a556..03750531e 100644 --- a/kubectl-fdb/cmd/recover_single_dc_cluster.go +++ b/kubectl-fdb/cmd/recover_single_dc_cluster.go @@ -1,5 +1,5 @@ /* - * recover_multi_region_cluster.go + * recover_single_dc_cluster.go * * This source file is part of the FoundationDB open source project * From a5a519c4e5dc13b0e9cb8b1303811e2b07582893 Mon Sep 17 00:00:00 2001 From: "Johannes M. Scheuermann" Date: Tue, 8 Sep 2026 11:43:45 +0200 Subject: [PATCH 4/4] Update e2e test setup and code review comments --- e2e/fixtures/fdb_cluster.go | 9 +- .../operator_plugin_test.go | 334 +++++++----------- kubectl-fdb/cmd/recover.go | 18 +- kubectl-fdb/cmd/recover_single_dc_cluster.go | 2 +- 4 files changed, 138 insertions(+), 225 deletions(-) diff --git a/e2e/fixtures/fdb_cluster.go b/e2e/fixtures/fdb_cluster.go index 635b92c77..c0e40f566 100644 --- a/e2e/fixtures/fdb_cluster.go +++ b/e2e/fixtures/fdb_cluster.go @@ -843,8 +843,11 @@ func (fdbCluster *FdbCluster) SetPodsAsUnschedulable(ctx context.Context, pods [ unschedulableProcessGroups, ) - // context aware. - time.Sleep(5 * time.Second) + select { + case <-ctx.Done(): + return + case <-time.After(5 * time.Second): + } for _, pod := range pods { fetchedPod := &corev1.Pod{} @@ -853,8 +856,6 @@ func (fdbCluster *FdbCluster) SetPodsAsUnschedulable(ctx context.Context, pods [ if err != nil { continue } - // check if error is absent -> case not created, just deleted - gomega.Expect(err).NotTo(gomega.HaveOccurred()) // Try deleting the Pod as a workaround until the operator handles all cases. if fetchedPod.Spec.NodeName != "" && fetchedPod.DeletionTimestamp.IsZero() { diff --git a/e2e/test_operator_plugin/operator_plugin_test.go b/e2e/test_operator_plugin/operator_plugin_test.go index e127ad8ca..77a06d6de 100644 --- a/e2e/test_operator_plugin/operator_plugin_test.go +++ b/e2e/test_operator_plugin/operator_plugin_test.go @@ -72,7 +72,6 @@ var _ = Describe("Operator Plugin", Label("e2e", "pr"), func() { }) AfterEach(func(ctx SpecContext) { - // Delete the cluster. Expect(fdbCluster.Delete(ctx)).NotTo(HaveOccurred()) }) @@ -101,104 +100,98 @@ var _ = Describe("Operator Plugin", Label("e2e", "pr"), func() { }) When("all Pods in the primary and satellites are down", func() { - var useDNS bool var fdbCluster *fixtures.HaFdbCluster AfterEach(func(ctx SpecContext) { - // Delete the cluster. fdbCluster.Delete(ctx) }) - JustBeforeEach(func(ctx SpecContext) { - clusterConfig = fixtures.DefaultClusterConfigWithHaMode( - fixtures.HaFourZoneSingleSat, - false, - ) - fdbCluster = factory.CreateFdbHaCluster(ctx, clusterConfig) - - var errGroup errgroup.Group - // Enable DNS names in the cluster file for the whole cluster. - for _, cluster := range fdbCluster.GetAllClusters() { - target := cluster - errGroup.Go(func() error { - return target.SetUseDNSInClusterFile(ctx, useDNS) - }) - } - Expect(errGroup.Wait()).NotTo(HaveOccurred()) - - for _, cluster := range fdbCluster.GetAllClusters() { - Expect(cluster.GetCluster(ctx).UseDNSInClusterFile()).To(Equal(useDNS)) - } + // Default case is to run with DNS enabled. The test case with IPs enabled can run into issues when + // the underlying Kubernetes cluster deletes pods. + // Because of the above issues the test case is currently disabled (marked as pending) and can be used + // to run the test manually if needed. + DescribeTableSubtree("should recover the coordinators", func(shouldUseDNS bool) { + JustBeforeEach(func(ctx SpecContext) { + clusterConfig = fixtures.DefaultClusterConfigWithHaMode( + fixtures.HaFourZoneSingleSat, + false, + ) + fdbCluster = factory.CreateFdbHaCluster(ctx, clusterConfig) + + var errGroup errgroup.Group + // Enable DNS names in the cluster file for the whole cluster. + for _, cluster := range fdbCluster.GetAllClusters() { + target := cluster + errGroup.Go(func() error { + return target.SetUseDNSInClusterFile(ctx, shouldUseDNS) + }) + } + Expect(errGroup.Wait()).NotTo(HaveOccurred()) - // This tests is a destructive test where the cluster will stop working for some period. - primary := fdbCluster.GetPrimary() - primary.SetSkipReconciliation(ctx, true) + for _, cluster := range fdbCluster.GetAllClusters() { + Expect(cluster.GetCluster(ctx).UseDNSInClusterFile()).To(Equal(shouldUseDNS)) + } - primarySatellite := fdbCluster.GetPrimarySatellite() - primarySatellite.SetSkipReconciliation(ctx, true) + // This tests is a destructive test where the cluster will stop working for some period. + primary := fdbCluster.GetPrimary() + primary.SetSkipReconciliation(ctx, true) - remoteSatellite := fdbCluster.GetRemoteSatellite() - remoteSatellite.SetSkipReconciliation(ctx, true) + primarySatellite := fdbCluster.GetPrimarySatellite() + primarySatellite.SetSkipReconciliation(ctx, true) - remote := fdbCluster.GetRemote() - remote.SetSkipReconciliation(ctx, true) + remoteSatellite := fdbCluster.GetRemoteSatellite() + remoteSatellite.SetSkipReconciliation(ctx, true) - var wg errgroup.Group - log.Println("Delete Pods in primary") - wg.Go(func() error { - return factory.GetControllerRuntimeClient(). - DeleteAllOf(ctx, &corev1.Pod{}, ctrlClient.MatchingLabels(primary.GetResourceLabels()), ctrlClient.InNamespace(primary.Namespace())) - }) + remote := fdbCluster.GetRemote() + remote.SetSkipReconciliation(ctx, true) - log.Println("Delete Pods in primary satellite") - wg.Go(func() error { - return factory.GetControllerRuntimeClient(). - DeleteAllOf(ctx, &corev1.Pod{}, ctrlClient.MatchingLabels(primarySatellite.GetResourceLabels()), ctrlClient.InNamespace(primarySatellite.Namespace())) - }) + var wg errgroup.Group + log.Println("Delete Pods in primary") + wg.Go(func() error { + return factory.GetControllerRuntimeClient(). + DeleteAllOf(ctx, &corev1.Pod{}, ctrlClient.MatchingLabels(primary.GetResourceLabels()), ctrlClient.InNamespace(primary.Namespace())) + }) - log.Println("Delete Pods in remote satellite") - wg.Go(func() error { - return factory.GetControllerRuntimeClient(). - DeleteAllOf(ctx, &corev1.Pod{}, ctrlClient.MatchingLabels(remoteSatellite.GetResourceLabels()), ctrlClient.InNamespace(remoteSatellite.Namespace())) - }) + log.Println("Delete Pods in primary satellite") + wg.Go(func() error { + return factory.GetControllerRuntimeClient(). + DeleteAllOf(ctx, &corev1.Pod{}, ctrlClient.MatchingLabels(primarySatellite.GetResourceLabels()), ctrlClient.InNamespace(primarySatellite.Namespace())) + }) - Expect(wg.Wait()).NotTo(HaveOccurred()) - // Wait a short amount of time to let the cluster see that the primary and primary satellite is down. - time.Sleep(5 * time.Second) + log.Println("Delete Pods in remote satellite") + wg.Go(func() error { + return factory.GetControllerRuntimeClient(). + DeleteAllOf(ctx, &corev1.Pod{}, ctrlClient.MatchingLabels(remoteSatellite.GetResourceLabels()), ctrlClient.InNamespace(remoteSatellite.Namespace())) + }) - // Ensure that all the pods are deleted. - Eventually(func(g Gomega) []corev1.Pod { - pods := &corev1.PodList{} - g.Expect(factory.GetControllerRuntimeClient().List(ctx, pods, ctrlClient.MatchingLabels(primary.GetResourceLabels()), ctrlClient.InNamespace(remoteSatellite.Namespace()))). - To(Succeed()) + Expect(wg.Wait()).NotTo(HaveOccurred()) + // Wait a short amount of time to let the cluster see that the primary and primary satellite is down. + time.Sleep(5 * time.Second) - return pods.Items - }).WithTimeout(5 * time.Minute).WithPolling(10 * time.Second).Should(BeEmpty()) + // Ensure that all the pods are deleted. + Eventually(func(g Gomega) []corev1.Pod { + pods := &corev1.PodList{} + g.Expect(factory.GetControllerRuntimeClient().List(ctx, pods, ctrlClient.MatchingLabels(primary.GetResourceLabels()), ctrlClient.InNamespace(remoteSatellite.Namespace()))). + To(Succeed()) - Eventually(func(g Gomega) []corev1.Pod { - pods := &corev1.PodList{} - g.Expect(factory.GetControllerRuntimeClient().List(ctx, pods, ctrlClient.MatchingLabels(primarySatellite.GetResourceLabels()), ctrlClient.InNamespace(remoteSatellite.Namespace()))). - To(Succeed()) + return pods.Items + }).WithTimeout(5 * time.Minute).WithPolling(10 * time.Second).Should(BeEmpty()) - return pods.Items - }).WithTimeout(5 * time.Minute).WithPolling(10 * time.Second).Should(BeEmpty()) + Eventually(func(g Gomega) []corev1.Pod { + pods := &corev1.PodList{} + g.Expect(factory.GetControllerRuntimeClient().List(ctx, pods, ctrlClient.MatchingLabels(primarySatellite.GetResourceLabels()), ctrlClient.InNamespace(remoteSatellite.Namespace()))). + To(Succeed()) - Eventually(func(g Gomega) []corev1.Pod { - pods := &corev1.PodList{} - g.Expect(factory.GetControllerRuntimeClient().List(ctx, pods, ctrlClient.MatchingLabels(remoteSatellite.GetResourceLabels()), ctrlClient.InNamespace(remoteSatellite.Namespace()))). - To(Succeed()) + return pods.Items + }).WithTimeout(5 * time.Minute).WithPolling(10 * time.Second).Should(BeEmpty()) - return pods.Items - }).WithTimeout(5 * time.Minute).WithPolling(10 * time.Second).Should(BeEmpty()) - }) + Eventually(func(g Gomega) []corev1.Pod { + pods := &corev1.PodList{} + g.Expect(factory.GetControllerRuntimeClient().List(ctx, pods, ctrlClient.MatchingLabels(remoteSatellite.GetResourceLabels()), ctrlClient.InNamespace(remoteSatellite.Namespace()))). + To(Succeed()) - // Default case is to run with DNS enabled. The test case with IPs enabled can run into issues when - // the underlying Kubernetes cluster deletes pods. - // Because of the above issues the test case is currently disabled (marked as pending) and can be used - // to run the test manually if needed. - PWhen("DNS is disabled", func() { - BeforeEach(func(_ SpecContext) { - useDNS = false + return pods.Items + }).WithTimeout(5 * time.Minute).WithPolling(10 * time.Second).Should(BeEmpty()) }) It("should recover the coordinators", func(ctx SpecContext) { @@ -220,6 +213,11 @@ var _ = Describe("Operator Plugin", Label("e2e", "pr"), func() { false, ) log.Println("stdout:", stdout, "stderr:", stderr) + if shouldUseDNS && strings.Contains(stderr, "Error determining public address") { + Skip( + "plugin was not able to determine public address, this means that all coordinators are probably gone", + ) + } Expect(err).NotTo(HaveOccurred()) // Ensure the cluster is available again. @@ -233,110 +231,63 @@ var _ = Describe("Operator Plugin", Label("e2e", "pr"), func() { // Ensure that the cluster is able to reconcile Expect(remote.WaitForReconciliation(ctx)).To(Succeed()) - log.Println( - "new connection string:", - remote.GetCluster(ctx).Status.ConnectionString, - ) - connectionString, err := fdbv1beta2.ParseConnectionString( - remote.GetCluster(ctx).Status.ConnectionString, - ) - Expect(err).NotTo(HaveOccurred()) - - for _, coordinator := range connectionString.Coordinators { - address, err := fdbv1beta2.ParseProcessAddress(coordinator) - Expect(err).NotTo(HaveOccurred()) - Expect(address.StringAddress).To(BeEmpty()) - } - }) - }) - - When("DNS names in the cluster file are used", func() { - BeforeEach(func(_ SpecContext) { - useDNS = true - }) - - It("should recover the coordinators", func(ctx SpecContext) { - remote := fdbCluster.GetRemote() - // Pick one operator pod and execute the recovery command - operatorPod := factory.RandomPickOnePod( - factory.GetOperatorPods(ctx, remote.Namespace()).Items, - ) - log.Println("operatorPod:", operatorPod.Name) - stdout, stderr, err := factory.ExecuteCmdOnPod( - ctx, - &operatorPod, - "manager", - fmt.Sprintf( - "kubectl-fdb -n %s recover multi-region --version-check=false --wait=false %s", - remote.Namespace(), - remote.Name(), - ), - false, - ) - log.Println("stdout:", stdout, "stderr:", stderr) - if strings.Contains(stderr, "Error determining public address") { - Skip( - "plugin was not able to determine public address, this means that all coordinators are probably gone", - ) + var currentConnectionString string + if shouldUseDNS { + currentConnectionString = remote.GetStatus(ctx).Cluster.ConnectionString + } else { + currentConnectionString = remote.GetCluster(ctx).Status.ConnectionString } - Expect(err).NotTo(HaveOccurred()) - - // Ensure the cluster is available again. - Eventually(func() bool { - return remote.GetStatus(ctx).Client.DatabaseStatus.Available - }).WithTimeout(2 * time.Minute).WithPolling(1 * time.Second).Should(BeTrue()) - - currentConnectionString := remote.GetStatus(ctx).Cluster.ConnectionString log.Println("new connection string:", currentConnectionString) connectionString, err := fdbv1beta2.ParseConnectionString(currentConnectionString) Expect(err).NotTo(HaveOccurred()) for _, coordinator := range connectionString.Coordinators { address, err := fdbv1beta2.ParseProcessAddress(coordinator) - log.Println("address", address) Expect(err).NotTo(HaveOccurred()) - Expect(address.StringAddress).NotTo(BeEmpty()) + if shouldUseDNS { + log.Println("address", address) + Expect(address.StringAddress).NotTo(BeEmpty()) + } else { + Expect(address.StringAddress).To(BeEmpty()) + } } }) - }) + }, + PEntry("DNS is disabled", false), + Entry("DNS is enabled", true), + ) }) When("a majority of coordinators are down in a single dc cluster", func() { - var useDNS bool var fdbCluster *fixtures.FdbCluster AfterEach(func(ctx SpecContext) { - // Delete the cluster. Expect(fdbCluster.Delete(ctx)).NotTo(HaveOccurred()) }) - JustBeforeEach(func(ctx SpecContext) { - clusterConfig = fixtures.DefaultClusterConfig(false) - clusterConfig.UseDNS = ptr.To(useDNS) - fdbCluster = factory.CreateFdbCluster(ctx, clusterConfig) - coordinators := fdbCluster.GetCoordinators(ctx) - minimumFaultDomains := fdbCluster.GetCluster(ctx).MinimumFaultDomains() - downCoordinators := make([]corev1.Pod, 0, minimumFaultDomains) - for _, coordinator := range coordinators { - if len(downCoordinators) >= minimumFaultDomains { - break - } - - downCoordinators = append(downCoordinators, coordinator) - } - - // Set those coordinators as unschedulable to simulate that those coordinators are down. Another option - // would be to create a network partition. - fdbCluster.SetPodsAsUnschedulable(ctx, downCoordinators) - }) - // Default case is to run with DNS enabled. The test case with IPs enabled can run into issues when // the underlying Kubernetes cluster deletes pods. // Because of the above issues the test case is currently disabled (marked as pending) and can be used // to run the test manually if needed. - PWhen("DNS is disabled", func() { - BeforeEach(func(_ SpecContext) { - useDNS = false + DescribeTableSubtree("should recover the coordinators", func(shouldUseDNS bool) { + JustBeforeEach(func(ctx SpecContext) { + clusterConfig = fixtures.DefaultClusterConfig(false) + clusterConfig.UseDNS = ptr.To(shouldUseDNS) + fdbCluster = factory.CreateFdbCluster(ctx, clusterConfig) + coordinators := fdbCluster.GetCoordinators(ctx) + minimumFaultDomains := fdbCluster.GetCluster(ctx).MinimumFaultDomains() + downCoordinators := make([]corev1.Pod, 0, minimumFaultDomains) + for _, coordinator := range coordinators { + if len(downCoordinators) >= minimumFaultDomains { + break + } + + downCoordinators = append(downCoordinators, coordinator) + } + + // Set those coordinators as unschedulable to simulate that those coordinators are down. Another option + // would be to create a network partition. + fdbCluster.SetPodsAsUnschedulable(ctx, downCoordinators) }) It("should recover the coordinators", func(ctx SpecContext) { @@ -357,6 +308,11 @@ var _ = Describe("Operator Plugin", Label("e2e", "pr"), func() { false, ) log.Println("stdout:", stdout, "stderr:", stderr) + if shouldUseDNS && strings.Contains(stderr, "Error determining public address") { + Skip( + "plugin was not able to determine public address, this means that all coordinators are probably gone", + ) + } Expect(err).NotTo(HaveOccurred()) // Ensure the cluster is available again. @@ -382,58 +338,16 @@ var _ = Describe("Operator Plugin", Label("e2e", "pr"), func() { for _, coordinator := range connectionString.Coordinators { address, err := fdbv1beta2.ParseProcessAddress(coordinator) Expect(err).NotTo(HaveOccurred()) - Expect(address.StringAddress).To(BeEmpty()) - } - }) - }) - - When("DNS names in the cluster file are used", func() { - BeforeEach(func(_ SpecContext) { - useDNS = true - }) - - It("should recover the coordinators", func(ctx SpecContext) { - // Pick one operator pod and execute the recovery command - operatorPod := factory.RandomPickOnePod( - factory.GetOperatorPods(ctx, fdbCluster.Namespace()).Items, - ) - log.Println("operatorPod:", operatorPod.Name) - stdout, stderr, err := factory.ExecuteCmdOnPod( - ctx, - &operatorPod, - "manager", - fmt.Sprintf( - "kubectl-fdb -n %s recover single-dc --version-check=false --wait=false %s", - fdbCluster.Namespace(), - fdbCluster.Name(), - ), - false, - ) - log.Println("stdout:", stdout, "stderr:", stderr) - if strings.Contains(stderr, "Error determining public address") { - Skip( - "plugin was not able to determine public address, this means that all coordinators are probably gone", - ) - } - Expect(err).NotTo(HaveOccurred()) - - // Ensure the cluster is available again. - Eventually(func() bool { - return fdbCluster.GetStatus(ctx).Client.DatabaseStatus.Available - }).WithTimeout(2 * time.Minute).WithPolling(1 * time.Second).Should(BeTrue()) - - currentConnectionString := fdbCluster.GetStatus(ctx).Cluster.ConnectionString - log.Println("new connection string:", currentConnectionString) - connectionString, err := fdbv1beta2.ParseConnectionString(currentConnectionString) - Expect(err).NotTo(HaveOccurred()) - - for _, coordinator := range connectionString.Coordinators { - address, err := fdbv1beta2.ParseProcessAddress(coordinator) - log.Println("address", address) - Expect(err).NotTo(HaveOccurred()) - Expect(address.StringAddress).NotTo(BeEmpty()) + if shouldUseDNS { + Expect(address.StringAddress).NotTo(BeEmpty()) + } else { + Expect(address.StringAddress).To(BeEmpty()) + } } }) - }) + }, + PEntry("DNS is disabled", false), + Entry("DNS is enabled", true), + ) }) }) diff --git a/kubectl-fdb/cmd/recover.go b/kubectl-fdb/cmd/recover.go index d77d83a3e..c5dfe61cf 100644 --- a/kubectl-fdb/cmd/recover.go +++ b/kubectl-fdb/cmd/recover.go @@ -27,7 +27,6 @@ import ( "log" "os" "path" - "strings" "time" fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/v2/api/v1beta2" @@ -321,6 +320,9 @@ func restartFdbserverInCluster( return nil } +// checkIfClusterIsUnavailableAndMajorityOfCoordinatorsAreUnreachable checks if the majority of the coordinators are +// unreachable and if the cluster is unavailable. This is a safeguard to reduce the risk of running the recovery +// commands against a healthy cluster. func checkIfClusterIsUnavailableAndMajorityOfCoordinatorsAreUnreachable( ctx context.Context, kubeClient client.Client, @@ -344,15 +346,11 @@ func checkIfClusterIsUnavailableAndMajorityOfCoordinatorsAreUnreachable( break } - time.Sleep(5 * time.Second) - } - - // If DNS is used for the cluster file, we could hit cases where no DNS entry can be resolved, in this case we could - // assume that the cluster is also down. The error from the client side is the following: - // Error: error getting status: Error determining public address. - // ERROR: Unable to bind to network (1512) - if err != nil && strings.Contains(err.Error(), "Error determining public address") { - return err + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(5 * time.Second): + } } return err diff --git a/kubectl-fdb/cmd/recover_single_dc_cluster.go b/kubectl-fdb/cmd/recover_single_dc_cluster.go index 03750531e..70b71e2c1 100644 --- a/kubectl-fdb/cmd/recover_single_dc_cluster.go +++ b/kubectl-fdb/cmd/recover_single_dc_cluster.go @@ -84,7 +84,7 @@ func newRecoverSingleDCClusterCmd(streams genericiooptions.IOStreams) *cobra.Com if wait { confirmed := confirmAction( fmt.Sprintf( - "WARNING:\nThe cluster: %s/%s will be force recovered.\nOnly perform those steps if you are unable to recover the coordinator pods.\nPerforming this action could lead to data loss.\n At leas one coordinator must be active and running to copy the coordinator state.\n", + "WARNING:\nThe cluster: %s/%s will be force recovered.\nOnly perform those steps if you are unable to recover the coordinator pods.\nPerforming this action could lead to data loss.\n At least one coordinator must be active and running to copy the coordinator state.\n", namespace, clusterName, ),