Add env var support (#324)

Co-authored-by: Filipe Regadas <[email protected]>
spotify · Dec 21, 2023 · f61f437 · f61f437
1 parent 10fb176
commit f61f437
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 10 deletions.
diff --git a/pkg/flink/handler.go b/pkg/flink/handler.go
@@ -40,11 +40,12 @@ import (
 )
 
 type FlinkTaskContext struct {
-	ClusterName ClusterName
-	Namespace   string
-	Annotations map[string]string
-	Labels      map[string]string
-	Job         flinkIdl.FlinkJob
+	ClusterName          ClusterName
+	Namespace            string
+	Annotations          map[string]string
+	Labels               map[string]string
+	EnvironmentVariables map[string]string
+	Job                  flinkIdl.FlinkJob
 }
 
 type FlinkTaskExecContext interface {
@@ -92,11 +93,12 @@ func NewFlinkTaskContext(ctx context.Context, taskCtx FlinkTaskExecContext) (*Fl
 	}
 
 	return &FlinkTaskContext{
-		ClusterName: cn,
-		Namespace:   taskMetadata.GetNamespace(),
-		Annotations: GetDefaultAnnotations(taskMetadata),
-		Labels:      GetDefaultLabels(taskMetadata),
-		Job:         job,
+		ClusterName:          cn,
+		Namespace:            taskMetadata.GetNamespace(),
+		Annotations:          GetDefaultAnnotations(taskMetadata),
+		Labels:               GetDefaultLabels(taskMetadata),
+		EnvironmentVariables: GetDefaultEnvironmentVariables(taskMetadata),
+		Job:                  job,
 	}, nil
 }
 

diff --git a/pkg/flink/types.go b/pkg/flink/types.go
@@ -63,6 +63,15 @@ func GetDefaultLabels(taskCtx pluginsCore.TaskExecutionMetadata) Labels {
 	)
 }
 
+type EnvVars map[string]string
+
+func GetDefaultEnvironmentVariables(taskCtx pluginsCore.TaskExecutionMetadata) EnvVars {
+	return utils.UnionMaps(
+		config.GetK8sPluginConfig().DefaultEnvVars,
+		utils.CopyMap(taskCtx.GetEnvironmentVariables()),
+	)
+}
+
 type ClusterName string
 
 func (cn ClusterName) Validate() error {