Sanitize jsonb column from unicode null character, hide jobFailureInfo jsonb map behind enableJobRunFailureInfoMap (disabled by default) (#4829)

masipauskas · Martynas Asipauskas · web-flow · commit aa8bbed45a41 · 2026-04-10T16:23:51.000+01:00
* Unicode null termination characters in /dev/termination-log will no
longer break lookout ingester.
* We're gating the error categorisation error map behind feature flag
(as we're likely to change field shape due to load on the DB).

Co-authored-by: Martynas Asipauskas &lt;Martynas.Asipauskas@gresearch.co.uk&gt;
diff --git a/.gitignore b/.gitignore
@@ -113,3 +113,5 @@ tmp/
 tmp.*
 *.tmp.*
 .aider*
+
+.claude/
diff --git a/internal/lookout/pruner/pruner_test.go b/internal/lookout/pruner/pruner_test.go
@@ -134,7 +134,7 @@ func TestPruneDb(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.testName, func(t *testing.T) {
 			err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error {
-				converter := instructions.NewInstructionConverter(metrics.Get().Metrics, "armadaproject.io/", []string{}, &compress.NoOpCompressor{})
+				converter := instructions.NewInstructionConverter(metrics.Get().Metrics, "armadaproject.io/", []string{}, &compress.NoOpCompressor{}, false)
 				store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10)
 
 				ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 5*time.Minute)
diff --git a/internal/lookout/repository/getjoberror_test.go b/internal/lookout/repository/getjoberror_test.go
@@ -16,7 +16,7 @@ import (
 
 func TestGetJobError(t *testing.T) {
 	err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error {
-		converter := instructions.NewInstructionConverter(metrics.Get().Metrics, userAnnotationPrefix, []string{}, &compress.NoOpCompressor{})
+		converter := instructions.NewInstructionConverter(metrics.Get().Metrics, userAnnotationPrefix, []string{}, &compress.NoOpCompressor{}, false)
 		store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10)
 		errMsg := "some bad error happened!"
 		_ = NewJobSimulator(converter, store).
diff --git a/internal/lookout/repository/getjobrundebugmessage_test.go b/internal/lookout/repository/getjobrundebugmessage_test.go
@@ -16,7 +16,7 @@ import (
 
 func TestGetJobRunDebugMessage(t *testing.T) {
 	err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error {
-		converter := instructions.NewInstructionConverter(metrics.Get().Metrics, userAnnotationPrefix, []string{}, &compress.NoOpCompressor{})
+		converter := instructions.NewInstructionConverter(metrics.Get().Metrics, userAnnotationPrefix, []string{}, &compress.NoOpCompressor{}, false)
 		store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10)
 
 		debugMessageStrings := []string{
diff --git a/internal/lookout/repository/getjobrunerror_test.go b/internal/lookout/repository/getjobrunerror_test.go
@@ -16,7 +16,7 @@ import (
 
 func TestGetJobRunError(t *testing.T) {
 	err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error {
-		converter := instructions.NewInstructionConverter(metrics.Get().Metrics, userAnnotationPrefix, []string{}, &compress.NoOpCompressor{})
+		converter := instructions.NewInstructionConverter(metrics.Get().Metrics, userAnnotationPrefix, []string{}, &compress.NoOpCompressor{}, false)
 		store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10)
 
 		errorStrings := []string{
diff --git a/internal/lookout/repository/getjobs_test.go b/internal/lookout/repository/getjobs_test.go
@@ -59,7 +59,7 @@ var (
 func withGetJobsSetup(f func(*instructions.InstructionConverter, *lookoutdb.LookoutDb, *SqlGetJobsRepository, *clock.FakeClock) error) error {
 	testClock := clock.NewFakeClock(time.Now())
 	return lookout.WithLookoutDb(func(db *pgxpool.Pool) error {
-		converter := instructions.NewInstructionConverter(metrics.Get().Metrics, userAnnotationPrefix, []string{}, &compress.NoOpCompressor{})
+		converter := instructions.NewInstructionConverter(metrics.Get().Metrics, userAnnotationPrefix, []string{}, &compress.NoOpCompressor{}, false)
 		store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10)
 		repo := NewSqlGetJobsRepository(db)
 		repo.clock = testClock
diff --git a/internal/lookout/repository/getjobspec_test.go b/internal/lookout/repository/getjobspec_test.go
@@ -18,7 +18,7 @@ import (
 
 func TestGetJobSpec(t *testing.T) {
 	err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error {
-		converter := instructions.NewInstructionConverter(metrics.Get().Metrics, userAnnotationPrefix, []string{}, &compress.NoOpCompressor{})
+		converter := instructions.NewInstructionConverter(metrics.Get().Metrics, userAnnotationPrefix, []string{}, &compress.NoOpCompressor{}, false)
 		store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10)
 
 		job := NewJobSimulator(converter, store).
@@ -54,7 +54,7 @@ func TestGetJobSpec(t *testing.T) {
 func TestMIGRATEDGetJobSpec(t *testing.T) {
 	var migratedResult *api.Job
 	err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error {
-		converter := instructions.NewInstructionConverter(metrics.Get().Metrics, userAnnotationPrefix, []string{}, &compress.NoOpCompressor{})
+		converter := instructions.NewInstructionConverter(metrics.Get().Metrics, userAnnotationPrefix, []string{}, &compress.NoOpCompressor{}, false)
 		store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10)
 
 		_ = NewJobSimulator(converter, store).
diff --git a/internal/lookout/repository/groupjobs_test.go b/internal/lookout/repository/groupjobs_test.go
@@ -22,7 +22,7 @@ import (
 
 func withGroupJobsSetup(f func(*instructions.InstructionConverter, *lookoutdb.LookoutDb, *SqlGroupJobsRepository) error) error {
 	return lookout.WithLookoutDb(func(db *pgxpool.Pool) error {
-		converter := instructions.NewInstructionConverter(metrics.Get().Metrics, userAnnotationPrefix, []string{}, &compress.NoOpCompressor{})
+		converter := instructions.NewInstructionConverter(metrics.Get().Metrics, userAnnotationPrefix, []string{}, &compress.NoOpCompressor{}, false)
 		store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10)
 		repo := NewSqlGroupJobsRepository(db)
 		return f(converter, store, repo)
diff --git a/internal/lookoutingester/configuration/types.go b/internal/lookoutingester/configuration/types.go
@@ -49,6 +49,9 @@ type LookoutIngesterConfiguration struct {
 	Profiling *profilingconfig.ProfilingConfig
 	// List of Regexes which will identify fatal errors when inserting into postgres
 	FatalInsertionErrors []string
+	// When true, populate FailureInfo on terminal job run errors from the proto FailureInfo field.
+	// When false, set FailureInfo to an empty map for terminal errors.
+	EnableJobRunFailureInfoMap bool
 }
 
 func (c *LookoutIngesterConfiguration) GetUserAnnotationPrefix() string {
diff --git a/internal/lookoutingester/dbloadtester/simulator.go b/internal/lookoutingester/dbloadtester/simulator.go
@@ -79,7 +79,7 @@ func Setup(lookoutIngesterConfig configuration.LookoutIngesterConfiguration, tes
 		panic(errors.WithMessage(err, "Error creating compressor"))
 	}
 
-	converter := instructions.NewInstructionConverter(m.Metrics, lookoutIngesterConfig.UserAnnotationPrefix, []string{}, compressor)
+	converter := instructions.NewInstructionConverter(m.Metrics, lookoutIngesterConfig.UserAnnotationPrefix, []string{}, compressor, lookoutIngesterConfig.EnableJobRunFailureInfoMap)
 
 	submitJobTemplate := &v1.PodSpec{}
 	err = clientUtil.BindJsonOrYaml(testConfig.JobTemplateFile, submitJobTemplate)
diff --git a/internal/lookoutingester/ingester.go b/internal/lookoutingester/ingester.go
@@ -65,6 +65,7 @@ func Run(config *configuration.LookoutIngesterConfiguration) {
 		config.GetUserAnnotationPrefix(),
 		config.Annotations.BlocklistAnnotations,
 		compressor,
+		config.EnableJobRunFailureInfoMap,
 	)
 
 	ingester := ingest.NewIngestionPipeline[*model.InstructionSet, *armadaevents.EventSequence](
diff --git a/internal/lookoutingester/instructions/instructions.go b/internal/lookoutingester/instructions/instructions.go
@@ -40,10 +40,11 @@ type HasNodeName interface {
 }
 
 type InstructionConverter struct {
-	metrics              *metrics.Metrics
-	userAnnotationPrefix string
-	blocklistAnnotations map[string]struct{}
-	compressor           compress.Compressor
+	metrics                    *metrics.Metrics
+	userAnnotationPrefix       string
+	blocklistAnnotations       map[string]struct{}
+	compressor                 compress.Compressor
+	enableJobRunFailureInfoMap bool
 }
 
 type jobResources struct {
@@ -53,16 +54,17 @@ type jobResources struct {
 	Gpu              int64
 }
 
-func NewInstructionConverter(m *metrics.Metrics, userAnnotationPrefix string, blocklistAnnotations []string, compressor compress.Compressor) *InstructionConverter {
+func NewInstructionConverter(m *metrics.Metrics, userAnnotationPrefix string, blocklistAnnotations []string, compressor compress.Compressor, enableJobRunFailureInfoMap bool) *InstructionConverter {
 	blocked := make(map[string]struct{}, len(blocklistAnnotations))
 	for _, b := range blocklistAnnotations {
 		blocked[strings.ToLower(b)] = struct{}{}
 	}
 	return &InstructionConverter{
-		metrics:              m,
-		userAnnotationPrefix: userAnnotationPrefix,
-		blocklistAnnotations: blocked,
-		compressor:           compressor,
+		metrics:                    m,
+		userAnnotationPrefix:       userAnnotationPrefix,
+		blocklistAnnotations:       blocked,
+		compressor:                 compressor,
+		enableJobRunFailureInfoMap: enableJobRunFailureInfoMap,
 	}
 }
 
@@ -176,7 +178,7 @@ func (c *InstructionConverter) handleSubmitJob(
 	}
 
 	annotations := event.GetObjectMeta().GetAnnotations()
-	userAnnotations := extractUserAnnotations(c.userAnnotationPrefix, c.blocklistAnnotations, annotations)
+	userAnnotations := extractUserAnnotations(event.JobId, c.userAnnotationPrefix, c.blocklistAnnotations, annotations)
 	// Prefer the proto field; fall back to annotation for events produced before the field existed.
 	externalJobUri := event.ExternalJobUri
 	if externalJobUri == "" {
@@ -215,7 +217,7 @@ filters out any keys in blocklistAnnotations (case-insensitive),
 and truncates keys and values to their maximal lengths as specified by
 maxAnnotationKeyLen and maxAnnotationValLen.
 */
-func extractUserAnnotations(userAnnotationPrefix string, blocklistAnnotations map[string]struct{}, jobAnnotations map[string]string) map[string]string {
+func extractUserAnnotations(jobId string, userAnnotationPrefix string, blocklistAnnotations map[string]struct{}, jobAnnotations map[string]string) map[string]string {
 	result := make(map[string]string, len(jobAnnotations))
 	n := len(userAnnotationPrefix)
 	for k, v := range jobAnnotations {
@@ -225,13 +227,24 @@ func extractUserAnnotations(userAnnotationPrefix string, blocklistAnnotations ma
 		if strings.HasPrefix(k, userAnnotationPrefix) {
 			k = k[n:]
 		}
-		key := util.Truncate(k, maxAnnotationKeyLen)
-		val := util.Truncate(v, maxAnnotationValLen)
+		key := sanitizeForJsonb(k)
+		val := sanitizeForJsonb(v)
+		if key != k || val != v {
+			log.Warnf("Sanitized annotation for job %s: key %q (was %q), value %q (was %q)", jobId, key, k, val, v)
+		}
+		key = util.Truncate(key, maxAnnotationKeyLen)
+		val = util.Truncate(val, maxAnnotationValLen)
 		result[key] = val
 	}
 	return result
 }
 
+// sanitizeForJsonb removes null bytes (\x00) from s.
+// PostgreSQL jsonb rejects strings containing \u0000 with SQLSTATE 22P05.
+func sanitizeForJsonb(s string) string {
+	return strings.ReplaceAll(s, "\x00", "")
+}
+
 func (c *InstructionConverter) handleReprioritiseJob(_ time.Time, event *armadaevents.ReprioritisedJob, update *model.InstructionSet) error {
 	jobUpdate := model.UpdateJobInstruction{
 		JobId:    event.JobId,
@@ -454,7 +467,11 @@ func (c *InstructionConverter) handleJobRunErrors(ts time.Time, event *armadaeve
 		// may be retried and the final FailureInfo is the one that matters.
 		if e.Terminal {
 			if fi := e.GetFailureInfo(); fi != nil {
-				jobRunUpdate.FailureInfo = failureInfoToMap(fi)
+				if c.enableJobRunFailureInfoMap {
+					jobRunUpdate.FailureInfo = failureInfoToMap(event.JobId, fi)
+				} else {
+					jobRunUpdate.FailureInfo = map[string]any{}
+				}
 			}
 		}
 		update.JobRunsToUpdate = append(update.JobRunsToUpdate, jobRunUpdate)
@@ -486,13 +503,18 @@ func (c *InstructionConverter) handleStandaloneIngressInfo(event *armadaevents.S
 // failureInfoToMap converts a FailureInfo proto to a JSON-friendly map for JSONB storage.
 // Only non-zero fields are included. After a JSON round-trip through PostgreSQL,
 // numeric fields arrive as float64 - see failureInfoToSwagger for the read side.
-func failureInfoToMap(fi *armadaevents.FailureInfo) map[string]any {
+func failureInfoToMap(jobId string, fi *armadaevents.FailureInfo) map[string]any {
 	m := make(map[string]any, 4)
 	if fi.GetExitCode() != 0 {
 		m["exitCode"] = fi.GetExitCode()
 	}
 	if fi.GetTerminationMessage() != "" {
-		m["terminationMessage"] = util.Truncate(fi.GetTerminationMessage(), maxTerminationMessageLen)
+		terminationMessage := util.Truncate(fi.GetTerminationMessage(), maxTerminationMessageLen)
+		sanitizedTerminationMessage := sanitizeForJsonb(terminationMessage)
+		if sanitizedTerminationMessage != terminationMessage {
+			log.Warnf("Sanitized termination message for job %s: %q (was %q)", jobId, sanitizedTerminationMessage, terminationMessage)
+		}
+		m["terminationMessage"] = sanitizedTerminationMessage
 	}
 	if len(fi.GetCategories()) > 0 {
 		m["categories"] = fi.GetCategories()
diff --git a/internal/lookoutingester/instructions/instructions_test.go b/internal/lookoutingester/instructions/instructions_test.go

-Original file line number
+Diff line change
 tmp.*
 *.tmp.*
 .aider*
++
 +.claude/
Original file line number	Diff line number	Diff line change
`@@ -49,6 +49,9 @@ type LookoutIngesterConfiguration struct {`
`49`	`49`	`Profiling *profilingconfig.ProfilingConfig`
`50`	`50`	`// List of Regexes which will identify fatal errors when inserting into postgres`
`51`	`51`	`FatalInsertionErrors []string`
	`52`	`+ // When true, populate FailureInfo on terminal job run errors from the proto FailureInfo field.`
	`53`	`+ // When false, set FailureInfo to an empty map for terminal errors.`
	`54`	`+ EnableJobRunFailureInfoMap bool`
`52`	`55`	`}`
`53`	`56`
`54`	`57`	`func (c *LookoutIngesterConfiguration) GetUserAnnotationPrefix() string {`
Original file line number	Diff line number	Diff line change
`@@ -79,7 +79,7 @@ func Setup(lookoutIngesterConfig configuration.LookoutIngesterConfiguration, tes`
`79`	`79`	`panic(errors.WithMessage(err, "Error creating compressor"))`
`80`	`80`	`}`
`81`	`81`
`82`		`- converter := instructions.NewInstructionConverter(m.Metrics, lookoutIngesterConfig.UserAnnotationPrefix, []string{}, compressor)`
	`82`	`+ converter := instructions.NewInstructionConverter(m.Metrics, lookoutIngesterConfig.UserAnnotationPrefix, []string{}, compressor, lookoutIngesterConfig.EnableJobRunFailureInfoMap)`
`83`	`83`
`84`	`84`	`submitJobTemplate := &v1.PodSpec{}`
`85`	`85`	`err = clientUtil.BindJsonOrYaml(testConfig.JobTemplateFile, submitJobTemplate)`