fix(cache): merge CacheRuntime resources onto the CacheRuntimeClass baseline

docs/en/samples/cacheruntime/cacheruntime_spec_update.md

@@ -77,15 +77,22 @@ spec:
 
 `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.
+1. the `resources` declared by the CacheRuntimeClass template form the baseline;
+2. the `resources` set on the CacheRuntime are overlaid on that baseline key by key —
+   `requests` and `limits` merge by resource name, `claims` by claim name;
+3. when neither declares anything, nothing is synced and the workload keeps its current resources.
+
+The CacheRuntime therefore expresses only the differences it wants from the template. Naming
+`limits.memory` alone moves that one value and leaves the template's `requests.cpu`,
+`requests.memory` and `limits.cpu` in place.
 
 **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.
+- ⚠️ A key the CacheRuntimeClass template declares can be **overridden but not removed**: omitting it
+  from the CacheRuntime leaves the template's value in place rather than leaving the component
+  unconstrained. To relax a limit, set the value you want explicitly; to drop a requirement
+  altogether, change the CacheRuntimeClass template, which is where the runtime's own requirements
+  are described.
 - ⚠️ **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,18 +73,23 @@ spec:
         memory: 16Gi
 ```
 
-**取值优先级**：
+**取值方式**：
 
-每次 reconcile 时，`resources` 按以下顺序取值：
+每次 reconcile 时，`resources` 按以下方式取值：
 
-1. CacheRuntime 上设置的值（只要声明了 `requests` 或 `limits`）；
-2. 否则取 CacheRuntimeClass 模板中声明的值；
+1. 以 CacheRuntimeClass 模板中声明的 `resources` 为基线；
+2. 将 CacheRuntime 上设置的 `resources` 按 key 叠加到该基线上——`requests` 与 `limits` 按资源名
+   合并，`claims` 按 claim 名合并；
 3. 两者都未声明时不做同步，工作负载保持当前的资源配置。
 
+也就是说，CacheRuntime 只需要表达相对模板的差异。只设置 `limits.memory` 时，仅该值被修改，模板中的
+`requests.cpu`、`requests.memory` 和 `limits.cpu` 保持不变。
+
 **限制**：
 - ⚠️ 不能超过节点可用资源
