[Bugfix] Fix Resources Copy mechanism to prevent invalid pod creation

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## [master](https://github.com/arangodb/kube-arangodb/tree/master) (N/A)
 - (Feature) Extract Scheduler API
 - (Bugfix) Fix Image Discovery
+- (Bugfix) Fix Resources Copy mechanism to prevent invalid pod creation
 
 ## [1.2.38](https://github.com/arangodb/kube-arangodb/tree/1.2.38) (2024-02-22)
 - (Feature) Extract GRPC Server

pkg/util/k8sutil/resources/resources.go

@@ -33,6 +33,16 @@ func ApplyContainerResourceRequirements(container *core.Container, resources cor
 	container.Resources.Requests = ApplyContainerResourceList(container.Resources.Requests, resources.Requests)
 }
 
+// MergeContainerResource updates resources from `from` to `to` ResourceList
+func MergeContainerResource(to core.ResourceRequirements, from core.ResourceRequirements) core.ResourceRequirements {
+	var r core.ResourceRequirements
+
+	r.Limits = MergeContainerResourceList(to.Limits, from.Limits)
+	r.Requests = MergeContainerResourceList(to.Requests, from.Requests)
+
+	return r
+}
+
 // ApplyContainerResource adds non-existing resources from `from` to `to` ResourceList
 func ApplyContainerResource(to core.ResourceRequirements, from core.ResourceRequirements) core.ResourceRequirements {
 	var r core.ResourceRequirements
@@ -43,6 +53,27 @@ func ApplyContainerResource(to core.ResourceRequirements, from core.ResourceRequ
 	return r
 }
 
+// MergeContainerResourceList updates resources from `from` to `to` ResourceList
+func MergeContainerResourceList(to core.ResourceList, from core.ResourceList) core.ResourceList {
+	if len(from) == 0 {
+		return to
+	}
+
+	if to == nil {
+		to = core.ResourceList{}
+	}
+
+	for k, v := range from {
+		if v.IsZero() {
+			delete(to, k)
+		} else {
+			to[k] = v
+		}
+	}
+
+	return to
+}
+
 // ApplyContainerResourceList adds non-existing resources from `from` to `to` ResourceList
 func ApplyContainerResourceList(to core.ResourceList, from core.ResourceList) core.ResourceList {
 	if len(from) == 0 {
@@ -69,16 +100,30 @@ func UpscaleContainerResourceRequirements(container *core.Container, resources c
 
 	container.Resources.Limits = UpscaleContainerResourceList(container.Resources.Limits, resources.Limits)
 	container.Resources.Requests = UpscaleContainerResourceList(container.Resources.Requests, resources.Requests)
+
+	// Ensure that Limits are always higher or equals requests
+	container.Resources.Limits = UpscaleOptionalContainerResourceList(container.Resources.Limits, container.Resources.Requests)
 }
 
-// UpscaleContainerResource scales up resources from `from` to `to` ResourceList
-func UpscaleContainerResource(to core.ResourceRequirements, from core.ResourceRequirements) core.ResourceRequirements {
-	var r core.ResourceRequirements
+// UpscaleOptionalContainerResourceList scales up resources from `from` to `to` ResourceList if they exists in `to`
+func UpscaleOptionalContainerResourceList(to core.ResourceList, from core.ResourceList) core.ResourceList {
+	if len(from) == 0 {
+		return to
+	}
+
+	if to == nil {
+		to = core.ResourceList{}
+	}
 
-	r.Limits = UpscaleContainerResourceList(to.Limits, from.Limits)
-	r.Requests = UpscaleContainerResourceList(to.Requests, from.Requests)
+	for k, v := range from {
+		if n, ok := to[k]; ok {
+			if n.Cmp(v) < 0 {
+				to[k] = v
+			}
+		}
+	}
 
-	return r
+	return to
 }
 
 // UpscaleContainerResourceList scales up resources from `from` to `to` ResourceList

pkg/util/k8sutil/resources/resources_test.go

@@ -163,6 +163,9 @@ func Test_ApplyContainerResourceRequirements(t *testing.T) {
 }
 
 func Test_UpsertContainerResourceRequirements(t *testing.T) {
+	v0, err := resource.ParseQuantity("512Mi")
+	require.NoError(t, err)
+
 	v1, err := resource.ParseQuantity("1Gi")
 	require.NoError(t, err)
 
@@ -174,6 +177,39 @@ func Test_UpsertContainerResourceRequirements(t *testing.T) {
 
 	var container core.Container
 
+	t.Run("Ensure limits are applied optionally", func(t *testing.T) {
+		UpscaleContainerResourceRequirements(&container, core.ResourceRequirements{
+			Requests: core.ResourceList{
+				core.ResourceMemory: v1,
+			},
+		})
+
+		require.Len(t, container.Resources.Requests, 1)
+		require.Contains(t, container.Resources.Requests, core.ResourceMemory)
+		require.Equal(t, v1, container.Resources.Requests[core.ResourceMemory])
+
+		require.Len(t, container.Resources.Limits, 0)
+	})
+
+	t.Run("Ensure limits are upscaled optionally", func(t *testing.T) {
+		UpscaleContainerResourceRequirements(&container, core.ResourceRequirements{
+			Limits: core.ResourceList{
+				core.ResourceMemory: v0,
+			},
+			Requests: core.ResourceList{
+				core.ResourceMemory: v1,
+			},
+		})
+
+		require.Len(t, container.Resources.Requests, 1)
+		require.Contains(t, container.Resources.Requests, core.ResourceMemory)
+		require.Equal(t, v1, container.Resources.Requests[core.ResourceMemory])
+
+		require.Len(t, container.Resources.Limits, 1)
+		require.Contains(t, container.Resources.Limits, core.ResourceMemory)
+		require.Equal(t, v1, container.Resources.Limits[core.ResourceMemory])
+	})
+
 	t.Run("Ensure limits are copied", func(t *testing.T) {
 		UpscaleContainerResourceRequirements(&container, core.ResourceRequirements{
 			Limits: core.ResourceList{