Fix duplicate env vars in container (#358)

* removed duplicate execution of AddFlyteCustomizationsToContainer

Signed-off-by: Daniel Rammer <daniel@union.ai>

* removed duplicate call to AddFlyteCustomizationsToContainer when creating container

Signed-off-by: Daniel Rammer <daniel@union.ai>

* removed dead code

Signed-off-by: Daniel Rammer <daniel@union.ai>

* added unit test for duplicate environment variables

Signed-off-by: Daniel Rammer <daniel@union.ai>

---------

Signed-off-by: Daniel Rammer <daniel@union.ai>

go/tasks/pluginmachinery/flytek8s/container_helper.go

@@ -3,8 +3,6 @@ package flytek8s
 import (
 	"context"
 
-	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/core"
-
 	"github.com/flyteorg/flyteplugins/go/tasks/errors"
 	pluginscore "github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/core"
 	"github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/core/template"
@@ -194,12 +192,26 @@ func ApplyResourceOverrides(resources, platformResources v1.ResourceRequirements
 	return resources
 }
 
-// BuildRawContainer constructs a Container based on the definition passed by the taskContainer and
-// TaskExecutionMetadata.
-func BuildRawContainer(ctx context.Context, taskContainer *core.Container, taskExecMetadata pluginscore.TaskExecutionMetadata) (*v1.Container, error) {
+// BuildRawContainer constructs a Container based on the definition passed by the TaskExecutionContext.
+func BuildRawContainer(ctx context.Context, tCtx pluginscore.TaskExecutionContext) (*v1.Container, error) {
+	taskTemplate, err := tCtx.TaskReader().Read(ctx)
+	if err != nil {
+		logger.Warnf(ctx, "failed to read task information when trying to construct container, err: %s", err.Error())
+		return nil, err
+	}
+
+	// validate arguments
+	taskContainer := taskTemplate.GetContainer()
+	if taskContainer == nil {
+		return nil, errors.Errorf(errors.BadTaskSpecification, "unable to create container with no definition in TaskTemplate")
+	}
+	if tCtx.TaskExecutionMetadata().GetOverrides() == nil || tCtx.TaskExecutionMetadata().GetOverrides().GetResources() == nil {
+		return nil, errors.Errorf(errors.BadTaskSpecification, "resource requirements not found for container task, required!")
+	}
+
 	// Make the container name the same as the pod name, unless it violates K8s naming conventions
 	// Container names are subject to the DNS-1123 standard
-	containerName := taskExecMetadata.GetTaskExecutionID().GetGeneratedName()
+	containerName := tCtx.TaskExecutionMetadata().GetTaskExecutionID().GetGeneratedName()
 	if errs := validation.IsDNS1123Label(containerName); len(errs) > 0 {
 		containerName = rand.String(4)
 	}
@@ -225,22 +237,8 @@ func BuildRawContainer(ctx context.Context, taskContainer *core.Container, taskE
 // ToK8sContainer builds a Container based on the definition passed by the TaskExecutionContext. This involves applying
 // all Flyte configuration including k8s plugins and resource requests.
 func ToK8sContainer(ctx context.Context, tCtx pluginscore.TaskExecutionContext) (*v1.Container, error) {
-	taskTemplate, err := tCtx.TaskReader().Read(ctx)
-	if err != nil {
-		logger.Warnf(ctx, "failed to read task information when trying to construct container, err: %s", err.Error())
-		return nil, err
-	}
-
-	// validate arguments
-	if taskTemplate.GetContainer() == nil {
-		return nil, errors.Errorf(errors.BadTaskSpecification, "unable to create container with no definition in TaskTemplate")
-	}
-	if tCtx.TaskExecutionMetadata().GetOverrides() == nil || tCtx.TaskExecutionMetadata().GetOverrides().GetResources() == nil {
-		return nil, errors.Errorf(errors.BadTaskSpecification, "resource requirements not found for container task, required!")
-	}
-
 	// build raw container
-	container, err := BuildRawContainer(ctx, taskTemplate.GetContainer(), tCtx.TaskExecutionMetadata())
+	container, err := BuildRawContainer(ctx, tCtx)
 	if err != nil {
 		return nil, err
 	}

go/tasks/pluginmachinery/flytek8s/pod_helper.go

@@ -134,7 +134,7 @@ func BuildRawPod(ctx context.Context, tCtx pluginsCore.TaskExecutionContext) (*v
 	switch target := taskTemplate.GetTarget().(type) {
 	case *core.TaskTemplate_Container:
 		// handles tasks defined by a single container
-		c, err := ToK8sContainer(ctx, tCtx)
+		c, err := BuildRawContainer(ctx, tCtx)
 		if err != nil {
 			return nil, nil, "", err
 		}

go/tasks/pluginmachinery/flytek8s/pod_helper_test.go

@@ -531,6 +531,26 @@ func TestToK8sPod(t *testing.T) {
 		assert.Equal(t, val3, *p.DNSConfig.Options[3].Value)
 		assert.Equal(t, []string{"ns1.svc.cluster-domain.example", "my.dns.search.suffix"}, p.DNSConfig.Searches)
 	})
+
+	t.Run("environmentVariables", func(t *testing.T) {
+		assert.NoError(t, config.SetK8sPluginConfig(&config.K8sPluginConfig{
+			DefaultEnvVars: map[string]string{
+				"foo": "bar",
+			},
+		}))
+		x := dummyExecContext(&v1.ResourceRequirements{})
+		p, _, _, err := ToK8sPodSpec(ctx, x)
+		assert.NoError(t, err)
+		for _, c := range p.Containers {
+			uniqueVariableNames := make(map[string]string)
+			for _, envVar := range c.Env {
+				if _, ok := uniqueVariableNames[envVar.Name]; ok {
+					t.Errorf("duplicate environment variable %s", envVar.Name)
+				}
+				uniqueVariableNames[envVar.Name] = envVar.Value
+			}
+		}
+	})
 }
 
 func TestDemystifyPending(t *testing.T) {