-- ⚠️ 当 CacheRuntimeClass 模板声明了 `resources` 时，从 CacheRuntime 中删除 `resources` **不会**
-  让组件变为不受限，而是回退到模板中的值。如需放宽限制，请显式设置目标值，而不是删除该字段。
+- ⚠️ CacheRuntimeClass 模板中声明的 key **可以覆盖，但不能删除**：从 CacheRuntime 中省略某个 key 不会
+  让组件变为不受限，而是保留模板中的值。如需放宽限制，请显式设置目标值；如需彻底去掉某项资源要求，
+  请修改 CacheRuntimeClass 模板——运行时自身的资源要求本就描述在那里。
 - ⚠️ **Kubernetes 版本要求**：需要 K8s >= 1.27 且启用 `InPlacePodVerticalScaling` Feature Gate
   ```bash
   # 检查 Feature Gate 是否启用

pkg/ddc/cache/engine/sync.go

@@ -229,28 +229,26 @@ func (e *CacheEngine) syncRuntimeSpec(ctx cruntime.ReconcileRequestContext, runt
 }
 
 // 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.
+// component's workload. It mirrors the creation path: the values set on the CacheRuntime
+// are overlaid key by key on the CacheRuntimeClass template baseline, so a key the
+// CacheRuntime does not name keeps the template value the creation path rendered into the
+// workload instead of being dropped from it. 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()
+	var templateResources corev1.ResourceRequirements
+	if componentDefinition != nil && len(componentDefinition.Template.Spec.Containers) > 0 {
+		templateResources = componentDefinition.Template.Spec.Containers[0].Resources
 	}
 
-	if componentDefinition == nil || len(componentDefinition.Template.Spec.Containers) == 0 {
+	desired := mergeResourceRequirements(templateResources, runtimeResources)
+	if desired.Requests == nil && desired.Limits == nil && desired.Claims == nil {
 		return nil
 	}
 
-	templateResources := componentDefinition.Template.Spec.Containers[0].Resources
-	if templateResources.Requests == nil && templateResources.Limits == nil {
-		return nil
-	}
-
-	return templateResources.DeepCopy()
+	return &desired
 }
 
 func (e *CacheEngine) syncDatasetCacheStates(ctx cruntime.ReconcileRequestContext, runtime *datav1alpha1.CacheRuntime, runtimeClass *datav1alpha1.CacheRuntimeClass) (err error) {

pkg/ddc/cache/engine/sync_test.go

@@ -932,6 +932,76 @@ var _ = Describe("CacheEngine Sync Tests", Label("pkg.ddc.cache.engine.sync_test
 			})
 		})
 
+		// Issue #6173: a CacheRuntime that names one key used to replace the container's
+		// whole ResourceRequirements, so raising the memory limit alone left the container
+		// with no CPU request and no CPU limit at all.
+		Context("when the CacheRuntime only names part of the template's resources", func() {
+			fullTemplateResources := corev1.ResourceRequirements{
+				Requests: corev1.ResourceList{
+					corev1.ResourceCPU:    resource.MustParse("1"),
+					corev1.ResourceMemory: resource.MustParse("2Gi"),
+				},
+				Limits: corev1.ResourceList{
+					corev1.ResourceCPU:    resource.MustParse("2"),
+					corev1.ResourceMemory: resource.MustParse("4Gi"),
+				},
+			}
+
+			workerWorkload := func() *workloadv1alpha1.AdvancedStatefulSet {
+				sts := &workloadv1alpha1.AdvancedStatefulSet{}
+				key := types.NamespacedName{Name: workerSts, Namespace: "default"}
+				Expect(fakeClient.Get(ctx.Context, key, sts)).To(Succeed())
+				return sts
+			}
+
+			// The CacheRuntime raises the memory limit and names nothing else.
+			BeforeEach(func() {
+				runtimeObj.Spec.Worker.Resources = corev1.ResourceRequirements{
+					Limits: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("8Gi")},
+				}
+			})
+
+			Context("and the template declares a full set of requirements", func() {
+				// Re-seed over the limits-only template the outer BeforeEach installs, so
+				// the spec has requirements left to lose.
+				BeforeEach(func() {
+					runtimeClass.Topology.Worker.Template.Spec.Containers[0].Resources = *fullTemplateResources.DeepCopy()
+
+					sts := workerWorkload()
+					sts.Spec.Template.Spec.Containers[0].Resources = *fullTemplateResources.DeepCopy()
+					Expect(fakeClient.Update(ctx.Context, sts)).To(Succeed())
+				})
+
+				It("should move the key it names and keep the template's other requirements", func() {
+					Expect(engine.syncRuntimeSpec(ctx, runtimeObj, runtimeClass)).To(Succeed())
+
+					synced := workerWorkload().Spec.Template.Spec.Containers[0].Resources
+					Expect(synced.Limits.Memory().String()).To(Equal("8Gi"))
+					Expect(synced.Limits.Cpu().String()).To(Equal("2"))
+					Expect(synced.Requests.Cpu().String()).To(Equal("1"))
+					Expect(synced.Requests.Memory().String()).To(Equal("2Gi"))
+				})
+			})
+
+			Context("and neither side declares requests", func() {
+				// The template the outer BeforeEach installs is limits-only. A merge that
+				// turned the absent requests into an empty map instead of leaving them nil
+				// would never compare equal to the workload, and the sync would patch on
+				// every reconcile.
+				It("should leave requests unset and stop patching once converged", func() {
+					Expect(engine.syncRuntimeSpec(ctx, runtimeObj, runtimeClass)).To(Succeed())
+
+					synced := workerWorkload().Spec.Template.Spec.Containers[0].Resources
+					Expect(synced.Limits.Memory().String()).To(Equal("8Gi"))
+					Expect(synced.Requests).To(BeNil())
+
+					converged := workerWorkload().ResourceVersion
+					Expect(engine.syncRuntimeSpec(ctx, runtimeObj, runtimeClass)).To(Succeed())
+					Expect(workerWorkload().ResourceVersion).To(Equal(converged))
+				})
+			})
+		})
+
 		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.

pkg/ddc/cache/engine/transform_common.go

@@ -127,10 +127,11 @@ func (e *CacheEngine) transformComponentPodTemplate(runtimeCompSpec datav1alpha1
 			podTemplate.Spec.Containers[0].ImagePullPolicy = (corev1.PullPolicy)(runtimeCompSpec.RuntimeVersion.ImagePullPolicy)
 		}
 
-		// use runtime component resources if specified, otherwise use default resources
-		if runtimeCompSpec.Resources.Limits != nil || runtimeCompSpec.Resources.Requests != nil {
-			podTemplate.Spec.Containers[0].Resources = runtimeCompSpec.Resources
-		}
+		// Overlay the runtime component resources on the CacheRuntimeClass template
+		// baseline key by key: a partially specified resources only moves the keys it
+		// names and leaves the rest of the template's requirements in place.
+		podTemplate.Spec.Containers[0].Resources = mergeResourceRequirements(
+			podTemplate.Spec.Containers[0].Resources, runtimeCompSpec.Resources)
 
 		if runtimeCompSpec.Args != nil {
 			podTemplate.Spec.Containers[0].Args = runtimeCompSpec.Args
@@ -148,3 +149,66 @@ func (e *CacheEngine) transformComponentPodTemplate(runtimeCompSpec datav1alpha1
 		componentValue.PodTemplateSpec.Spec.InitContainers[0].Env = append(addEnvs, componentValue.PodTemplateSpec.Spec.InitContainers[0].Env...)
 	}
 }
+
+// mergeResourceRequirements overlays the resources declared on a CacheRuntime component
+// on top of the baseline rendered from the CacheRuntimeClass template, key by key.
+//
+// The two are not alternatives: the template carries the runtime's own requirements and
+// the CacheRuntime only expresses the deltas an owner wants for their instance, so
+// replacing the whole struct would silently drop every requirement the CacheRuntime does
+// not restate. A key the overlay does not name keeps its template value; a key it names
+// wins, including a key the template never declared.
+//
+// The corollary is that a key set by the template cannot be removed by omitting it from
+// the CacheRuntime, only overridden. Removing a requirement is a change to the template,
+// which is where the runtime's requirements are described in the first place.
+//
+// Claims are name-keyed rather than merged by resource name: an overlay claim replaces
+// the template claim with the same name and any other claim is appended, preserving the
+// template's order.
+func mergeResourceRequirements(base, overlay corev1.ResourceRequirements) corev1.ResourceRequirements {
+	merged := *base.DeepCopy()
+	merged.Limits = mergeResourceList(merged.Limits, overlay.Limits)
+	merged.Requests = mergeResourceList(merged.Requests, overlay.Requests)
+	merged.Claims = mergeResourceClaims(merged.Claims, overlay.Claims)
+	return merged
+}
+
+// mergeResourceList applies the overlay entries onto base, modifying it in place, and
+// returns it. Callers own base: mergeResourceRequirements hands over a deep copy. A nil
+// base is kept nil when the overlay declares nothing, so that an untouched component
+// still compares equal to the workload it was rendered into.
+func mergeResourceList(base, overlay corev1.ResourceList) corev1.ResourceList {
+	if len(overlay) == 0 {
+		return base
+	}
+	if base == nil {
+		base = corev1.ResourceList{}
+	}
+	for name, quantity := range overlay {
+		base[name] = quantity.DeepCopy()
+	}
+	return base
+}
+
+// mergeResourceClaims applies the overlay claims onto base by name, modifying it in
+// place, and returns it.
+func mergeResourceClaims(base, overlay []corev1.ResourceClaim) []corev1.ResourceClaim {
+	if len(overlay) == 0 {
+		return base
+	}
+	for _, claim := range overlay {
+		replaced := false
+		for i := range base {
+			if base[i].Name == claim.Name {
+				base[i] = *claim.DeepCopy()
+				replaced = true
+				break
+			}
+		}
+		if !replaced {
+			base = append(base, *claim.DeepCopy())
+		}
+	}
+	return base
+}