fix(cache): charge tiered store memory quota on every reconcile

pkg/ddc/cache/component/advanced_statefulset_manager.go

@@ -19,7 +19,6 @@ package component
 import (
 	"context"
 	"fmt"
-	"reflect"
 
 	workloadv1alpha1 "github.com/fluid-cloudnative/advanced-statefulset/api/workload/v1alpha1"
 	datav1alpha1 "github.com/fluid-cloudnative/fluid/api/v1alpha1"
@@ -29,6 +28,7 @@ import (
 	"github.com/go-logr/logr"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -293,8 +293,11 @@ func (s *AdvancedStatefulSetManager) updateResources(asts *workloadv1alpha1.Adva
 
 	container := &asts.Spec.Template.Spec.Containers[0]
 
-	// Directly compare and replace, nil is also a valid value
-	if !reflect.DeepEqual(container.Resources, resources) {
+	// Compare semantically rather than structurally: resource.Quantity caches the
+	// string it was parsed from, and a quantity produced by arithmetic carries an
+	// empty cache. reflect.DeepEqual would read that as a change on every reconcile
+	// even when the value is identical. nil is still a valid desired value.
+	if !equality.Semantic.DeepEqual(container.Resources, resources) {
 		logger.Info("resources changed, will update")
 		container.Resources = *resources.DeepCopy()
 		return true

pkg/ddc/cache/component/sync_component_spec_test.go

@@ -18,6 +18,7 @@ package component
 
 import (
 	"context"
+	"encoding/json"
 
 	workloadv1alpha1 "github.com/fluid-cloudnative/advanced-statefulset/api/workload/v1alpha1"
 	datav1alpha1 "github.com/fluid-cloudnative/fluid/api/v1alpha1"
@@ -509,5 +510,55 @@ var _ = Describe("AdvancedStatefulSetManager SyncComponentSpec", func() {
 			result := manager.updateResources(emptyAsts, resources, GinkgoLogr)
 			Expect(result).To(BeFalse())
 		})
+
+		It("should not report a change when an equal quantity was produced by arithmetic", func() {
+			// A quantity built by adding, as the tiered store quota is, carries no cached
+			// string, while the one decoded from the workload does. Comparing the two
+			// structurally reports a difference that is not there, which makes every
+			// reconcile patch the workload again.
+			sum := resource.MustParse("4Gi")
+			sum.Add(resource.MustParse("8Gi"))
+
+			decoded := corev1.ResourceRequirements{}
+			Expect(json.Unmarshal([]byte(`{"limits":{"memory":"12Gi"}}`), &decoded)).To(Succeed())
+
+			asts := &workloadv1alpha1.AdvancedStatefulSet{
+				Spec: workloadv1alpha1.AdvancedStatefulSetSpec{
+					Template: corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{{Name: "test", Resources: decoded}},
+						},
+					},
+				},
+			}
+			recomputed := corev1.ResourceRequirements{
+				Limits: corev1.ResourceList{corev1.ResourceMemory: sum},
+			}
+
+			Expect(manager.updateResources(asts, recomputed, GinkgoLogr)).To(BeFalse())
+		})
+
+		It("should still report a change when the quantity really differs", func() {
+			decoded := corev1.ResourceRequirements{}
+			Expect(json.Unmarshal([]byte(`{"limits":{"memory":"12Gi"}}`), &decoded)).To(Succeed())
+
+			asts := &workloadv1alpha1.AdvancedStatefulSet{
+				Spec: workloadv1alpha1.AdvancedStatefulSetSpec{
+					Template: corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{{Name: "test", Resources: decoded}},
+						},
+					},
+				},
+			}
+			bigger := resource.MustParse("4Gi")
+			bigger.Add(resource.MustParse("16Gi"))
+
+			Expect(manager.updateResources(asts, corev1.ResourceRequirements{
+				Limits: corev1.ResourceList{corev1.ResourceMemory: bigger},
+			}, GinkgoLogr)).To(BeTrue())
+			limit := asts.Spec.Template.Spec.Containers[0].Resources.Limits[corev1.ResourceMemory]
+			Expect(limit.String()).To(Equal("20Gi"))
+		})
 	})
 })

pkg/ddc/cache/engine/sync.go

