Re-create missing pod which has incompleted downscale operation

pkg/k8sutil/status.go

@@ -23,6 +23,7 @@ import (
 	"emperror.dev/errors"
 	"github.com/go-logr/logr"
 
+	"github.com/lestrrat-go/backoff/v2"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -46,33 +47,36 @@ func IsMarkedForDeletion(m metav1.ObjectMeta) bool {
 
 // UpdateBrokerConfigurationBackup updates the broker status with a backup from kafka broker configurations
 func UpdateBrokerConfigurationBackup(c client.Client, cluster *banzaicloudv1beta1.KafkaCluster) error {
-	needsUpdate, err := generateBrokerConfigurationBackups(cluster)
-	if err != nil {
-		return err
-	}
-	if !needsUpdate {
-		return nil
-	}
-	ctx := context.Background()
-	if err := c.Status().Update(ctx, cluster); err != nil {
-		if !apierrors.IsConflict(err) {
-			return errors.WrapIff(err, "could not update Kafka broker(s) configuration backup state")
-		}
-		if err := c.Get(ctx, types.NamespacedName{Namespace: cluster.Namespace, Name: cluster.Name}, cluster); err != nil {
-			return errors.WrapIf(err, "could not get config for updating status")
-		}
+	p := backoff.Constant(
+		backoff.WithInterval(time.Millisecond*100),
+		backoff.WithMaxRetries(3),
+	)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	b := p.Start(ctx)
+	var errAPI error
+	for backoff.Continue(b) {
 		needsUpdate, err := generateBrokerConfigurationBackups(cluster)
 		if err != nil {
 			return err
 		}
 		if !needsUpdate {
 			return nil
 		}
-		if err = c.Status().Update(ctx, cluster); err != nil {
-			return errors.WrapIff(err, "could not update Kafka broker(s) configuration backup state")
+		errAPI = c.Status().Update(ctx, cluster)
+		if err != nil {
+			if apierrors.IsConflict(errAPI) {
+				if err := c.Get(ctx, types.NamespacedName{Namespace: cluster.Namespace, Name: cluster.Name}, cluster); err != nil {
+					return errors.WrapIf(err, "could not get config for updating status")
+				}
+			}
+		} else {
+			return nil
 		}
 	}
-	return nil
+
+	return errAPI
 }
 
 func generateBrokerConfigurationBackups(cluster *banzaicloudv1beta1.KafkaCluster) (bool, error) {

pkg/util/backoff/backoff_test.go

@@ -15,59 +15,69 @@
 package backoff
 
 import (
+	"context"
 	"errors"
 	"testing"
 	"time"
 
-	"github.com/lestrrat-go/backoff"
+	"github.com/lestrrat-go/backoff/v2"
 )
 
-func TestRetry(t *testing.T) {
-	count := 0
-	config := &ConstantBackoffConfig{
-		Delay:          time.Duration(1) * time.Second,
-		MaxElapsedTime: time.Duration(10) * time.Second,
-		MaxRetries:     3,
-	}
-	policy := NewConstantBackoffPolicy(config)
-
-	err := Retry(func() error {
-		count++
-		return errors.New("err")
-	}, policy)
-	if err == nil {
-		t.Error("Expected error, got nil")
-	}
+type MyError struct {
+	error
+	number int
+}
 
-	if count != 4 {
-		t.Error("Expected function to run 4 times, got:", count)
+func TestExponentialBackOff(t *testing.T) {
+	p := backoff.Constant(
+		backoff.WithInterval(time.Millisecond*10),
+		backoff.WithMaxRetries(2),
+	)
+	flakyFunc := func(a int) (int, error) {
+		switch {
+		case a%3 == 0:
+			return 3, errors.New("error 3")
+		case a%4 == 0:
+			return 4, nil
+		case a%5 == 0:
+			return 5, MyError{number: 5}
+		}
+		return 0, errors.New(`invalid`)
 	}
 
-	count = 0
-	err = Retry(func() error {
-		count++
-		return nil
-	}, policy)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
 
-	if err != nil {
-		t.Error("Expected nil, got error")
+	startNum := 3
+	retryFunc := func() int {
+		b := p.Start(ctx)
+		var err error
+		var x int
+		for backoff.Continue(b) {
+			x, err = flakyFunc(startNum)
+			switch startNum {
+			case 3:
+				if x != 3 || err.Error() != "error 3" {
+					t.Errorf("backOff error 3")
+				}
+			case 4:
+				if x != 4 || err != nil {
+					t.Errorf("backOff error 4")
+				}
+			case 5:
+				if x != 5 || !errors.Is(err, MyError{number: 5}) {
+					t.Errorf("backOff error 5")
+				}
+			default:
+				t.Errorf("more retry then maxRetry")
+			}
+			startNum++
+		}
+		return startNum
 	}
 
-	if count != 1 {
-		t.Error("Expected function to run once, got:", count)
+	if ret := retryFunc(); ret != 6 {
+		t.Errorf("not enough iteration")
 	}
 
-	count = 0
-	err = Retry(func() error {
-		count++
-		return MarkErrorPermanent(errors.New("permanent"))
-	}, policy)
-	switch {
-	case err == nil:
-		t.Error("Expected permanent error, got nil")
-	case !backoff.IsPermanentError(err):
-		t.Error("Expected true for is permanent error, got false")
-	case count != 1:
-		t.Error("Expected function to only try once, got:", count)
-	}
 }