diff --git a/pkg/ddc/cache/engine/dataset.go b/pkg/ddc/cache/engine/dataset.go index 35abdd1dd0a..8ebcc5a11ec 100644 --- a/pkg/ddc/cache/engine/dataset.go +++ b/pkg/ddc/cache/engine/dataset.go @@ -39,14 +39,28 @@ func (e *CacheEngine) BindToDataset(runtime *datav1alpha1.CacheRuntime, runtimeC return } - return e.UpdateDatasetStatus(datav1alpha1.BoundDatasetPhase, runtime, runtimeClass) + return e.UpdateDatasetStatus(datav1alpha1.BoundDatasetPhase, runtime, runtimeClass, true) } -func (e *CacheEngine) UpdateDatasetStatus(phase datav1alpha1.DatasetPhase, runtime *datav1alpha1.CacheRuntime, runtimeClass *datav1alpha1.CacheRuntimeClass) (err error) { +// UpdateDatasetStatus transitions the Dataset to phase. fetchCacheStates controls whether it +// execs into the master pod for fresh cache states (only relevant for BoundDatasetPhase); the +// caller decides this so that rate-limiting via permitSync stays a concern of Sync alone. +func (e *CacheEngine) UpdateDatasetStatus(phase datav1alpha1.DatasetPhase, runtime *datav1alpha1.CacheRuntime, runtimeClass *datav1alpha1.CacheRuntimeClass, fetchCacheStates bool) (err error) { var cacheStates common.CacheStateList - // only update cache states for BoundDatasetPhase - if phase == datav1alpha1.BoundDatasetPhase { + current, err := utils.GetDataset(e.Client, e.name, e.namespace) + if err != nil { + return err + } + if current.Status.Phase == phase { + // already in the desired phase, nothing to do + return nil + } + + // GetCacheStates execs into the master pod with a floor of MinExecutionTimeoutSeconds, + // so callers keep it behind the same rate limiter that bounds other engine RPCs, and it's + // only attempted for BoundDatasetPhase. + if phase == datav1alpha1.BoundDatasetPhase && fetchCacheStates { e.Log.V(1).Info("Start to update cache states") cacheStates, err = e.GetCacheStates(runtime, runtimeClass) if err != nil { diff --git a/pkg/ddc/cache/engine/sync.go b/pkg/ddc/cache/engine/sync.go index c7c45c6c24e..d870f24b68e 100644 --- a/pkg/ddc/cache/engine/sync.go +++ b/pkg/ddc/cache/engine/sync.go @@ -86,16 +86,33 @@ func (e *CacheEngine) Sync(ctx cruntime.ReconcileRequestContext) (err error) { if !runtimeReady { // update dataset status when runtime not ready - err = e.UpdateDatasetStatus(datav1alpha1.FailedDatasetPhase, runtime, runtimeClass) + err = e.UpdateDatasetStatus(datav1alpha1.FailedDatasetPhase, runtime, runtimeClass, permitSyncEngineStatus) if err != nil { return err } - } else if permitSyncEngineStatus { - // sync dataset cache states when runtime is ready and sync permitted - e.Log.Info("sync dataset cache states") - err = e.syncDatasetCacheStates(ctx, runtime, runtimeClass) - if err != nil { - return err + } else { + dataset, getErr := utils.GetDataset(e.Client, e.name, e.namespace) + if getErr != nil { + return getErr + } + + if dataset.Status.Phase == datav1alpha1.FailedDatasetPhase { + // the runtime recovered from a previous outage but the dataset was left in Failed + // phase because the phase is otherwise only restored to Bound by the mount flow, + // which does not run on a normal reconcile. Restore it here, keeping the cache-states + // exec behind the same sync limiter as syncDatasetCacheStates below. + e.Log.Info("runtime is ready again, restoring dataset phase from Failed to Bound") + err = e.UpdateDatasetStatus(datav1alpha1.BoundDatasetPhase, runtime, runtimeClass, permitSyncEngineStatus) + if err != nil { + return err + } + } else if permitSyncEngineStatus { + // sync dataset cache states when runtime is ready and sync permitted + e.Log.Info("sync dataset cache states") + err = e.syncDatasetCacheStates(ctx, runtime, runtimeClass) + if err != nil { + return err + } } } diff --git a/pkg/ddc/cache/engine/sync_test.go b/pkg/ddc/cache/engine/sync_test.go index 509a417dbb6..a3ee3d571c9 100644 --- a/pkg/ddc/cache/engine/sync_test.go +++ b/pkg/ddc/cache/engine/sync_test.go @@ -19,9 +19,11 @@ package engine import ( "context" "os" + "reflect" "time" "github.com/fluid-cloudnative/fluid/pkg/common" + "github.com/fluid-cloudnative/fluid/pkg/utils" "github.com/agiledragon/gomonkey/v2" "github.com/go-logr/logr" @@ -295,7 +297,7 @@ var _ = Describe("CacheEngine Sync Tests", Label("pkg.ddc.cache.engine.sync_test engine.Client = fake.NewClientBuilder(). WithScheme(scheme). WithObjects(dataset, runtimeObj, runtimeClass, configMap, masterSts, workerSts, clientDs). - WithStatusSubresource(runtimeObj). + WithStatusSubresource(dataset, runtimeObj). Build() }) @@ -338,6 +340,121 @@ var _ = Describe("CacheEngine Sync Tests", Label("pkg.ddc.cache.engine.sync_test }) }) + Context("when runtime is ready but dataset was left Failed by a previous outage", func() { + BeforeEach(func() { + dataset.Status.Phase = datav1alpha1.FailedDatasetPhase + dataset.Status.Conditions = []datav1alpha1.DatasetCondition{ + { + Type: datav1alpha1.DatasetReady, + Status: corev1.ConditionFalse, + }, + } + + masterReplicas := int32(1) + masterSts := &workloadv1alpha1.AdvancedStatefulSet{ + ObjectMeta: metav1.ObjectMeta{Name: "test-runtime-master", Namespace: "default"}, + Spec: workloadv1alpha1.AdvancedStatefulSetSpec{ + Replicas: &masterReplicas, + Template: corev1.PodTemplateSpec{ + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: "master", Image: "test-master:latest"}}, + }, + }, + }, + Status: workloadv1alpha1.AdvancedStatefulSetStatus{ReadyReplicas: 1, CurrentReplicas: 1, AvailableReplicas: 1}, + } + + workerReplicas := int32(2) + workerSts := &workloadv1alpha1.AdvancedStatefulSet{ + ObjectMeta: metav1.ObjectMeta{Name: "test-runtime-worker", Namespace: "default"}, + Spec: workloadv1alpha1.AdvancedStatefulSetSpec{ + Replicas: &workerReplicas, + Template: corev1.PodTemplateSpec{ + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: "worker", Image: "test-worker:latest"}}, + }, + }, + }, + Status: workloadv1alpha1.AdvancedStatefulSetStatus{ReadyReplicas: 2, CurrentReplicas: 2, AvailableReplicas: 2}, + } + + clientDs := &appsv1.DaemonSet{ + ObjectMeta: metav1.ObjectMeta{Name: "test-runtime-client", Namespace: "default"}, + Spec: appsv1.DaemonSetSpec{ + Template: corev1.PodTemplateSpec{ + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: "client", Image: "test-client:latest"}}, + }, + }, + }, + Status: appsv1.DaemonSetStatus{NumberReady: 0, DesiredNumberScheduled: 0}, + } + + engine.Client = fake.NewClientBuilder(). + WithScheme(CacheEngineTestScheme). + WithObjects(dataset, runtimeObj, runtimeClass, masterSts, workerSts, clientDs). + WithStatusSubresource(dataset, runtimeObj). + Build() + }) + + It("should restore the dataset phase to Bound", func() { + err := engine.Sync(ctx) + Expect(err).NotTo(HaveOccurred()) + + updatedDataset := &datav1alpha1.Dataset{} + err = engine.Client.Get(context.Background(), types.NamespacedName{ + Name: "test-runtime", + Namespace: "default", + }, updatedDataset) + Expect(err).NotTo(HaveOccurred()) + Expect(updatedDataset.Status.Phase).To(Equal(datav1alpha1.BoundDatasetPhase)) + + idx, cond := utils.GetDatasetCondition(updatedDataset.Status.Conditions, datav1alpha1.DatasetReady) + Expect(idx).NotTo(Equal(-1)) + Expect(cond.Status).To(Equal(corev1.ConditionTrue)) + }) + + Context("and the sync limiter is closed", func() { + var patches *gomonkey.Patches + var getCacheStatesCalled bool + + BeforeEach(func() { + engine.syncRetryDuration = defaultSyncRetryDuration + engine.timeOfLastSync = time.Now() + + getCacheStatesCalled = false + // Patched at the GetCacheStates level, not NewCacheFileUtil: this Context has no + // ReportSummary execution entries configured, so a real call would fail before ever + // reaching the exec layer. The point here is only whether GetCacheStates is invoked at all. + patches = gomonkey.ApplyMethod(reflect.TypeOf(engine), "GetCacheStates", + func(_ *CacheEngine, _ *datav1alpha1.CacheRuntime, _ *datav1alpha1.CacheRuntimeClass) (common.CacheStateList, error) { + getCacheStatesCalled = true + return common.CacheStateList{}, nil + }) + }) + + AfterEach(func() { + if patches != nil { + patches.Reset() + } + }) + + It("should still restore the dataset phase to Bound without fetching cache states", func() { + err := engine.Sync(ctx) + Expect(err).NotTo(HaveOccurred()) + + updatedDataset := &datav1alpha1.Dataset{} + err = engine.Client.Get(context.Background(), types.NamespacedName{ + Name: "test-runtime", + Namespace: "default", + }, updatedDataset) + Expect(err).NotTo(HaveOccurred()) + Expect(updatedDataset.Status.Phase).To(Equal(datav1alpha1.BoundDatasetPhase)) + Expect(getCacheStatesCalled).To(BeFalse(), "GetCacheStates should be skipped while the sync limiter is closed") + }) + }) + }) + Context("when runtime is ready with ReportSummary configured", func() { var patches *gomonkey.Patches