@@ -17,10 +17,12 @@
 package engine
 
 import (
+	"context"
 	"os"
 	"reflect"
 	"time"
 
+	workloadv1alpha1 "github.com/fluid-cloudnative/advanced-statefulset/api/workload/v1alpha1"
 	datav1alpha1 "github.com/fluid-cloudnative/fluid/api/v1alpha1"
 	"github.com/fluid-cloudnative/fluid/pkg/common"
 	"github.com/fluid-cloudnative/fluid/pkg/ddc/cache/component"
@@ -29,6 +31,7 @@ import (
 	"github.com/fluid-cloudnative/fluid/pkg/utils/dataset/lifecycle"
 	"github.com/fluid-cloudnative/fluid/pkg/utils/kubeclient"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/util/retry"
@@ -211,9 +214,28 @@ func (e *CacheEngine) syncRuntimeSpec(ctx cruntime.ReconcileRequestContext, runt
 			Namespace: e.namespace,
 		}
 		manager := component.NewComponentHelper(common.ComponentTypeWorker, e.Client)
+		// Mirror the creation path: the tiered store memory quota is charged on top
+		// of the resolved baseline, whether that came from the CacheRuntime or from
+		// the CacheRuntimeClass template. See TransformRuntimeTieredStore.
+		//
+		// The quota is read back from the workload rather than recomputed from the
+		// spec: tieredStore is not a supported update field, and SyncComponentSpec
+		// only patches resources, so a spec-derived quota would move the container's
+		// memory while the tmpfs volumes kept their original size.
+		workerResources := desiredComponentResources(runtime.Spec.Worker.Resources, runtimeClass.Topology.Worker)
+		if workerResources != nil {
+			quota, err := e.chargedWorkerTieredStoreMemoryQuota(ctx.Context, workerIdentity)
+			if err != nil {
+				e.Log.Error(err, "failed to read the tiered store memory quota charged to the worker",
+					"component", workerIdentity.Name)
+				return err
+			}
+			resources := withTieredStoreMemoryQuota(*workerResources, quota)
+			workerResources = &resources
+		}
 		workerSpec := component.ComponentSpec{
 			Version:   runtime.Spec.Worker.RuntimeVersion,
-			Resources: desiredComponentResources(runtime.Spec.Worker.Resources, runtimeClass.Topology.Worker),
+			Resources: workerResources,
 			Replicas:  &runtime.Spec.Worker.Replicas,
 		}
 		if err := manager.SyncComponentSpec(ctx.Context, workerIdentity, workerSpec); err != nil {
@@ -253,6 +275,17 @@ func desiredComponentResources(runtimeResources corev1.ResourceRequirements, com
 	return templateResources.DeepCopy()
 }
 
+// chargedWorkerTieredStoreMemoryQuota reads back the tiered store memory quota that
+// the creation path charged to the worker workload.
+func (e *CacheEngine) chargedWorkerTieredStoreMemoryQuota(ctx context.Context, identity *common.ComponentIdentity) (resource.Quantity, error) {
+	workers := &workloadv1alpha1.AdvancedStatefulSet{}
+	key := types.NamespacedName{Name: identity.Name, Namespace: identity.Namespace}
+	if err := e.Get(ctx, key, workers); err != nil {
+		return resource.Quantity{}, err
+	}
+	return chargedTieredStoreMemoryQuota(&workers.Spec.Template.Spec), nil
+}
+
 func (e *CacheEngine) syncDatasetCacheStates(ctx cruntime.ReconcileRequestContext, runtime *datav1alpha1.CacheRuntime, runtimeClass *datav1alpha1.CacheRuntimeClass) (err error) {
 	cacheStates, err := e.GetCacheStates(runtime, runtimeClass)
 	if err != nil {

pkg/ddc/cache/engine/sync_test.go

@@ -889,6 +889,21 @@ var _ = Describe("CacheEngine Sync Tests", Label("pkg.ddc.cache.engine.sync_test
 			return limit.String()
 		}
 
+		// tmpfsSizeOf reports the size limit of the workload's memory-backed tiered
+		// store volume, i.e. the quota the container's memory has to cover.
+		tmpfsSizeOf := func(stsName string) string {
+			sts := &workloadv1alpha1.AdvancedStatefulSet{}
+			key := types.NamespacedName{Name: stsName, Namespace: "default"}
+			Expect(fakeClient.Get(ctx.Context, key, sts)).To(Succeed())
+			for _, volume := range sts.Spec.Template.Spec.Volumes {
+				emptyDir := volume.EmptyDir
+				if emptyDir != nil && emptyDir.Medium == corev1.StorageMediumMemory && emptyDir.SizeLimit != nil {
+					return emptyDir.SizeLimit.String()
+				}
+			}
+			return "<none>"
+		}
+
 		BeforeEach(func() {
 			seedTemplateResources(masterSts, &runtimeClass.Topology.Master.Template)
 			seedTemplateResources(workerSts, &runtimeClass.Topology.Worker.Template)
@@ -984,5 +999,166 @@ var _ = Describe("CacheEngine Sync Tests", Label("pkg.ddc.cache.engine.sync_test
 				Expect(memLimitOf(workerSts)).To(Equal("8Gi"))
 			})
 		})
+
+		Context("when the tiered store baseline comes from the template", func() {
+			// With the CacheRuntime declaring no resources, the baseline the quota is
+			// charged on top of is the CacheRuntimeClass template value. The quota must
+			// still be added, otherwise the worker is short by the quota it was created
+			// with the moment a sync runs.
+			BeforeEach(func() {
+				runtimeObj.Spec.Worker.TieredStore = datav1alpha1.RuntimeTieredStore{
+					Levels: []datav1alpha1.RuntimeTieredStoreLevel{
+						{ProcessMemory: &datav1alpha1.ProcessMemoryMediumSource{Quota: resource.MustParse("8Gi")}},
+					},
+				}
+				Expect(fakeClient.Update(ctx.Context, runtimeObj)).To(Succeed())
+			})
+
+			It("should charge the quota on top of the template value", func() {
+				Expect(runtimeObj.Spec.Worker.Resources.Limits).To(BeNil())
+				Expect(runtimeObj.Spec.Worker.Resources.Requests).To(BeNil())
+
+				// The workload carries the template value plus the quota, as creation left it.
+				key := types.NamespacedName{Name: workerSts, Namespace: "default"}
+				sts := &workloadv1alpha1.AdvancedStatefulSet{}
+				Expect(fakeClient.Get(ctx.Context, key, sts)).To(Succeed())
+				sts.Spec.Template.Spec.Containers[0].Resources = corev1.ResourceRequirements{
+					Limits: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("10Gi")},
+				}
+				chargedQuota := resource.MustParse("8Gi")
+				sts.Spec.Template.Spec.Volumes = []corev1.Volume{{
+					Name: "tiered-store-level-0-memory",
+					VolumeSource: corev1.VolumeSource{
+						EmptyDir: &corev1.EmptyDirVolumeSource{
+							Medium:    corev1.StorageMediumMemory,
+							SizeLimit: &chargedQuota,
+						},
+					},
+				}}
+				Expect(fakeClient.Update(ctx.Context, sts)).To(Succeed())
+
+				syncRuntime, err := engine.getRuntime()
+				Expect(err).NotTo(HaveOccurred())
+				Expect(engine.syncRuntimeSpec(ctx, syncRuntime, runtimeClass)).To(Succeed())
+
+				// template 2Gi + quota 8Gi, not the bare template value.
+				Expect(memLimitOf(workerSts)).To(Equal("10Gi"))
+			})
+		})
+
+		Context("when the CacheRuntime declares a memory tiered store", func() {
+			// The creation path derives the worker's memory as
+			// <user baseline> + <tiered store memory quota>. A later sync recomputes
+			// the desired state from the same spec and must land on the same value,
+			// otherwise the quota silently disappears on the second reconcile.
+			BeforeEach(func() {
+				runtimeObj.Spec.Worker.Resources = corev1.ResourceRequirements{
+					Limits: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("4Gi")},
+				}
+				runtimeObj.Spec.Worker.TieredStore = datav1alpha1.RuntimeTieredStore{
+					Levels: []datav1alpha1.RuntimeTieredStoreLevel{
+						{ProcessMemory: &datav1alpha1.ProcessMemoryMediumSource{Quota: resource.MustParse("8Gi")}},
+					},
+				}
+				Expect(fakeClient.Update(ctx.Context, runtimeObj)).To(Succeed())
+			})
+
+			It("should keep the tiered store quota when syncing after creation", func() {
+				// Creation reconcile: derive the desired state and create the workload.
+				createRuntime, err := engine.getRuntime()
+				Expect(err).NotTo(HaveOccurred())
+
+				// Read the baseline before transforming: the transform must not be
+				// trusted to leave the runtime spec alone.
+				baseline := createRuntime.Spec.Worker.Resources.Limits[corev1.ResourceMemory].DeepCopy()
+
+				value, err := engine.transform(dataset, createRuntime, runtimeClass)
+				Expect(err).NotTo(HaveOccurred())
+				desired := value.Worker.PodTemplateSpec.Spec.Containers[0].Resources.Limits[corev1.ResourceMemory]
+
+				// Guard against a vacuous comparison: creation must actually charge the quota.
+				Expect(desired.Cmp(baseline)).To(Equal(1), "creation path must charge the tiered store quota")
+
+				key := types.NamespacedName{Name: workerSts, Namespace: "default"}
+				seeded := &workloadv1alpha1.AdvancedStatefulSet{}
+				Expect(fakeClient.Get(ctx.Context, key, seeded)).To(Succeed())
+				Expect(fakeClient.Delete(ctx.Context, seeded)).To(Succeed())
+
+				_, err = engine.SetupWorkerComponent(value.Worker)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(memLimitOf(workerSts)).To(Equal(desired.String()))
+
+				// Sync reconcile: re-read the runtime the way a fresh reconcile would,
+				// so the sync can only rely on what is persisted in the spec.
+				syncRuntime, err := engine.getRuntime()
+				Expect(err).NotTo(HaveOccurred())
+				Expect(engine.syncRuntimeSpec(ctx, syncRuntime, runtimeClass)).To(Succeed())
+
+				Expect(memLimitOf(workerSts)).To(Equal(desired.String()))
+			})
+
+			// createWorker runs the creation reconcile and returns the memory the
+			// creation path derives, i.e. baseline + quota.
+			createWorker := func() string {
+				createRuntime, err := engine.getRuntime()
+				Expect(err).NotTo(HaveOccurred())
+				value, err := engine.transform(dataset, createRuntime, runtimeClass)
+				Expect(err).NotTo(HaveOccurred())
+				desired := value.Worker.PodTemplateSpec.Spec.Containers[0].Resources.Limits[corev1.ResourceMemory]
+
+				key := types.NamespacedName{Name: workerSts, Namespace: "default"}
+				seeded := &workloadv1alpha1.AdvancedStatefulSet{}
+				Expect(fakeClient.Get(ctx.Context, key, seeded)).To(Succeed())
+				Expect(fakeClient.Delete(ctx.Context, seeded)).To(Succeed())
+				_, err = engine.SetupWorkerComponent(value.Worker)
+				Expect(err).NotTo(HaveOccurred())
+				return desired.String()
+			}
+
+			syncOnce := func() {
+				syncRuntime, err := engine.getRuntime()
+				Expect(err).NotTo(HaveOccurred())
+				Expect(engine.syncRuntimeSpec(ctx, syncRuntime, runtimeClass)).To(Succeed())
+			}
+
+			It("should not let an edited tiered store quota half-apply", func() {
+				// tieredStore is not a supported update field, and SyncComponentSpec only
+				// patches resources. A spec-derived quota would move the container's memory
+				// to baseline+16Gi while the tmpfs volume stayed at the 8Gi it was created
+				// with.
+				desired := createWorker()
+				Expect(tmpfsSizeOf(workerSts)).To(Equal("8Gi"))
+
+				edited, err := engine.getRuntime()
+				Expect(err).NotTo(HaveOccurred())
+				edited.Spec.Worker.TieredStore.Levels[0].ProcessMemory.Quota = resource.MustParse("16Gi")
+				Expect(fakeClient.Update(ctx.Context, edited)).To(Succeed())
+
+				syncOnce()
+
+				Expect(memLimitOf(workerSts)).To(Equal(desired))
+				Expect(tmpfsSizeOf(workerSts)).To(Equal("8Gi"))
+			})
+
+			It("should recharge a quota that an earlier release stripped", func() {
+				// Workloads created before this fix have the bare baseline on the container
+				// and the full quota on the tmpfs volume. Recovering the quota from the
+				// volume rather than the container is what lets them be repaired in place.
+				desired := createWorker()
+
+				key := types.NamespacedName{Name: workerSts, Namespace: "default"}
+				stripped := &workloadv1alpha1.AdvancedStatefulSet{}
+				Expect(fakeClient.Get(ctx.Context, key, stripped)).To(Succeed())
+				stripped.Spec.Template.Spec.Containers[0].Resources = corev1.ResourceRequirements{
+					Limits: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("4Gi")},
+				}
+				Expect(fakeClient.Update(ctx.Context, stripped)).To(Succeed())
+				Expect(memLimitOf(workerSts)).To(Equal("4Gi"))
+
+				syncOnce()
+
+				Expect(memLimitOf(workerSts)).To(Equal(desired))
+			})
+		})
 	})
 })