Emit account_id and job metadata as Kubernetes pod labels

Rather than weaving the account_id job variable into the pod name (which
risks the 63-char limit and OpsLevel-side naming conventions), stamp
descriptive labels on the job pod for observability:

  app.kubernetes.io/name:  opslevel-job
  opslevel.com/account-id: <account_id>  (when the job sets it)
  opslevel.com/job-id:     <job id>
  opslevel.com/mode:       <faktory|api>

These enable native DataDog filtering via podLabelsAsTags with no pod
name parsing. Labels are added after building the PDB label selector so
they stay out of the selector, which the instance label already makes
unique.

Adds a getRunnerJobVariable helper (named explicitly since a similar
helper is anticipated for egress proxies).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: wesleyjellis

src/pkg/k8s.go

@@ -111,6 +111,16 @@ func NewJobRunner(runnerId string, path string) *JobRunner {
 	}
 }
 
+// getRunnerJobVariable returns the value of the job variable with the given key, if present.
+func getRunnerJobVariable(configs []opslevel.RunnerJobVariable, key string) string {
+	for _, config := range configs {
+		if config.Key == key {
+			return config.Value
+		}
+	}
+	return ""
+}
+
 // getPodEnv returns the env vars to inject into a container for the given
 // scope. Variables with no Scope set are visible to every container; variables
 // with a Scope are only visible to containers running in that scope.
@@ -372,6 +382,16 @@ func (s *JobRunner) Run(ctx context.Context, job opslevel.RunnerJob, stdout, std
 			Outcome: opslevel.RunnerJobOutcomeEnumFailed,
 		}
 	}
+	// Descriptive labels for observability (e.g. DataDog podLabelsAsTags), so
+	// running pods can be filtered by account/job/mode without parsing the pod
+	// name. Added after building the selector so they stay out of the PDB
+	// selector, which the instance label already makes unique.
+	labels["app.kubernetes.io/name"] = "opslevel-job"
+	labels["opslevel.com/job-id"] = id
+	labels["opslevel.com/mode"] = viper.GetString("mode")
+	if accountId := getRunnerJobVariable(job.Variables, "account_id"); accountId != "" {
+		labels["opslevel.com/account-id"] = accountId
+	}
 	// TODO: manage pods based on image for re-use?
 	cfgMap, err := s.CreateConfigMap(ctx, s.getConfigMapObject(identifier, job))
 	if err != nil {

src/pkg/k8s_test.go

@@ -193,6 +193,33 @@ func TestGetPodEnv_FiltersByScope(t *testing.T) {
 	autopilot.Equals(t, []string{"BOTH", "MAIN_ONLY"}, mainKeys)
 }
 
+func TestGetRunnerJobVariable_ReturnsMatchingValue(t *testing.T) {
+	// Arrange
+	vars := []opslevel.RunnerJobVariable{
+		{Key: "FOO", Value: "bar"},
+		{Key: "account_id", Value: "acct-123"},
+	}
+
+	// Act
+	value := getRunnerJobVariable(vars, "account_id")
+
+	// Assert
+	autopilot.Equals(t, "acct-123", value)
+}
+
+func TestGetRunnerJobVariable_ReturnsEmptyWhenMissing(t *testing.T) {
+	// Arrange
+	vars := []opslevel.RunnerJobVariable{
+		{Key: "FOO", Value: "bar"},
+	}
+
+	// Act
+	value := getRunnerJobVariable(vars, "account_id")
+
+	// Assert
+	autopilot.Equals(t, "", value)
+}
+
 func TestGetPodObject_NoInitCommands(t *testing.T) {
 	// Arrange
 	runner := &JobRunner{