fix(cache): preserve CacheRuntimeClass template resources when unset

docs/en/samples/cacheruntime/cacheruntime_spec_update.md

@@ -73,8 +73,19 @@ spec:
         memory: 16Gi
 ```
 
+**Resolution order**:
+
+`resources` is resolved as follows on every reconcile:
+
+1. the value set on the CacheRuntime, if it declares any `requests` or `limits`;
+2. otherwise the value declared by the CacheRuntimeClass template;
+3. otherwise nothing is synced and the workload keeps its current resources.
+
 **Limitations**:
 - ⚠️ Cannot exceed the node's available resources.
+- ⚠️ When the CacheRuntimeClass template declares `resources`, removing `resources` from the
+  CacheRuntime does **not** leave the component unconstrained — it falls back to the template value.
+  To relax a limit, set the value you want explicitly instead of removing the field.
 - ⚠️ **Kubernetes version requirement**: K8s >= 1.27 with the `InPlacePodVerticalScaling` Feature Gate enabled.
   ```bash
   # Check if the Feature Gate is enabled

docs/zh/samples/cacheruntime/cacheruntime_spec_update.md

@@ -73,8 +73,18 @@ spec:
         memory: 16Gi
 ```
 
+**取值优先级**：
+
+每次 reconcile 时，`resources` 按以下顺序取值：
+
+1. CacheRuntime 上设置的值（只要声明了 `requests` 或 `limits`）；
+2. 否则取 CacheRuntimeClass 模板中声明的值；
+3. 两者都未声明时不做同步，工作负载保持当前的资源配置。
+
 **限制**：
 - ⚠️ 不能超过节点可用资源
+- ⚠️ 当 CacheRuntimeClass 模板声明了 `resources` 时，从 CacheRuntime 中删除 `resources` **不会**
+  让组件变为不受限，而是回退到模板中的值。如需放宽限制，请显式设置目标值，而不是删除该字段。
 - ⚠️ **Kubernetes 版本要求**：需要 K8s >= 1.27 且启用 `InPlacePodVerticalScaling` Feature Gate
   ```bash
   # 检查 Feature Gate 是否启用

pkg/ddc/cache/component/advanced_statefulset_manager.go

@@ -216,8 +216,10 @@ func (s *AdvancedStatefulSetManager) SyncComponentSpec(ctx context.Context, iden
 	}
 
 	// 3. Update resources if specified
