fix: Do not wait for all pods to be ready in dataloss recovery (#259)

When recovering after dataloss, there might be other brokers which are
not yet recovered. So we cannot wait for all brokers to be ready.

go-chaos/cmd/dataloss_sim.go

@@ -17,6 +17,7 @@ package cmd
 import (
 	"github.com/spf13/cobra"
 	"github.com/zeebe-io/zeebe-chaos/go-chaos/internal"
+	"time"
 )
 
 func init() {
@@ -103,10 +104,20 @@ var datalossRecover = &cobra.Command{
 			panic(err)
 		}
 
-		err = k8Client.AwaitReadiness()
+		pod, err := internal.GetBrokerPodForNodeId(k8Client, int32(nodeId))
+
+		if err != nil {
+			internal.LogInfo("Failed to get pod with nodeId %d %s", nodeId, err)
+			panic(err)
+		}
+
+		// The pod is restarting after dataloss, so it takes longer to be ready
+		err = k8Client.AwaitPodReadiness(pod.Name, 10*time.Minute)
+
 		if err != nil {
 			internal.LogInfo("%s", err)
+			panic(err)
 		}
-		internal.LogInfo("Restarted broker %d", nodeId)
+		internal.LogInfo("Broker %d is recovered", nodeId)
 	},
 }

go-chaos/internal/dataloss_sim_helper.go

@@ -73,7 +73,7 @@ func (c K8Client) ApplyInitContainerPatch() error {
               "-c"
             ],
             "args": [
-              "while true; do block=$(cat /etc/config/block_${K8S_NAME##*-}); if [ $block == \"false\" ]; then break; fi; echo "Startup is blocked."; sleep 10; done"
+              "while true; do block=$(cat /etc/config/block_${K8S_NAME##*-}); if [ $block == \"false\" ]; then break; fi; echo \"Startup is blocked.\"; sleep 10; done"
             ],
             "env": [
               {

go-chaos/internal/pods.go

@@ -184,6 +184,29 @@ func (c K8Client) checkIfGatewaysAreRunning() (bool, error) {
 	return true, nil
 }
 
+func (c K8Client) AwaitPodReadiness(podName string, timeout time.Duration) error {
+	timedOut := time.After(timeout)
+	ticker := time.Tick(1 * time.Second)
+
+	// Keep checking until we're timed out
+	for {
+		select {
+		case <-timedOut:
+			return errors.New(fmt.Sprintf("Pod %s is not ready with in given timeout %v", podName, timeout))
+		case <-ticker:
+			// check status of pod on every tick (1 second)
+			pod, err := c.Clientset.CoreV1().Pods(c.GetCurrentNamespace()).Get(context.TODO(), podName, metav1.GetOptions{})
+			if err != nil {
+				LogVerbose("Failed to get pod %s. Will retry", pod.Name)
+			} else if pod.Status.ContainerStatuses[0].Ready { // assuming there is only one container
+				return nil
+			} else {
+				LogVerbose("Pod %s is in phase %s, but not ready. Wait for some seconds", pod.Name, pod.Status.Phase)
+			}
+		}
+	}
+}
+
 func (c K8Client) RestartPod(podName string) error {
 	return c.Clientset.CoreV1().Pods(c.GetCurrentNamespace()).Delete(context.TODO(), podName, metav1.DeleteOptions{})
 }