fix(dataset): recreate the ThinRuntime of a reference dataset stuck in NotBound

A reference dataset (spec.mounts[0].mountPoint: dataset://<ns>/<name>) could stay
in NotBound forever, without any ThinRuntime and without any error being surfaced.
Two independent gaps combined to wedge it:

1. CreateRuntimeForReferenceDatasetIfNotExist treated a terminating ThinRuntime as
   an existing one. GetThinRuntime is a plain Get, so it also returns objects which
   already have a deletionTimestamp, e.g. a leftover runtime whose controller was
   scaled to 0 and could not remove its finalizer. The helper concluded that the
   runtime already exists, tried to adopt the dying object by overwriting its
   ownerReferences and returned success, so no runtime was ever created once the
   object finally disappeared. A runtime which is being deleted is now reported as
   an error (which is not a conflict error, so it is not swallowed by the
   surrounding retry.RetryOnConflict), letting the caller requeue until the object
   is really gone and a fresh runtime can be created.

2. The dataset controller did not watch the ThinRuntime it creates, so nothing
   re-triggered the dataset after the runtime vanished. The controller now Owns()
   the ThinRuntime.

The owner based watch resolves a dependent through the kind and apiVersion of its
ownerReference. Both were read from the dataset's TypeMeta, which a typed client
may hand back empty, so they now fall back to the well-known values of the Dataset
CRD.

The chart role for the dataset controller already allows watching thinruntimes and
the generated config/rbac/role.yaml is unchanged by the added kubebuilder marker.

While verifying the change, the pre-existing spec "should requeue after the resync
period when runtime scaleout returns an error" turned out to be flaky (~1% of runs,
depending on binary layout): it reset the patch installed by BeforeEach and applied
a second patch on the very same function, and gomonkey leaves the first patch in
effect in a measurable share of such reset-then-reapply sequences (2217 out of
20000 iterations of an isolated probe). The spec then observed a successful scaleout
and no requeue. It now switches behaviour through a variable which the single patch
delegates to, instead of patching twice; 500 consecutive runs are green.

Fixes #6136

Signed-off-by: cheyang <cheyang@163.com>

Author: cheyang

pkg/controllers/v1alpha1/dataset/dataset_controller.go