-	if s.updateResources(astsToUpdate, newSpec.Resources, logger) {
-		needsUpdate = true
+	if newSpec.Resources != nil {
+		if s.updateResources(astsToUpdate, *newSpec.Resources, logger) {
+			needsUpdate = true
+		}
 	}
 
 	// Skip patching if no changes detected

pkg/ddc/cache/component/component_manager.go

@@ -44,7 +44,7 @@ type ComponentSpec struct {
 	// Version contains image and pull policy information
 	Version datav1alpha1.VersionSpec
 	// Resources contains CPU and memory resource requirements
-	Resources corev1.ResourceRequirements
+	Resources *corev1.ResourceRequirements
 }
 
 func NewComponentHelper(componentType common.ComponentType, client client.Client) ComponentManager {

pkg/ddc/cache/component/sync_component_spec_test.go

@@ -233,7 +233,7 @@ var _ = Describe("AdvancedStatefulSetManager SyncComponentSpec", func() {
 		Context("when updating resources", func() {
 			It("should update both requests and limits", func() {
 				spec := ComponentSpec{
-					Resources: corev1.ResourceRequirements{
+					Resources: &corev1.ResourceRequirements{
 						Requests: corev1.ResourceList{
 							corev1.ResourceCPU:    resource.MustParse("4"),
 							corev1.ResourceMemory: resource.MustParse("8Gi"),
@@ -265,7 +265,7 @@ var _ = Describe("AdvancedStatefulSetManager SyncComponentSpec", func() {
 
 			It("should not update when resources unchanged", func() {
 				spec := ComponentSpec{
-					Resources: corev1.ResourceRequirements{
+					Resources: &corev1.ResourceRequirements{
 						Requests: corev1.ResourceList{
 							corev1.ResourceCPU:    resource.MustParse("2"),
 							corev1.ResourceMemory: resource.MustParse("4Gi"),
@@ -302,7 +302,7 @@ var _ = Describe("AdvancedStatefulSetManager SyncComponentSpec", func() {
 						Image:    "fluid-cache",
 						ImageTag: "v1.1.0",
 					},
-					Resources: corev1.ResourceRequirements{
+					Resources: &corev1.ResourceRequirements{
 						Requests: corev1.ResourceList{
 							corev1.ResourceCPU: resource.MustParse("4"),
 						},

pkg/ddc/cache/engine/sync.go

@@ -193,15 +193,9 @@ func (e *CacheEngine) syncRuntimeSpec(ctx cruntime.ReconcileRequestContext, runt
 			Namespace: e.namespace,
 		}
 		manager := component.NewComponentHelper(common.ComponentTypeMaster, e.Client)
-		// Only sync resources if they are explicitly set (not zero-value)
-		// This prevents overwriting template defaults when user hasn't specified resources
-		var resources corev1.ResourceRequirements
-		if runtime.Spec.Master.Resources.Requests != nil || runtime.Spec.Master.Resources.Limits != nil {
-			resources = runtime.Spec.Master.Resources
-		}
 		masterSpec := component.ComponentSpec{
 			Version:   runtime.Spec.Master.RuntimeVersion,
-			Resources: resources,
+			Resources: desiredComponentResources(runtime.Spec.Master.Resources, runtimeClass.Topology.Master),
 			Replicas:  &runtime.Spec.Master.Replicas,
 		}
 		if err := manager.SyncComponentSpec(ctx.Context, masterIdentity, masterSpec); err != nil {
@@ -217,15 +211,9 @@ func (e *CacheEngine) syncRuntimeSpec(ctx cruntime.ReconcileRequestContext, runt
 			Namespace: e.namespace,
 		}
 		manager := component.NewComponentHelper(common.ComponentTypeWorker, e.Client)
-		// Only sync resources if they are explicitly set (not zero-value)
-		// This prevents overwriting template defaults when user hasn't specified resources
-		var workerResources corev1.ResourceRequirements
-		if runtime.Spec.Worker.Resources.Requests != nil || runtime.Spec.Worker.Resources.Limits != nil {
-			workerResources = runtime.Spec.Worker.Resources
-		}
 		workerSpec := component.ComponentSpec{
 			Version:   runtime.Spec.Worker.RuntimeVersion,
-			Resources: workerResources,
+			Resources: desiredComponentResources(runtime.Spec.Worker.Resources, runtimeClass.Topology.Worker),
 			Replicas:  &runtime.Spec.Worker.Replicas,
 		}
 		if err := manager.SyncComponentSpec(ctx.Context, workerIdentity, workerSpec); err != nil {
@@ -240,6 +228,31 @@ func (e *CacheEngine) syncRuntimeSpec(ctx cruntime.ReconcileRequestContext, runt
 	return nil
 }
 
+// desiredComponentResources resolves the resources that should be synced to a
+// component's workload. A value set on the CacheRuntime wins; when the CacheRuntime
+// sets none, the CacheRuntimeClass template value is used, which is what the creation
+// path rendered into the workload. A nil return means neither declares resources, and
+// the workload's current resources are left untouched.
+//
+// Only the first container is considered, matching the creation path, which also only
+// fills in resources for Containers[0].
+func desiredComponentResources(runtimeResources corev1.ResourceRequirements, componentDefinition *datav1alpha1.RuntimeComponentDefinition) *corev1.ResourceRequirements {
+	if runtimeResources.Requests != nil || runtimeResources.Limits != nil {
+		return runtimeResources.DeepCopy()
+	}
+
+	if componentDefinition == nil || len(componentDefinition.Template.Spec.Containers) == 0 {
+		return nil
+	}
+
+	templateResources := componentDefinition.Template.Spec.Containers[0].Resources
+	if templateResources.Requests == nil && templateResources.Limits == nil {
+		return nil
+	}
+
+	return templateResources.DeepCopy()
+}
+
 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

@@ -33,6 +33,7 @@ import (
 	cruntime "github.com/fluid-cloudnative/fluid/pkg/runtime"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@@ -858,4 +859,130 @@ var _ = Describe("CacheEngine Sync Tests", Label("pkg.ddc.cache.engine.sync_test
 			})
 		})
 	})
+
+	Describe("syncRuntimeSpec", func() {
+		const masterSts, workerSts = "test-runtime-master", "test-runtime-worker"
+
+		// templateResources mirrors the value the creation path derives from the
+		// CacheRuntimeClass template, i.e. what a sync must leave untouched.
+		templateResources := corev1.ResourceRequirements{
+			Limits: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("2Gi")},
+		}
+
+		// seedTemplateResources reproduces the post-creation state: the template declares
+		// resources and the already-created workload carries them.
+		seedTemplateResources := func(stsName string, template *corev1.PodTemplateSpec) {
+			template.Spec.Containers[0].Resources = *templateResources.DeepCopy()
+
+			sts := &workloadv1alpha1.AdvancedStatefulSet{}
+			key := types.NamespacedName{Name: stsName, Namespace: "default"}
+			Expect(fakeClient.Get(ctx.Context, key, sts)).To(Succeed())
+			sts.Spec.Template.Spec.Containers[0].Resources = *templateResources.DeepCopy()
+			Expect(fakeClient.Update(ctx.Context, sts)).To(Succeed())
+		}
+
+		memLimitOf := func(stsName string) string {
+			sts := &workloadv1alpha1.AdvancedStatefulSet{}
+			key := types.NamespacedName{Name: stsName, Namespace: "default"}
+			Expect(fakeClient.Get(ctx.Context, key, sts)).To(Succeed())
+			limit := sts.Spec.Template.Spec.Containers[0].Resources.Limits[corev1.ResourceMemory]
+			return limit.String()
+		}
+
+		BeforeEach(func() {
+			seedTemplateResources(masterSts, &runtimeClass.Topology.Master.Template)
+			seedTemplateResources(workerSts, &runtimeClass.Topology.Worker.Template)
+		})
+
+		Context("when the CacheRuntime does not specify resources", func() {
+			It("should leave the template's resources untouched", func() {
+				Expect(runtimeObj.Spec.Master.Resources.Limits).To(BeNil())
+				Expect(runtimeObj.Spec.Master.Resources.Requests).To(BeNil())
+				Expect(runtimeObj.Spec.Worker.Resources.Limits).To(BeNil())
+				Expect(runtimeObj.Spec.Worker.Resources.Requests).To(BeNil())
+
+				Expect(engine.syncRuntimeSpec(ctx, runtimeObj, runtimeClass)).To(Succeed())
+
+				Expect(memLimitOf(masterSts)).To(Equal("2Gi"))
+				Expect(memLimitOf(workerSts)).To(Equal("2Gi"))
+			})
+		})
+
+		Context("when the CacheRuntime specifies resources", func() {
+			It("should apply them to the master workload only", func() {
+				runtimeObj.Spec.Master.Resources = corev1.ResourceRequirements{
+					Limits: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("4Gi")},
+				}
+
+				Expect(engine.syncRuntimeSpec(ctx, runtimeObj, runtimeClass)).To(Succeed())
+
+				Expect(memLimitOf(masterSts)).To(Equal("4Gi"))
+				Expect(memLimitOf(workerSts)).To(Equal("2Gi"))
+			})
+
+			It("should apply them to the worker workload only", func() {
+				runtimeObj.Spec.Worker.Resources = corev1.ResourceRequirements{
+					Limits: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("4Gi")},
+				}
+
+				Expect(engine.syncRuntimeSpec(ctx, runtimeObj, runtimeClass)).To(Succeed())
+
+				Expect(memLimitOf(workerSts)).To(Equal("4Gi"))
+				Expect(memLimitOf(masterSts)).To(Equal("2Gi"))
+			})
+		})
+
+		Context("when the workload no longer matches the template", func() {
+			// setWorkloadMemLimit edits the workload behind the runtime's back, standing in
+			// for a workload that drifted from the template for any reason.
+			setWorkloadMemLimit := func(stsName, limit string) {
+				sts := &workloadv1alpha1.AdvancedStatefulSet{}
+				key := types.NamespacedName{Name: stsName, Namespace: "default"}
+				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(limit)},
+				}
+				Expect(fakeClient.Update(ctx.Context, sts)).To(Succeed())
+			}
+
+			It("should restore the template value when the CacheRuntime specifies none", func() {
+				setWorkloadMemLimit(workerSts, "8Gi")
+				Expect(runtimeObj.Spec.Worker.Resources.Limits).To(BeNil())
+				Expect(runtimeObj.Spec.Worker.Resources.Requests).To(BeNil())
+
+				Expect(engine.syncRuntimeSpec(ctx, runtimeObj, runtimeClass)).To(Succeed())
+
+				Expect(memLimitOf(workerSts)).To(Equal("2Gi"))
+			})
+
+			It("should still let the CacheRuntime win over the template", func() {
+				setWorkloadMemLimit(workerSts, "8Gi")
+				runtimeObj.Spec.Worker.Resources = corev1.ResourceRequirements{
+					Limits: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("4Gi")},
+				}
+
+				Expect(engine.syncRuntimeSpec(ctx, runtimeObj, runtimeClass)).To(Succeed())
+
+				Expect(memLimitOf(workerSts)).To(Equal("4Gi"))
+			})
+		})
+
+		Context("when neither the CacheRuntime nor the template specifies resources", func() {
+			It("should leave the workload's resources untouched", func() {
+				runtimeClass.Topology.Worker.Template.Spec.Containers[0].Resources = corev1.ResourceRequirements{}
+
+				sts := &workloadv1alpha1.AdvancedStatefulSet{}
+				key := types.NamespacedName{Name: workerSts, Namespace: "default"}
+				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("8Gi")},
+				}
+				Expect(fakeClient.Update(ctx.Context, sts)).To(Succeed())
+
+				Expect(engine.syncRuntimeSpec(ctx, runtimeObj, runtimeClass)).To(Succeed())
+
+				Expect(memLimitOf(workerSts)).To(Equal("8Gi"))
+			})
+		})
+	})
 })