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"
@@ -63,6 +63,16 @@ func (s *AdvancedStatefulSetManager) GetNodeAffinity(identity *common.ComponentI
 	return affinity, nil
 }
 
+func (s *AdvancedStatefulSetManager) GetPodSpec(ctx context.Context, identity *common.ComponentIdentity) (*corev1.PodSpec, error) {
+	asts := &workloadv1alpha1.AdvancedStatefulSet{}
+	err := s.client.Get(ctx, types.NamespacedName{Name: identity.Name, Namespace: identity.Namespace}, asts)
+	if err != nil {
+		return nil, err
+	}
+
+	return asts.Spec.Template.Spec.DeepCopy(), nil
+}
+
 func (s *AdvancedStatefulSetManager) reconcileStatefulSet(ctx context.Context, component *common.CacheRuntimeComponentValue) error {
 	logger := log.FromContext(ctx)
 	logger.Info("start to reconciling advanced statefulset workload")
@@ -293,8 +303,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/component_manager.go

@@ -34,6 +34,10 @@ type ComponentManager interface {
 	GetNodeAffinity(identity *common.ComponentIdentity) (*corev1.NodeAffinity, error)
 	// SyncComponentSpec synchronizes component specification changes to the workload
 	SyncComponentSpec(ctx context.Context, identity *common.ComponentIdentity, newSpec ComponentSpec) error
+	// GetPodSpec returns a copy of the pod template spec of the component's workload,
+	// so callers can inspect what the workload carries without depending on the
+	// concrete workload type.
+	GetPodSpec(ctx context.Context, identity *common.ComponentIdentity) (*corev1.PodSpec, error)
 }
 
 // ComponentSpec represents the specification that can be synchronized to a component

pkg/ddc/cache/component/component_test.go

@@ -292,6 +292,46 @@ var _ = Describe("AdvancedStatefulSetManager", func() {
 			Expect(err).To(HaveOccurred())
 		})
 	})
+
+	Describe("GetPodSpec", func() {
+		It("should return the pod spec of the workload", func() {
+			Expect(manager.Reconciler(ctx, component)).To(Succeed())
+
+			podSpec, err := manager.GetPodSpec(ctx, &common.ComponentIdentity{
+				Name:      component.Name,
+				Namespace: component.Namespace,
+			})
+			Expect(err).NotTo(HaveOccurred())
+			Expect(podSpec).NotTo(BeNil())
+			Expect(podSpec.Containers).To(HaveLen(1))
+			Expect(podSpec.Containers[0].Name).To(Equal("master"))
+			Expect(podSpec.Containers[0].Image).To(Equal("test-image:latest"))
+		})
+
+		It("should return a copy the caller cannot use to mutate the workload", func() {
+			Expect(manager.Reconciler(ctx, component)).To(Succeed())
+			identity := &common.ComponentIdentity{
+				Name:      component.Name,
+				Namespace: component.Namespace,
+			}
+
+			podSpec, err := manager.GetPodSpec(ctx, identity)
+			Expect(err).NotTo(HaveOccurred())
+			podSpec.Containers[0].Image = "mutated:latest"
+
+			reread, err := manager.GetPodSpec(ctx, identity)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(reread.Containers[0].Image).To(Equal("test-image:latest"))
+		})
+
+		It("should return error when AdvancedStatefulSet doesn't exist", func() {
+			_, err := manager.GetPodSpec(ctx, &common.ComponentIdentity{
+				Name:      component.Name,
+				Namespace: component.Namespace,
+			})
+			Expect(err).To(HaveOccurred())
+		})
+	})
 })
 
 var _ = Describe("DaemonSetManager", func() {
@@ -464,4 +504,44 @@ var _ = Describe("DaemonSetManager", func() {
 			Expect(err).To(HaveOccurred())
 		})
 	})
+
+	Describe("GetPodSpec", func() {
+		It("should return the pod spec of the workload", func() {
+			Expect(manager.Reconciler(ctx, component)).To(Succeed())
+
+			podSpec, err := manager.GetPodSpec(ctx, &common.ComponentIdentity{
+				Name:      component.Name,
+				Namespace: component.Namespace,
+			})
+			Expect(err).NotTo(HaveOccurred())
+			Expect(podSpec).NotTo(BeNil())
+			Expect(podSpec.Containers).To(HaveLen(1))
+			Expect(podSpec.Containers[0].Name).To(Equal("worker"))
+			Expect(podSpec.Containers[0].Image).To(Equal("test-image:latest"))
+		})
+
+		It("should return a copy the caller cannot use to mutate the workload", func() {
+			Expect(manager.Reconciler(ctx, component)).To(Succeed())
+			identity := &common.ComponentIdentity{
+				Name:      component.Name,
+				Namespace: component.Namespace,
+			}
+
+			podSpec, err := manager.GetPodSpec(ctx, identity)
+			Expect(err).NotTo(HaveOccurred())
+			podSpec.Containers[0].Image = "mutated:latest"
+
+			reread, err := manager.GetPodSpec(ctx, identity)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(reread.Containers[0].Image).To(Equal("test-image:latest"))
+		})
+
+		It("should return error when DaemonSet doesn't exist", func() {
+			_, err := manager.GetPodSpec(ctx, &common.ComponentIdentity{
+				Name:      component.Name,
+				Namespace: component.Namespace,
+			})
+			Expect(err).To(HaveOccurred())
+		})
+	})
 })

pkg/ddc/cache/component/daemonset_manager.go

@@ -59,6 +59,15 @@ func (s *DaemonSetManager) GetNodeAffinity(identity *common.ComponentIdentity) (
 	return affinity, nil
 }
 
+func (s *DaemonSetManager) GetPodSpec(ctx context.Context, identity *common.ComponentIdentity) (*corev1.PodSpec, error) {
+	ds, err := kubeclient.GetDaemonset(s.client, identity.Name, identity.Namespace)
+	if err != nil {
+		return nil, err
+	}
+
+	return ds.Spec.Template.Spec.DeepCopy(), nil
+}
+
 func (s *DaemonSetManager) reconcileDaemonSet(ctx context.Context, component *common.CacheRuntimeComponentValue) error {
 	logger := log.FromContext(ctx)
 	logger.Info("start to reconciling ds workload")

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

@@ -211,9 +211,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 {
+			podSpec, err := manager.GetPodSpec(ctx.Context, workerIdentity)
+			if err != nil {
+				e.Log.Error(err, "failed to read the worker pod spec", "component", workerIdentity.Name)
+				return err
+			}
+			resources := withTieredStoreMemoryQuota(*workerResources, chargedTieredStoreMemoryQuota(podSpec))
+			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 {