Restart processes only once when doing a setting rollout and include uptime from processes (#2164)

* Restart processes only once when doing a setting rollout and include uptime from processes

Author: johscheuer

controllers/bounce_processes.go

@@ -162,12 +162,18 @@ func (c bounceProcesses) reconcile(_ context.Context, r *FoundationDBClusterReco
 
 	logger.Info("Bouncing processes", "addresses", addresses, "upgrading", upgrading)
 	r.Recorder.Event(cluster, corev1.EventTypeNormal, "BouncingProcesses", fmt.Sprintf("Bouncing processes: %v", addresses))
-	err = adminClient.KillProcesses(addresses)
+	if upgrading {
+		// When upgrading, we want to issue two restart commands to increase the probability that we are restarting all
+		// processes in the cluster.
+		err = adminClient.KillProcessesForUpgrade(addresses)
+	} else {
+		err = adminClient.KillProcesses(addresses)
+	}
 	if err != nil {
 		return &requeue{curError: err}
 	}
 
-	// Reset the SecondsSinceLastRecovered sine the operator just restarted some processes, which will could cause a recovery.
+	// Reset the SecondsSinceLastRecovered since the operator just restarted some processes, which could cause a recovery.
 	status.Cluster.RecoveryState.SecondsSinceLastRecovered = 0.0
 
 	// If the cluster was upgraded we will requeue and let the update_status command set the correct version.

e2e/test_operator/operator_test.go

@@ -2619,4 +2619,48 @@ var _ = Describe("Operator", Label("e2e", "pr"), func() {
 			Expect(factory.GetControllerRuntimeClient().Update(context.Background(), pod)).NotTo(HaveOccurred())
 		})
 	})
+
+	When("a new knob for storage servers is rolled out to the cluster", func() {
+		var initialCustomParameters fdbv1beta2.FoundationDBCustomParameters
+
+		BeforeEach(func() {
+			initialCustomParameters = fdbCluster.GetCustomParameters(
+				fdbv1beta2.ProcessClassStorage,
+			)
+
+			Expect(
+				fdbCluster.SetCustomParameters(
+					fdbv1beta2.ProcessClassStorage,
+					append(
+						initialCustomParameters,
+						"knob_read_sampling_enabled=true",
+					),
+					false,
+				),
+			).NotTo(HaveOccurred())
+		})
+
+		AfterEach(func() {
+			Expect(fdbCluster.SetCustomParameters(
+				fdbv1beta2.ProcessClassStorage,
+				initialCustomParameters,
+				true,
+			)).NotTo(HaveOccurred())
+		})
+
+		It("should update the locality with the substituted environment variable", func() {
+			Eventually(func(g Gomega) bool {
+				for _, process := range fdbCluster.GetStatus().Cluster.Processes {
+					// We change the knob only for storage processes.
+					if process.ProcessClass != fdbv1beta2.ProcessClassStorage {
+						continue
+					}
+
+					g.Expect(process.CommandLine).To(ContainSubstring("knob_read_sampling_enabled"))
+				}
+
+				return true
+			}).WithTimeout(5 * time.Minute).WithPolling(5 * time.Second).Should(BeTrue())
+		})
+	})
 })

fdbclient/admin_client.go

@@ -434,18 +434,43 @@ func (client *cliAdminClient) CanSafelyRemove(addresses []fdbv1beta2.ProcessAddr
 	return fdbstatus.CanSafelyRemoveFromStatus(client.log, client, addresses, status)
 }
 
+func getKillCommand(addresses []fdbv1beta2.ProcessAddress, isUpgrade bool) string {
+	var killCmd strings.Builder
+
+	addrString := fdbv1beta2.ProcessAddressesStringWithoutFlags(addresses, " ")
+	killCmd.WriteString("kill; kill ")
+	killCmd.WriteString(addrString)
+
+	if isUpgrade {
+		killCmd.WriteString("; sleep 1; kill ")
+		killCmd.WriteString(addrString)
+	}
+	killCmd.WriteString("; sleep 5")
+
+	return killCmd.String()
+}
+
 // KillProcesses restarts processes
 func (client *cliAdminClient) KillProcesses(addresses []fdbv1beta2.ProcessAddress) error {
 	if len(addresses) == 0 {
 		return nil
 	}
 
-	killCommand := fmt.Sprintf(
-		"kill; kill %[1]s; sleep 1; kill %[1]s; sleep 5",
-		fdbv1beta2.ProcessAddressesStringWithoutFlags(addresses, " "),
-	)
 	// Run the kill command once with the max timeout to reduce the risk of multiple recoveries happening.
-	_, err := client.runCommand(cliCommand{command: killCommand, timeout: client.getTimeout()})
+	_, err := client.runCommand(cliCommand{command: getKillCommand(addresses, false), timeout: client.getTimeout()})
+
+	return err
+}
+
+// KillProcessesForUpgrade restarts processes for upgrades, this will issue 2 kill commands to make sure all
+// processes are restarted.
+func (client *cliAdminClient) KillProcessesForUpgrade(addresses []fdbv1beta2.ProcessAddress) error {
+	if len(addresses) == 0 {
+		return nil
+	}
+
+	// Run the kill command once with the max timeout to reduce the risk of multiple recoveries happening.
+	_, err := client.runCommand(cliCommand{command: getKillCommand(addresses, true), timeout: client.getTimeout()})
 
 	return err
 }

