Merge branch 'master' into real-time-deck-support

flyteorg · Jun 29, 2023 · 4e0adf1 · 4e0adf1
2 parents 96e958e + efb69c3
commit 4e0adf1
Show file tree

Hide file tree

Showing 4 changed files with 25 additions and 14 deletions.
diff --git a/Dockerfile b/Dockerfile
@@ -3,7 +3,7 @@
 # 
 # TO OPT OUT OF UPDATES, SEE https://github.com/lyft/boilerplate/blob/master/Readme.rst
 
-FROM --platform=${BUILDPLATFORM} golang:1.18-alpine3.16 as builder
+FROM --platform=${BUILDPLATFORM} golang:1.19-alpine3.16 as builder
 
 ARG TARGETARCH
 ENV GOARCH "${TARGETARCH}"

diff --git a/pkg/controller/nodes/subworkflow/launchplan.go b/pkg/controller/nodes/subworkflow/launchplan.go
@@ -67,14 +67,15 @@ func (l *launchPlanHandler) StartLaunchPlan(ctx context.Context, nCtx handler.No
 	}
 
 	launchCtx := launchplan.LaunchContext{
-		ParentNodeExecution: parentNodeExecutionID,
-		MaxParallelism:      nCtx.ExecutionContext().GetExecutionConfig().MaxParallelism,
-		SecurityContext:     nCtx.ExecutionContext().GetSecurityContext(),
-		RawOutputDataConfig: nCtx.ExecutionContext().GetRawOutputDataConfig().RawOutputDataConfig,
-		Labels:              nCtx.ExecutionContext().GetLabels(),
-		Annotations:         nCtx.ExecutionContext().GetAnnotations(),
-		Interruptible:       nCtx.ExecutionContext().GetExecutionConfig().Interruptible,
-		OverwriteCache:      nCtx.ExecutionContext().GetExecutionConfig().OverwriteCache,
+		ParentNodeExecution:  parentNodeExecutionID,
+		MaxParallelism:       nCtx.ExecutionContext().GetExecutionConfig().MaxParallelism,
+		SecurityContext:      nCtx.ExecutionContext().GetSecurityContext(),
+		RawOutputDataConfig:  nCtx.ExecutionContext().GetRawOutputDataConfig().RawOutputDataConfig,
+		Labels:               nCtx.ExecutionContext().GetLabels(),
+		Annotations:          nCtx.ExecutionContext().GetAnnotations(),
+		Interruptible:        nCtx.ExecutionContext().GetExecutionConfig().Interruptible,
+		OverwriteCache:       nCtx.ExecutionContext().GetExecutionConfig().OverwriteCache,
+		EnvironmentVariables: nCtx.ExecutionContext().GetExecutionConfig().EnvironmentVariables,
 	}
 
 	if nCtx.ExecutionContext().GetExecutionConfig().RecoveryExecution.WorkflowExecutionIdentifier != nil {

diff --git a/pkg/controller/nodes/subworkflow/launchplan/admin.go b/pkg/controller/nodes/subworkflow/launchplan/admin.go
@@ -102,6 +102,14 @@ func (a *adminLaunchPlanExecutor) Launch(ctx context.Context, launchCtx LaunchCo
 		}
 	}
 
+	environmentVariables := make([]*core.KeyValuePair, 0, len(launchCtx.EnvironmentVariables))
+	for k, v := range launchCtx.EnvironmentVariables {
+		environmentVariables = append(environmentVariables, &core.KeyValuePair{
+			Key:   k,
+			Value: v,
+		})
+	}
+
 	req := &admin.ExecutionCreateRequest{
 		Project: executionID.Project,
 		Domain:  executionID.Domain,
@@ -122,6 +130,7 @@ func (a *adminLaunchPlanExecutor) Launch(ctx context.Context, launchCtx LaunchCo
 			RawOutputDataConfig: launchCtx.RawOutputDataConfig,
 			Interruptible:       interruptible,
 			OverwriteCache:      launchCtx.OverwriteCache,
+			Envs:                &admin.Envs{Values: environmentVariables},
 		},
 	}
 

diff --git a/pkg/controller/nodes/subworkflow/launchplan/launchplan.go b/pkg/controller/nodes/subworkflow/launchplan/launchplan.go
@@ -25,11 +25,12 @@ type LaunchContext struct {
 	// MaxParallelism
 	MaxParallelism uint32
 	// RawOutputDataConfig
-	RawOutputDataConfig *admin.RawOutputDataConfig
-	Annotations         map[string]string
-	Labels              map[string]string
-	Interruptible       *bool
-	OverwriteCache      bool
+	RawOutputDataConfig  *admin.RawOutputDataConfig
+	Annotations          map[string]string
+	Labels               map[string]string
+	Interruptible        *bool
+	OverwriteCache       bool
+	EnvironmentVariables map[string]string
 }
 
 // Executor interface to be implemented by the remote system that can allow workflow launching capabilities