Bugfix/emitter bucket envvar

pkg/core/environment_variable.go

@@ -1,12 +1,16 @@
 package core
 
+import "go.uber.org/zap"
+
 type (
 	EnvironmentVariable struct {
 		Name       string
 		Kind       string
 		ResourceID string
 		Value      string
 	}
+
+	EnvironmentVariables []EnvironmentVariable
 )
 
 type EnvironmentVariableValue string
@@ -25,3 +29,34 @@ var (
 	CONNECTION_STRING EnvironmentVariableValue = "connection_string"
 	BUCKET_NAME       EnvironmentVariableValue = "bucket_name"
 )
+
+var InternalStorageVariable = EnvironmentVariable{
+	Name:       KLOTHO_PROXY_ENV_VAR_NAME,
+	Kind:       InternalKind,
+	ResourceID: KlothoPayloadName,
+	Value:      string(BUCKET_NAME),
+}
+
+// Add the given environment variable to the list. If a variable of the same name already exists, replace it.
+func (vars *EnvironmentVariables) Add(v EnvironmentVariable) {
+	if *vars == nil {
+		*vars = make(EnvironmentVariables, 0)
+	}
+	for i, e := range *vars {
+		if e.Name == v.Name {
+			if e.Value != v.Value || e.ResourceID != v.ResourceID {
+				zap.S().Debugf("Replacing variable %+v with %+v", e, v)
+			}
+			(*vars)[i] = v
+			return
+		}
+	}
+	*vars = append(*vars, v)
+}
+
+// AddAll is a convenience over `Add` to add many variables.
+func (vars *EnvironmentVariables) AddAll(vs EnvironmentVariables) {
+	for _, v := range vs {
+		vars.Add(v)
+	}
+}

pkg/core/exec_unit.go

@@ -18,7 +18,7 @@ type (
 		Name                 string
 		files                ConcurrentMap[string, File]
 		Executable           Executable
-		EnvironmentVariables []EnvironmentVariable
+		EnvironmentVariables EnvironmentVariables
 		DockerfilePath       string
 	}
 

pkg/env_var/plugin.go

@@ -68,7 +68,7 @@ func (p EnvVarInjection) Transform(result *core.CompilationResult, deps *core.De
 						errs.Append(err)
 						continue
 					}
-					unit.EnvironmentVariables = append(unit.EnvironmentVariables, directiveResult.variables...)
+					unit.EnvironmentVariables.AddAll(directiveResult.variables)
 				}
 			}
 		}
