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

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())
+	}
 }