feat: add labels from job spec to job labels environment (#187)

* feat: update input compiler to include labels from the job spec

* fix: add job label enrichment for hooks as well

core/scheduler/service/executor_input_compiler.go

@@ -136,20 +136,21 @@ func (i InputCompiler) Compile(ctx context.Context, job *scheduler.JobWithDetail
 		"job_name":  job.Job.Name.String(),
 		"job_id":    job.Job.ID.String(),
 	}
-	jobAttributionLabels := getJobLabelsString(jobLabelsToAdd)
-	if jobLabels, ok := confs[JobAttributionLabelsKey]; ok {
-		if len(jobLabels) == 0 {
-			confs[JobAttributionLabelsKey] = jobAttributionLabels
-		} else {
-			confs[JobAttributionLabelsKey] = jobLabels + "," + jobAttributionLabels
+
+	if job.JobMetadata != nil {
+		for key, value := range job.JobMetadata.Labels {
+			if _, ok := jobLabelsToAdd[key]; !ok {
+				jobLabelsToAdd[key] = value
+			}
 		}
-	} else {
-		confs[JobAttributionLabelsKey] = jobAttributionLabels
 	}
 
+	jobAttributionLabels := getJobLabelsString(jobLabelsToAdd)
+
 	if config.Executor.Type == scheduler.ExecutorTask {
+		enriched := i.getEnrichedWithJobLabelsValue(confs, jobAttributionLabels)
 		return &scheduler.ExecutorInput{
-			Configs: utils.MergeMaps(confs, systemDefinedVars),
+			Configs: utils.MergeMaps(enriched, systemDefinedVars),
 			Secrets: secretConfs,
 			Files:   fileMap,
 		}, nil
@@ -167,13 +168,33 @@ func (i InputCompiler) Compile(ctx context.Context, job *scheduler.JobWithDetail
 		return nil, err
 	}
 
+	enriched := i.getEnrichedWithJobLabelsValue(hookConfs, jobAttributionLabels)
 	return &scheduler.ExecutorInput{
-		Configs: utils.MergeMaps(hookConfs, systemDefinedVars),
+		Configs: utils.MergeMaps(enriched, systemDefinedVars),
 		Secrets: hookSecrets,
 		Files:   fileMap,
 	}, nil
 }
 
+func (InputCompiler) getEnrichedWithJobLabelsValue(config map[string]string, incomingValue string) map[string]string {
+	output := make(map[string]string)
+	for key, value := range config {
+		output[key] = value
+	}
+
+	if existingValue, ok := output[JobAttributionLabelsKey]; ok {
+		if len(existingValue) == 0 {
+			output[JobAttributionLabelsKey] = incomingValue
+		} else {
+			output[JobAttributionLabelsKey] = existingValue + "," + incomingValue
+		}
+	} else {
+		output[JobAttributionLabelsKey] = incomingValue
+	}
+
+	return output
+}
+
 func (i InputCompiler) compileConfigs(configs map[string]string, templateCtx map[string]any) (map[string]string, map[string]string, error) {
 	conf, secretsConfig := splitConfigWithSecrets(configs)
 

core/scheduler/service/executor_input_compiler_test.go

@@ -3,6 +3,7 @@ package service_test
 import (
 	"context"
 	"fmt"
+	"sort"
 	"strings"
 	"testing"
 	"time"
@@ -183,6 +184,12 @@ func TestExecutorCompiler(t *testing.T) {
 				Schedule: &scheduler.Schedule{
 					Interval: "0 * * * *",
 				},
+				JobMetadata: &scheduler.JobMetadata{
+					Labels: map[string]string{
+						"user-specified-label-key1": "user-specified-label-value-for-test-1",
+						"user-specified-label-key2": "user-specified-label-value-for-test-2",
+					},
+				},
 			}
 			config := scheduler.RunConfig{
 				Executor: scheduler.Executor{
@@ -272,7 +279,9 @@ func TestExecutorCompiler(t *testing.T) {
 					"project=proj1": true,
 					"namespace=ns1": true,
 					"job_name=job1": true,
-					"job_id=00000000-0000-0000-0000-000000000000": true,
+					"job_id=00000000-0000-0000-0000-000000000000":                     true,
+					"user-specified-label-key1=user-specified-label-value-for-test-1": true,
+					"user-specified-label-key2=user-specified-label-value-for-test-2": true,
 				}
 
 				for _, v := range strings.Split(inputExecutorResp.Configs["JOB_LABELS"], ",") {
@@ -419,10 +428,18 @@ func TestExecutorCompiler(t *testing.T) {
 					"EXECUTION_TIME":  executedAt.Format(time.RFC3339),
 					"JOB_DESTINATION": job.Destination,
 					"hook.compiled":   "hook.val.compiled",
+					"JOB_LABELS":      "job_id=00000000-0000-0000-0000-000000000000,job_name=job1,namespace=ns1,project=proj1",
 				},
 				Secrets: map[string]string{"secret.hook.compiled": "hook.s.val.compiled"},
 				Files:   compiledFile,
 			}
+
+			assert.NotNil(t, inputExecutorResp)
+			if value, ok := inputExecutorResp.Configs["JOB_LABELS"]; ok {
+				splitValues := strings.Split(value, ",")
+				sort.Strings(splitValues)
+				inputExecutorResp.Configs["JOB_LABELS"] = strings.Join(splitValues, ",")
+			}
 			assert.Equal(t, expectedInputExecutor, inputExecutorResp)
 		})
 		t.Run("compileConfigs for Executor type Hook should fail if error in hook compilation", func(t *testing.T) {