Skip to content

fix(cache): restore Dataset to Bound after CacheRuntime recovers from an outage #6162

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
pujitha24 wants to merge 4 commits into
base: master
Choose a base branch
from
+160 −12
Open

fix(cache): restore Dataset to Bound after CacheRuntime recovers from an outage #6162

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
49b2a77
fix(cache): restore Dataset to Bound after CacheRuntime recovers from…
pujitha24 Aug 15, 2026
2e2b62a
fix(cache): keep dataset phase-restore cheap and rate-limited
pujitha24 Aug 17, 2026
426ed33
fix(cache): move permitSync check out of UpdateDatasetStatus
pujitha24 Aug 17, 2026
1594e4f
fix(cache): make the closed-limiter regression test actually assert t…
pujitha24 Aug 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 13 additions & 2 deletions pkg/ddc/cache/engine/dataset.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,11 +42,22 @@
return e.UpdateDatasetStatus(datav1alpha1.BoundDatasetPhase, runtime, runtimeClass)
}

func (e *CacheEngine) UpdateDatasetStatus(phase datav1alpha1.DatasetPhase, runtime *datav1alpha1.CacheRuntime, runtimeClass *datav1alpha1.CacheRuntimeClass) (err error) {

Check failure on line 45 in pkg/ddc/cache/engine/dataset.go

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this method to reduce its Cognitive Complexity from 16 to the 15 allowed. 


See more on https://sonarcloud.io/project/issues?id=fluid-cloudnative_fluid&issues=AaAOVL1OobttII-QTt4U&open=AaAOVL1OobttII-QTt4U&pullRequest=6162
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 keep it behind the same rate limiter that bounds other engine RPCs, and only
// attempt it for BoundDatasetPhase.
if phase == datav1alpha1.BoundDatasetPhase && e.permitSync() {

@xliuqq xliuqq Aug 17, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the permitSync() method should be only called in sync method, it will make the sync logic clear.

e.Log.V(1).Info("Start to update cache states")
cacheStates, err = e.GetCacheStates(runtime, runtimeClass)
if err != nil {
Expand Down
30 changes: 24 additions & 6 deletions pkg/ddc/cache/engine/sync.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,12 +90,30 @@ func (e *CacheEngine) Sync(ctx cruntime.ReconcileRequestContext) (err error) {
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. UpdateDatasetStatus
// keeps this cheap: it only execs into the master pod for cache states when the
// sync limiter permits it.
e.Log.Info("runtime is ready again, restoring dataset phase from Failed to Bound")
err = e.UpdateDatasetStatus(datav1alpha1.BoundDatasetPhase, runtime, runtimeClass)

@cheyang cheyang Aug 17, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

UpdateDatasetStatus(BoundDatasetPhase, ...) isn't a cheap phase write. For the Bound case, dataset.go:49-55 calls GetCacheStates, which execs into the master pod via ExecCommandInContainerWithTimeout with a 20s timeout floor (MinExecutionTimeoutSeconds). That call sits outside permitSyncEngineStatus, and only the else if below still carries the guard, so the restore path issues exactly the sort of unthrottled RPC that the comment on line 38 says the limiter exists to bound.

Alluxio is a useful comparison here, since it does the same recovery and is also called outside the limiter (base/syncs.go:63 into alluxio/health_check.go:81). That's fine there for two reasons the cache engine's version lacks: its UpdateDatasetStatus writes only phase, condition, mounts and runtimes with no RPC, and it wraps the transition in if phase != dataset.Status.Phase, so calling it again costs nothing. Cache states get refreshed separately by UpdateCacheOfDataset().

I measured this rather than guessing, and I want to be straight about the size of it. In a unit test, driving three not-ready to ready flaps inside a single 5s window produces 3 execs where at most 1 should happen. On a real cluster it's much milder: across five induced worker outages in 47s, the change added one exec attempt (6 versus 5 without it), because a real recovery cycle takes longer than 5s and so tends to get its own window. So this is not a production hazard, and I'm not claiming it is.

Even so, I'd rather see it handled in this PR than carried forward, because the fix is small and lives in code you're already touching. Making the restore phase-only and moving the idempotence check into the helper does it, and then this call site no longer needs the GetDataset above either:

current, err := utils.GetDataset(e.Client, e.name, e.namespace)
if err != nil {
	return err
}
if current.Status.Phase == phase {
	return nil
}
// pod exec with a 20s timeout floor, so keep it behind the limiter
if phase == datav1alpha1.BoundDatasetPhase && e.permitSync() {
	cacheStates, err = e.GetCacheStates(runtime, runtimeClass)
	...
}

One approach I'd avoid: simply wrapping the restore in permitSyncEngineStatus. That makes recovery wait on the limiter rather than making it cheap, so it swaps this for a slower fix.

Smaller point about the current shape: on the reconcile that restores the phase, syncDatasetCacheStates gets skipped entirely, because the restore takes the if and the sync sits in the else if.

Harness and captured output, if it's useful: https://github.com/cheyang/fluid/tree/verify/cacheruntime-dataset-phase-restore/docs/verification/cacheruntime-dataset-phase-restore

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
}
}
}
Comment on lines +93 to 117

Expand Down
98 changes: 97 additions & 1 deletion pkg/ddc/cache/engine/sync_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import (
"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"
Expand Down Expand Up @@ -295,7 +296,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()
})

Expand Down Expand Up @@ -338,6 +339,101 @@ 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))

@cheyang cheyang Aug 17, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This pins the phase but not the condition. UpdateDatasetStatus also flips DatasetReady back to ConditionTrue, and the condition is what IsSetupDone and the other consumers actually read, so it seems worth asserting too:

idx, cond := utils.GetDatasetCondition(updatedDataset.Status.Conditions, datav1alpha1.DatasetReady)
Expect(idx).NotTo(Equal(-1))
Expect(cond.Status).To(Equal(corev1.ConditionTrue))

It would also help to seed the Dataset with a DatasetReady/False condition in the BeforeEach. That's what a real outage leaves behind, and it's the reason Setup never re-runs, so the fixture would then match the state the fix is actually for.

One path this case can't reach: the shared fixture leaves syncRetryDuration at its zero value, which makes permitSync() always return true and hides the closed-limiter branch completely. Setting it to defaultSyncRetryDuration covers the case where the limiter is shut.


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() {
BeforeEach(func() {
engine.syncRetryDuration = defaultSyncRetryDuration
engine.timeOfLastSync = time.Now()
})

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

Context("when runtime is ready with ReportSummary configured", func() {
var patches *gomonkey.Patches

Expand Down