diff --git a/pkg/ddc/cache/component/advanced_statefulset_manager.go b/pkg/ddc/cache/component/advanced_statefulset_manager.go index 3dc43376b90..e9fb9e08df0 100644 --- a/pkg/ddc/cache/component/advanced_statefulset_manager.go +++ b/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 diff --git a/pkg/ddc/cache/component/component_manager.go b/pkg/ddc/cache/component/component_manager.go index 7f18149a8fe..808aff25d22 100644 --- a/pkg/ddc/cache/component/component_manager.go +++ b/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 diff --git a/pkg/ddc/cache/component/component_test.go b/pkg/ddc/cache/component/component_test.go index 5478fe145b5..e5e4263c039 100644 --- a/pkg/ddc/cache/component/component_test.go +++ b/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()) + }) + }) }) diff --git a/pkg/ddc/cache/component/daemonset_manager.go b/pkg/ddc/cache/component/daemonset_manager.go index 01c652dd00b..9cda1997fa3 100644 --- a/pkg/ddc/cache/component/daemonset_manager.go +++ b/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") diff --git a/pkg/ddc/cache/component/sync_component_spec_test.go b/pkg/ddc/cache/component/sync_component_spec_test.go index d719169e36b..719b3df7272 100644 --- a/pkg/ddc/cache/component/sync_component_spec_test.go +++ b/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")) + }) }) }) diff --git a/pkg/ddc/cache/engine/sync.go b/pkg/ddc/cache/engine/sync.go index a197649546d..29b8849cd06 100644 --- a/pkg/ddc/cache/engine/sync.go +++ b/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 { diff --git a/pkg/ddc/cache/engine/sync_test.go b/pkg/ddc/cache/engine/sync_test.go index f307f40c13a..5d5eead08c2 100644 --- a/pkg/ddc/cache/engine/sync_test.go +++ b/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 "" + } + 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 + // + . 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)) + }) + }) }) }) diff --git a/pkg/ddc/cache/engine/transform_tiered_store.go b/pkg/ddc/cache/engine/transform_tiered_store.go index ac451b45ee5..e3048309940 100644 --- a/pkg/ddc/cache/engine/transform_tiered_store.go +++ b/pkg/ddc/cache/engine/transform_tiered_store.go @@ -18,12 +18,18 @@ package engine import ( "fmt" + "strings" datav1alpha1 "github.com/fluid-cloudnative/fluid/api/v1alpha1" "github.com/fluid-cloudnative/fluid/pkg/utils" corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/resource" ) +// tieredStoreVolumeNamePrefix prefixes every volume TransformRuntimeTieredStore +// generates, so the quota charged to a workload can later be recovered from it. +const tieredStoreVolumeNamePrefix = "tiered-store-level-" + // TransformRuntimeTieredStore transforms the tiered store configuration to worker pod spec func (e *CacheEngine) TransformRuntimeTieredStore(tieredStore *datav1alpha1.RuntimeTieredStore, podSpec *corev1.PodSpec) error { if len(tieredStore.Levels) == 0 { @@ -101,21 +107,10 @@ func (e *CacheEngine) handleProcessMemory(podSpec *corev1.PodSpec, container *co totalQuota := memoryMediumSource.Quota.DeepCopy() // add totalQuota to memory resources only when memory is restricted. - if container.Resources.Requests != nil { - if currentRequest, exists := container.Resources.Requests[corev1.ResourceMemory]; exists && !currentRequest.IsZero() { - currentRequest.Add(totalQuota) - container.Resources.Requests[corev1.ResourceMemory] = currentRequest - } - } - if container.Resources.Limits != nil { - if currentLimit, exists := container.Resources.Limits[corev1.ResourceMemory]; exists && !currentLimit.IsZero() { - currentLimit.Add(totalQuota) - container.Resources.Limits[corev1.ResourceMemory] = currentLimit - } - } + container.Resources = withTieredStoreMemoryQuota(container.Resources, totalQuota) // add an memory emptyDir for /dev/shm in the container - volumeName := fmt.Sprintf("tiered-store-level-%d-memory", levelIndex) + volumeName := getMemoryTieredStoreVolumeName(levelIndex) mountPath := GetMemoryTieredStoreMountPath(levelIndex) volume := corev1.Volume{ Name: volumeName, @@ -150,7 +145,7 @@ func (e *CacheEngine) handleHostPath(podSpec *corev1.PodSpec, container *corev1. // Process each path and corresponding quota for i, hostPath := range hostPathMediumSource.Paths { - volumeName := fmt.Sprintf("tiered-store-level-%d-index-%d", levelIndex, i) + volumeName := getTieredStoreVolumeName(levelIndex, i) mountPath := GetHostPathTieredStoreMountPath(levelIndex, i) volume := corev1.Volume{ @@ -184,7 +179,7 @@ func (e *CacheEngine) handleEmptyDir(podSpec *corev1.PodSpec, container *corev1. return fmt.Errorf("emptyDir quota cannot be zero for empty dir medium source at level index %d", levelIndex) } - volumeName := fmt.Sprintf("tiered-store-level-%d-index-%d", levelIndex, 0) + volumeName := getTieredStoreVolumeName(levelIndex, 0) mountPath := GetEmptyDirTieredStoreMountPath(levelIndex) quota := emptyDirMediumSource.Quota.DeepCopy() @@ -211,21 +206,53 @@ func (e *CacheEngine) handleEmptyDir(podSpec *corev1.PodSpec, container *corev1. // For Memory-backed EmptyDir (tmpfs), add quota to container memory resources // This ensures proper resource accounting and prevents excessive memory usage if emptyDirMediumSource.Medium == corev1.StorageMediumMemory { - // Only add to resources if the container already has memory constraints - // If no memory resources are set, the container is unconstrained and we don't need to add - if container.Resources.Requests != nil { - if currentRequest, exists := container.Resources.Requests[corev1.ResourceMemory]; exists && !currentRequest.IsZero() { - currentRequest.Add(quota) - container.Resources.Requests[corev1.ResourceMemory] = currentRequest - } + container.Resources = withTieredStoreMemoryQuota(container.Resources, quota) + } + + return nil +} + +// chargedTieredStoreMemoryQuota sums the tiered store memory quota already charged +// to a workload's container memory, recovering it from the tmpfs volumes the +// creation path wrote. It covers exactly the levels TransformRuntimeTieredStore +// charges -- process memory and Memory-medium emptyDir -- both of which carry the +// quota as the volume's size limit. +// +// The workload is authoritative rather than the CacheRuntime spec, because +// tieredStore is not a supported update field. Recomputing the quota from an edited +// spec would move the container's memory while the tmpfs volumes it must cover keep +// the size they were created with. +func chargedTieredStoreMemoryQuota(podSpec *corev1.PodSpec) resource.Quantity { + total := *resource.NewQuantity(0, resource.BinarySI) + for _, volume := range podSpec.Volumes { + if !strings.HasPrefix(volume.Name, tieredStoreVolumeNamePrefix) { + continue } - if container.Resources.Limits != nil { - if currentLimit, exists := container.Resources.Limits[corev1.ResourceMemory]; exists && !currentLimit.IsZero() { - currentLimit.Add(quota) - container.Resources.Limits[corev1.ResourceMemory] = currentLimit - } + emptyDir := volume.EmptyDir + if emptyDir == nil || emptyDir.Medium != corev1.StorageMediumMemory || emptyDir.SizeLimit == nil { + continue } + total.Add(*emptyDir.SizeLimit) } + return total +} - return nil +// withTieredStoreMemoryQuota returns base with quota added to its memory request +// and limit. Absent or zero constraints are left untouched, so a container +// without memory constraints stays unconstrained. +func withTieredStoreMemoryQuota(base corev1.ResourceRequirements, quota resource.Quantity) corev1.ResourceRequirements { + result := *base.DeepCopy() + if quota.IsZero() { + return result + } + for _, list := range []corev1.ResourceList{result.Requests, result.Limits} { + if list == nil { + continue + } + if cur, exists := list[corev1.ResourceMemory]; exists && !cur.IsZero() { + cur.Add(quota) + list[corev1.ResourceMemory] = cur + } + } + return result } diff --git a/pkg/ddc/cache/engine/transform_tiered_store_test.go b/pkg/ddc/cache/engine/transform_tiered_store_test.go index 44099d9ae2e..d32ed6e3c11 100644 --- a/pkg/ddc/cache/engine/transform_tiered_store_test.go +++ b/pkg/ddc/cache/engine/transform_tiered_store_test.go @@ -17,6 +17,8 @@ limitations under the License. package engine import ( + "fmt" + . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -116,7 +118,7 @@ var _ = Describe("CacheEngine TransformRuntimeTieredStore Tests", Label("pkg.ddc // Verify volume and volume mount are created for ProcessMemory Expect(podSpec.Volumes).To(HaveLen(1)) - Expect(podSpec.Volumes[0].Name).To(Equal("tiered-store-level-0-memory")) + Expect(podSpec.Volumes[0].Name).To(Equal(getMemoryTieredStoreVolumeName(0))) Expect(podSpec.Volumes[0].EmptyDir).NotTo(BeNil()) Expect(podSpec.Volumes[0].EmptyDir.Medium).To(Equal(corev1.StorageMediumMemory)) @@ -207,13 +209,13 @@ var _ = Describe("CacheEngine TransformRuntimeTieredStore Tests", Label("pkg.ddc // Verify volume is created Expect(podSpec.Volumes).To(HaveLen(1)) - Expect(podSpec.Volumes[0].Name).To(Equal("tiered-store-level-0-index-0")) + Expect(podSpec.Volumes[0].Name).To(Equal(getTieredStoreVolumeName(0, 0))) Expect(podSpec.Volumes[0].HostPath).NotTo(BeNil()) Expect(podSpec.Volumes[0].HostPath.Path).To(Equal("/mnt/cache1")) // Verify volume mount is created Expect(podSpec.Containers[0].VolumeMounts).To(HaveLen(1)) - Expect(podSpec.Containers[0].VolumeMounts[0].Name).To(Equal("tiered-store-level-0-index-0")) + Expect(podSpec.Containers[0].VolumeMounts[0].Name).To(Equal(getTieredStoreVolumeName(0, 0))) Expect(podSpec.Containers[0].VolumeMounts[0].MountPath).To(ContainSubstring("tiered-store")) }) @@ -239,8 +241,8 @@ var _ = Describe("CacheEngine TransformRuntimeTieredStore Tests", Label("pkg.ddc // Verify 3 volumes are created Expect(podSpec.Volumes).To(HaveLen(3)) for i := 0; i < 3; i++ { - Expect(podSpec.Volumes[i].Name).To(Equal("tiered-store-level-0-index-" + string(rune('0'+i)))) - Expect(podSpec.Volumes[i].HostPath.Path).To(Equal("/mnt/cache" + string(rune('1'+i)))) + Expect(podSpec.Volumes[i].Name).To(Equal(getTieredStoreVolumeName(0, i))) + Expect(podSpec.Volumes[i].HostPath.Path).To(Equal(fmt.Sprintf("/mnt/cache%d", i+1))) } // Verify 3 volume mounts are created @@ -285,7 +287,7 @@ var _ = Describe("CacheEngine TransformRuntimeTieredStore Tests", Label("pkg.ddc // Verify volume is created Expect(podSpec.Volumes).To(HaveLen(1)) - Expect(podSpec.Volumes[0].Name).To(Equal("tiered-store-level-0-index-0")) + Expect(podSpec.Volumes[0].Name).To(Equal(getTieredStoreVolumeName(0, 0))) Expect(podSpec.Volumes[0].EmptyDir).NotTo(BeNil()) Expect(podSpec.Volumes[0].EmptyDir.Medium).To(Equal(corev1.StorageMediumDefault)) @@ -434,4 +436,101 @@ var _ = Describe("CacheEngine TransformRuntimeTieredStore Tests", Label("pkg.ddc }) }) }) + + // chargedTieredStoreMemoryQuota recovers from a workload the quota that + // TransformRuntimeTieredStore charged to container memory. It must select exactly + // the volumes that carry that quota, because the sync path adds whatever it returns + // on top of the user's baseline. + Describe("chargedTieredStoreMemoryQuota", func() { + memoryVolume := func(name, size string) corev1.Volume { + quota := resource.MustParse(size) + return corev1.Volume{ + Name: name, + VolumeSource: corev1.VolumeSource{ + EmptyDir: &corev1.EmptyDirVolumeSource{ + Medium: corev1.StorageMediumMemory, + SizeLimit: "a, + }, + }, + } + } + + It("should return zero for a pod spec without volumes", func() { + empty := chargedTieredStoreMemoryQuota(&corev1.PodSpec{}) + Expect(empty.IsZero()).To(BeTrue()) + }) + + It("should sum the memory-backed tiered store volumes", func() { + podSpec := &corev1.PodSpec{Volumes: []corev1.Volume{ + memoryVolume("tiered-store-level-0-memory", "8Gi"), + memoryVolume("tiered-store-level-1-index-0", "2Gi"), + }} + + quota := chargedTieredStoreMemoryQuota(podSpec) + Expect(quota.String()).To(Equal("10Gi")) + }) + + It("should round-trip the quota that the transform charged", func() { + podSpec := &corev1.PodSpec{Containers: []corev1.Container{{ + Name: "worker", + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("4Gi")}, + }, + }}} + tieredStore := &datav1alpha1.RuntimeTieredStore{ + Levels: []datav1alpha1.RuntimeTieredStoreLevel{ + {ProcessMemory: &datav1alpha1.ProcessMemoryMediumSource{Quota: resource.MustParse("8Gi")}}, + }, + } + + Expect(engine.TransformRuntimeTieredStore(tieredStore, podSpec)).To(Succeed()) + + // The container was charged 4Gi + 8Gi, and the quota is recoverable from the volume. + limit := podSpec.Containers[0].Resources.Limits[corev1.ResourceMemory] + Expect(limit.String()).To(Equal("12Gi")) + recovered := chargedTieredStoreMemoryQuota(podSpec) + Expect(recovered.String()).To(Equal("8Gi")) + }) + + It("should ignore volumes the transform never charges to memory", func() { + hostPathQuota := resource.MustParse("100Gi") + diskQuota := resource.MustParse("50Gi") + podSpec := &corev1.PodSpec{Volumes: []corev1.Volume{ + memoryVolume("tiered-store-level-0-memory", "8Gi"), + // hostPath levels are not charged to container memory + { + Name: "tiered-store-level-1-index-0", + VolumeSource: corev1.VolumeSource{HostPath: &corev1.HostPathVolumeSource{Path: "/mnt/disk"}}, + }, + // a disk-backed emptyDir level is not charged either + { + Name: "tiered-store-level-2-index-0", + VolumeSource: corev1.VolumeSource{ + EmptyDir: &corev1.EmptyDirVolumeSource{SizeLimit: &diskQuota}, + }, + }, + // a memory volume with no size limit contributes nothing + { + Name: "tiered-store-level-3-memory", + VolumeSource: corev1.VolumeSource{ + EmptyDir: &corev1.EmptyDirVolumeSource{Medium: corev1.StorageMediumMemory}, + }, + }, + // a tmpfs the CacheRuntimeClass template declares itself is not a tiered + // store level, and charging it would inflate the container on every sync + { + Name: "user-defined-shm", + VolumeSource: corev1.VolumeSource{ + EmptyDir: &corev1.EmptyDirVolumeSource{ + Medium: corev1.StorageMediumMemory, + SizeLimit: &hostPathQuota, + }, + }, + }, + }} + + quota := chargedTieredStoreMemoryQuota(podSpec) + Expect(quota.String()).To(Equal("8Gi")) + }) + }) }) diff --git a/pkg/ddc/cache/engine/util.go b/pkg/ddc/cache/engine/util.go index 00dc52eb3ca..52ca9ff85a1 100644 --- a/pkg/ddc/cache/engine/util.go +++ b/pkg/ddc/cache/engine/util.go @@ -129,3 +129,13 @@ func GetEmptyDirTieredStoreMountPath(levelIndex int) string { func getTieredStoreMountPath(levelIndex int, pathIndex int, mediumType string) string { return fmt.Sprintf("/etc/fluid/mount/tiered-store/level-%d-index-%d-%s", levelIndex, pathIndex, mediumType) } + +// getTieredStoreVolumeName generates the volume name for an indexed tiered store medium +func getTieredStoreVolumeName(levelIndex int, pathIndex int) string { + return fmt.Sprintf("%s%d-index-%d", tieredStoreVolumeNamePrefix, levelIndex, pathIndex) +} + +// getMemoryTieredStoreVolumeName generates the volume name for the process memory medium +func getMemoryTieredStoreVolumeName(levelIndex int) string { + return fmt.Sprintf("%s%d-memory", tieredStoreVolumeNamePrefix, levelIndex) +} diff --git a/pkg/ddc/cache/engine/util_test.go b/pkg/ddc/cache/engine/util_test.go index 440f0ce69f2..43df8d75361 100644 --- a/pkg/ddc/cache/engine/util_test.go +++ b/pkg/ddc/cache/engine/util_test.go @@ -194,3 +194,29 @@ var _ = Describe("getSecretVolumeName Tests", Label("pkg.ddc.cache.engine.util_t }) }) }) + +var _ = Describe("getTieredStoreVolumeName Tests", Label("pkg.ddc.cache.engine.util_test.go"), func() { + // The literals are pinned here on purpose: the names are rendered into the + // workload, and chargedTieredStoreMemoryQuota recovers the quota by matching + // them, so a rename is a breaking change rather than an implementation detail. + It("should generate the volume name for an indexed medium", func() { + Expect(getTieredStoreVolumeName(0, 0)).To(Equal("tiered-store-level-0-index-0")) + Expect(getTieredStoreVolumeName(1, 2)).To(Equal("tiered-store-level-1-index-2")) + Expect(getTieredStoreVolumeName(0, 10)).To(Equal("tiered-store-level-0-index-10")) + }) + + It("should generate the volume name for the process memory medium", func() { + Expect(getMemoryTieredStoreVolumeName(0)).To(Equal("tiered-store-level-0-memory")) + Expect(getMemoryTieredStoreVolumeName(3)).To(Equal("tiered-store-level-3-memory")) + }) + + It("should generate names chargedTieredStoreMemoryQuota recognises", func() { + Expect(getTieredStoreVolumeName(0, 0)).To(HavePrefix(tieredStoreVolumeNamePrefix)) + Expect(getMemoryTieredStoreVolumeName(0)).To(HavePrefix(tieredStoreVolumeNamePrefix)) + }) + + It("should always generate valid DNS-1035 label names", func() { + Expect(validation.IsDNS1035Label(getTieredStoreVolumeName(9, 9))).To(BeEmpty()) + Expect(validation.IsDNS1035Label(getMemoryTieredStoreVolumeName(9))).To(BeEmpty()) + }) +})