@@ -88,13 +88,13 @@ func validateValue(kind string, value string) bool {
 
 type EnvironmentVariableDirectiveResult struct {
 	kind      string
-	variables []core.EnvironmentVariable
+	variables core.EnvironmentVariables
 }
 
 func ParseDirectiveToEnvVars(cap *annotation.Capability) (EnvironmentVariableDirectiveResult, error) {
 	overallKind := ""
 	envVars := cap.Directives.Object(core.EnvironmentVariableDirective)
-	foundVars := []core.EnvironmentVariable{}
+	foundVars := core.EnvironmentVariables{}
 	if envVars == nil {
 		return EnvironmentVariableDirectiveResult{}, nil
 	}
@@ -133,7 +133,7 @@ func ParseDirectiveToEnvVars(cap *annotation.Capability) (EnvironmentVariableDir
 			Value:      value,
 			ResourceID: cap.ID,
 		}
-		foundVars = append(foundVars, foundVariable)
+		foundVars.Add(foundVariable)
 	}
 
 	return EnvironmentVariableDirectiveResult{kind: overallKind, variables: foundVars}, nil

pkg/env_var/plugin_test.go

@@ -12,7 +12,7 @@ import (
 func Test_envVarPlugin(t *testing.T) {
 	type testResult struct {
 		resource core.CloudResource
-		envVars  []core.EnvironmentVariable
+		envVars  core.EnvironmentVariables
 	}
 	tests := []struct {
 		name    string
@@ -34,12 +34,13 @@ func Test_envVarPlugin(t *testing.T) {
 const a = 1`,
 			want: testResult{
 				resource: &core.Persist{Kind: core.PersistRedisNodeKind, Name: "myRedisNode"},
-				envVars: []core.EnvironmentVariable{{
-					Name:       "REDIS_NODE_HOST",
-					Kind:       "persist_redis_node",
-					ResourceID: "myRedisNode",
-					Value:      "host",
-				},
+				envVars: core.EnvironmentVariables{
+					{
+						Name:       "REDIS_NODE_HOST",
+						Kind:       "persist_redis_node",
+						ResourceID: "myRedisNode",
+						Value:      "host",
+					},
 					{
 						Name:       "REDIS_NODE_PORT",
 						Kind:       "persist_redis_node",
@@ -63,12 +64,13 @@ const a = 1`,
 const a = 1`,
 			want: testResult{
 				resource: &core.Persist{Kind: core.PersistRedisClusterKind, Name: "myRedisCluster"},
-				envVars: []core.EnvironmentVariable{{
-					Name:       "REDIS_HOST",
-					Kind:       "persist_redis_cluster",
-					ResourceID: "myRedisCluster",
-					Value:      "host",
-				},
+				envVars: core.EnvironmentVariables{
+					{
+						Name:       "REDIS_HOST",
+						Kind:       "persist_redis_cluster",
+						ResourceID: "myRedisCluster",
+						Value:      "host",
+					},
 					{
 						Name:       "REDIS_PORT",
 						Kind:       "persist_redis_cluster",
@@ -91,12 +93,13 @@ const a = 1`,
 const a = 1`,
 			want: testResult{
 				resource: &core.Persist{Kind: core.PersistORMKind, Name: "myOrm"},
-				envVars: []core.EnvironmentVariable{{
-					Name:       "ORM_CONNECTION_STRING",
-					Kind:       "persist_orm",
-					ResourceID: "myOrm",
-					Value:      "connection_string",
-				},
+				envVars: core.EnvironmentVariables{
+					{
+						Name:       "ORM_CONNECTION_STRING",
+						Kind:       "persist_orm",
+						ResourceID: "myOrm",
+						Value:      "connection_string",
+					},
 				},
 			},
 		},
@@ -215,12 +218,13 @@ func Test_parseDirectiveToEnvVars(t *testing.T) {
 const a = 1`,
 			want: EnvironmentVariableDirectiveResult{
 				kind: string(core.PersistRedisNodeKind),
-				variables: []core.EnvironmentVariable{{
-					Name:       "REDIS_NODE_HOST",
-					Kind:       "persist_redis_node",
-					ResourceID: "myRedisNode",
-					Value:      "host",
-				},
+				variables: core.EnvironmentVariables{
+					{
+						Name:       "REDIS_NODE_HOST",
+						Kind:       "persist_redis_node",
+						ResourceID: "myRedisNode",
+						Value:      "host",
+					},
 					{
 						Name:       "REDIS_NODE_PORT",
 						Kind:       "persist_redis_node",

pkg/exec_unit/plugin_exec_unit.go

@@ -27,7 +27,7 @@ func (p ExecUnitPlugin) Transform(result *core.CompilationResult, deps *core.Dep
 	cfg := p.Config.GetExecutionUnit(unit.Name)
 
 	for key, value := range cfg.EnvironmentVariables {
-		unit.EnvironmentVariables = append(unit.EnvironmentVariables, core.EnvironmentVariable{
+		unit.EnvironmentVariables.Add(core.EnvironmentVariable{
 			Name:  key,
 			Value: value,
 		})

pkg/exec_unit/plugin_exec_unit_test.go

@@ -13,7 +13,7 @@ func Test_environmentVarsAddedToUnit(t *testing.T) {
 	tests := []struct {
 		name         string
 		envVars      map[string]string
-		want         []core.EnvironmentVariable
+		want         core.EnvironmentVariables
 		wantExecUnit bool
 	}{
 		{
@@ -30,7 +30,7 @@ func Test_environmentVarsAddedToUnit(t *testing.T) {
 				"key1": "value1",
 				"key2": "value2",
 			},
-			want: []core.EnvironmentVariable{
+			want: core.EnvironmentVariables{
 				{
 					Name:  "key1",
 					Value: "value1",

pkg/infra/kubernetes/exec_unit.go

@@ -326,7 +326,7 @@ func (unit *HelmExecUnit) getServiceName() string {
 	return unit.Name
 }
 
-func (unit *HelmExecUnit) addEnvsVarToDeployment(envVars []core.EnvironmentVariable) ([]Value, error) {
+func (unit *HelmExecUnit) addEnvsVarToDeployment(envVars core.EnvironmentVariables) ([]Value, error) {
 	values := []Value{}
 
 	log := zap.L().Sugar().With(logging.FileField(unit.Deployment), zap.String("unit", unit.Name))
@@ -376,7 +376,7 @@ func (unit *HelmExecUnit) addEnvsVarToDeployment(envVars []core.EnvironmentVaria
 	return values, nil
 }
 
-func (unit *HelmExecUnit) addEnvVarToPod(envVars []core.EnvironmentVariable) ([]Value, error) {
+func (unit *HelmExecUnit) addEnvVarToPod(envVars core.EnvironmentVariables) ([]Value, error) {
 	values := []Value{}
 
 	log := zap.L().Sugar().With(logging.FileField(unit.Pod), zap.String("unit", unit.Name))

pkg/infra/kubernetes/exec_unit_test.go

@@ -428,7 +428,7 @@ func Test_addEnvVarToDeployment(t *testing.T) {
 	tests := []struct {
 		name    string
 		file    string
-		envVars []core.EnvironmentVariable
+		envVars core.EnvironmentVariables
 		want    result
 		wantErr bool
 	}{
@@ -451,7 +451,7 @@ spec:
       containers:
       - name: nginx
         image: nginx:1.14.2`,
-			envVars: []core.EnvironmentVariable{{Name: "SEQUELIZEDB_PERSIST_ORM_CONNECTION"}},
+			envVars: core.EnvironmentVariables{{Name: "SEQUELIZEDB_PERSIST_ORM_CONNECTION"}},
 			want: result{
 				values: []Value{
 					{
@@ -546,13 +546,13 @@ func Test_addEnvVarToPod(t *testing.T) {
 	tests := []struct {
 		name    string
 		file    string
-		envVars []core.EnvironmentVariable
+		envVars core.EnvironmentVariables
 		want    result
 		wantErr bool
 	}{
 		{
 			name:    "Basic Pod",
-			envVars: []core.EnvironmentVariable{{Name: "SEQUELIZEDB_PERSIST_ORM_CONNECTION"}},
+			envVars: core.EnvironmentVariables{{Name: "SEQUELIZEDB_PERSIST_ORM_CONNECTION"}},
 			file: `apiVersion: v1
 kind: Pod
 metadata:

pkg/infra/kubernetes/persist.go

@@ -33,8 +33,8 @@ func (unit *HelmExecUnit) handlePersistForExecUnit(deps *core.Dependencies) ([]V
 	return values, nil
 }
 
-func generateEnvVars(deps *core.Dependencies, name string) []core.EnvironmentVariable {
-	envVars := []core.EnvironmentVariable{}
+func generateEnvVars(deps *core.Dependencies, name string) core.EnvironmentVariables {
+	envVars := core.EnvironmentVariables{}
 	for _, target := range deps.Downstream(core.ResourceKey{Name: name, Kind: core.ExecutionUnitKind}) {
 		switch target.Kind {
 		case string(core.PersistORMKind):

pkg/infra/kubernetes/persist_test.go

@@ -284,25 +284,25 @@ func Test_generateEnvVarsForPersist(t *testing.T) {
 		name     string
 		unit     *core.ExecutionUnit
 		resource core.CloudResource
-		values   []core.EnvironmentVariable
+		values   core.EnvironmentVariables
 	}{
 		{
 			name:     "Wrong dependencies",
 			unit:     &core.ExecutionUnit{Name: "main"},
 			resource: &core.Persist{Name: "file", Kind: core.PersistFileKind},
-			values:   []core.EnvironmentVariable{},
+			values:   core.EnvironmentVariables{},
 		},
 		{
 			name:     "orm dependency",
 			unit:     &core.ExecutionUnit{Name: "main"},
 			resource: &core.Persist{Name: "orm", Kind: core.PersistORMKind},
-			values:   []core.EnvironmentVariable{{Name: "ORM_PERSIST_ORM_CONNECTION", Kind: "persist_orm", ResourceID: "orm", Value: string(core.CONNECTION_STRING)}},
+			values:   core.EnvironmentVariables{{Name: "ORM_PERSIST_ORM_CONNECTION", Kind: "persist_orm", ResourceID: "orm", Value: string(core.CONNECTION_STRING)}},
 		},
 		{
 			name:     "redis node dependency",
 			unit:     &core.ExecutionUnit{Name: "main"},
 			resource: &core.Persist{Name: "redisNode", Kind: core.PersistRedisNodeKind},
-			values: []core.EnvironmentVariable{
+			values: core.EnvironmentVariables{
 				{Name: "REDISNODE_PERSIST_REDIS_HOST", Kind: "persist_redis_node", ResourceID: "redisNode", Value: string(core.HOST)},
 				{Name: "REDISNODE_PERSIST_REDIS_PORT", Kind: "persist_redis_node", ResourceID: "redisNode", Value: string(core.PORT)},
 			},
@@ -311,7 +311,7 @@ func Test_generateEnvVarsForPersist(t *testing.T) {
 			name:     "redis cluster dependency",
 			unit:     &core.ExecutionUnit{Name: "main"},
 			resource: &core.Persist{Name: "redisCluster", Kind: core.PersistRedisClusterKind},
-			values: []core.EnvironmentVariable{
+			values: core.EnvironmentVariables{
 				{Name: "REDISCLUSTER_PERSIST_REDIS_HOST", Kind: "persist_redis_cluster", ResourceID: "redisCluster", Value: string(core.HOST)},
 				{Name: "REDISCLUSTER_PERSIST_REDIS_PORT", Kind: "persist_redis_cluster", ResourceID: "redisCluster", Value: string(core.PORT)},
 			},

pkg/lang/golang/plugin_fs.go

@@ -79,7 +79,7 @@ func (p *PersistFsPlugin) transformFS(f *core.SourceFile, cap *core.Annotation,
 		Value:      "bucket_url",
 	}
 
-	unit.EnvironmentVariables = append(unit.EnvironmentVariables, fsEnvVar)
+	unit.EnvironmentVariables.Add(fsEnvVar)
 
 	args, _ := getArguments(result.expression)
 	// Generate the new node content before replacing the node. We just set it so we can compile correctly

pkg/lang/golang/plugin_secrets.go

@@ -99,7 +99,7 @@ func (p *PersistSecretsPlugin) transformSecret(f *core.SourceFile, cap *core.Ann
 		ResourceID: cap.Capability.ID,
 		Value:      "secret_name",
 	}
-	unit.EnvironmentVariables = append(unit.EnvironmentVariables, secretEnvVar)
+	unit.EnvironmentVariables.Add(secretEnvVar)
 
 	args[1].Content = fmt.Sprintf(`"awssecretsmanager://" + os.Getenv("%s") + "?region=" + os.Getenv("AWS_REGION") + queryParams`, secretEnvVar.Name)
 

pkg/lang/javascript/aws_runtime/_emitter.ts

@@ -12,7 +12,7 @@ import * as crypto from 'crypto'
 import { addInflight } from './dispatcher'
 import { Readable } from 'stream'
 
-const payloadBucketPhysicalName = process.env.KLOTHO_S3_PREFIX + '{{.PayloadsBucketName}}'
+const payloadBucketPhysicalName = process.env.KLOTHO_PROXY_RESOURCE_NAME
 const appName = '{{.AppName}}'
 
 // The account-level ARN for sns. The topics must be account-wide unique

pkg/lang/javascript/aws_runtime/aws.go

@@ -181,6 +181,12 @@ func (r *AwsRuntime) AddRedisClusterRuntimeFiles(unit *core.ExecutionUnit) error
 }
 
 func (r *AwsRuntime) AddPubsubRuntimeFiles(unit *core.ExecutionUnit) error {
+	unit.EnvironmentVariables.Add(core.InternalStorageVariable)
+	err := r.AddFsRuntimeFiles(unit, core.InternalStorageVariable.Name, "payload")
+	if err != nil {
+		return err
+	}
+
 	return r.AddRuntimeFiles(unit, pubsubRuntimeFiles)
 }
 
@@ -198,14 +204,8 @@ func (r *AwsRuntime) AddProxyRuntimeFiles(unit *core.ExecutionUnit, proxyType st
 		proxyFile = proxyLambda
 
 		// We also need to add the Fs files because exec to exec calls in aws use s3
-		proxyEnvVar := core.EnvironmentVariable{
-			Name:       core.KLOTHO_PROXY_ENV_VAR_NAME,
-			Kind:       core.InternalKind,
-			ResourceID: core.KlothoPayloadName,
-			Value:      string(core.BUCKET_NAME),
-		}
-		unit.EnvironmentVariables = append(unit.EnvironmentVariables, proxyEnvVar)
-		err := r.AddFsRuntimeFiles(unit, proxyEnvVar.Name, "payload")
+		unit.EnvironmentVariables.Add(core.InternalStorageVariable)
+		err := r.AddFsRuntimeFiles(unit, core.InternalStorageVariable.Name, "payload")
 		if err != nil {
 			return err
 		}
@@ -232,14 +232,8 @@ func (r *AwsRuntime) AddExecRuntimeFiles(unit *core.ExecutionUnit, result *core.
 		DockerFile = dockerfileLambda
 		Dispatcher = dispatcherLambda
 
-		proxyEnvVar := core.EnvironmentVariable{
-			Name:       core.KLOTHO_PROXY_ENV_VAR_NAME,
-			Kind:       core.InternalKind,
-			ResourceID: core.KlothoPayloadName,
-			Value:      string(core.BUCKET_NAME),
-		}
-		unit.EnvironmentVariables = append(unit.EnvironmentVariables, proxyEnvVar)
-		err := r.AddFsRuntimeFiles(unit, proxyEnvVar.Name, "payload")
+		unit.EnvironmentVariables.Add(core.InternalStorageVariable)
+		err := r.AddFsRuntimeFiles(unit, core.InternalStorageVariable.Name, "payload")
 		if err != nil {
 			return err
 		}