Rename k8s resources envs. Add the same options to shuffle app

functions/onprem/orborus/orborus.go

@@ -799,13 +799,13 @@ func buildResourcesFromEnv() corev1.ResourceRequirements {
 
 	items := []item{
 		// kubernetes requests
-		{env: "KUBERNETES_CPU_REQUEST", rn: corev1.ResourceCPU, to: &reqs},
+		{env: "SHUFFLE_WORKER_CPU_REQUEST", rn: corev1.ResourceCPU, to: &reqs},
-		{env: "KUBERNETES_MEMORY_REQUEST", rn: corev1.ResourceMemory, to: &reqs},
+		{env: "SHUFFLE_WORKER_MEMORY_REQUEST", rn: corev1.ResourceMemory, to: &reqs},
-		{env: "KUBERNETES_EPHEMERAL_STORAGE_REQUEST", rn: corev1.ResourceEphemeralStorage, to: &reqs},
+		{env: "SHUFFLE_WORKER_EPHEMERAL_STORAGE_REQUEST", rn: corev1.ResourceEphemeralStorage, to: &reqs},
 		// kubernetes limits
-		{env: "KUBERNETES_CPU_LIMIT", rn: corev1.ResourceCPU, to: &lims},
+		{env: "SHUFFLE_WORKER_CPU_LIMIT", rn: corev1.ResourceCPU, to: &lims},
-		{env: "KUBERNETES_MEMORY_LIMIT", rn: corev1.ResourceMemory, to: &lims},
+		{env: "SHUFFLE_WORKER_MEMORY_LIMIT", rn: corev1.ResourceMemory, to: &lims},
-		{env: "KUBERNETES_EPHEMERAL_STORAGE_LIMIT", rn: corev1.ResourceEphemeralStorage, to: &lims},
+		{env: "SHUFFLE_WORKER_EPHEMERAL_STORAGE_LIMIT", rn: corev1.ResourceEphemeralStorage, to: &lims},
 	}
 
 	for _, it := range items {
@@ -1072,6 +1072,34 @@ func deployK8sWorker(image string, identifier string, env []string) error {
 	env = append(env, fmt.Sprintf("IS_KUBERNETES=true"))
 	env = append(env, fmt.Sprintf("KUBERNETES_NAMESPACE=%s", os.Getenv("KUBERNETES_NAMESPACE")))
 
+	// worker resource env
+	for _, k := range []string{
+		"SHUFFLE_WORKER_CPU_REQUEST",
+		"SHUFFLE_WORKER_MEMORY_REQUEST",
+		"SHUFFLE_WORKER_EPHEMERAL_STORAGE_REQUEST",
+		"SHUFFLE_WORKER_CPU_LIMIT",
+		"SHUFFLE_WORKER_MEMORY_LIMIT",
+		"SHUFFLE_WORKER_EPHEMERAL_STORAGE_LIMIT",
+	} {
+		if v := os.Getenv(k); v != "" {
+			env = append(env, fmt.Sprintf("%s=%s", k, v))
+		}
+	}
+
+	// app resource env
+	for _, k := range []string{
+		"SHUFFLE_APP_CPU_REQUEST",
+		"SHUFFLE_APP_MEMORY_REQUEST",
+		"SHUFFLE_APP_EPHEMERAL_STORAGE_REQUEST",
+		"SHUFFLE_APP_CPU_LIMIT",
+		"SHUFFLE_APP_MEMORY_LIMIT",
+		"SHUFFLE_APP_EPHEMERAL_STORAGE_LIMIT",
+	} {
+		if v := os.Getenv(k); v != "" {
+			env = append(env, fmt.Sprintf("%s=%s", k, v))
+		}
+	}
+
 	if len(os.Getenv("KUBERNETES_SERVICE_HOST")) > 0 {
 		env = append(env, fmt.Sprintf("KUBERNETES_SERVICE_HOST=%s", os.Getenv("KUBERNETES_SERVICE_HOST")))
 	}

functions/onprem/worker/worker.go

@@ -42,6 +42,7 @@ import (
 	//k8s deps
 	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/util/intstr"
 	"k8s.io/client-go/kubernetes"
@@ -639,6 +640,7 @@ func deployk8sApp(image string, identifier string, env []string) error {
 								},
 							},
 							SecurityContext: containerSecurityContext,
+							Resources:       buildResourcesFromEnv(),
 						},
 					},
 					DNSPolicy:          corev1.DNSClusterFirst,
@@ -2375,6 +2377,49 @@ func buildEnvVars(envMap map[string]string) []corev1.EnvVar {
 	}
 	return envVars
 }
+
+func buildResourcesFromEnv() corev1.ResourceRequirements {
+	reqs := corev1.ResourceList{}
+	lims := corev1.ResourceList{}
+
+	type item struct {
+		env string
+		rn  corev1.ResourceName
+		to  *corev1.ResourceList
+	}
+
+	items := []item{
+		// kubernetes requests
+		{env: "SHUFFLE_APP_CPU_REQUEST", rn: corev1.ResourceCPU, to: &reqs},
+		{env: "SHUFFLE_APP_MEMORY_REQUEST", rn: corev1.ResourceMemory, to: &reqs},
+		{env: "SHUFFLE_APP_EPHEMERAL_STORAGE_REQUEST", rn: corev1.ResourceEphemeralStorage, to: &reqs},
+		// kubernetes limits
+		{env: "SHUFFLE_APP_CPU_LIMIT", rn: corev1.ResourceCPU, to: &lims},
+		{env: "SHUFFLE_APP_MEMORY_LIMIT", rn: corev1.ResourceMemory, to: &lims},
+		{env: "SHUFFLE_APP_EPHEMERAL_STORAGE_LIMIT", rn: corev1.ResourceEphemeralStorage, to: &lims},
+	}
+
+	for _, it := range items {
+		if v := strings.TrimSpace(os.Getenv(it.env)); v != "" {
+			if q, err := resource.ParseQuantity(v); err == nil {
+				(*it.to)[it.rn] = q
+			} else {
+				log.Printf("[WARN] Cannot parse %s=%q as resource quantity: %v", it.env, v, err)
+			}
+		}
+	}
+
+	rr := corev1.ResourceRequirements{}
+	if len(reqs) > 0 {
+		rr.Requests = reqs
+	}
+	if len(lims) > 0 {
+		rr.Limits = lims
+	}
+
+	return rr
+}
+
 func getWorkerBackendExecution(auth string, executionId string) (*shuffle.WorkflowExecution, error) {
 	backendUrl := os.Getenv("BASE_URL")
 	if len(backendUrl) == 0 {
@@ -2385,9 +2430,9 @@ func getWorkerBackendExecution(auth string, executionId string) (*shuffle.Workfl
 
 	streamResultUrl := fmt.Sprintf("%s/api/v1/streams/results", backendUrl)
 	topClient := shuffle.GetExternalClient(backendUrl)
-	requestData := shuffle.ActionResult {
+	requestData := shuffle.ActionResult{
 		Authorization: auth,
-		ExecutionId: executionId,
+		ExecutionId:   executionId,
 	}
 
 	data, err := json.Marshal(requestData)