diff --git a/go.mod b/go.mod index 54385e09f73..96f55521ed3 100644 --- a/go.mod +++ b/go.mod @@ -51,6 +51,7 @@ require ( github.com/IBM/pgxpoolprometheus v1.1.2 github.com/Masterminds/semver/v3 v3.4.0 github.com/benbjohnson/immutable v0.4.3 + github.com/cenkalti/backoff/v4 v4.3.0 github.com/charmbracelet/glamour v0.10.0 github.com/go-openapi/errors v0.22.6 github.com/go-openapi/strfmt v0.25.0 @@ -121,7 +122,6 @@ require ( github.com/blang/semver/v4 v4.0.0 // indirect github.com/buger/jsonparser v1.1.1 // indirect github.com/caarlos0/log v0.5.4 // indirect - github.com/cenkalti/backoff/v4 v4.3.0 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/charmbracelet/colorprofile v0.4.1 // indirect github.com/charmbracelet/lipgloss v1.1.1-0.20250404203927-76690c660834 // indirect diff --git a/internal/broadside/db/postgres.go b/internal/broadside/db/postgres.go index 463241eb968..1495795142d 100644 --- a/internal/broadside/db/postgres.go +++ b/internal/broadside/db/postgres.go @@ -117,7 +117,7 @@ func (p *PostgresDatabase) InitialiseSchema(ctx context.Context) error { } decompressor := &compress.NoOpDecompressor{} - p.lookoutDb = lookoutdb.NewLookoutDb(p.pool, nil, lookoutingestermetrics.Get(), 16, 12) + p.lookoutDb = lookoutdb.NewLookoutDb(p.pool, nil, lookoutingestermetrics.Get()) if p.features.HotColdSplit { tables := repository.NewTablesWithJobTable("job_all") p.jobsRepository = repository.NewSqlGetJobsRepositoryWithTables(p.pool, tables) diff --git a/internal/common/config/pulsar.go b/internal/common/config/pulsar.go index be0989f75f7..96c2d55592c 100644 --- a/internal/common/config/pulsar.go +++ b/internal/common/config/pulsar.go @@ -63,7 +63,8 @@ type PulsarConfig struct { // The pulsar topic that messages will be published to if a sink cannot store them after DeadLetterMaxAttempts attempts DeadLetterTopic string // Number of consecutive Sink.Store attempts before a message is published to DeadLetterTopic and acked. - // If set, must be at least 2: a value of 1 would dead-letter on the first failure with no retry at all. + // If unset, dead-lettering is not configured. If set, must be at least 2: a value of 1 would + // dead-letter on the first failure with no retry at all. DeadLetterMaxAttempts int `validate:"omitempty,gte=2"` } diff --git a/internal/common/ingest/ingestion_pipeline.go b/internal/common/ingest/ingestion_pipeline.go index 75707e93a1f..edbce730b1b 100644 --- a/internal/common/ingest/ingestion_pipeline.go +++ b/internal/common/ingest/ingestion_pipeline.go @@ -1,12 +1,12 @@ package ingest import ( - "context" "fmt" "sync" "time" "github.com/apache/pulsar-client-go/pulsar" + "github.com/cenkalti/backoff/v4" "github.com/pkg/errors" "github.com/armadaproject/armada/internal/common/armadacontext" @@ -48,6 +48,15 @@ type Sink[T HasPulsarMessageIds] interface { // Store should persist the sink. The store is responsible for retrying failed attempts and should only return an error // When it is satisfied that operation cannot be retries. Store(ctx *armadacontext.Context, msg T) error + // Serialize renders msg as a self-contained byte payload for the dead-letter topic. + // Only called when the pipeline is about to give up on msg, never on the happy path. + Serialize(msg T) ([]byte, error) +} + +// deadLetterPublisher is implemented by *pulsarutils.DeadLetterPublisher; declared here so tests can inject a fake. +type deadLetterPublisher interface { + Publish(ctx *armadacontext.Context, payload []byte, meta pulsarutils.DeadLetterMetadata) error + Close() } // IngestionPipeline is a pipeline that reads message from pulsar and inserts them into a sink. The pipeline will @@ -76,6 +85,7 @@ type IngestionPipeline[T HasPulsarMessageIds, U utils.ArmadaEvent] struct { converter InstructionConverter[T, U] sink Sink[T] consumer pulsar.Consumer // for test purposes only + deadLetterPublisher deadLetterPublisher } // NewIngestionPipeline creates an IngestionPipeline that processes all pulsar messages @@ -111,6 +121,46 @@ func NewIngestionPipeline[T HasPulsarMessageIds, U utils.ArmadaEvent]( } } +// defaultDeadLetterMaxAttempts is used when PulsarConfig.DeadLetterMaxAttempts is unset (0), +// i.e. dead-lettering has not been explicitly configured. It still bounds retries so that a +// persistently failing message doesn't block the pipeline forever; the message is dead-lettered +// (if DeadLetterTopic is set) or, if not, left unacked for redelivery - see deadLetterMaxAttempts. +const defaultDeadLetterMaxAttempts = 5 + +// deadLetterMaxAttempts returns the effective number of Sink.Store attempts before a message is +// given up on, defaulting to defaultDeadLetterMaxAttempts when PulsarConfig.DeadLetterMaxAttempts +// is unset. +func (i *IngestionPipeline[T, U]) deadLetterMaxAttempts() int { + if i.pulsarConfig.DeadLetterMaxAttempts > 0 { + return i.pulsarConfig.DeadLetterMaxAttempts + } + return defaultDeadLetterMaxAttempts +} + +// newBackOff returns a fresh jittered exponential backoff sequence, starting at +// i.pulsarConfig.BackoffTime and capped at i.pulsarConfig.MaxBackoffTime (falling back to +// BackoffTime, i.e. no growth, if MaxBackoffTime is unset). Randomization and growth can be +// tuned via i.pulsarConfig.BackoffRandomizationFactor and BackoffMultiplier; if unset, the +// backoff library's own defaults are used. +func (i *IngestionPipeline[T, U]) newBackOff() *backoff.ExponentialBackOff { + maxInterval := i.pulsarConfig.MaxBackoffTime + if maxInterval <= 0 { + maxInterval = i.pulsarConfig.BackoffTime + } + opts := []backoff.ExponentialBackOffOpts{ + backoff.WithInitialInterval(i.pulsarConfig.BackoffTime), + backoff.WithMaxInterval(maxInterval), + backoff.WithMaxElapsedTime(0), + } + if i.pulsarConfig.BackoffRandomizationFactor >= 0 { + opts = append(opts, backoff.WithRandomizationFactor(i.pulsarConfig.BackoffRandomizationFactor)) + } + if i.pulsarConfig.BackoffMultiplier > 0 { + opts = append(opts, backoff.WithMultiplier(i.pulsarConfig.BackoffMultiplier)) + } + return backoff.NewExponentialBackOff(opts...) +} + // Run will run the ingestion pipeline until the supplied context is shut down func (i *IngestionPipeline[T, U]) Run(ctx *armadacontext.Context) error { // Waitgroup that wil fire when the pipeline has been torn down @@ -125,6 +175,21 @@ func (i *IngestionPipeline[T, U]) Run(ctx *armadacontext.Context) error { i.consumer = consumer defer closePulsar() + if i.pulsarConfig.DeadLetterTopic != "" { + deadLetterPublisher, err := pulsarutils.NewDeadLetterPublisher( + client, + i.pulsarConfig.DeadLetterTopic, + i.pulsarConfig.CompressionType, + i.pulsarConfig.CompressionLevel, + i.pulsarConfig.SendTimeout, + ) + if err != nil { + return errors.WithMessage(err, "error creating dead-letter publisher") + } + i.deadLetterPublisher = deadLetterPublisher + defer deadLetterPublisher.Close() + } + if i.pulsarConfig.DelayMonitor.Enabled { err := i.startProcessingDelayMonitor(ctx, client) if err != nil { @@ -153,7 +218,7 @@ func (i *IngestionPipeline[T, U]) Run(ctx *armadacontext.Context) error { pulsarMessages <- msg lastReceivedTime = time.Now() case <-ticker.C: - timeSinceLastReceived := time.Now().Sub(lastReceivedTime) + timeSinceLastReceived := time.Since(lastReceivedTime) if timeSinceLastReceived > timeout { log.Infof("%s - Last pulsar message received %s ago", i.pulsarTopic, timeSinceLastReceived) } @@ -205,7 +270,7 @@ func (i *IngestionPipeline[T, U]) Run(ctx *armadacontext.Context) error { for batch := range preprocessedEventBatches { start := time.Now() converted := i.converter.Convert(ctx, batch) - taken := time.Now().Sub(start) + taken := time.Since(start) log.Infof("%s - Processed %d pulsar messages in %dms", i.pulsarTopic, len(batch.MessageIds), taken.Milliseconds()) instructions <- converted } @@ -214,32 +279,95 @@ func (i *IngestionPipeline[T, U]) Run(ctx *armadacontext.Context) error { // Publish messages to sink then ACK on pulsar go func() { + loop: for msg := range instructions { - // The sink is responsible for retrying any messages so if we get a message here we know we can give up - // and just ACK the ids start := time.Now() - err := i.sink.Store(ctx, msg) - taken := time.Now().Sub(start) - if err != nil { - log.WithError(err).Warnf("%s - Error inserting messages", i.pulsarTopic) - } else { - log.Infof("%s - Inserted %d pulsar messages in %dms", i.pulsarTopic, len(msg.GetMessageIDs()), taken.Milliseconds()) - } - if errors.Is(err, context.DeadlineExceeded) { - // This occurs when we're shutting down- it's a signal to stop processing immediately - break - } else { - for _, msgId := range msg.GetMessageIDs() { + storeBackoff := i.newBackOff() + dropped := false + deadLettered := false + succeeded := util.RetryUntilSuccessOrExhausted( + ctx, + i.deadLetterMaxAttempts(), + func() error { + return i.sink.Store(ctx, msg) + }, + func(attempt int, err error) { + i.metrics.RecordPulsarMessageStoreRetry() + wait := storeBackoff.NextBackOff() + log.WithError(err).Warnf("%s - Error inserting %d messages (ids: %v); will retry after %s", + i.pulsarTopic, len(msg.GetMessageIDs()), msg.GetMessageIDs(), wait) + // This sleep is not ctx-aware: RetryUntilSuccessOrExhausted only checks ctx + // before the next performAction call, not during this wait. On shutdown, the + // sleep runs to completion before cancellation is noticed, delaying shutdown + // by up to `wait`. Deemed acceptable since wait is bounded by backoff config. + time.Sleep(wait) + }, + func(lastErr error) { + if i.deadLetterPublisher == nil { + // No dead-letter topic configured: leave the message unacked for + // redelivery rather than dropping it silently. + dropped = true + log.WithError(lastErr).Warnf("%s - Exhausted %d attempts inserting %d messages (ids: %v); no dead-letter topic configured, leaving unacked for redelivery", + i.pulsarTopic, i.deadLetterMaxAttempts(), len(msg.GetMessageIDs()), msg.GetMessageIDs()) + return + } + log.WithError(lastErr).Warnf("%s - Exhausted %d attempts inserting %d messages (ids: %v); publishing to dead-letter topic", + i.pulsarTopic, i.deadLetterMaxAttempts(), len(msg.GetMessageIDs()), msg.GetMessageIDs()) + payload, err := i.sink.Serialize(msg) + if err != nil { + log.WithError(err).Warnf("%s - Error serializing dead-lettered messages (ids: %v); publishing error text instead", + i.pulsarTopic, msg.GetMessageIDs()) + payload = []byte(err.Error()) + } + meta := pulsarutils.DeadLetterMetadata{ + OriginalTopic: i.pulsarTopic, + Subscription: i.pulsarSubscriptionName, + Attempts: i.deadLetterMaxAttempts(), + LastError: lastErr.Error(), + MessageIDs: pulsarutils.MessageIdsToStrings(msg.GetMessageIDs()), + } + dlqBackoff := i.newBackOff() util.RetryUntilSuccess( - armadacontext.Background(), - func() error { return i.consumer.AckID(msgId) }, + ctx, + func() error { + err := i.deadLetterPublisher.Publish(ctx, payload, meta) + if err == nil { + deadLettered = true + } + return err + }, func(err error) { - log.WithError(err).Warnf("%s - Pulsar ack failed; backing off for %s", i.pulsarTopic, i.pulsarConfig.BackoffTime) - time.Sleep(i.pulsarConfig.BackoffTime) + wait := dlqBackoff.NextBackOff() + log.WithError(err).Warnf("%s - Dead-letter publish failed; backing off for %s", i.pulsarTopic, wait) + time.Sleep(wait) }, ) - i.metrics.RecordPulsarMessageProcessed() - } + if deadLettered { + i.metrics.RecordPulsarMessageDeadLettered() + } + }, + ) + if !succeeded && !deadLettered && (ctx.Err() != nil || dropped) { + // Either ctx was cancelled (e.g. ingester shutdown) while retrying, or attempts + // were exhausted with no dead-letter topic configured; the message is left + // unacked for redelivery rather than dropped. If the message was successfully + // dead-lettered, it must still be acked below even if ctx was cancelled + // immediately afterwards, otherwise it is redelivered and dead-lettered again + // on restart. + break loop + } + taken := time.Since(start) + log.Infof("%s - Inserted %d pulsar messages in %dms", i.pulsarTopic, len(msg.GetMessageIDs()), taken.Milliseconds()) + for _, msgId := range msg.GetMessageIDs() { + util.RetryUntilSuccess( + armadacontext.Background(), + func() error { return i.consumer.AckID(msgId) }, + func(err error) { + log.WithError(err).Warnf("%s - Pulsar ack failed; backing off for %s", i.pulsarTopic, i.pulsarConfig.BackoffTime) + time.Sleep(i.pulsarConfig.BackoffTime) + }, + ) + i.metrics.RecordPulsarMessageProcessed() } } wg.Done() diff --git a/internal/common/ingest/ingestion_pipeline_test.go b/internal/common/ingest/ingestion_pipeline_test.go index fc3c502c89a..799dcf085eb 100644 --- a/internal/common/ingest/ingestion_pipeline_test.go +++ b/internal/common/ingest/ingestion_pipeline_test.go @@ -1,6 +1,7 @@ package ingest import ( + "fmt" "sync" "testing" "time" @@ -172,6 +173,35 @@ func (p *mockPulsarConsumer) Close() { // do nothing } +type fakeDeadLetterPublisher struct { + mutex sync.Mutex + published []pulsarutils.DeadLetterMetadata + shouldFail bool +} + +func newFakeDeadLetterPublisher() *fakeDeadLetterPublisher { + return &fakeDeadLetterPublisher{} +} + +func (f *fakeDeadLetterPublisher) Publish(_ *armadacontext.Context, _ []byte, meta pulsarutils.DeadLetterMetadata) error { + if f.shouldFail { + return assert.AnError + } + f.mutex.Lock() + defer f.mutex.Unlock() + f.published = append(f.published, meta) + return nil +} + +func (f *fakeDeadLetterPublisher) Close() {} + +func (f *fakeDeadLetterPublisher) assertPublishedCount(t *testing.T, count int) { + t.Helper() + f.mutex.Lock() + defer f.mutex.Unlock() + assert.Len(t, f.published, count) +} + type simpleMessage struct { id pulsar.MessageID size int @@ -345,10 +375,60 @@ func TestRun_ControlPlaneEvents_LimitsProcessingBatchSize(t *testing.T) { } } +func TestRun_ControlPlaneEvents_ExhaustsRetriesThenDeadLettersAndAcks(t *testing.T) { + ctx, cancel := armadacontext.WithDeadline(armadacontext.Background(), time.Now().Add(10*time.Second)) + messages := []pulsar.Message{ + pulsarutils.NewPulsarMessage(1, baseTime, marshal(t, f.UpsertExecutorSettingsCordon)), + } + mockConsumer := newMockPulsarConsumer(t, messages, cancel) + converter := newSimpleControlPlaneEventConverter(t) + sink := &alwaysFailingSink{} + dlq := newFakeDeadLetterPublisher() + + pipeline := testControlPlaneEventsPipeline(mockConsumer, converter, sink) + pipeline.pulsarConfig.DeadLetterMaxAttempts = 3 + pipeline.pulsarConfig.BackoffTime = time.Millisecond + pipeline.deadLetterPublisher = dlq + + err := pipeline.Run(ctx) + assert.NoError(t, err) + + mockConsumer.assertDidAck(messages) + dlq.assertPublishedCount(t, 1) +} + +func TestRun_ControlPlaneEvents_CancelledMidRetry_NotAckedNotDeadLettered(t *testing.T) { + ctx, cancel := armadacontext.WithCancel(armadacontext.Background()) + messages := []pulsar.Message{ + pulsarutils.NewPulsarMessage(1, baseTime, marshal(t, f.UpsertExecutorSettingsCordon)), + } + mockConsumer := newMockPulsarConsumer(t, messages, func() {}) + converter := newSimpleControlPlaneEventConverter(t) + sink := &alwaysFailingSink{} + dlq := newFakeDeadLetterPublisher() + + pipeline := testControlPlaneEventsPipeline(mockConsumer, converter, sink) + pipeline.pulsarConfig.DeadLetterMaxAttempts = 1000000 + pipeline.pulsarConfig.BackoffTime = 50 * time.Millisecond + pipeline.deadLetterPublisher = dlq + + go func() { + time.Sleep(200 * time.Millisecond) + cancel() + }() + + err := pipeline.Run(ctx) + assert.NoError(t, err) + + assert.Empty(t, mockConsumer.acked) + dlq.assertPublishedCount(t, 0) +} + func testControlPlaneEventsPipeline(consumer pulsar.Consumer, converter InstructionConverter[*simpleMessages, *controlplaneevents.Event], sink Sink[*simpleMessages]) *IngestionPipeline[*simpleMessages, *controlplaneevents.Event] { return &IngestionPipeline[*simpleMessages, *controlplaneevents.Event]{ pulsarConfig: commonconfig.PulsarConfig{ - BackoffTime: time.Second, + BackoffTime: time.Second, + DeadLetterMaxAttempts: 5, }, pulsarTopic: controlPlaneEventsTopic, pulsarSubscriptionName: "subscription", @@ -362,6 +442,7 @@ func testControlPlaneEventsPipeline(consumer pulsar.Consumer, converter Instruct sink: sink, metrics: testMetrics, consumer: consumer, + deadLetterPublisher: newFakeDeadLetterPublisher(), } } @@ -390,6 +471,20 @@ func (s *simpleSink) Store(_ *armadacontext.Context, msg *simpleMessages) error return nil } +func (s *simpleSink) Serialize(msg *simpleMessages) ([]byte, error) { + return fmt.Appendf(nil, "%+v", msg), nil +} + +type alwaysFailingSink struct{} + +func (s *alwaysFailingSink) Store(_ *armadacontext.Context, _ *simpleMessages) error { + return assert.AnError +} + +func (s *alwaysFailingSink) Serialize(msg *simpleMessages) ([]byte, error) { + return fmt.Appendf(nil, "%+v", msg), nil +} + func (s *simpleSink) assertDidProcess(messages []pulsar.Message) { s.t.Helper() for _, msg := range messages { @@ -569,7 +664,8 @@ func TestRun_MultipleSimultaneousIngesters(t *testing.T) { func testJobSetEventsPipeline(consumer pulsar.Consumer, converter InstructionConverter[*simpleMessages, *armadaevents.EventSequence], sink Sink[*simpleMessages]) *IngestionPipeline[*simpleMessages, *armadaevents.EventSequence] { return &IngestionPipeline[*simpleMessages, *armadaevents.EventSequence]{ pulsarConfig: commonconfig.PulsarConfig{ - BackoffTime: time.Second, + BackoffTime: time.Second, + DeadLetterMaxAttempts: 5, }, pulsarTopic: jobSetEventsTopic, pulsarSubscriptionName: "subscription", @@ -583,6 +679,7 @@ func testJobSetEventsPipeline(consumer pulsar.Consumer, converter InstructionCon sink: sink, metrics: testMetrics, consumer: consumer, + deadLetterPublisher: newFakeDeadLetterPublisher(), } } diff --git a/internal/common/ingest/metrics/metrics.go b/internal/common/ingest/metrics/metrics.go index 3cad747b1c0..6da01c46953 100644 --- a/internal/common/ingest/metrics/metrics.go +++ b/internal/common/ingest/metrics/metrics.go @@ -37,6 +37,8 @@ type Metrics struct { pulsarConnectionError prometheus.Counter pulsarMessageError *prometheus.CounterVec pulsarMessagesProcessed prometheus.Counter + pulsarMessageStoreRetries prometheus.Counter + pulsarMessagesDeadLettered prometheus.Counter pulsarMessagePublishTime *prometheus.GaugeVec pulsarMessageProcessingDelay *prometheus.GaugeVec eventsProcessed *prometheus.CounterVec @@ -65,6 +67,14 @@ func NewMetricsWithRegistry(prefix string, registerer prometheus.Registerer) *Me Name: prefix + "pulsar_messages_processed", Help: "Number of pulsar messages processed", } + pulsarMessageStoreRetriesOpts := prometheus.CounterOpts{ + Name: prefix + "pulsar_message_store_retries", + Help: "Number of times a sink Store call failed and was retried before the batch was acked", + } + pulsarMessagesDeadLetteredOpts := prometheus.CounterOpts{ + Name: prefix + "pulsar_messages_dead_lettered", + Help: "Number of message batches that exhausted retries and were published to the dead-letter topic", + } pulsarMessagePublishTime := prometheus.GaugeOpts{ Name: prefix + "pulsar_message_publish_time", Help: "Publish time of pulsar message being processed", @@ -94,6 +104,8 @@ func NewMetricsWithRegistry(prefix string, registerer prometheus.Registerer) *Me pulsarMessageProcessingDelay: factory.NewGaugeVec(pulsarMessageProcessingDelayOpts, []string{"subscription", "partition"}), pulsarMessagePublishTime: factory.NewGaugeVec(pulsarMessagePublishTime, []string{"subscription", "partition"}), pulsarMessagesProcessed: factory.NewCounter(pulsarMessagesProcessedOpts), + pulsarMessageStoreRetries: factory.NewCounter(pulsarMessageStoreRetriesOpts), + pulsarMessagesDeadLettered: factory.NewCounter(pulsarMessagesDeadLetteredOpts), eventsProcessed: factory.NewCounterVec(eventsProcessedOpts, []string{"queue", "eventType", "msgType"}), uncompressedEventBytesTotal: factory.NewCounterVec(uncompressedEventBytesTotalOpts, []string{"queue", "event_type"}), estimatedCompressedEventBytesTotal: factory.NewCounterVec(estimatedCompressedEventBytesTotalOpts, []string{"queue", "event_type"}), @@ -116,6 +128,14 @@ func (m *Metrics) RecordPulsarMessageProcessed() { m.pulsarMessagesProcessed.Inc() } +func (m *Metrics) RecordPulsarMessageStoreRetry() { + m.pulsarMessageStoreRetries.Inc() +} + +func (m *Metrics) RecordPulsarMessageDeadLettered() { + m.pulsarMessagesDeadLettered.Inc() +} + func (m *Metrics) RecordPulsarMessagePublishTime(subscriptionName string, partition int, publishTime time.Time) { partitionStr := strconv.Itoa(partition) m.pulsarMessagePublishTime.WithLabelValues(subscriptionName, partitionStr).Set(float64(publishTime.UTC().Unix())) diff --git a/internal/common/ingest/retry.go b/internal/common/ingest/retry.go deleted file mode 100644 index 810b0c346a2..00000000000 --- a/internal/common/ingest/retry.go +++ /dev/null @@ -1,35 +0,0 @@ -package ingest - -import ( - "time" - - log "github.com/armadaproject/armada/internal/common/logging" -) - -// WithRetry executes the supplied action until it either completes successfully or it returns false, indicating that -// the error is fatal -func WithRetry(action func() (bool, error), intialBackoff time.Duration, maxBackOff time.Duration) error { - backOff := intialBackoff - for { - retry, err := action() - if err == nil { - return nil - } - if retry { - backOff = min(2*backOff, maxBackOff) - log.WithError(err).Warnf("Retryable error encountered, will wait for %s before retrying", backOff) - time.Sleep(backOff) - } else { - // Non retryable error - return err - } - } -} - -// min returns the minimum of two durations -func min(d1 time.Duration, d2 time.Duration) time.Duration { - if d1.Nanoseconds() < d2.Nanoseconds() { - return d1 - } - return d2 -} diff --git a/internal/eventingester/ingester.go b/internal/eventingester/ingester.go index 5d55fb0fd95..1b1c89e3329 100644 --- a/internal/eventingester/ingester.go +++ b/internal/eventingester/ingester.go @@ -2,7 +2,6 @@ package eventingester import ( "regexp" - "time" "github.com/apache/pulsar-client-go/pulsar" "github.com/pkg/errors" @@ -73,7 +72,7 @@ func Run(config *configuration.EventIngesterConfiguration) { dbNames = append(dbNames, "replica") } - eventDb := store.NewRedisEventStore(dbs, dbNames, config.EventRetentionPolicy, fatalRegexes, 100*time.Millisecond, 60*time.Second) + eventDb := store.NewRedisEventStore(dbs, dbNames, config.EventRetentionPolicy, fatalRegexes) g, ctx := armadacontext.ErrGroup(app.CreateContextWithShutdown()) diff --git a/internal/eventingester/store/doc.go b/internal/eventingester/store/doc.go new file mode 100644 index 00000000000..7a7138ca69b --- /dev/null +++ b/internal/eventingester/store/doc.go @@ -0,0 +1,10 @@ +// Package store provides the [ingest.Sink] implementation for the event +// ingester. [RedisEventStore] writes batches of job-set events to one or +// more Redis instances, sharded by job set, with event-retention expiry. +// [RedisEventStore.Store] attempts each write once and returns immediately +// on error; retry-then-dead-letter policy is owned by the shared +// [ingest.IngestionPipeline.Run] ack-path. Errors classified as +// non-retryable by isRetryableRedisError are wrapped with +// [util.ErrNonRetryable] so the ack-path dead-letters immediately +// instead of exhausting its retry budget first. +package store diff --git a/internal/eventingester/store/eventstore.go b/internal/eventingester/store/eventstore.go index e965fff202d..9fdf952a6d0 100644 --- a/internal/eventingester/store/eventstore.go +++ b/internal/eventingester/store/eventstore.go @@ -1,6 +1,7 @@ package store import ( + "encoding/json" "fmt" "regexp" "time" @@ -12,6 +13,8 @@ import ( "github.com/armadaproject/armada/internal/common/constants" "github.com/armadaproject/armada/internal/common/ingest" log "github.com/armadaproject/armada/internal/common/logging" + "github.com/armadaproject/armada/internal/common/pulsarutils" + "github.com/armadaproject/armada/internal/common/util" "github.com/armadaproject/armada/internal/eventingester/configuration" "github.com/armadaproject/armada/internal/eventingester/metrics" "github.com/armadaproject/armada/internal/eventingester/model" @@ -22,24 +25,20 @@ const ( ) type RedisEventStore struct { - dbs []redis.UniversalClient - dbNames []string - eventRetention configuration.EventRetentionPolicy - intialRetryBackoff time.Duration - maxRetryBackoff time.Duration - maxRows int - maxSize int - fatalErrors []*regexp.Regexp + dbs []redis.UniversalClient + dbNames []string + eventRetention configuration.EventRetentionPolicy + maxRows int + maxSize int + fatalErrors []*regexp.Regexp } -func NewRedisEventStore(dbs []redis.UniversalClient, dbNames []string, eventRetention configuration.EventRetentionPolicy, fatalErrors []*regexp.Regexp, intialRetryBackoff time.Duration, maxRetryBackoff time.Duration) ingest.Sink[*model.BatchUpdate] { +func NewRedisEventStore(dbs []redis.UniversalClient, dbNames []string, eventRetention configuration.EventRetentionPolicy, fatalErrors []*regexp.Regexp) ingest.Sink[*model.BatchUpdate] { return &RedisEventStore{ - dbs: dbs, - dbNames: dbNames, - eventRetention: eventRetention, - fatalErrors: fatalErrors, - intialRetryBackoff: intialRetryBackoff, - maxRetryBackoff: maxRetryBackoff, + dbs: dbs, + dbNames: dbNames, + eventRetention: eventRetention, + fatalErrors: fatalErrors, } } @@ -77,6 +76,17 @@ func (repo *RedisEventStore) Store(ctx *armadacontext.Context, update *model.Bat return result.ErrorOrNil() } +// Serialize renders update as JSON for the dead-letter topic. +func (repo *RedisEventStore) Serialize(update *model.BatchUpdate) ([]byte, error) { + return json.Marshal(struct { + MessageIds []string + Events []*model.Event + }{ + MessageIds: pulsarutils.MessageIdsToStrings(update.MessageIds), + Events: update.Events, + }) +} + type eventData struct { key string data []byte @@ -88,27 +98,28 @@ func (repo *RedisEventStore) doStore(ctx *armadacontext.Context, update []*model return nil } - return ingest.WithRetry(func() (bool, error) { - var data []eventData - uniqueJobSets := make(map[string]bool) + var data []eventData + uniqueJobSets := make(map[string]bool) - for _, e := range update { - key := getJobSetEventsKey(e.Queue, e.Jobset) - data = append(data, eventData{key: key, data: e.Event}) - uniqueJobSets[key] = true - } + for _, e := range update { + key := getJobSetEventsKey(e.Queue, e.Jobset) + data = append(data, eventData{key: key, data: e.Event}) + uniqueJobSets[key] = true + } - for i, db := range repo.dbs { - r, e := repo.writeToRedis(ctx, db, data, uniqueJobSets, repo.dbNames[i]) - if e != nil { - return r, fmt.Errorf("error with redis %s: %v", repo.dbNames[i], e) + for i, db := range repo.dbs { + if err := repo.writeToRedis(ctx, db, data, uniqueJobSets, repo.dbNames[i]); err != nil { + if !repo.isRetryableRedisError(err) { + log.WithError(err).Warnf("Non-retryable error writing to redis %s; returning immediately", repo.dbNames[i]) + return fmt.Errorf("%w: error with redis %s: %v", util.ErrNonRetryable, repo.dbNames[i], err) } + return fmt.Errorf("error with redis %s: %v", repo.dbNames[i], err) } - return false, nil - }, repo.intialRetryBackoff, repo.maxRetryBackoff) + } + return nil } -func (repo *RedisEventStore) writeToRedis(ctx *armadacontext.Context, db redis.UniversalClient, data []eventData, uniqueJobSets map[string]bool, redisName string) (bool, error) { +func (repo *RedisEventStore) writeToRedis(ctx *armadacontext.Context, db redis.UniversalClient, data []eventData, uniqueJobSets map[string]bool, redisName string) error { start := time.Now() pipe := db.Pipeline() for _, e := range data { @@ -128,17 +139,17 @@ func (repo *RedisEventStore) writeToRedis(ctx *armadacontext.Context, db redis.U cmders, err := pipe.Exec(ctx) if err != nil { metrics.RecordWriteDuration(redisName, "failed_write", time.Since(start)) - return repo.isRetryableRedisError(err), err + return err } err = populateRedisSequenceIds(cmders, data) if err != nil { metrics.RecordWriteDuration(redisName, "failed_get_sequence_id", time.Since(start)) - return true, err + return err } metrics.RecordWriteDuration(redisName, "success", time.Since(start)) - return false, nil + return nil } func populateRedisSequenceIds(cmders []redis.Cmder, data []eventData) error { diff --git a/internal/eventingester/store/eventstore_test.go b/internal/eventingester/store/eventstore_test.go index 6b4aed802ad..d4ce4a6d006 100644 --- a/internal/eventingester/store/eventstore_test.go +++ b/internal/eventingester/store/eventstore_test.go @@ -1,13 +1,16 @@ package store import ( + "encoding/json" "testing" "time" + "github.com/apache/pulsar-client-go/pulsar" "github.com/redis/go-redis/v9" "github.com/stretchr/testify/assert" "github.com/armadaproject/armada/internal/common/armadacontext" + "github.com/armadaproject/armada/internal/common/pulsarutils" "github.com/armadaproject/armada/internal/eventingester/configuration" "github.com/armadaproject/armada/internal/eventingester/model" ) @@ -46,6 +49,35 @@ func TestReportEvents(t *testing.T) { }) } +func TestSerialize(t *testing.T) { + ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 10*time.Second) + defer cancel() + withRedisEventStore(ctx, func(r *RedisEventStore) { + update := &model.BatchUpdate{ + Events: []*model.Event{ + {Queue: "testQueue", Jobset: "testJobset", Event: []byte{1}}, + }, + MessageIds: []pulsar.MessageID{pulsarutils.NewMessageId(1), pulsarutils.NewMessageId(2)}, + } + + bytes, err := r.Serialize(update) + assert.NoError(t, err) + + var decoded struct { + MessageIds []string + Events []*model.Event + } + err = json.Unmarshal(bytes, &decoded) + assert.NoError(t, err) + + assert.Equal(t, []string{ + pulsarutils.NewMessageId(1).String(), + pulsarutils.NewMessageId(2).String(), + }, decoded.MessageIds) + assert.Equal(t, update.Events, decoded.Events) + }) +} + func withRedisEventStore(ctx *armadacontext.Context, action func(es *RedisEventStore)) { client := redis.NewClient(&redis.Options{Addr: "localhost:6379", DB: 5}) defer client.FlushDB(ctx) diff --git a/internal/lookout/pruner/pruner_test.go b/internal/lookout/pruner/pruner_test.go index 0ee6e29cf47..e1a97f7f461 100644 --- a/internal/lookout/pruner/pruner_test.go +++ b/internal/lookout/pruner/pruner_test.go @@ -211,7 +211,7 @@ func TestPruneDb(t *testing.T) { 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{}) - store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get()) ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 5*time.Minute) defer cancel() diff --git a/internal/lookout/pruner/reconcile_zombies_test.go b/internal/lookout/pruner/reconcile_zombies_test.go index 71d97d73dec..03cc380c513 100644 --- a/internal/lookout/pruner/reconcile_zombies_test.go +++ b/internal/lookout/pruner/reconcile_zombies_test.go @@ -143,7 +143,7 @@ func TestReconcileZombieJobs(t *testing.T) { t.Run(tc.name, func(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { converter := instructions.NewInstructionConverter(metrics.Get().Metrics, "armadaproject.io/", []string{}, &compress.NoOpCompressor{}) - store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get()) ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 5*time.Minute) defer cancel() @@ -184,7 +184,7 @@ func TestReconcileZombieJobs(t *testing.T) { func TestReconcileZombieJobsLeavesNonZombiesAlone(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { converter := instructions.NewInstructionConverter(metrics.Get().Metrics, "armadaproject.io/", []string{}, &compress.NoOpCompressor{}) - store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get()) ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 5*time.Minute) defer cancel() @@ -234,7 +234,7 @@ func TestReconcileZombieJobsLeavesNonZombiesAlone(t *testing.T) { func TestReconcileZombieJobsCountsNullFinishedZombies(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { converter := instructions.NewInstructionConverter(metrics.Get().Metrics, "armadaproject.io/", []string{}, &compress.NoOpCompressor{}) - store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get()) ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 5*time.Minute) defer cancel() @@ -274,7 +274,7 @@ func TestReconcileZombieJobsCountsNullFinishedZombies(t *testing.T) { func TestPruneDbRunsZombieReconciliation(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { converter := instructions.NewInstructionConverter(metrics.Get().Metrics, "armadaproject.io/", []string{}, &compress.NoOpCompressor{}) - store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get()) ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 5*time.Minute) defer cancel() diff --git a/internal/lookout/repository/getjoberror_test.go b/internal/lookout/repository/getjoberror_test.go index c4eba41b45d..a05e73d6cf1 100644 --- a/internal/lookout/repository/getjoberror_test.go +++ b/internal/lookout/repository/getjoberror_test.go @@ -17,7 +17,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{}) - store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get()) errMsg := "some bad error happened!" _ = NewJobSimulator(converter, store). Submit(queue, jobSet, owner, namespace, baseTime, &JobOptions{ diff --git a/internal/lookout/repository/getjobrundebugmessage_test.go b/internal/lookout/repository/getjobrundebugmessage_test.go index 758ab6569de..88c8e1d4084 100644 --- a/internal/lookout/repository/getjobrundebugmessage_test.go +++ b/internal/lookout/repository/getjobrundebugmessage_test.go @@ -17,7 +17,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{}) - store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get()) debugMessageStrings := []string{ "some bad error happened!", diff --git a/internal/lookout/repository/getjobrunerror_test.go b/internal/lookout/repository/getjobrunerror_test.go index e131bce4b72..ac2ee9a5e4a 100644 --- a/internal/lookout/repository/getjobrunerror_test.go +++ b/internal/lookout/repository/getjobrunerror_test.go @@ -17,7 +17,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{}) - store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get()) errorStrings := []string{ "some bad error happened!", diff --git a/internal/lookout/repository/getjobrunschedulerterminationreason_test.go b/internal/lookout/repository/getjobrunschedulerterminationreason_test.go index 215b7e967a3..ca7e4381125 100644 --- a/internal/lookout/repository/getjobrunschedulerterminationreason_test.go +++ b/internal/lookout/repository/getjobrunschedulerterminationreason_test.go @@ -19,7 +19,7 @@ import ( func TestGetJobRunSchedulerTerminationReason(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { converter := instructions.NewInstructionConverter(metrics.Get().Metrics, userAnnotationPrefix, []string{}, &compress.NoOpCompressor{}) - store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get()) _ = NewJobSimulator(converter, store). Submit(queue, jobSet, owner, namespace, baseTime, basicJobOpts). @@ -49,7 +49,7 @@ func TestGetJobRunSchedulerTerminationReason(t *testing.T) { func TestGetJobRunSchedulerTerminationReasonNoPreemptingJob(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { converter := instructions.NewInstructionConverter(metrics.Get().Metrics, userAnnotationPrefix, []string{}, &compress.NoOpCompressor{}) - store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get()) _ = NewJobSimulator(converter, store). Submit(queue, jobSet, owner, namespace, baseTime, basicJobOpts). @@ -87,7 +87,7 @@ func TestGetJobRunSchedulerTerminationReasonNotFound(t *testing.T) { func TestGetJobRunSchedulerTerminationReasonNullForNonPreemptedRun(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { converter := instructions.NewInstructionConverter(metrics.Get().Metrics, userAnnotationPrefix, []string{}, &compress.NoOpCompressor{}) - store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get()) _ = NewJobSimulator(converter, store). Submit(queue, jobSet, owner, namespace, baseTime, basicJobOpts). diff --git a/internal/lookout/repository/getjobs_test.go b/internal/lookout/repository/getjobs_test.go index 9af4742fd97..159cf072e7e 100644 --- a/internal/lookout/repository/getjobs_test.go +++ b/internal/lookout/repository/getjobs_test.go @@ -60,7 +60,7 @@ func withGetJobsSetup(f func(*instructions.InstructionConverter, *lookoutdb.Look testClock := clock.NewFakeClock(time.Now()) return lookout.WithLookoutDb(func(db *pgxpool.Pool) error { converter := instructions.NewInstructionConverter(metrics.Get().Metrics, userAnnotationPrefix, []string{}, &compress.NoOpCompressor{}) - store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get()) repo := NewSqlGetJobsRepository(db) repo.clock = testClock return f(converter, store, repo, testClock) diff --git a/internal/lookout/repository/getjobspec_test.go b/internal/lookout/repository/getjobspec_test.go index 1ee785edbb4..6477d1f764b 100644 --- a/internal/lookout/repository/getjobspec_test.go +++ b/internal/lookout/repository/getjobspec_test.go @@ -19,7 +19,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{}) - store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get()) job := NewJobSimulator(converter, store). Submit(queue, jobSet, owner, namespace, baseTime, &JobOptions{ @@ -55,7 +55,7 @@ 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{}) - store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get()) _ = NewJobSimulator(converter, store). Submit(queue, jobSet, owner, namespace, baseTime, &JobOptions{ diff --git a/internal/lookout/repository/groupjobs_test.go b/internal/lookout/repository/groupjobs_test.go index 028a6cadf49..710ca8a30fc 100644 --- a/internal/lookout/repository/groupjobs_test.go +++ b/internal/lookout/repository/groupjobs_test.go @@ -23,7 +23,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{}) - store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get()) repo := NewSqlGroupJobsRepository(db) return f(converter, store, repo) }) diff --git a/internal/lookoutingester/benchmark/benchmark.go b/internal/lookoutingester/benchmark/benchmark.go index af96f7dff7d..50f7bc6c581 100644 --- a/internal/lookoutingester/benchmark/benchmark.go +++ b/internal/lookoutingester/benchmark/benchmark.go @@ -49,7 +49,7 @@ func benchmarkSubmissions1000(b *testing.B, config configuration.LookoutIngester JobsToCreate: jobsToCreate, } withDbBenchmark(b, config, func(b *testing.B, db *pgxpool.Pool) { - ldb := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + ldb := lookoutdb.NewLookoutDb(db, nil, metrics.Get()) b.StartTimer() err := ldb.Store(armadacontext.TODO(), instructions) if err != nil { @@ -67,7 +67,7 @@ func benchmarkSubmissions10000(b *testing.B, config configuration.LookoutIngeste JobsToCreate: jobsToCreate, } withDbBenchmark(b, config, func(b *testing.B, db *pgxpool.Pool) { - ldb := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + ldb := lookoutdb.NewLookoutDb(db, nil, metrics.Get()) b.StartTimer() err := ldb.Store(armadacontext.TODO(), instructions) if err != nil { @@ -99,7 +99,7 @@ func benchmarkUpdates1000(b *testing.B, config configuration.LookoutIngesterConf } withDbBenchmark(b, config, func(b *testing.B, db *pgxpool.Pool) { - ldb := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + ldb := lookoutdb.NewLookoutDb(db, nil, metrics.Get()) err := ldb.Store(armadacontext.TODO(), initialInstructions) if err != nil { panic(err) @@ -135,7 +135,7 @@ func benchmarkUpdates10000(b *testing.B, config configuration.LookoutIngesterCon } withDbBenchmark(b, config, func(b *testing.B, db *pgxpool.Pool) { - ldb := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + ldb := lookoutdb.NewLookoutDb(db, nil, metrics.Get()) err := ldb.Store(armadacontext.TODO(), initialInstructions) if err != nil { panic(err) diff --git a/internal/lookoutingester/configuration/types.go b/internal/lookoutingester/configuration/types.go index 7bebcadf7c6..3b1be802f73 100644 --- a/internal/lookoutingester/configuration/types.go +++ b/internal/lookoutingester/configuration/types.go @@ -45,9 +45,6 @@ type LookoutIngesterConfiguration struct { // The annotation before storing in the db UserAnnotationPrefix string } - // Between each attempt to store data in the database, there is an exponential backoff (starting out as 1s). - // MaxBackoff caps this backoff to whatever it is specified (in seconds) - MaxBackoff int // If non-nil, configures pprof profiling Profiling *profilingconfig.ProfilingConfig // List of Regexes which will identify fatal errors when inserting into postgres diff --git a/internal/lookoutingester/dbloadtester/simulator.go b/internal/lookoutingester/dbloadtester/simulator.go index 160df2401cd..f5dfc561b96 100644 --- a/internal/lookoutingester/dbloadtester/simulator.go +++ b/internal/lookoutingester/dbloadtester/simulator.go @@ -71,7 +71,7 @@ func Setup(lookoutIngesterConfig configuration.LookoutIngesterConfiguration, tes fatalRegexes[i] = rgx } - lookoutDb := lookoutdb.NewLookoutDb(db, fatalRegexes, m, lookoutIngesterConfig.MaxBackoff, 0) + lookoutDb := lookoutdb.NewLookoutDb(db, fatalRegexes, m) // To avoid load testing the compression algorithm, the compressor is configured not to compress. compressor, err := compress.NewZlibCompressor(math.MaxInt) diff --git a/internal/lookoutingester/ingester.go b/internal/lookoutingester/ingester.go index b243832be87..1e2dda8e30a 100644 --- a/internal/lookoutingester/ingester.go +++ b/internal/lookoutingester/ingester.go @@ -43,7 +43,7 @@ func Run(config *configuration.LookoutIngesterConfiguration) { fatalRegexes[i] = rgx } - lookoutDb := lookoutdb.NewLookoutDb(db, fatalRegexes, m, config.MaxBackoff, 0) + lookoutDb := lookoutdb.NewLookoutDb(db, fatalRegexes, m) compressor, err := compress.NewZlibCompressor(config.MinJobSpecCompressionSize) if err != nil { diff --git a/internal/lookoutingester/lookoutdb/doc.go b/internal/lookoutingester/lookoutdb/doc.go index aa1ed64fd73..63400931c93 100644 --- a/internal/lookoutingester/lookoutdb/doc.go +++ b/internal/lookoutingester/lookoutdb/doc.go @@ -3,4 +3,11 @@ // job updates, job-run creations/updates, and error records) into SQL writes // against the Lookout database, with batched and scalar fallback paths, update // conflation, and terminal-state filtering. +// +// [LookoutDb.Store] attempts each write once and returns immediately on +// error; retry-then-dead-letter policy is owned by the shared +// [ingest.IngestionPipeline.Run] ack-path. Errors classified as +// non-retryable by [armadaerrors.IsRetryablePostgresError] are wrapped with +// [util.ErrNonRetryable] so the ack-path dead-letters immediately +// instead of exhausting its retry budget first. package lookoutdb diff --git a/internal/lookoutingester/lookoutdb/insertion.go b/internal/lookoutingester/lookoutdb/insertion.go index e7c946c67bd..0e5e53425f2 100644 --- a/internal/lookoutingester/lookoutdb/insertion.go +++ b/internal/lookoutingester/lookoutdb/insertion.go @@ -1,6 +1,7 @@ package lookoutdb import ( + "encoding/json" "fmt" "regexp" "time" @@ -13,6 +14,8 @@ import ( "github.com/armadaproject/armada/internal/common/database/lookout" commonmetrics "github.com/armadaproject/armada/internal/common/ingest/metrics" log "github.com/armadaproject/armada/internal/common/logging" + "github.com/armadaproject/armada/internal/common/pulsarutils" + "github.com/armadaproject/armada/internal/common/util" "github.com/armadaproject/armada/internal/lookoutingester/metrics" "github.com/armadaproject/armada/internal/lookoutingester/model" ) @@ -20,17 +23,13 @@ import ( type LookoutDb struct { db *pgxpool.Pool metrics *metrics.Metrics - maxBackoff int - maxRetries int fatalErrors []*regexp.Regexp } -func NewLookoutDb(db *pgxpool.Pool, fatalErrors []*regexp.Regexp, metrics *metrics.Metrics, maxBackoff int, maxRetries int) *LookoutDb { +func NewLookoutDb(db *pgxpool.Pool, fatalErrors []*regexp.Regexp, metrics *metrics.Metrics) *LookoutDb { return &LookoutDb{ db: db, metrics: metrics, - maxBackoff: maxBackoff, - maxRetries: maxRetries, fatalErrors: fatalErrors, } } @@ -96,6 +95,25 @@ func (l *LookoutDb) Store(ctx *armadacontext.Context, instructions *model.Instru return nil } +// Serialize renders instructions as JSON for the dead-letter topic. +func (l *LookoutDb) Serialize(instructions *model.InstructionSet) ([]byte, error) { + return json.Marshal(struct { + JobsToCreate []*model.CreateJobInstruction + JobsToUpdate []*model.UpdateJobInstruction + JobRunsToCreate []*model.CreateJobRunInstruction + JobRunsToUpdate []*model.UpdateJobRunInstruction + JobErrorsToCreate []*model.CreateJobErrorInstruction + MessageIds []string + }{ + JobsToCreate: instructions.JobsToCreate, + JobsToUpdate: instructions.JobsToUpdate, + JobRunsToCreate: instructions.JobRunsToCreate, + JobRunsToUpdate: instructions.JobRunsToUpdate, + JobErrorsToCreate: instructions.JobErrorsToCreate, + MessageIds: pulsarutils.MessageIdsToStrings(instructions.MessageIds), + }) +} + func (l *LookoutDb) CreateJobs(ctx *armadacontext.Context, instructions []*model.CreateJobInstruction) error { if len(instructions) == 0 { return nil @@ -139,8 +157,11 @@ func (l *LookoutDb) UpdateJobs(ctx *armadacontext.Context, instructions []*model return nil } start := time.Now() - instructions = l.filterEventsForTerminalJobs(ctx, l.db, instructions, l.metrics) - err := l.UpdateJobsBatch(ctx, instructions) + instructions, err := l.filterEventsForTerminalJobs(ctx, l.db, instructions, l.metrics) + if err != nil { + return err + } + err = l.UpdateJobsBatch(ctx, instructions) if err != nil { log.WithError(err).Warn("Updating jobs via batch failed, will attempt to insert serially (this might be slow).") if scalarErr := l.UpdateJobsScalar(ctx, instructions); scalarErr != nil { @@ -245,7 +266,7 @@ func (l *LookoutDb) CreateJobErrors(ctx *armadacontext.Context, instructions []* } func (l *LookoutDb) CreateJobsBatch(ctx *armadacontext.Context, instructions []*model.CreateJobInstruction) error { - return l.withDatabaseRetryInsert(ctx, func() error { + return l.executeDbInsert(ctx, func() error { tmpTable := "job_create_tmp" createTmp := func(tx pgx.Tx) error { @@ -420,7 +441,7 @@ func (l *LookoutDb) CreateJobsScalar(ctx *armadacontext.Context, instructions [] if ctx.Err() != nil { return ctx.Err() } - err := l.withDatabaseRetryInsert(ctx, func() error { + err := l.executeDbInsert(ctx, func() error { _, err := l.db.Exec(ctx, sqlStatement, i.JobId, i.Queue, @@ -456,7 +477,7 @@ func (l *LookoutDb) CreateJobsScalar(ctx *armadacontext.Context, instructions [] } func (l *LookoutDb) UpdateJobsBatch(ctx *armadacontext.Context, instructions []*model.UpdateJobInstruction) error { - return l.withDatabaseRetryInsert(ctx, func() error { + return l.executeDbInsert(ctx, func() error { tmpTable := "job_update_tmp" createTmp := func(tx pgx.Tx) error { @@ -555,7 +576,7 @@ func (l *LookoutDb) UpdateJobsScalar(ctx *armadacontext.Context, instructions [] if ctx.Err() != nil { return ctx.Err() } - err := l.withDatabaseRetryInsert(ctx, func() error { + err := l.executeDbInsert(ctx, func() error { _, err := l.db.Exec(ctx, sqlStatement, i.JobId, i.Priority, @@ -583,7 +604,7 @@ func (l *LookoutDb) UpdateJobsScalar(ctx *armadacontext.Context, instructions [] } func (l *LookoutDb) CreateJobSpecsBatch(ctx *armadacontext.Context, instructions []*model.CreateJobInstruction) error { - return l.withDatabaseRetryInsert(ctx, func() error { + return l.executeDbInsert(ctx, func() error { tmpTable := "job_spec_create_tmp" createTmp := func(tx pgx.Tx) error { @@ -646,7 +667,7 @@ func (l *LookoutDb) CreateJobSpecsScalar(ctx *armadacontext.Context, instruction if ctx.Err() != nil { return ctx.Err() } - err := l.withDatabaseRetryInsert(ctx, func() error { + err := l.executeDbInsert(ctx, func() error { _, err := l.db.Exec(ctx, sqlStatement, i.JobId, i.JobProto, @@ -667,7 +688,7 @@ func (l *LookoutDb) CreateJobSpecsScalar(ctx *armadacontext.Context, instruction } func (l *LookoutDb) CreateJobRunsBatch(ctx *armadacontext.Context, instructions []*model.CreateJobRunInstruction) error { - return l.withDatabaseRetryInsert(ctx, func() error { + return l.executeDbInsert(ctx, func() error { tmpTable := "job_run_create_tmp" createTmp := func(tx pgx.Tx) error { @@ -762,7 +783,7 @@ func (l *LookoutDb) CreateJobRunsScalar(ctx *armadacontext.Context, instructions if ctx.Err() != nil { return ctx.Err() } - err := l.withDatabaseRetryInsert(ctx, func() error { + err := l.executeDbInsert(ctx, func() error { _, err := l.db.Exec(ctx, sqlStatement, i.RunId, i.JobId, @@ -790,7 +811,7 @@ func (l *LookoutDb) CreateJobRunsScalar(ctx *armadacontext.Context, instructions } func (l *LookoutDb) UpdateJobRunsBatch(ctx *armadacontext.Context, instructions []*model.UpdateJobRunInstruction) error { - return l.withDatabaseRetryInsert(ctx, func() error { + return l.executeDbInsert(ctx, func() error { tmpTable := "job_run_update_tmp" createTmp := func(tx pgx.Tx) error { @@ -904,7 +925,7 @@ func (l *LookoutDb) UpdateJobRunsScalar(ctx *armadacontext.Context, instructions if ctx.Err() != nil { return ctx.Err() } - err := l.withDatabaseRetryInsert(ctx, func() error { + err := l.executeDbInsert(ctx, func() error { _, err := l.db.Exec(ctx, sqlStatement, i.RunId, i.Node, @@ -937,7 +958,7 @@ func (l *LookoutDb) UpdateJobRunsScalar(ctx *armadacontext.Context, instructions func (l *LookoutDb) CreateJobErrorsBatch(ctx *armadacontext.Context, instructions []*model.CreateJobErrorInstruction) error { tmpTable := "job_error_create_tmp" - return l.withDatabaseRetryInsert(ctx, func() error { + return l.executeDbInsert(ctx, func() error { createTmp := func(tx pgx.Tx) error { _, err := tx.Exec(ctx, fmt.Sprintf(` CREATE TEMPORARY TABLE %s ( @@ -993,7 +1014,7 @@ func (l *LookoutDb) CreateJobErrorsScalar(ctx *armadacontext.Context, instructio if ctx.Err() != nil { return ctx.Err() } - err := l.withDatabaseRetryInsert(ctx, func() error { + err := l.executeDbInsert(ctx, func() error { _, err := l.db.Exec(ctx, sqlStatement, i.JobId, i.Error) @@ -1163,20 +1184,21 @@ type updateInstructionsForJob struct { // The proper solution here is to make it so once a job is terminal, no more events are generated for it, but until // that day we have to manually filter them out here. // NOTE: this function will retry querying the database for as long as possible in order to determine which jobs are -// in the terminal state. If, however, the database returns a non-retryable error it will give up and simply not -// filter out any events as the job state is undetermined. +// in the terminal state. If the database returns a non-retryable error, the job state is undetermined and it +// returns an error rather than silently skipping the filter, since proceeding unfiltered risks a stale update +// overwriting a terminal one. func (l *LookoutDb) filterEventsForTerminalJobs( ctx *armadacontext.Context, db *pgxpool.Pool, instructions []*model.UpdateJobInstruction, m *metrics.Metrics, -) []*model.UpdateJobInstruction { +) ([]*model.UpdateJobInstruction, error) { jobIds := make([]string, len(instructions)) for i, instruction := range instructions { jobIds[i] = instruction.JobId } queryStart := time.Now() - rowsRaw, err := l.withDatabaseRetryQuery(ctx, func() (interface{}, error) { + rowsRaw, err := l.executeDbQuery(ctx, func() (interface{}, error) { terminalStates := []int{ lookout.JobSucceededOrdinal, lookout.JobFailedOrdinal, @@ -1188,8 +1210,8 @@ func (l *LookoutDb) filterEventsForTerminalJobs( }) if err != nil { m.RecordDBError(commonmetrics.DBOperationRead) - log.WithError(err).Warnf("Cannot retrieve job state from the database- Cancelled jobs may not be filtered out") - return instructions + log.WithError(err).Warnf("Cannot retrieve job state from the database- unable to determine which jobs are terminal") + return nil, err } rows := rowsRaw.(pgx.Rows) @@ -1233,57 +1255,36 @@ func (l *LookoutDb) filterEventsForTerminalJobs( filtered = append(filtered, updateInstructions.instructions...) } } - return filtered + return filtered, nil } else { - return instructions + return instructions, nil } } -func (l *LookoutDb) withDatabaseRetryInsert(ctx *armadacontext.Context, executeDb func() error) error { - _, err := l.withDatabaseRetryQuery(ctx, func() (interface{}, error) { +func (l *LookoutDb) executeDbInsert(ctx *armadacontext.Context, executeDb func() error) error { + _, err := l.executeDbQuery(ctx, func() (interface{}, error) { return nil, executeDb() }) return err } -// Executes a database function, retrying until it either succeeds, exceeds the -// retry limit, encounters a non-retryable error, or the context is cancelled. -func (l *LookoutDb) withDatabaseRetryQuery(ctx *armadacontext.Context, executeDb func() (interface{}, error)) (interface{}, error) { - backOff := 1 - retries := 0 - for { - res, err := executeDb() - - if err == nil { - return res, nil - } - - if ctx.Err() != nil { - return nil, ctx.Err() - } +// executeDbQuery runs a database function once. Retry-then-dead-letter policy is owned by the +// IngestionPipeline ack-path (see internal/common/ingest). Errors classified as +// non-retryable are wrapped with util.ErrNonRetryable so that the ack-path skips +// straight to dead-lettering instead of exhausting its retry budget first. +func (l *LookoutDb) executeDbQuery(ctx *armadacontext.Context, executeDb func() (interface{}, error)) (interface{}, error) { + res, err := executeDb() + if err == nil { + return res, nil + } - if armadaerrors.IsRetryablePostgresError(err, l.fatalErrors) { - retries++ - if l.maxRetries > 0 && retries >= l.maxRetries { - return nil, fmt.Errorf("exceeded max retries (%d): %w", l.maxRetries, err) - } - backOff = min(2*backOff, l.maxBackoff) - log.WithError(err).Warnf("Retryable error encountered executing sql (attempt %d/%d), will wait for %d seconds before retrying.", retries, l.maxRetries, backOff) - select { - case <-time.After(time.Duration(backOff) * time.Second): - case <-ctx.Done(): - return nil, ctx.Err() - } - } else { - // Non retryable error - return nil, err - } + if ctx.Err() != nil { + return nil, ctx.Err() } -} -func min(a int, b int) int { - if a < b { - return a + if !armadaerrors.IsRetryablePostgresError(err, l.fatalErrors) { + log.WithError(err).Warnf("Non-retryable error encountered executing sql; returning immediately") + return nil, fmt.Errorf("%w: %w", util.ErrNonRetryable, err) } - return b + return nil, err } diff --git a/internal/lookoutingester/lookoutdb/insertion_hotcold_test.go b/internal/lookoutingester/lookoutdb/insertion_hotcold_test.go index 07ca3c4dac9..008f448f19b 100644 --- a/internal/lookoutingester/lookoutdb/insertion_hotcold_test.go +++ b/internal/lookoutingester/lookoutdb/insertion_hotcold_test.go @@ -47,7 +47,7 @@ func countInPartition(t *testing.T, db *pgxpool.Pool, partition, jobId string) i func TestHotCold_StoreRoutesTerminalJobToTerminatedPartition(t *testing.T) { err := withLookoutHCDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) createInstructions := &model.InstructionSet{ JobsToCreate: []*model.CreateJobInstruction{makeCreateJobInstruction(JobId)}, @@ -103,7 +103,7 @@ func TestHotCold_StoreRoutesTerminalJobToTerminatedPartition(t *testing.T) { func TestHotCold_StoreKeepsRunningJobInActivePartition(t *testing.T) { err := withLookoutHCDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) instructions := &model.InstructionSet{ JobsToCreate: []*model.CreateJobInstruction{makeCreateJobInstruction(JobId)}, @@ -129,7 +129,7 @@ func TestHotCold_StoreKeepsRunningJobInActivePartition(t *testing.T) { func TestHotCold_MultipleJobsDistributedAcrossPartitions(t *testing.T) { err := withLookoutHCDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) activeIds := []string{"job-active-1", "job-active-2"} terminalTargets := map[string]int32{ @@ -199,7 +199,7 @@ func TestHotCold_MultipleJobsDistributedAcrossPartitions(t *testing.T) { func TestHotCold_FailedJobStoresErrorAndRoutesToTerminatedPartition(t *testing.T) { err := withLookoutHCDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) instructions := &model.InstructionSet{ JobsToCreate: []*model.CreateJobInstruction{makeCreateJobInstruction(JobId)}, @@ -229,7 +229,7 @@ func TestHotCold_FailedJobStoresErrorAndRoutesToTerminatedPartition(t *testing.T func TestHotCold_ParentJobTableReturnsAllJobs(t *testing.T) { err := withLookoutHCDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) createInstructions := []*model.CreateJobInstruction{ makeCreateJobInstruction("job-a"), @@ -287,7 +287,7 @@ func TestHotCold_ParentJobTableReturnsAllJobs(t *testing.T) { func TestHotCold_TerminalStateQueryPrunesActivePartition(t *testing.T) { err := withLookoutHCDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) require.NoError(t, ldb.Store(armadacontext.Background(), &model.InstructionSet{ JobsToCreate: []*model.CreateJobInstruction{ @@ -328,7 +328,7 @@ func TestHotCold_TerminalStateQueryPrunesActivePartition(t *testing.T) { func TestHotCold_ConflatedTerminalUpdatesProduceSingleTerminatedRow(t *testing.T) { err := withLookoutHCDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) require.NoError(t, ldb.Store(armadacontext.Background(), &model.InstructionSet{ JobsToCreate: []*model.CreateJobInstruction{makeCreateJobInstruction(JobId)}, @@ -389,7 +389,7 @@ func TestHotCold_ConflatedTerminalUpdatesProduceSingleTerminatedRow(t *testing.T // (active) partition and would insert a duplicate routed there. func TestHotCold_CreateSuppressedWhenJobExistsInOtherPartition(t *testing.T) { err := withLookoutHCDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) // A row for the job already exists in job_terminated (e.g. from an // out-of-order or replayed terminal event). @@ -423,7 +423,7 @@ func TestHotCold_CreateSuppressedWhenJobExistsInOtherPartition(t *testing.T) { // exists there (e.g. a Queued create arriving after a Leased row is present). func TestHotCold_CreateSuppressedWhenJobExistsInSamePartition(t *testing.T) { err := withLookoutHCDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) // An active-state row (Leased) already exists in job_active. _, err := db.Exec(armadacontext.Background(), diff --git a/internal/lookoutingester/lookoutdb/insertion_test.go b/internal/lookoutingester/lookoutdb/insertion_test.go index 6f087617de6..875bdb8a402 100644 --- a/internal/lookoutingester/lookoutdb/insertion_test.go +++ b/internal/lookoutingester/lookoutdb/insertion_test.go @@ -239,7 +239,7 @@ var expectedJobRunAfterUpdate = JobRunRow{ func TestCreateJobsBatch(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) // Insert err := ldb.CreateJobsBatch(armadacontext.Background(), defaultInstructionSet().JobsToCreate) assert.Nil(t, err) @@ -268,7 +268,7 @@ func TestCreateJobsBatch(t *testing.T) { func TestUpdateJobsBatch(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) // Insert err := ldb.CreateJobsBatch(armadacontext.Background(), defaultInstructionSet().JobsToCreate) assert.Nil(t, err) @@ -303,7 +303,7 @@ func TestUpdateJobsBatch(t *testing.T) { func TestUpdateJobsScalar(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) // Insert err := ldb.CreateJobsBatch(armadacontext.Background(), defaultInstructionSet().JobsToCreate) assert.Nil(t, err) @@ -442,7 +442,7 @@ func TestUpdateJobsWithTerminal(t *testing.T) { LatestRunId: pointer.String(RunId), }} - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) // Insert err := ldb.CreateJobs(armadacontext.Background(), initial) @@ -475,7 +475,7 @@ func TestUpdateJobsWithTerminal(t *testing.T) { func TestCreateJobsScalar(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) // Simple create err := ldb.CreateJobsScalar(armadacontext.Background(), defaultInstructionSet().JobsToCreate) assert.NoError(t, err) @@ -505,7 +505,7 @@ func TestCreateJobsScalar(t *testing.T) { func TestCreateJobRunsBatch(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) // Need to make sure we have a job, so we can satisfy PK err := ldb.CreateJobsBatch(armadacontext.Background(), defaultInstructionSet().JobsToCreate) assert.Nil(t, err) @@ -538,7 +538,7 @@ func TestCreateJobRunsBatch(t *testing.T) { func TestCreateJobRunsScalar(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) // Need to make sure we have a job, so we can satisfy PK err := ldb.CreateJobsBatch(armadacontext.Background(), defaultInstructionSet().JobsToCreate) assert.Nil(t, err) @@ -572,7 +572,7 @@ func TestCreateJobRunsScalar(t *testing.T) { func TestUpdateJobRunsBatch(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) // Need to make sure we have a job and run err := ldb.CreateJobsBatch(armadacontext.Background(), defaultInstructionSet().JobsToCreate) assert.Nil(t, err) @@ -611,7 +611,7 @@ func TestUpdateJobRunsBatch(t *testing.T) { func TestUpdateJobRunsScalar(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) // Need to make sure we have a job and run err := ldb.CreateJobsBatch(armadacontext.Background(), defaultInstructionSet().JobsToCreate) assert.Nil(t, err) @@ -650,7 +650,7 @@ func TestUpdateJobRunsScalar(t *testing.T) { func TestCreateJobErrorsBatch(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) // Insert err := ldb.CreateJobErrorsBatch(armadacontext.Background(), defaultInstructionSet().JobErrorsToCreate) @@ -688,7 +688,7 @@ func TestCreateJobErrorsBatch(t *testing.T) { func TestCreateJobErrorsScalar(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) // Insert err := ldb.CreateJobErrorsScalar(armadacontext.Background(), defaultInstructionSet().JobErrorsToCreate) @@ -708,7 +708,7 @@ func TestCreateJobErrorsScalar(t *testing.T) { func TestStoreWithEmptyInstructionSet(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) err := ldb.Store(armadacontext.Background(), &model.InstructionSet{ MessageIds: []pulsar.MessageID{pulsarutils.NewMessageId(1)}, }) @@ -728,7 +728,7 @@ func TestStoreWithEmptyInstructionSet(t *testing.T) { // messages and they are re-processed on the next run. func TestStoreReturnsErrorOnContextCancellation(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) ctx, cancel := armadacontext.WithCancel(armadacontext.Background()) cancel() err := ldb.Store(ctx, defaultInstructionSet()) @@ -745,7 +745,7 @@ func TestStoreReturnsErrorOnContextCancellation(t *testing.T) { // cancellation and allowing the pipeline to ack messages whose DB writes never succeeded. func TestScalarMethodsReturnErrorOnContextCancellation(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) ctx, cancel := armadacontext.WithCancel(armadacontext.Background()) cancel() @@ -762,7 +762,7 @@ func TestScalarMethodsReturnErrorOnContextCancellation(t *testing.T) { func TestStore(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) // Do the update err := ldb.Store(armadacontext.Background(), defaultInstructionSet()) assert.NoError(t, err) @@ -780,7 +780,7 @@ func TestStore(t *testing.T) { // ensure executor update doesn't write over scheduler termination reason update func TestSchedulerTerminationReasonNotOverwritten(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) err := ldb.CreateJobsBatch(armadacontext.Background(), defaultInstructionSet().JobsToCreate) assert.Nil(t, err) @@ -970,7 +970,7 @@ func TestStoreNullValue(t *testing.T) { instructions.JobRunsToUpdate[0].Error = errorMsg instructions.JobRunsToUpdate[0].Debug = debugMsg - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) // Do the update err := ldb.Store(armadacontext.Background(), instructions) assert.NoError(t, err) @@ -988,7 +988,7 @@ func TestStoreNullValue(t *testing.T) { func TestStoreEventsForAlreadyTerminalJobs(t *testing.T) { err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { - ldb := NewLookoutDb(db, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(db, fatalErrors, m) baseInstructions := &model.InstructionSet{ JobsToCreate: []*model.CreateJobInstruction{ @@ -1039,7 +1039,7 @@ func TestStoreEventsForAlreadyTerminalJobs(t *testing.T) { } func TestRecordTerminalStateUpdates(t *testing.T) { - ldb := NewLookoutDb(nil, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(nil, fatalErrors, m) instructions := []*model.UpdateJobInstruction{ {JobId: "job1", State: pointer.Int32(lookout.JobSucceededOrdinal)}, @@ -1056,12 +1056,37 @@ func TestRecordTerminalStateUpdates(t *testing.T) { } func TestRecordTerminalStateUpdates_Empty(t *testing.T) { - ldb := NewLookoutDb(nil, fatalErrors, m, 10, 10) + ldb := NewLookoutDb(nil, fatalErrors, m) // Neither nil nor empty should panic ldb.recordStateUpdates(nil) ldb.recordStateUpdates([]*model.UpdateJobInstruction{}) } +func TestSerialize(t *testing.T) { + ldb := NewLookoutDb(nil, fatalErrors, m) + instructionSet := &model.InstructionSet{ + JobsToCreate: []*model.CreateJobInstruction{makeCreateJobInstruction(JobId)}, + MessageIds: []pulsar.MessageID{pulsarutils.NewMessageId(1), pulsarutils.NewMessageId(2)}, + } + + bytes, err := ldb.Serialize(instructionSet) + assert.NoError(t, err) + + var decoded struct { + JobsToCreate []*model.CreateJobInstruction + MessageIds []string + } + err = json.Unmarshal(bytes, &decoded) + assert.NoError(t, err) + + assert.Equal(t, []string{ + pulsarutils.NewMessageId(1).String(), + pulsarutils.NewMessageId(2).String(), + }, decoded.MessageIds) + assert.Len(t, decoded.JobsToCreate, 1) + assert.Equal(t, JobId, decoded.JobsToCreate[0].JobId) +} + func makeCreateJobInstruction(jobId string) *model.CreateJobInstruction { return &model.CreateJobInstruction{ JobId: jobId, diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go index 91b6f8ab1bb..97f5a1702c3 100644 --- a/internal/scheduler/scheduler_test.go +++ b/internal/scheduler/scheduler_test.go @@ -3624,8 +3624,6 @@ func TestCycleConsistency(t *testing.T) { schedulerDb := scheduleringester.NewSchedulerDb( db, nil, - time.Second, - time.Second, 10*time.Second, ) diff --git a/internal/scheduleringester/dbops.go b/internal/scheduleringester/dbops.go index 6a6468c8b2a..0ffb5d3eb9d 100644 --- a/internal/scheduleringester/dbops.go +++ b/internal/scheduleringester/dbops.go @@ -161,6 +161,9 @@ type DbOperation interface { CanBeAppliedBefore(DbOperation) bool // GetOperation returns the Operation/grouping that this DbOperation belongs to. GetOperation() Operation + // SerializeForDLQ returns a JSON-marshalable representation of the op for the dead-letter + // topic. This payload is for inspection/manual-replay purposes, not automatic round-tripping. + SerializeForDLQ() any } // AppendDbOperation appends a sql operation, @@ -728,114 +731,239 @@ func (a InsertJobs) GetOperation() Operation { return JobSetOperation } +func (a InsertJobs) SerializeForDLQ() any { + return a +} + func (a InsertRuns) GetOperation() Operation { return JobSetOperation } +func (a InsertRuns) SerializeForDLQ() any { + return a +} + func (a UpdateJobSetPriorities) GetOperation() Operation { return JobSetOperation } +func (a UpdateJobSetPriorities) SerializeForDLQ() any { + return jobSetKeyMapToPairs(a) +} + func (a MarkJobSetsCancelRequested) GetOperation() Operation { return JobSetOperation } +func (a MarkJobSetsCancelRequested) SerializeForDLQ() any { + return struct { + CancelUser string + CancelReason string + JobSets []jobSetKeyPair[*JobSetCancelAction] + }{a.cancelUser, a.cancelReason, jobSetKeyMapToPairs(a.jobSets)} +} + func (a MarkJobsCancelRequested) GetOperation() Operation { return JobSetOperation } +func (a MarkJobsCancelRequested) SerializeForDLQ() any { + return struct { + CancelUser string + CancelReason string + JobIds []jobSetKeyPair[[]string] + }{a.cancelUser, a.cancelReason, jobSetKeyMapToPairs(a.jobIds)} +} + func (a MarkRunsForJobPreemptRequested) GetOperation() Operation { return JobSetOperation } +func (a MarkRunsForJobPreemptRequested) SerializeForDLQ() any { + return jobSetKeyMapToPairs(a) +} + func (a UpdateJobSchedulingInfo) GetOperation() Operation { return JobSetOperation } +func (a UpdateJobSchedulingInfo) SerializeForDLQ() any { + return a +} + func (a UpdateJobQueuedState) GetOperation() Operation { return JobSetOperation } +func (a UpdateJobQueuedState) SerializeForDLQ() any { + return a +} + func (a MarkJobsCancelled) GetOperation() Operation { return JobSetOperation } +func (a MarkJobsCancelled) SerializeForDLQ() any { + return a +} + func (a MarkJobsSucceeded) GetOperation() Operation { return JobSetOperation } +func (a MarkJobsSucceeded) SerializeForDLQ() any { + return a +} + func (a MarkJobsFailed) GetOperation() Operation { return JobSetOperation } +func (a MarkJobsFailed) SerializeForDLQ() any { + return a +} + func (a *UpdateJobPriorities) GetOperation() Operation { return JobSetOperation } +func (a *UpdateJobPriorities) SerializeForDLQ() any { + return struct { + Key JobReprioritiseKey + JobIds []string + }{a.key, a.jobIds} +} + func (a MarkRunsSucceeded) GetOperation() Operation { return JobSetOperation } +func (a MarkRunsSucceeded) SerializeForDLQ() any { + return a +} + func (a MarkRunsFailed) GetOperation() Operation { return JobSetOperation } +func (a MarkRunsFailed) SerializeForDLQ() any { + return a +} + func (a MarkRunsRunning) GetOperation() Operation { return JobSetOperation } +func (a MarkRunsRunning) SerializeForDLQ() any { + return a +} + func (a MarkRunsPending) GetOperation() Operation { return JobSetOperation } +func (a MarkRunsPending) SerializeForDLQ() any { + return a +} + func (a MarkRunsPreempted) GetOperation() Operation { return JobSetOperation } +func (a MarkRunsPreempted) SerializeForDLQ() any { + return a +} + func (a InsertJobRunErrors) GetOperation() Operation { return JobSetOperation } +func (a InsertJobRunErrors) SerializeForDLQ() any { + return a +} + func (a MarkJobsValidated) GetOperation() Operation { return JobSetOperation } +func (a MarkJobsValidated) SerializeForDLQ() any { + return a +} + func (a *InsertPartitionMarker) GetOperation() Operation { return JobSetOperation } +func (a *InsertPartitionMarker) SerializeForDLQ() any { + return struct { + Markers []*schedulerdb.Marker + }{a.markers} +} + func (a UpsertExecutorSettings) GetOperation() Operation { return ControlPlaneOperation } +func (a UpsertExecutorSettings) SerializeForDLQ() any { + return a +} + func (a DeleteExecutorSettings) GetOperation() Operation { return ControlPlaneOperation } +func (a DeleteExecutorSettings) SerializeForDLQ() any { + return a +} + func (pe PreemptExecutor) GetOperation() Operation { return ControlPlaneOperation } +func (pe PreemptExecutor) SerializeForDLQ() any { + return pe +} + func (ce CancelExecutor) GetOperation() Operation { return ControlPlaneOperation } +func (ce CancelExecutor) SerializeForDLQ() any { + return ce +} + func (ne PreemptNode) GetOperation() Operation { return ControlPlaneOperation } +func (ne PreemptNode) SerializeForDLQ() any { + return nodeOnExecutorMapToPairs(ne) +} + func (cn CancelNode) GetOperation() Operation { return ControlPlaneOperation } +func (cn CancelNode) SerializeForDLQ() any { + return nodeOnExecutorMapToPairs(cn) +} + func (pq PreemptQueue) GetOperation() Operation { return ControlPlaneOperation } +func (pq PreemptQueue) SerializeForDLQ() any { + return pq +} + func (cq CancelQueue) GetOperation() Operation { return ControlPlaneOperation } +func (cq CancelQueue) SerializeForDLQ() any { + return cq +} + type executorOperation interface { affectsExecutor(string) bool } diff --git a/internal/scheduleringester/ingester.go b/internal/scheduleringester/ingester.go index fe0578f6028..0e68ba5e172 100644 --- a/internal/scheduleringester/ingester.go +++ b/internal/scheduleringester/ingester.go @@ -31,7 +31,7 @@ func Run(config Configuration) error { if err != nil { panic(errors.WithMessage(err, "Error opening connection to postgres")) } - schedulerDb := NewSchedulerDb(db, svcMetrics, 100*time.Millisecond, 60*time.Second, 5*time.Second) + schedulerDb := NewSchedulerDb(db, svcMetrics, 5*time.Second) jobSetEventsConverter, err := NewJobSetEventsInstructionConverter(svcMetrics) if err != nil { diff --git a/internal/scheduleringester/schedulerdb.go b/internal/scheduleringester/schedulerdb.go index f2a3855ef2d..88c5be07217 100644 --- a/internal/scheduleringester/schedulerdb.go +++ b/internal/scheduleringester/schedulerdb.go @@ -1,6 +1,7 @@ package scheduleringester import ( + "encoding/json" "fmt" "time" @@ -12,8 +13,8 @@ import ( "github.com/armadaproject/armada/internal/common/armadacontext" "github.com/armadaproject/armada/internal/common/database" - "github.com/armadaproject/armada/internal/common/ingest" "github.com/armadaproject/armada/internal/common/ingest/metrics" + "github.com/armadaproject/armada/internal/common/pulsarutils" "github.com/armadaproject/armada/internal/common/slices" schedulerdb "github.com/armadaproject/armada/internal/scheduler/database" "github.com/armadaproject/armada/internal/scheduler/schedulerobjects" @@ -28,56 +29,101 @@ const ( // SchedulerDb writes DbOperations into postgres. type SchedulerDb struct { // Connection to the postgres database. - db *pgxpool.Pool - metrics *metrics.Metrics - initialBackOff time.Duration - maxBackOff time.Duration - lockTimeout time.Duration + db *pgxpool.Pool + metrics *metrics.Metrics + lockTimeout time.Duration } func NewSchedulerDb( db *pgxpool.Pool, metrics *metrics.Metrics, - initialBackOff time.Duration, - maxBackOff time.Duration, lockTimeout time.Duration, ) *SchedulerDb { return &SchedulerDb{ - db: db, - metrics: metrics, - initialBackOff: initialBackOff, - maxBackOff: maxBackOff, - lockTimeout: lockTimeout, + db: db, + metrics: metrics, + lockTimeout: lockTimeout, } } // Store persists all operations in the database. -// This function retires until it either succeeds or encounters a terminal error. // This function locks the postgres table to avoid write conflicts; see acquireLock() for details. func (s *SchedulerDb) Store(ctx *armadacontext.Context, instructions *DbOperationsWithMessageIds) error { - return ingest.WithRetry(func() (bool, error) { - err := pgx.BeginTxFunc(ctx, s.db, pgx.TxOptions{ - IsoLevel: pgx.ReadCommitted, - AccessMode: pgx.ReadWrite, - DeferrableMode: pgx.Deferrable, - }, func(tx pgx.Tx) error { - lockCtx, cancel := armadacontext.WithTimeout(ctx, s.lockTimeout) - defer cancel() - if scope, err := getLockKey(instructions.Ops); err == nil { - // The lock is released automatically on transaction rollback/commit. - if err := s.acquireLock(lockCtx, tx, scope); err != nil { - return err - } + return pgx.BeginTxFunc(ctx, s.db, pgx.TxOptions{ + IsoLevel: pgx.ReadCommitted, + AccessMode: pgx.ReadWrite, + DeferrableMode: pgx.Deferrable, + }, func(tx pgx.Tx) error { + lockCtx, cancel := armadacontext.WithTimeout(ctx, s.lockTimeout) + defer cancel() + if scope, err := getLockKey(instructions.Ops); err == nil { + // The lock is released automatically on transaction rollback/commit. + if err := s.acquireLock(lockCtx, tx, scope); err != nil { + return err } - for _, dbOp := range instructions.Ops { - if err := s.WriteDbOp(ctx, tx, dbOp); err != nil { - return err - } + } + for _, dbOp := range instructions.Ops { + if err := s.WriteDbOp(ctx, tx, dbOp); err != nil { + return err } - return nil - }) - return true, err - }, s.initialBackOff, s.maxBackOff) + } + return nil + }) +} + +// Serialize renders instructions as JSON for the dead-letter topic. Each op is rendered as its +// concrete type name plus its data, since DbOperation is an interface and would otherwise +// serialize to an uninformative "{}". This payload is for inspection/manual-replay purposes, +// not automatic round-tripping. +func (s *SchedulerDb) Serialize(instructions *DbOperationsWithMessageIds) ([]byte, error) { + type serializedOp struct { + Type string + Data any + } + ops := make([]serializedOp, len(instructions.Ops)) + for i, op := range instructions.Ops { + ops[i] = serializedOp{ + Type: fmt.Sprintf("%T", op), + Data: op.SerializeForDLQ(), + } + } + return json.Marshal(struct { + Ops []serializedOp + MessageIds []string + }{ + Ops: ops, + MessageIds: pulsarutils.MessageIdsToStrings(instructions.MessageIds), + }) +} + +type jobSetKeyPair[V any] struct { + Key JobSetKey + Value V +} + +// jobSetKeyMapToPairs converts a map keyed by JobSetKey (a struct, and so not directly +// JSON-marshalable as an object key) into a slice of key/value pairs. +func jobSetKeyMapToPairs[V any](m map[JobSetKey]V) []jobSetKeyPair[V] { + pairs := make([]jobSetKeyPair[V], 0, len(m)) + for k, v := range m { + pairs = append(pairs, jobSetKeyPair[V]{Key: k, Value: v}) + } + return pairs +} + +type nodeOnExecutorPair[V any] struct { + Key NodeOnExecutor + Value V +} + +// nodeOnExecutorMapToPairs converts a map keyed by NodeOnExecutor (a struct, and so not +// directly JSON-marshalable as an object key) into a slice of key/value pairs. +func nodeOnExecutorMapToPairs[V any](m map[NodeOnExecutor]V) []nodeOnExecutorPair[V] { + pairs := make([]nodeOnExecutorPair[V], 0, len(m)) + for k, v := range m { + pairs = append(pairs, nodeOnExecutorPair[V]{Key: k, Value: v}) + } + return pairs } // acquireLock acquires a postgres advisory lock, thus preventing concurrent writes. diff --git a/internal/scheduleringester/schedulerdb_test.go b/internal/scheduleringester/schedulerdb_test.go index e49d7965c20..8ee0f3883c5 100644 --- a/internal/scheduleringester/schedulerdb_test.go +++ b/internal/scheduleringester/schedulerdb_test.go @@ -1178,7 +1178,7 @@ func TestStore(t *testing.T) { ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 5*time.Second) defer cancel() err := schedulerdb.WithTestDb(func(q *schedulerdb.Queries, db *pgxpool.Pool) error { - schedulerDb := NewSchedulerDb(db, metrics.NewMetrics("test"), time.Second, time.Second, 10*time.Second) + schedulerDb := NewSchedulerDb(db, metrics.NewMetrics("test"), 10*time.Second) err := schedulerDb.Store(ctx, &DbOperationsWithMessageIds{Ops: ops}) require.NoError(t, err) diff --git a/internal/scheduleringester/serialize_test.go b/internal/scheduleringester/serialize_test.go new file mode 100644 index 00000000000..aef6b9da2af --- /dev/null +++ b/internal/scheduleringester/serialize_test.go @@ -0,0 +1,147 @@ +package scheduleringester + +import ( + "encoding/json" + "testing" + "time" + + "github.com/apache/pulsar-client-go/pulsar" + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + + "github.com/armadaproject/armada/internal/common/pulsarutils" + schedulerdb "github.com/armadaproject/armada/internal/scheduler/database" + "github.com/armadaproject/armada/pkg/controlplaneevents" +) + +// allDbOperations returns one instance of every concrete DbOperation type, keyed by a +// human-readable name. TestAllDbOperationsCovered guards this set against silently going stale +// as DbOperation types are added or removed in dbops.go. +func allDbOperations() map[string]DbOperation { + return map[string]DbOperation{ + "InsertJobs": InsertJobs{"job1": &JobInsertion{}}, + "InsertRuns": InsertRuns{"run1": &JobRunDetails{}}, + "UpdateJobSetPriorities": UpdateJobSetPriorities{{queue: "queue1", jobSet: "set1"}: 1}, + "MarkJobSetsCancelRequested": MarkJobSetsCancelRequested{ + cancelUser: "user1", + cancelReason: "reason1", + jobSets: map[JobSetKey]*JobSetCancelAction{{queue: "queue1", jobSet: "set1"}: {cancelQueued: true}}, + }, + "MarkJobsCancelRequested": MarkJobsCancelRequested{ + cancelUser: "user1", + cancelReason: "reason1", + jobIds: map[JobSetKey][]string{{queue: "queue1", jobSet: "set1"}: {"job1"}}, + }, + "MarkJobsCancelled": MarkJobsCancelled{"job1": time.Now()}, + "MarkJobsSucceeded": MarkJobsSucceeded{"job1": true}, + "MarkJobsFailed": MarkJobsFailed{"job1": true}, + "UpdateJobSchedulingInfo": UpdateJobSchedulingInfo{"job1": &JobSchedulingInfoUpdate{}}, + "UpdateJobQueuedState": UpdateJobQueuedState{"job1": &JobQueuedStateUpdate{}}, + "MarkRunsSucceeded": MarkRunsSucceeded{"run1": time.Now()}, + "MarkRunsFailed": MarkRunsFailed{"run1": &JobRunFailed{}}, + "MarkRunsForJobPreemptRequested": MarkRunsForJobPreemptRequested{{queue: "queue1", jobSet: "set1"}: {"job1": "run1"}}, + "MarkRunsRunning": MarkRunsRunning{"run1": time.Now()}, + "MarkRunsPending": MarkRunsPending{"run1": time.Now()}, + "MarkRunsPreempted": MarkRunsPreempted{"run1": time.Now()}, + "InsertJobRunErrors": InsertJobRunErrors{"run1": &schedulerdb.JobRunError{}}, + "UpdateJobPriorities": &UpdateJobPriorities{ + key: JobReprioritiseKey{JobSetKey: JobSetKey{queue: "queue1", jobSet: "set1"}, Priority: 1}, + jobIds: []string{"job1"}, + }, + "MarkJobsValidated": MarkJobsValidated{"job1": {"pool1"}}, + "InsertPartitionMarker": &InsertPartitionMarker{markers: []*schedulerdb.Marker{{}}}, + "UpsertExecutorSettings": UpsertExecutorSettings{ + "executor1": {ExecutorID: "executor1", Cordoned: true}, + }, + "DeleteExecutorSettings": DeleteExecutorSettings{"executor1": {ExecutorID: "executor1"}}, + "PreemptExecutor": PreemptExecutor{"executor1": {Name: "executor1"}}, + "CancelExecutor": CancelExecutor{"executor1": {Name: "executor1"}}, + "PreemptNode": PreemptNode{{Node: "node1", Executor: "executor1"}: {Name: "node1"}}, + "CancelNode": CancelNode{{Node: "node1", Executor: "executor1"}: {Name: "node1"}}, + "PreemptQueue": PreemptQueue{"queue1": {Name: "queue1"}}, + "CancelQueue": CancelQueue{"queue1": {Name: "queue1", JobStates: []controlplaneevents.ActiveJobState{controlplaneevents.ActiveJobState_QUEUED}}}, + } +} + +// TestAllDbOperationsCovered guards against allDbOperations silently going stale: it fails if +// the number of concrete types it constructs no longer matches the number of DbOperation +// implementations, which would happen if a DbOperation type were added to dbops.go without +// also being added here. +func TestAllDbOperationsCovered(t *testing.T) { + assert.Len(t, allDbOperations(), 28) +} + +// TestSerializeForDLQ_NeverEmpty asserts that every current concrete DbOperation type's +// SerializeForDLQ implementation produces non-empty JSON output. +func TestSerializeForDLQ_NeverEmpty(t *testing.T) { + for name, op := range allDbOperations() { + t.Run(name, func(t *testing.T) { + data := op.SerializeForDLQ() + bytes, err := json.Marshal(data) + assert.NoError(t, err) + assert.NotEqual(t, "{}", string(bytes), "SerializeForDLQ produced empty output for %s", name) + }) + } +} + +// TestSerialize_UnexportedFieldsSurface confirms that DbOperation types whose data lives in +// unexported struct fields (and so does not appear in the source struct's JSON output) is +// nonetheless surfaced by SerializeForDLQ. +func TestSerialize_UnexportedFieldsSurface(t *testing.T) { + db := &SchedulerDb{} + + instructions := &DbOperationsWithMessageIds{ + Ops: []DbOperation{ + MarkJobSetsCancelRequested{ + cancelUser: "user1", + cancelReason: "reason1", + jobSets: map[JobSetKey]*JobSetCancelAction{{queue: "queue1", jobSet: "set1"}: {cancelQueued: true}}, + }, + MarkJobsCancelRequested{ + cancelUser: "user2", + cancelReason: "reason2", + jobIds: map[JobSetKey][]string{{queue: "queue1", jobSet: "set1"}: {"job1"}}, + }, + &UpdateJobPriorities{ + key: JobReprioritiseKey{JobSetKey: JobSetKey{queue: "queue1", jobSet: "set1"}, Priority: 5}, + jobIds: []string{"job1", "job2"}, + }, + &InsertPartitionMarker{markers: []*schedulerdb.Marker{{GroupID: uuid.New(), PartitionID: 3}}}, + }, + } + + bytes, err := db.Serialize(instructions) + assert.NoError(t, err) + + s := string(bytes) + assert.Contains(t, s, "user1") + assert.Contains(t, s, "reason1") + assert.Contains(t, s, "user2") + assert.Contains(t, s, "reason2") + assert.Contains(t, s, "job1") + assert.Contains(t, s, "job2") + assert.Contains(t, s, `"Priority":5`) + assert.Contains(t, s, `"PartitionID":3`) +} + +// TestSerialize_MessageIdsAsStrings confirms MessageIds round-trips as an array of strings, +// not "{}", which would happen if pulsar.MessageID (an interface) were marshalled directly. +func TestSerialize_MessageIdsAsStrings(t *testing.T) { + db := &SchedulerDb{} + instructions := &DbOperationsWithMessageIds{ + Ops: []DbOperation{MarkJobsSucceeded{"job1": true}}, + MessageIds: []pulsar.MessageID{pulsarutils.NewMessageId(1), pulsarutils.NewMessageId(2)}, + } + + bytes, err := db.Serialize(instructions) + assert.NoError(t, err) + + var decoded struct { + MessageIds []string + } + assert.NoError(t, json.Unmarshal(bytes, &decoded)) + assert.Equal(t, []string{ + pulsarutils.NewMessageId(1).String(), + pulsarutils.NewMessageId(2).String(), + }, decoded.MessageIds) +}