diff --git a/src/pkg/k8s.go b/src/pkg/k8s.go
index dbbf6aa..9505494 100644
--- a/src/pkg/k8s.go
+++ b/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,21 @@ 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")
+	// Variable keys arrive uppercased/sanitized by the OpsLevel API (see
+	// RunnerJobType#format_key_as_valid_env_var_name), so match the uppercase key.
+	if accountId := getRunnerJobVariable(job.Variables, "ACCOUNT_ID"); accountId != "" {
+		labels["opslevel.com/account-id"] = accountId
+	}
+	if jobType := getRunnerJobVariable(job.Variables, "JOB_TYPE"); jobType != "" {
+		labels["opslevel.com/job-type"] = jobType
+	}
 	// TODO: manage pods based on image for re-use?
 	cfgMap, err := s.CreateConfigMap(ctx, s.getConfigMapObject(identifier, job))
 	if err != nil {
diff --git a/src/pkg/k8s_test.go b/src/pkg/k8s_test.go
index f7322e5..76d1338 100644
--- a/src/pkg/k8s_test.go
+++ b/src/pkg/k8s_test.go
@@ -193,6 +193,50 @@ func TestGetPodEnv_FiltersByScope(t *testing.T) {
 	autopilot.Equals(t, []string{"BOTH", "MAIN_ONLY"}, mainKeys)
 }
 
+func TestGetRunnerJobVariable_ReturnsMatchingValue(t *testing.T) {
+	// Arrange
+	// The OpsLevel API uppercases/sanitizes variable keys before sending them,
+	// so the runner matches the uppercase key.
+	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 TestGetRunnerJobVariable_IsCaseSensitive(t *testing.T) {
+	// Arrange
+	// Guards the bug where the runner looked up "account_id" but the API sends
+	// "ACCOUNT_ID"; the lowercase key must NOT match.
+	vars := []opslevel.RunnerJobVariable{
+		{Key: "ACCOUNT_ID", Value: "acct-123"},
+	}
+
+	// Act
+	value := getRunnerJobVariable(vars, "account_id")
+
+	// Assert
+	autopilot.Equals(t, "", value)
+}
+
 func TestGetPodObject_NoInitCommands(t *testing.T) {
 	// Arrange
 	runner := &JobRunner{