fdbclient/admin_client_test.go

@@ -1293,4 +1293,19 @@ protocol fdb00b071010000`,
 			})
 		})
 	})
+
+	When("getting the kill command", func() {
+		addresses := []fdbv1beta2.ProcessAddress{{
+			IPAddress: net.ParseIP("192.168.0.2"),
+			Port:      4500,
+		}}
+
+		When("the cluster is upgraded ", func() {
+			Expect(getKillCommand(addresses, true)).To(Equal("kill; kill 192.168.0.2:4500; sleep 1; kill 192.168.0.2:4500; sleep 5"))
+		})
+
+		When("the cluster is not upgraded", func() {
+			Expect(getKillCommand(addresses, false)).To(Equal("kill; kill 192.168.0.2:4500; sleep 5"))
+		})
+	})
 })

pkg/fdbadminclient/admin_client.go

@@ -56,6 +56,10 @@ type AdminClient interface {
 	// KillProcesses restarts processes
 	KillProcesses(addresses []fdbv1beta2.ProcessAddress) error
 
+	// KillProcessesForUpgrade restarts processes for upgrades, this will issue 2 kill commands to make sure all
+	// processes are restarted.
+	KillProcessesForUpgrade(addresses []fdbv1beta2.ProcessAddress) error
+
 	// ChangeCoordinators changes the coordinator set
 	ChangeCoordinators(addresses []fdbv1beta2.ProcessAddress) (string, error)
 

pkg/fdbadminclient/mock/admin_client_mock.go

@@ -702,6 +702,12 @@ func (client *AdminClient) KillProcesses(addresses []fdbv1beta2.ProcessAddress)
 	return nil
 }
 
+// KillProcessesForUpgrade restarts processes for upgrades, this will issue 2 kill commands to make sure all
+// processes are restarted.
+func (client *AdminClient) KillProcessesForUpgrade(addresses []fdbv1beta2.ProcessAddress) error {
+	return client.KillProcesses(addresses)
+}
+
 // ChangeCoordinators changes the coordinator set
 func (client *AdminClient) ChangeCoordinators(addresses []fdbv1beta2.ProcessAddress) (string, error) {
 	adminClientMutex.Lock()

pkg/fdbstatus/status_checks.go

@@ -283,12 +283,10 @@ func GetMinimumUptimeAndAddressMap(logger logr.Logger, cluster *fdbv1beta2.Found
 		return 0, nil, err
 	}
 
-	useRecoveryState := runningVersion.SupportsRecoveryState() && recoveryStateEnabled
-
 	addressMap := make(map[fdbv1beta2.ProcessGroupID][]fdbv1beta2.ProcessAddress, len(status.Cluster.Processes))
 
 	minimumUptime := math.Inf(1)
-	if useRecoveryState {
+	if runningVersion.SupportsRecoveryState() && recoveryStateEnabled {
 		minimumUptime = status.Cluster.RecoveryState.SecondsSinceLastRecovered
 	}
 
@@ -307,7 +305,7 @@ func GetMinimumUptimeAndAddressMap(logger logr.Logger, cluster *fdbv1beta2.Found
 
 		addressMap[fdbv1beta2.ProcessGroupID(processGroupID)] = append(addressMap[fdbv1beta2.ProcessGroupID(process.Locality[fdbv1beta2.FDBLocalityInstanceIDKey])], process.Address)
 
-		if useRecoveryState || process.Excluded {
+		if process.Excluded {
 			continue
 		}
 
@@ -317,7 +315,10 @@ func GetMinimumUptimeAndAddressMap(logger logr.Logger, cluster *fdbv1beta2.Found
 			continue
 		}
 
+		// If the SecondsSinceLastRecovered is higher than the process uptime, we want to use the process uptime, e.g. in
+		// cases where only storage processes are restarted.
 		if process.UptimeSeconds < minimumUptime {
+			logger.V(1).Info("Process uptime is less than the last recovery", "processGroupID", process.Address, "minimumUptime", minimumUptime, "process.UptimeSeconds", process.UptimeSeconds)
 			minimumUptime = process.UptimeSeconds
 		}
 	}

pkg/fdbstatus/status_checks_test.go

@@ -842,7 +842,7 @@ var _ = Describe("status_checks", func() {
 					},
 				},
 			}),
-		Entry("when recovered since is enabled and version supports it",
+		Entry("when recovered since is enabled and version supports it, it should use the minimum of recovered since or process uptime",
 			&fdbv1beta2.FoundationDBCluster{
 				Spec: fdbv1beta2.FoundationDBClusterSpec{
 					Version: fdbv1beta2.Versions.SupportsRecoveryState.String(),
@@ -866,7 +866,7 @@ var _ = Describe("status_checks", func() {
 				},
 			},
 			true,
-			90.0,
+			30.0,
 			map[fdbv1beta2.ProcessGroupID][]fdbv1beta2.ProcessAddress{
 				"test": {
 					{