feat(hatchery:k8s): set memory and CPU limits for workers and services (#6187)

Author: richardlt

engine/hatchery/kubernetes/kubernetes.go

@@ -183,7 +183,15 @@ func (h *HatcheryKubernetes) SpawnWorker(ctx context.Context, spawnArgs hatchery
 		logJob = fmt.Sprintf("for workflow job %d,", spawnArgs.JobID)
 	}
 
+	cpu := h.Config.DefaultCPU
+	if cpu == "" {
+		cpu = "500m"
+	}
+
 	memory := int64(h.Config.DefaultMemory)
+	if memory == 0 {
+		memory = 1024
+	}
 	for _, r := range spawnArgs.Requirements {
 		if r.Type == sdk.MemoryRequirement {
 			var err error
@@ -192,9 +200,15 @@ func (h *HatcheryKubernetes) SpawnWorker(ctx context.Context, spawnArgs hatchery
 				log.Warn(ctx, "spawnKubernetesDockerWorker> %s unable to parse memory requirement %d: %v", logJob, memory, err)
 				return err
 			}
+			break
 		}
 	}
 
+	ephemeralStorage := h.Config.DefaultEphemeralStorage
+	if ephemeralStorage == "" {
+		ephemeralStorage = "1Gi"
+	}
+
 	workerConfig := h.GenerateWorkerConfig(ctx, h, spawnArgs)
 	udataParam := struct {
 		API string
@@ -283,7 +297,14 @@ func (h *HatcheryKubernetes) SpawnWorker(ctx context.Context, spawnArgs hatchery
 					Args:            []string{cmd},
 					Resources: apiv1.ResourceRequirements{
 						Requests: apiv1.ResourceList{
-							apiv1.ResourceMemory: resource.MustParse(fmt.Sprintf("%d", memory)),
+							apiv1.ResourceCPU:              resource.MustParse(cpu),
+							apiv1.ResourceMemory:           *resource.NewScaledQuantity(memory, resource.Mega),
+							apiv1.ResourceEphemeralStorage: resource.MustParse(ephemeralStorage),
+						},
+						Limits: apiv1.ResourceList{
+							apiv1.ResourceCPU:              resource.MustParse(cpu),
+							apiv1.ResourceMemory:           *resource.NewScaledQuantity(memory, resource.Mega),
+							apiv1.ResourceEphemeralStorage: resource.MustParse(ephemeralStorage),
 						},
 					},
 				},
@@ -313,28 +334,51 @@ func (h *HatcheryKubernetes) SpawnWorker(ctx context.Context, spawnArgs hatchery
 		podSchema.Spec.HostAliases[0].Hostnames[0] = "worker"
 	}
 
+	serviceCPU := h.Config.DefaultServiceCPU
+	if serviceCPU == "" {
+		serviceCPU = "256m"
+	}
+
+	serviceMemory := int64(h.Config.DefaultServiceMemory)
+	if serviceMemory == 0 {
+		serviceMemory = 512
+	}
+
+	serviceEphemeralStorage := h.Config.DefaultServiceEphemeralStorage
+	if serviceEphemeralStorage == "" {
+		serviceEphemeralStorage = "512Mi"
+	}
+
 	for i, serv := range services {
 		//name= <alias> => the name of the host put in /etc/hosts of the worker
 		//value= "postgres:latest env_1=blabla env_2=blabla"" => we can add env variables in requirement name
 		img, envm := hatchery.ParseRequirementModel(serv.Value)
 
-		servContainer := apiv1.Container{
-			Name:  fmt.Sprintf("service-%d-%s", serv.ID, strings.ToLower(serv.Name)),
-			Image: img,
-		}
-
 		if sm, ok := envm["CDS_SERVICE_MEMORY"]; ok {
-			mq, err := resource.ParseQuantity(sm)
+			var err error
+			serviceMemory, err = strconv.ParseInt(sm, 10, 64)
 			if err != nil {
 				log.Warn(ctx, "hatchery> kubernetes> SpawnWorker> Unable to parse CDS_SERVICE_MEMORY value '%s': %s", sm, err)
 				continue
 			}
-			servContainer.Resources = apiv1.ResourceRequirements{
+			delete(envm, "CDS_SERVICE_MEMORY")
+		}
+
+		servContainer := apiv1.Container{
+			Name:  fmt.Sprintf("service-%d-%s", serv.ID, strings.ToLower(serv.Name)),
+			Image: img,
+			Resources: apiv1.ResourceRequirements{
 				Requests: apiv1.ResourceList{
-					apiv1.ResourceMemory: mq,
+					apiv1.ResourceCPU:              resource.MustParse(serviceCPU),
+					apiv1.ResourceMemory:           *resource.NewScaledQuantity(serviceMemory, resource.Mega),
+					apiv1.ResourceEphemeralStorage: resource.MustParse(serviceEphemeralStorage),
 				},
-			}
-			delete(envm, "CDS_SERVICE_MEMORY")
+				Limits: apiv1.ResourceList{
+					apiv1.ResourceCPU:              resource.MustParse(serviceCPU),
+					apiv1.ResourceMemory:           *resource.NewScaledQuantity(serviceMemory, resource.Mega),
+					apiv1.ResourceEphemeralStorage: resource.MustParse(serviceEphemeralStorage),
+				},
+			},
 		}
 
 		if sa, ok := envm["CDS_SERVICE_ARGS"]; ok {

engine/hatchery/kubernetes/kubernetes_test.go

@@ -100,7 +100,7 @@ func TestHatcheryKubernetes_Status(t *testing.T) {
 
 			require.Equal(t, 2, len(podRequest.Spec.Containers))
 			require.Equal(t, "my-worker", podRequest.Spec.Containers[0].Name)
-			require.Equal(t, int64(4096), podRequest.Spec.Containers[0].Resources.Requests.Memory().Value())
+			require.Equal(t, int64(4096000000), podRequest.Spec.Containers[0].Resources.Requests.Memory().Value())
 			var foundEnv, foundSecret bool
 			for _, env := range podRequest.Spec.Containers[0].Env {
 				if env.Name == "PROVISION_ENV" && env.Value == "MYVALUE" {

engine/hatchery/kubernetes/types.go

@@ -19,8 +19,18 @@ var containerServiceNameRegexp = regexp.MustCompile(`service-([0-9]+)-(.*)`)
 // HatcheryConfiguration is the configuration for local hatchery
 type HatcheryConfiguration struct {
 	service.HatcheryCommonConfiguration `mapstructure:"commonConfiguration" toml:"commonConfiguration" json:"commonConfiguration"`
+	// DefaultCPU Worker default CPU
+	DefaultCPU string `mapstructure:"defaultCPU" toml:"defaultCPU" default:"500m" commented:"false" comment:"Worker default CPU" json:"defaultCPU"`
 	// DefaultMemory Worker default memory
 	DefaultMemory int `mapstructure:"defaultMemory" toml:"defaultMemory" default:"1024" commented:"false" comment:"Worker default memory in Mo" json:"defaultMemory"`
+	// DefaultEphemeralStorage Worker default ephemeral storage size
+	DefaultEphemeralStorage string `mapstructure:"defaultEphemeralStorage" toml:"defaultEphemeralStorage" default:"1Gi" commented:"false" comment:"Worker default ephemeral storage size" json:"defaultEphemeralStorage"`
+	// DefaultServiceCPU Service default CPU
+	DefaultServiceCPU string `mapstructure:"defaultServiceCPU" toml:"defaultServiceCPU" default:"256m" commented:"false" comment:"Service default CPU" json:"defaultServiceCPU"`
+	// DefaultServiceMemory Service default memory
+	DefaultServiceMemory int `mapstructure:"defaultServiceMemory" toml:"defaultServiceMemory" default:"512" commented:"false" comment:"Service default memory in Mo" json:"defaultServiceMemory"`
+	// DefaultServiceEphemeralStorage Service default ephemeral storage size
+	DefaultServiceEphemeralStorage string `mapstructure:"defaultServiceEphemeralStorage" toml:"defaultServiceEphemeralStorage" default:"512Mi" commented:"false" comment:"Service default ephemeral storage size" json:"defaultServiceEphemeralStorage"`
 	// Namespace is the kubernetes namespace in which workers are spawned"
 	Namespace string `mapstructure:"namespace" toml:"namespace" default:"cds" commented:"false" comment:"Kubernetes namespace in which workers are spawned" json:"namespace"`
 	// KubernetesMasterURL Address of kubernetes master