@@ -66,6 +66,7 @@ type reconcileRequestContext struct {
 
 // +kubebuilder:rbac:groups=data.fluid.io,resources=datasets,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=data.fluid.io,resources=datasets/status,verbs=get;update;patch
+// +kubebuilder:rbac:groups=data.fluid.io,resources=thinruntimes,verbs=get;list;watch;create;update;patch;delete
 
 func (r *DatasetReconciler) Reconcile(context context.Context, req ctrl.Request) (ctrl.Result, error) {
 	ctx := reconcileRequestContext{
@@ -266,6 +267,9 @@ func (r *DatasetReconciler) SetupWithManager(mgr ctrl.Manager, options controlle
 	return ctrl.NewControllerManagedBy(mgr).
 		WithOptions(options).
 		For(&datav1alpha1.Dataset{}).
+		// Watch the ThinRuntime created for a reference dataset, so that the owning dataset is
+		// reconciled again (and the runtime is re-created) once the runtime is deleted or lost.
+		Owns(&datav1alpha1.ThinRuntime{}).
 		Complete(r)
 }
 

pkg/controllers/v1alpha1/dataset/dataset_controller_ut_test.go

@@ -42,14 +42,21 @@ import (
 var _ = Describe("Dataset Controller Unit", func() {
 	var scheme *runtime.Scheme
 	var scaleoutPatch *gomonkey.Patches
+	// scaleoutResult is what the patched deploy.ScaleoutRuntimeControllerOnDemand returns. A spec which
+	// needs another result assigns this variable instead of patching the function a second time, because
+	// resetting and re-applying a patch on the very same function is not reliable.
+	var scaleoutResult func() (string, bool, error)
 
 	BeforeEach(func() {
 		scheme = runtime.NewScheme()
 		Expect(datav1alpha1.AddToScheme(scheme)).NotTo(HaveOccurred())
 		Expect(corev1.AddToScheme(scheme)).To(Succeed())
+		scaleoutResult = func() (string, bool, error) {
+			return "", false, nil
+		}
 		scaleoutPatch = gomonkey.ApplyFunc(deploy.ScaleoutRuntimeControllerOnDemand,
 			func(client.Client, types.NamespacedName, logr.Logger) (string, bool, error) {
-				return "", false, nil
+				return scaleoutResult()
 			})
 	})
 
@@ -209,11 +216,9 @@ var _ = Describe("Dataset Controller Unit", func() {
 		})
 
 		It("should requeue after the resync period when runtime scaleout returns an error", func() {
-			scaleoutPatch.Reset()
-			scaleoutPatch = gomonkey.ApplyFunc(deploy.ScaleoutRuntimeControllerOnDemand,
-				func(client.Client, types.NamespacedName, logr.Logger) (string, bool, error) {
-					return "", false, fmt.Errorf("scaleout failed")
-				})
+			scaleoutResult = func() (string, bool, error) {
+				return "", false, fmt.Errorf("scaleout failed")
+			}
 
 			dataset := &datav1alpha1.Dataset{
 				ObjectMeta: metav1.ObjectMeta{

pkg/controllers/v1alpha1/dataset/dataset_reconciler_test.go

@@ -251,6 +251,81 @@ var _ = Describe("DatasetReconciler (fake client)", func() {
 			Expect(result).To(Equal(ctrl.Result{}))
 		})
 
+		It("creates the ThinRuntime with a controller ownerReference the dataset controller can watch", func() {
+			// The dataset controller watches the ThinRuntime it creates via Owns(), which resolves
+			// the owner by the kind and the apiVersion of the ownerReference, so both must be set.
+			ds := datav1alpha1.Dataset{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:       "ref-ds-owner",
+					Namespace:  "default",
+					UID:        types.UID("ref-ds-owner-uid"),
+					Finalizers: []string{finalizer},
+				},
+				Spec: datav1alpha1.DatasetSpec{
+					Mounts: []datav1alpha1.Mount{
+						{Name: "m1", MountPoint: "dataset://default/physical-ds"},
+					},
+				},
+				Status: datav1alpha1.DatasetStatus{Phase: datav1alpha1.NotBoundDatasetPhase},
+			}
+			r := newTestReconciler(&ds)
+			ctx := makeReconcileCtx(r, ds)
+
+			_, err := r.reconcileDataset(ctx, false)
+			Expect(err).NotTo(HaveOccurred())
+
+			thinRuntime := &datav1alpha1.ThinRuntime{}
+			Expect(r.Get(ctx, types.NamespacedName{Namespace: "default", Name: "ref-ds-owner"}, thinRuntime)).To(Succeed())
+			Expect(thinRuntime.OwnerReferences).To(HaveLen(1))
+			Expect(thinRuntime.OwnerReferences[0].Kind).To(Equal(datav1alpha1.Datasetkind))
+			Expect(thinRuntime.OwnerReferences[0].APIVersion).To(Equal(datav1alpha1.GroupVersion.String()))
+			Expect(thinRuntime.OwnerReferences[0].UID).To(Equal(ds.UID))
+			Expect(thinRuntime.OwnerReferences[0].Controller).NotTo(BeNil())
+			Expect(*thinRuntime.OwnerReferences[0].Controller).To(BeTrue())
+		})
+
+		It("returns error when the ThinRuntime of the reference dataset is still terminating", func() {
+			// A leftover ThinRuntime stuck in Terminating must not be treated as a usable runtime:
+			// the reconcile has to fail so that it is retried once the object is really gone,
+			// instead of leaving the dataset silently without any runtime.
+			now := metav1.Now()
+			ds := datav1alpha1.Dataset{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:       "ref-ds-terminating",
+					Namespace:  "default",
+					UID:        types.UID("ref-ds-terminating-uid"),
+					Finalizers: []string{finalizer},
+				},
+				Spec: datav1alpha1.DatasetSpec{
+					Mounts: []datav1alpha1.Mount{
+						{Name: "m1", MountPoint: "dataset://default/physical-ds"},
+					},
+				},
+				Status: datav1alpha1.DatasetStatus{Phase: datav1alpha1.NotBoundDatasetPhase},
+			}
+			// The finalizer keeps the terminating runtime in the fake client's tracker.
+			terminatingRuntime := datav1alpha1.ThinRuntime{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "ref-ds-terminating",
+					Namespace:         "default",
+					DeletionTimestamp: &now,
+					Finalizers:        []string{"thin-runtime-controller-finalizer"},
+				},
+			}
+			r := newTestReconciler(&ds, &terminatingRuntime)
+			ctx := makeReconcileCtx(r, ds)
+
+			result, err := r.reconcileDataset(ctx, false)
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("terminating"))
+			Expect(result).To(Equal(ctrl.Result{}))
+
+			// The terminating runtime must not be adopted by the dataset.
+			stored := &datav1alpha1.ThinRuntime{}
+			Expect(r.Get(ctx, types.NamespacedName{Namespace: "default", Name: "ref-ds-terminating"}, stored)).To(Succeed())
+			Expect(stored.OwnerReferences).To(BeEmpty())
+		})
+
 		It("returns error when CreateRuntimeForReferenceDatasetIfNotExist fails", func() {
 			ds := datav1alpha1.Dataset{
 				ObjectMeta: metav1.ObjectMeta{

pkg/utils/dataset_runtime.go

@@ -18,6 +18,7 @@ package utils
 
 import (
 	"context"
+	"fmt"
 	"reflect"
 
 	datav1alpha1 "github.com/fluid-cloudnative/fluid/api/v1alpha1"
@@ -41,6 +42,29 @@ func GetRuntimeByCategory(runtimes []datav1alpha1.Runtime, category common.Categ
 	return -1, nil
 }
 
+// datasetControllerOwnerReference builds the controller ownerReference which points to the given dataset.
+// Kind and APIVersion come from the dataset's TypeMeta, and fall back to the well-known values of the Dataset
+// CRD when it is empty, which a typed client may hand back depending on how the object was read. The owner
+// based watch of the dataset controller resolves a dependent through those two fields, so they must be set.
+func datasetControllerOwnerReference(dataset *datav1alpha1.Dataset) metav1.OwnerReference {
+	kind := dataset.GetObjectKind().GroupVersionKind().Kind
+	if len(kind) == 0 {
+		kind = datav1alpha1.Datasetkind
+	}
+	apiVersion := dataset.APIVersion
+	if len(apiVersion) == 0 {
+		apiVersion = datav1alpha1.GroupVersion.String()
+	}
+
+	return metav1.OwnerReference{
+		Kind:       kind,
+		APIVersion: apiVersion,
+		Name:       dataset.GetName(),
+		UID:        dataset.GetUID(),
+		Controller: ptr.To(true),
+	}
+}
+
 // CreateRuntimeForReferenceDatasetIfNotExist creates runtime for ReferenceDataset
 func CreateRuntimeForReferenceDatasetIfNotExist(client client.Client, dataset *datav1alpha1.Dataset) (err error) {
 	err = retry.RetryOnConflict(retry.DefaultBackoff, func() error {
@@ -49,15 +73,17 @@ func CreateRuntimeForReferenceDatasetIfNotExist(client client.Client, dataset *d
 			dataset.GetNamespace())
 		// 1. if err is null, which indicates that the runtime exists, then return
 		if err == nil {
+			// 1.1 The runtime is being deleted, it can neither be adopted nor be re-created for now.
+			// Return an error (not a conflict error, so retry.RetryOnConflict won't swallow it) to
+			// let the caller requeue until the terminating runtime is really gone.
+			if HasDeletionTimestamp(runtime.ObjectMeta) {
+				return fmt.Errorf("the thinRuntime %s/%s is terminating, wait for it to be deleted before creating a new one for the reference dataset",
+					runtime.GetNamespace(), runtime.GetName())
+			}
+
 			runtimeToUpdate := runtime.DeepCopy()
 			runtimeToUpdate.SetOwnerReferences([]metav1.OwnerReference{
-				{
-					Kind:       dataset.GetObjectKind().GroupVersionKind().Kind,
-					APIVersion: dataset.APIVersion,
-					Name:       dataset.GetName(),
-					UID:        dataset.GetUID(),
-					Controller: ptr.To(true),
-				}})
+				datasetControllerOwnerReference(dataset)})
 			if !reflect.DeepEqual(runtimeToUpdate, runtime) {
 				err = client.Update(context.TODO(), runtimeToUpdate)
 				return err
@@ -72,13 +98,7 @@ func CreateRuntimeForReferenceDatasetIfNotExist(client client.Client, dataset *d
 					Name:      dataset.Name,
 					Namespace: dataset.Namespace,
 					OwnerReferences: []metav1.OwnerReference{
-						{
-							Kind:       dataset.GetObjectKind().GroupVersionKind().Kind,
-							APIVersion: dataset.APIVersion,
-							Name:       dataset.GetName(),
-							UID:        dataset.GetUID(),
-							Controller: ptr.To(true),
-						},
+						datasetControllerOwnerReference(dataset),
 					},
 					Labels: map[string]string{
 						common.LabelAnnotationDatasetId: GetDatasetId(dataset.GetNamespace(), dataset.GetName(), string(dataset.GetUID())),

pkg/utils/dataset_runtime_test.go

@@ -84,6 +84,7 @@ func mockThreeRuntimes(index int, category common.Category) []datav1alpha1.Runti
 
 func TestCreateRuntimeForReferenceDatasetIfNotExist(t *testing.T) {
 
+	deletionTimestamp := v1.Now()
 	thinRuntimes := []*datav1alpha1.ThinRuntime{
 		{
 			ObjectMeta: v1.ObjectMeta{
@@ -104,6 +105,16 @@ func TestCreateRuntimeForReferenceDatasetIfNotExist(t *testing.T) {
 				Name:      "ThinRuntimeExistWithOwnerReference",
 				Namespace: "default",
 			},
+		}, {
+			// A leftover runtime which is stuck in Terminating, e.g. because its controller is
+			// scaled to 0 and can not remove the finalizer. The finalizer is also required to keep
+			// the object in the fake client's tracker once a deletionTimestamp is set.
+			ObjectMeta: v1.ObjectMeta{
+				Name:              "ThinRuntimeTerminating",
+				Namespace:         "default",
+				DeletionTimestamp: &deletionTimestamp,
+				Finalizers:        []string{"thin-runtime-controller-finalizer"},
+			},
 		},
 	}
 	objs := []runtime.Object{}
@@ -148,6 +159,18 @@ func TestCreateRuntimeForReferenceDatasetIfNotExist(t *testing.T) {
 				},
 			},
 			wantErr: false,
+		}, {
+			// The runtime of the same name is still terminating, it can neither be adopted nor be
+			// re-created, so an error is expected to make the caller requeue.
+			name: "ThinRuntimeTerminating",
+			dataset: &datav1alpha1.Dataset{
+				ObjectMeta: v1.ObjectMeta{
+					Name:      "ThinRuntimeTerminating",
+					Namespace: "default",
+					UID:       "5b7bd2c9-e6e8-4c1e-9c9a-5d2b6ff6bd11",
+				},
+			},
+			wantErr: true,
 		},
 	}
 	for _, tt := range tests {
@@ -157,4 +180,16 @@ func TestCreateRuntimeForReferenceDatasetIfNotExist(t *testing.T) {
 			}
 		})
 	}
+
+	// The terminating runtime must be left untouched, especially it must not be adopted by the dataset.
+	terminatingRuntime, err := GetThinRuntime(fakeClient, "ThinRuntimeTerminating", "default")
+	if err != nil {
+		t.Fatalf("failed to get the terminating thinRuntime: %v", err)
+	}
+	if !HasDeletionTimestamp(terminatingRuntime.ObjectMeta) {
+		t.Errorf("expected the thinRuntime ThinRuntimeTerminating to be still terminating")
+	}
+	if len(terminatingRuntime.GetOwnerReferences()) != 0 {
+		t.Errorf("expected no ownerReference set on the terminating thinRuntime, but got %v", terminatingRuntime.GetOwnerReferences())
+	}
 }