Add mechanism to avoid to be stuck waiting for rpaasvalidation (#158)

* Add mechanism to avoid to be stuck waiting for rpaasvalidation * Use Revision to detect object changes
tsuru · Sep 9, 2024 · 1490b33 · 1490b33
1 parent ac8e627
commit 1490b33
Show file tree

Hide file tree

Showing 3 changed files with 161 additions and 8 deletions.
diff --git a/controllers/validation_controller.go b/controllers/validation_controller.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"path/filepath"
+	"time"
 
 	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
@@ -45,12 +46,7 @@ func (r *RpaasValidationReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 		return reconcile.Result{}, err
 	}
 
-	validationHash, err := generateSpecHash(&validation.Spec)
-	if err != nil {
-		return reconcile.Result{}, err
-	}
-
-	if validation.Status.RevisionHash == validationHash && validation.Status.Valid != nil {
+	if validation.Status.ObservedGeneration == validation.ObjectMeta.Generation && validation.Status.Valid != nil {
 		return reconcile.Result{}, nil
 	}
 
@@ -91,6 +87,11 @@ func (r *RpaasValidationReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 		return reconcile.Result{}, err
 	}
 
+	validationHash, err := generateSpecHash(&validation.Spec)
+	if err != nil {
+		return reconcile.Result{}, err
+	}
+
 	pod := newValidationPod(validationMergedWithFlavors, validationHash, plan, configMap)
 
 	existingPod, err := r.getPod(ctx, pod.Namespace, pod.Name)
@@ -110,11 +111,18 @@ func (r *RpaasValidationReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 		}
 	}
 
-	_, err = r.reconcilePod(ctx, pod)
+	hasChanged, err := r.reconcilePod(ctx, pod)
 	if err != nil {
 		return ctrl.Result{}, err
 	}
 
+	if !hasChanged {
+		return ctrl.Result{
+			Requeue:      true,
+			RequeueAfter: time.Second * 10,
+		}, nil
+	}
+
 	return ctrl.Result{}, nil
 }
 

diff --git a/internal/pkg/rpaas/validation/manager.go b/internal/pkg/rpaas/validation/manager.go
@@ -6,6 +6,7 @@ package validation
 import (
 	"context"
 	"fmt"
+	"reflect"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
@@ -333,14 +334,24 @@ func (v *validationManager) waitController(ctx context.Context, validation *v1al
 			return err
 		}
 	} else {
+		isLastStatus := existingValidation.Generation == existingValidation.Status.ObservedGeneration
+
+		if reflect.DeepEqual(existingValidation.Spec, validation.Spec) && existingValidation.Status.Valid != nil && isLastStatus {
+			if *existingValidation.Status.Valid {
+				return nil
+			}
+
+			return &rpaas.ValidationError{Msg: existingValidation.Status.Error}
+		}
+
 		validation.ObjectMeta.ResourceVersion = existingValidation.ResourceVersion
 		err = v.cli.Update(ctx, validation, &client.UpdateOptions{})
 		if err != nil {
 			return err
 		}
 	}
 
-	maxRetries := 30
+	maxRetries := 60
 
 	for retry := 0; retry < maxRetries; retry++ {
 		existingValidation = v1alpha1.RpaasValidation{}

diff --git a/internal/pkg/rpaas/validation/manager_test.go b/internal/pkg/rpaas/validation/manager_test.go
@@ -70,6 +70,82 @@ func TestUpdateBlock(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestUpdateBlockWithDelayedPodStart(t *testing.T) {
+	block := rpaas.ConfigurationBlock{
+		Name:    "http",
+		Content: "blah;",
+	}
+
+	cli := clientFake.NewClientBuilder().WithScheme(newScheme()).WithRuntimeObjects().Build()
+
+	baseManager := &fake.RpaasManager{
+		FakeUpdateBlock: func(instance string, updateBlock rpaas.ConfigurationBlock) error {
+			assert.Equal(t, instance, "blah")
+			assert.Equal(t, block, updateBlock)
+			return nil
+		},
+		FakeGetInstance: func(instanceName string) (*v1alpha1.RpaasInstance, error) {
+			return &v1alpha1.RpaasInstance{
+				ObjectMeta: v1.ObjectMeta{
+					Namespace: "default",
+					Name:      instanceName,
+				},
+				Spec: v1alpha1.RpaasInstanceSpec{},
+			}, nil
+		},
+	}
+
+	stop := fakeValidationControllerWithGenerationControl(cli, true, "")
+	defer stop()
+
+	validationMngr := New(baseManager, cli)
+
+	err := validationMngr.UpdateBlock(context.TODO(), "blah", block)
+	require.NoError(t, err)
+
+	err = validationMngr.UpdateBlock(context.TODO(), "blah", block)
+	require.NoError(t, err)
+}
+
+func TestUpdateBlockWithDelayedPodStartAndError(t *testing.T) {
+	block := rpaas.ConfigurationBlock{
+		Name:    "http",
+		Content: "blah;",
+	}
+
+	cli := clientFake.NewClientBuilder().WithScheme(newScheme()).WithRuntimeObjects().Build()
+
+	baseManager := &fake.RpaasManager{
+		FakeUpdateBlock: func(instance string, updateBlock rpaas.ConfigurationBlock) error {
+			assert.Equal(t, instance, "blah")
+			assert.Equal(t, block, updateBlock)
+			return nil
+		},
+		FakeGetInstance: func(instanceName string) (*v1alpha1.RpaasInstance, error) {
+			return &v1alpha1.RpaasInstance{
+				ObjectMeta: v1.ObjectMeta{
+					Namespace: "default",
+					Name:      instanceName,
+				},
+				Spec: v1alpha1.RpaasInstanceSpec{},
+			}, nil
+		},
+	}
+
+	stop := fakeValidationControllerWithGenerationControl(cli, false, "error")
+	defer stop()
+
+	validationMngr := New(baseManager, cli)
+
+	err := validationMngr.UpdateBlock(context.TODO(), "blah", block)
+	assert.NotNil(t, err)
+	assert.Equal(t, "error", err.Error())
+
+	err = validationMngr.UpdateBlock(context.TODO(), "blah", block)
+	assert.NotNil(t, err)
+	assert.Equal(t, "error", err.Error())
+}
+
 func TestUpdateBlockControllerError(t *testing.T) {
 	block := rpaas.ConfigurationBlock{
 		Name:    "http",
@@ -413,3 +489,61 @@ func fakeValidationController(cli client.Client, valid bool, errorMesssage strin
 
 	return stop
 }
+
+func fakeValidationControllerWithGenerationControl(cli client.Client, valid bool, errorMesssage string) (stop func()) {
+	var stopped int32
+	stop = func() {
+		atomic.StoreInt32(&stopped, 1)
+	}
+
+	list := v1alpha1.RpaasValidationList{}
+	err := cli.List(context.Background(), &list, &client.ListOptions{})
+	if err != nil {
+		fmt.Println("stop controller", err)
+	}
+
+	for _, item := range list.Items {
+		item.Generation = 1
+
+		cli.Update(context.Background(), &item)
+		if err != nil {
+			fmt.Println("failed to update", err)
+		}
+	}
+
+	time.Sleep(time.Millisecond * 100)
+
+	go func() {
+		for {
+			list := v1alpha1.RpaasValidationList{}
+			err := cli.List(context.Background(), &list, &client.ListOptions{})
+			if err != nil {
+				fmt.Println("stop controller", err)
+			}
+
+		itemsLoop:
+			for _, item := range list.Items {
+				if item.Status.Valid != nil {
+					continue itemsLoop
+				}
+
+				item.Status.Valid = &valid
+				item.Status.Error = errorMesssage
+				item.Status.ObservedGeneration = item.Generation
+
+				cli.Update(context.Background(), &item)
+				if err != nil {
+					fmt.Println("stop controller", err)
+				}
+			}
+
+			if atomic.LoadInt32(&stopped) == 1 {
+				break
+			}
+
+			time.Sleep(time.Millisecond * 100)
+		}
+	}()
+
+	return stop
+}