From 00e06cb40394655db771279e9195d6081207df65 Mon Sep 17 00:00:00 2001 From: Scot Wells Date: Wed, 14 Oct 2026 11:36:46 -0400 Subject: [PATCH] fix: keep indexed documents at their newest state Search results often describe a resource as it was created rather than as it is now, and resources deleted days ago keep appearing. The indexer's replicas share one JetStream queue group, each batches for a flush interval, and a flush replaces the whole document, so when one object's events land on different replicas the older event's batch can be enqueued last and overwrite the newer state. Dry-run creates are also indexed, leaving documents for objects that never existed. Key changes: - Record the newest state of every document in a JetStream KV bucket the indexer creates and every replica shares. A flush claims each write with a compare-and-set, applies the recorded state instead of its own when that is newer, and after enqueueing records the Meilisearch task that carries it. Meilisearch applies tasks in uid order, so a flush whose task would overwrite a newer state enqueued earlier re-enqueues that state behind its own task. Reading the current document from Meilisearch cannot arbitrate: its writes are queued tasks that take longer to apply than a flush interval under load. - Order operations within one buffer by resourceVersion, so a late or redelivered older event never displaces a newer buffered one. - Record a deleted object as a tombstone that wins over any later flushed write regardless of resourceVersion. Removal because an object stopped matching a policy stays an ordinary versioned state, so a newer matching write brings the document back. - Skip audit events whose request URI carries dryRun, and writes whose response has no resourceVersion. - Add stale_writes_superseded_total and order_repairs_total counters. Partitioning the stream by object instead would need a per-object token in the subject, which the Vector sink does not publish, a fixed replica count in place of the HPA, and would still regress on redelivery of a nacked older batch, so a version guard is needed either way. --- cmd/search/indexer/command.go | 43 +++ docs/components/resource-indexer.md | 36 +- internal/indexer/batcher.go | 302 +++++++++++------ internal/indexer/batcher_ordering_test.go | 307 ++++++++++++++++++ internal/indexer/consumer.go | 60 +++- internal/indexer/consumer_dryrun_test.go | 153 +++++++++ internal/indexer/consumer_scope_test.go | 4 +- internal/indexer/consumer_test.go | 10 +- internal/indexer/docstate.go | 210 ++++++++++++ internal/indexer/ordering.go | 288 ++++++++++++++++ internal/indexer/reindex_consumer.go | 13 +- internal/metrics/metrics.go | 28 ++ .../chainsaw-test.yaml | 4 +- 13 files changed, 1352 insertions(+), 106 deletions(-) create mode 100644 internal/indexer/batcher_ordering_test.go create mode 100644 internal/indexer/consumer_dryrun_test.go create mode 100644 internal/indexer/docstate.go create mode 100644 internal/indexer/ordering.go diff --git a/cmd/search/indexer/command.go b/cmd/search/indexer/command.go index bfe745e..c0f809f 100644 --- a/cmd/search/indexer/command.go +++ b/cmd/search/indexer/command.go @@ -29,6 +29,12 @@ import ( // process, so it is duplicated here to bound the indexing timeouts. const consumerAckWait = 300 * time.Second +// docStateTTLDefault mirrors the audit stream's maxAge in +// config/components/nats-streams/nats-streams.yaml. A record must outlive +// every event that could still be redelivered for its document, or a late +// event could be applied with nothing left to order it against. +const docStateTTLDefault = 7 * 24 * time.Hour + const ( // heartbeatMargin is how many heartbeat intervals an ackWait must fit. The // spare intervals absorb a lost or delayed tick without JetStream giving up on @@ -67,6 +73,12 @@ type ResourceIndexerOptions struct { NatsReindexStream string NatsReindexConsumerName string + // Document state bucket, shared by every replica so that writes to one + // document stay in resourceVersion order whichever replica flushes last. + NatsDocStateBucket string + NatsDocStateReplicas int + DocStateTTL time.Duration + // Meilisearch connection and timeout settings MeilisearchTaskWaitTimeout time.Duration MeilisearchTaskPollInterval time.Duration @@ -98,6 +110,9 @@ func NewResourceIndexerOptions() *ResourceIndexerOptions { NatsStreamName: "AUDIT_EVENTS", NatsReindexStream: "REINDEX_EVENTS", NatsReindexConsumerName: "search-reindexer", + NatsDocStateBucket: "SEARCH_INDEXER_DOC_STATE", + NatsDocStateReplicas: 1, + DocStateTTL: docStateTTLDefault, MeilisearchTaskWaitTimeout: 10 * time.Minute, MeilisearchTaskPollInterval: 500 * time.Millisecond, MeilisearchHTTPTimeout: 60 * time.Second, @@ -123,6 +138,9 @@ func (o *ResourceIndexerOptions) AddFlags(fs *pflag.FlagSet) { fs.StringVar(&o.NatsReindexStream, "nats-reindex-stream", o.NatsReindexStream, "The JetStream stream name for re-index messages.") fs.StringVar(&o.NatsReindexConsumerName, "nats-reindex-consumer-name", o.NatsReindexConsumerName, "The name of the re-index JetStream consumer (must match the manifest).") + fs.StringVar(&o.NatsDocStateBucket, "nats-doc-state-bucket", o.NatsDocStateBucket, "The JetStream KV bucket that records the newest indexed state of every document. It is created if missing and must be shared by all replicas.") + fs.IntVar(&o.NatsDocStateReplicas, "nats-doc-state-replicas", o.NatsDocStateReplicas, "The number of JetStream replicas for the document state bucket.") + fs.DurationVar(&o.DocStateTTL, "doc-state-ttl", o.DocStateTTL, "How long a document's recorded state is kept after its last write. Must cover the audit stream's retention so no redelivered event can outlive the record that orders it.") fs.StringVar(&o.NatsTLSCA, "nats-tls-ca", o.NatsTLSCA, "The path to the NATS TLS CA file.") fs.StringVar(&o.NatsTLSCert, "nats-tls-cert", o.NatsTLSCert, "The path to the NATS TLS certificate file.") fs.StringVar(&o.NatsTLSKey, "nats-tls-key", o.NatsTLSKey, "The path to the NATS TLS key file.") @@ -168,6 +186,15 @@ func (o *ResourceIndexerOptions) Validate() error { if o.NatsReindexConsumerName == "" { return fmt.Errorf("nats-reindex-consumer-name must be set") } + if o.NatsDocStateBucket == "" { + return fmt.Errorf("nats-doc-state-bucket must be set") + } + if o.NatsDocStateReplicas < 1 { + return fmt.Errorf("nats-doc-state-replicas must be at least 1") + } + if o.DocStateTTL < time.Hour { + return fmt.Errorf("doc-state-ttl must be at least 1h") + } if o.MeilisearchDomain == "" { return fmt.Errorf("meilisearch-domain must be set") } @@ -400,12 +427,28 @@ func Run(o *ResourceIndexerOptions, ctx context.Context) error { return fmt.Errorf("failed to create search client: %w", err) } + // The bucket is owned by the indexer rather than declared alongside the + // consumers because every replica, and the re-index path, must agree on + // the one arbiter for a document's newest state. + docStateKV, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{ + Bucket: o.NatsDocStateBucket, + Description: "Newest indexed state per document, shared by search indexer replicas to keep writes in resourceVersion order", + History: 1, + TTL: o.DocStateTTL, + Storage: jetstream.FileStorage, + Replicas: o.NatsDocStateReplicas, + }) + if err != nil { + return fmt.Errorf("failed to open document state bucket %s: %w", o.NatsDocStateBucket, err) + } + batchConfig := indexer.BatchConfig{ BatchSize: o.BatchSize, FlushInterval: o.FlushInterval, MaxConcurrentUploads: o.BatchMaxConcurrentUploads, MaxInFlightFlushes: o.BatchMaxInFlightFlushes, AckProgressInterval: o.BatchAckProgressInterval, + DocState: indexer.NewNATSDocStateStore(docStateKV), } // Create separate batchers for audit events and re-indexing events diff --git a/docs/components/resource-indexer.md b/docs/components/resource-indexer.md index 52081ef..b345f66 100644 --- a/docs/components/resource-indexer.md +++ b/docs/components/resource-indexer.md @@ -245,7 +245,41 @@ the buffer reaches its size or time threshold: Meilisearch applies an index's tasks in the order they are enqueued, so each batch enqueues its tasks only after the batch taken before it has enqueued. -This ordering holds within one indexer replica only. +That holds within one replica only; across replicas, which share one JetStream +queue group, the events for one object are spread out and an older event's +batch can be enqueued after a newer one's. + +A document state bucket (JetStream KV, created by the indexer) is the arbiter +across replicas. It records, per index and document, the newest state any +replica has accepted — its `resourceVersion`, whether it is indexed, removed +from the index or deleted, and a copy of the document — plus the latest +Meilisearch task known to carry a state for it. Reading the current document +from Meilisearch cannot play that role: writes are queued tasks that take +longer to apply than a flush interval under load, so a read misses exactly the +newer write it needs to see. Every flush: + +1. Claims each operation with a compare-and-set. An operation older than the + recorded state is replaced by that state, so a late create re-applies the + newest update instead of the create response. +2. Enqueues the claimed states and learns the task uid that carries them. +3. Records that task. If the bucket already names a newer state, it was + enqueued before this task and is about to be overwritten, so the flush + enqueues it again behind its own task. Only the flush holding the latest + recorded task repairs; a flush that fails between steps is nacked and the + redelivery runs the protocol from the newest recorded state. + +Within one buffer, an operation older than the one already buffered for its +document is dropped rather than replacing it. + +A deletion of the object itself (a `delete` verb, or a write with a +`deletionTimestamp`) is recorded as a tombstone that wins over every later +flushed write regardless of `resourceVersion`, since an object's uid is never +reused. Removal because the object stopped matching a policy is an ordinary +versioned state, so a newer matching write brings the document back. + +Dry-run writes are audited like real ones but persist nothing; the indexer +skips any event whose request URI carries `dryRun`, and any write whose +response lacks a `resourceVersion`. Create events that don't match any policy can be acknowledged immediately since the resource was never indexed. Update and delete events that don't match should diff --git a/internal/indexer/batcher.go b/internal/indexer/batcher.go index 6333fb6..cf9cafb 100644 --- a/internal/indexer/batcher.go +++ b/internal/indexer/batcher.go @@ -3,6 +3,7 @@ package indexer import ( "context" "errors" + "fmt" "sync" "time" @@ -65,6 +66,10 @@ type BatchConfig struct { // messages are still being worked on, extending ackWait for as long as the // flush is genuinely progressing. AckProgressInterval time.Duration + // DocState is the store that orders writes to one document across every + // batcher and replica that shares it. Nil uses a store private to this + // process, which is only suitable for tests and single-replica setups. + DocState DocStateStore } // SearchClient abstracts the search backend interactions. @@ -74,7 +79,8 @@ type SearchClient interface { WaitForTasks(tasks []*meilisearch.Task) (*meilisearch.Task, error) } -// UpsertOp is a single document upsert destined for one index. +// UpsertOp is a single document upsert destined for one index. The document's +// metadata.resourceVersion orders it against other writes to the same object. type UpsertOp struct { IndexUID string Doc any @@ -84,39 +90,60 @@ type UpsertOp struct { type DeleteOp struct { IndexUID string DocID string + // ResourceVersion orders the removal against other writes to the same + // object. It is zero when the event carried no object, as a Status + // response does. + ResourceVersion uint64 + // ObjectDeleted marks the object itself as gone. Such a removal wins over + // any write that flushes after it, whatever its resourceVersion, because + // nothing written for a deleted uid can be current. Without it the + // removal only means the object no longer belongs in this index, and a + // newer write brings it back. + ObjectDeleted bool } -type upsertItem struct { +// pendingOp is the latest operation buffered for one document in one index. +type pendingOp struct { indexUID string + docID string + rv uint64 + state docState doc any } -type deleteItem struct { - indexUID string - docID string +func (op pendingOp) record() docRecord { + return docRecord{RV: op.rv, State: op.state} } -// pendingOp is the latest operation buffered for one document in one index. -type pendingOp struct { - isDelete bool - upsert upsertItem - delete deleteItem +func (op pendingOp) isUpsert() bool { + return op.state == docIndexed } // batch is one flush's worth of operations and the messages they came from. type batch struct { - upserts []upsertItem - deletes []deleteItem - msgs []jetstream.Msg - turn enqueueTurn + ops []pendingOp + msgs []jetstream.Msg + turn enqueueTurn +} + +func (bt *batch) counts() (upserts, deletes int) { + for _, op := range bt.ops { + if op.isUpsert() { + upserts++ + } else { + deletes++ + } + } + return upserts, deletes } // flushType labels the batch's metrics and logs by the operations it carries. func (bt *batch) flushType() string { + upserts, deletes := bt.counts() switch { - case len(bt.upserts) > 0 && len(bt.deletes) > 0: + case upserts > 0 && deletes > 0: return flushTypeMixed - case len(bt.deletes) > 0: + case deletes > 0: return flushTypeDelete default: return flushTypeUpsert @@ -135,10 +162,12 @@ type enqueueTurn struct { type Batcher struct { client SearchClient batchConfig BatchConfig + docState DocStateStore - // pending holds the latest operation per index/docID, so an upsert and a - // delete for the same document can never both be buffered: whichever came - // last is the one flushed. pendingUpserts counts the upserts among them. + // pending holds the newest operation per index/docID, so an upsert and a + // delete for the same document can never both be buffered: whichever is + // newest by resourceVersion, or latest to arrive at the same version, is + // the one flushed. pendingUpserts counts the upserts among them. pending map[string]pendingOp pendingUpserts int @@ -187,12 +216,18 @@ func NewBatcher(client SearchClient, batchConfig BatchConfig) *Batcher { ackProgressInterval = DefaultAckProgressInterval } + docState := batchConfig.DocState + if docState == nil { + docState = NewMemoryDocStateStore() + } + lastEnqueue := make(chan struct{}) close(lastEnqueue) return &Batcher{ client: client, batchConfig: batchConfig, + docState: docState, ackProgressInterval: ackProgressInterval, lastEnqueue: lastEnqueue, pending: make(map[string]pendingOp), @@ -246,9 +281,15 @@ func (b *Batcher) Submit(msg *jetstream.Msg, upserts []UpsertOp, deletes []Delet // Deletes are buffered before upserts so that, within one message, a policy // that matches wins over one that does not for the same index. for _, op := range deletes { - b.setPending(op.IndexUID, op.DocID, pendingOp{ - isDelete: true, - delete: deleteItem{indexUID: op.IndexUID, docID: op.DocID}, + state := docUnindexed + if op.ObjectDeleted { + state = docDeleted + } + b.setPending(pendingOp{ + indexUID: op.IndexUID, + docID: op.DocID, + rv: op.ResourceVersion, + state: state, }) } @@ -261,8 +302,12 @@ func (b *Batcher) Submit(msg *jetstream.Msg, upserts []UpsertOp, deletes []Delet } } - b.setPending(op.IndexUID, docID, pendingOp{ - upsert: upsertItem{indexUID: op.IndexUID, doc: op.Doc}, + b.setPending(pendingOp{ + indexUID: op.IndexUID, + docID: docID, + rv: resourceVersionOf(op.Doc), + state: docIndexed, + doc: op.Doc, }) } @@ -297,24 +342,36 @@ func (b *Batcher) Submit(msg *jetstream.Msg, upserts []UpsertOp, deletes []Delet b.mu.Unlock() if due { - klog.Infof("Batch size reached, flushing %d upserts and %d deletes from %d unique messages", len(bt.upserts), len(bt.deletes), len(bt.msgs)) + upserts, deletes := bt.counts() + klog.Infof("Batch size reached, flushing %d upserts and %d deletes from %d unique messages", upserts, deletes, len(bt.msgs)) b.launchFlush(bt) } } -// setPending records op as the latest operation for docID in indexUID, -// replacing whatever was buffered for it. MUST hold lock. -func (b *Batcher) setPending(indexUID, docID string, op pendingOp) { - key := indexUID + "/" + docID - if prev, ok := b.pending[key]; ok && !prev.isDelete { - b.pendingUpserts-- +// setPending records op as the buffered operation for its document unless +// something at least as new is already buffered, in which case op is dropped: +// a redelivered or late-arriving older event must not displace a newer one +// that is waiting in the same window. MUST hold lock. +func (b *Batcher) setPending(op pendingOp) { + key := pendingKey(op.indexUID, op.docID) + if prev, ok := b.pending[key]; ok { + if !op.record().supersedes(prev.record()) { + return + } + if prev.isUpsert() { + b.pendingUpserts-- + } } - if !op.isDelete { + if op.isUpsert() { b.pendingUpserts++ } b.pending[key] = op } +func pendingKey(indexUID, docID string) string { + return indexUID + "/" + docID +} + // setPendingMetrics publishes the buffered operation counts. MUST hold lock. func (b *Batcher) setPendingMetrics() { searchmetrics.IndexerPendingOperations.WithLabelValues(flushTypeUpsert).Set(float64(b.pendingUpserts)) @@ -329,9 +386,9 @@ func (b *Batcher) QueueUpsert(indexUID string, doc any, msg *jetstream.Msg) { b.Submit(msg, []UpsertOp{{IndexUID: indexUID, Doc: doc}}, nil) } -// QueueDelete adds a document ID to the pending map and triggers an asynchronous -// flush if the batch size is reached. It is shorthand for a single-operation -// Submit; see QueueUpsert. +// QueueDelete removes a document from an index without ordering information, +// as for a document that no longer belongs there. It is shorthand for a +// single-operation Submit; see QueueUpsert. func (b *Batcher) QueueDelete(indexUID string, docID string, msg *jetstream.Msg) { b.Submit(msg, nil, []DeleteOp{{IndexUID: indexUID, DocID: docID}}) } @@ -468,17 +525,12 @@ func (b *Batcher) warnFlushBackpressure(flushType string, blocked time.Duration) // place in the enqueue order. MUST hold lock. func (b *Batcher) takeBatch() batch { bt := batch{ - upserts: make([]upsertItem, 0, b.pendingUpserts), - deletes: make([]deleteItem, 0, len(b.pending)-b.pendingUpserts), - msgs: b.msgs, - turn: enqueueTurn{prev: b.lastEnqueue, done: make(chan struct{})}, + ops: make([]pendingOp, 0, len(b.pending)), + msgs: b.msgs, + turn: enqueueTurn{prev: b.lastEnqueue, done: make(chan struct{})}, } for _, op := range b.pending { - if op.isDelete { - bt.deletes = append(bt.deletes, op.delete) - } else { - bt.upserts = append(bt.upserts, op.upsert) - } + bt.ops = append(bt.ops, op) } b.lastEnqueue = bt.turn.done @@ -547,9 +599,35 @@ func (b *Batcher) startAckProgress(msgs []jetstream.Msg, flushType string) (stop } } +// flushErrors collects the failures of one flush from its goroutines. +type flushErrors struct { + mu sync.Mutex + errs []error +} + +func (f *flushErrors) add(err error) { + f.mu.Lock() + defer f.mu.Unlock() + f.errs = append(f.errs, err) +} + +func (f *flushErrors) join() error { + f.mu.Lock() + defer f.mu.Unlock() + return errors.Join(f.errs...) +} + +// performFlush writes a batch to Meilisearch in three phases. First every +// operation is claimed in the document state store, which replaces any that +// an earlier flush anywhere has already superseded with the newer state. Then +// the resulting states are enqueued. Finally each is recorded against the task +// that carries it; where the store shows a newer state was enqueued before +// this task, the newer state is enqueued again so it lands last. See +// ordering.go for why each phase is needed. func (b *Batcher) performFlush(bt batch) { flushType := bt.flushType() - klog.Infof("Flushing batch of %d upserts and %d deletes to Meilisearch...", len(bt.upserts), len(bt.deletes)) + upserts, deletes := bt.counts() + klog.Infof("Flushing batch of %d upserts and %d deletes to Meilisearch...", upserts, deletes) searchmetrics.IndexerInFlightFlushes.Inc() defer searchmetrics.IndexerInFlightFlushes.Dec() @@ -558,78 +636,116 @@ func (b *Batcher) performFlush(bt batch) { searchmetrics.IndexerFlushDuration.WithLabelValues(flushType).Observe(time.Since(start).Seconds()) }() - upsertGroups := make(map[string][]any) - for _, item := range bt.upserts { - upsertGroups[item.indexUID] = append(upsertGroups[item.indexUID], item.doc) + // Heartbeat the batch for the whole time its tasks are outstanding. + stopAckProgress := b.startAckProgress(bt.msgs, flushType) + + // Flushes outlive a shutdown of the batcher on purpose (see + // acquireFlushSlot), so the store calls get their own lifetime. + ctx := context.Background() + + var wg sync.WaitGroup + var errs flushErrors + + claims, err := b.claimBatch(ctx, bt.ops) + if err != nil { + errs.add(fmt.Errorf("claiming document state: %w", err)) } - deleteGroups := make(map[string][]string) - for _, item := range bt.deletes { - deleteGroups[item.indexUID] = append(deleteGroups[item.indexUID], item.docID) + + var groups []*flushGroup + if err == nil { + groups = groupClaims(claims) + b.enqueueGroups(groups, &wg, &errs) } + close(bt.turn.done) - var wg, enqueueWG sync.WaitGroup - var errs []error - var errMu sync.Mutex + for round := 0; err == nil && len(groups) > 0; round++ { + repairs, recordErr := b.recordBatch(ctx, groups) + if recordErr != nil { + errs.add(fmt.Errorf("recording document state: %w", recordErr)) + break + } + if len(repairs) == 0 { + break + } + if round >= maxRepairRounds { + errs.add(fmt.Errorf("%d documents still carry a superseded state after %d repair rounds", len(repairs), maxRepairRounds)) + break + } + searchmetrics.IndexerOrderRepairs.Add(float64(len(repairs))) + klog.Infof("Re-enqueueing %d documents whose newer state was enqueued before this batch's", len(repairs)) + groups = groupClaims(repairs) + b.enqueueGroups(groups, &wg, &errs) + } - // Heartbeat the batch for the whole time its tasks are outstanding. - stopAckProgress := b.startAckProgress(bt.msgs, flushType) + wg.Wait() + stopAckProgress() + + if err := errs.join(); err != nil { + searchmetrics.IndexerFlushTotal.WithLabelValues(flushType, flushStatusFailure).Inc() + klog.Errorf("Failed to flush %d upserts and %d deletes from %d messages, nacking for redelivery in %s: %v", + upserts, deletes, len(bt.msgs), nakDelay, err) + b.nakBatch(bt.msgs, flushType) + return + } + + searchmetrics.IndexerFlushTotal.WithLabelValues(flushType, flushStatusSuccess).Inc() + klog.Infof("Successfully flushed %d upserts and %d deletes", upserts, deletes) + for _, msg := range bt.msgs { + msg.Ack() + } +} - // run enqueues one Meilisearch task and then waits for it. The keys in a - // batch are unique, so the order of its tasks relative to each other does - // not matter; only the enqueue barrier against the next batch does. - run := func(enqueue func() ([]*meilisearch.Task, error)) { +// enqueueGroups sends each group to Meilisearch and returns once every group +// has been enqueued, with the task uids recorded on the groups. Waiting for +// the tasks to finish continues in the background under wg. The keys in a +// group are unique, so the order of its tasks relative to each other does not +// matter; only the enqueue barrier against the next batch does. +func (b *Batcher) enqueueGroups(groups []*flushGroup, wg *sync.WaitGroup, errs *flushErrors) { + var enqueueWG sync.WaitGroup + for _, g := range groups { wg.Add(1) enqueueWG.Add(1) - go func() { + go func(g *flushGroup) { defer wg.Done() // 1. Enqueue - Limited by semaphore b.sem <- struct{}{} - tasks, err := enqueue() + tasks, err := g.enqueue(b.client) <-b.sem + if err == nil { + g.task = firstTaskUID(tasks) + } enqueueWG.Done() if err != nil { - errMu.Lock() - errs = append(errs, err) - errMu.Unlock() + errs.add(err) return } // 2. Wait (Polling) - NOT limited by semaphore - _, err = b.client.WaitForTasks(tasks) - if err != nil { - errMu.Lock() - errs = append(errs, err) - errMu.Unlock() + if _, err := b.client.WaitForTasks(tasks); err != nil { + errs.add(err) } - }() - } - - for indexUID, docs := range upsertGroups { - run(func() ([]*meilisearch.Task, error) { return b.client.AddDocumentsAsync(indexUID, docs) }) - } - for indexUID, docIDs := range deleteGroups { - run(func() ([]*meilisearch.Task, error) { return b.client.DeleteDocumentsAsync(indexUID, docIDs) }) + }(g) } enqueueWG.Wait() - close(bt.turn.done) - wg.Wait() - stopAckProgress() - - if len(errs) > 0 { - searchmetrics.IndexerFlushTotal.WithLabelValues(flushType, flushStatusFailure).Inc() - klog.Errorf("Failed to flush %d upserts and %d deletes from %d messages, nacking for redelivery in %s: %v", - len(bt.upserts), len(bt.deletes), len(bt.msgs), nakDelay, errors.Join(errs...)) - b.nakBatch(bt.msgs, flushType) - return - } +} - searchmetrics.IndexerFlushTotal.WithLabelValues(flushType, flushStatusSuccess).Inc() - klog.Infof("Successfully flushed %d upserts and %d deletes", len(bt.upserts), len(bt.deletes)) - for _, msg := range bt.msgs { - msg.Ack() +// firstTaskUID is the uid of the earliest task among those a group was +// enqueued as. A group spanning several chunks applies across a range of +// uids; claiming the lowest is conservative, since any task later than it is +// treated as able to land after the group. +func firstTaskUID(tasks []*meilisearch.Task) int64 { + var uid int64 + for _, t := range tasks { + if t == nil { + continue + } + if uid == 0 || t.TaskUID < uid { + uid = t.TaskUID + } } + return uid } // nakBatch returns a failed batch to JetStream for redelivery. Without it the diff --git a/internal/indexer/batcher_ordering_test.go b/internal/indexer/batcher_ordering_test.go new file mode 100644 index 0000000..62ade31 --- /dev/null +++ b/internal/indexer/batcher_ordering_test.go @@ -0,0 +1,307 @@ +package indexer + +import ( + "fmt" + "sync" + "testing" + "time" + + "github.com/meilisearch/meilisearch-go" + "github.com/nats-io/nats.go/jetstream" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// rvSearchClient models Meilisearch's task queue closely enough to test write +// ordering: every enqueue gets the next task uid, and an index's documents +// take on the state of its tasks in enqueue order. Each document is kept as +// its resourceVersion, so a test can read which write landed last. onAdd, when +// set, runs before an AddDocuments call is given its task uid, which lets a +// test hold one flush between claiming and enqueueing. +type rvSearchClient struct { + onAdd func(indexUID string, docs []any) + + mu sync.Mutex + nextTask int64 + log []string + docs map[string]map[string]map[string]any +} + +func newRVSearchClient() *rvSearchClient { + return &rvSearchClient{docs: make(map[string]map[string]map[string]any)} +} + +func (c *rvSearchClient) AddDocumentsAsync(indexUID string, documents []any) ([]*meilisearch.Task, error) { + if c.onAdd != nil { + c.onAdd(indexUID, documents) + } + + c.mu.Lock() + defer c.mu.Unlock() + + if c.docs[indexUID] == nil { + c.docs[indexUID] = make(map[string]map[string]any) + } + for _, d := range documents { + doc := d.(map[string]any) + uid := doc["uid"].(string) + c.docs[indexUID][uid] = doc + c.log = append(c.log, fmt.Sprintf("add %s/%s@%d", indexUID, uid, resourceVersionOf(doc))) + } + c.nextTask++ + return []*meilisearch.Task{{TaskUID: c.nextTask, IndexUID: indexUID}}, nil +} + +func (c *rvSearchClient) DeleteDocumentsAsync(indexUID string, documentIDs []string) ([]*meilisearch.Task, error) { + c.mu.Lock() + defer c.mu.Unlock() + + for _, id := range documentIDs { + delete(c.docs[indexUID], id) + c.log = append(c.log, "delete "+indexUID+"/"+id) + } + c.nextTask++ + return []*meilisearch.Task{{TaskUID: c.nextTask, IndexUID: indexUID}}, nil +} + +func (c *rvSearchClient) WaitForTasks(tasks []*meilisearch.Task) (*meilisearch.Task, error) { + return &meilisearch.Task{Status: "succeeded"}, nil +} + +func (c *rvSearchClient) Log() []string { + c.mu.Lock() + defer c.mu.Unlock() + return append([]string(nil), c.log...) +} + +// Doc returns the document indexed for uid, or nil if there is none. +func (c *rvSearchClient) Doc(indexUID, uid string) map[string]any { + c.mu.Lock() + defer c.mu.Unlock() + return c.docs[indexUID][uid] +} + +// rvDoc builds a document for uid at the given resourceVersion carrying one +// field whose value tells which write produced it. +func rvDoc(uid string, rv uint64, hostname string) map[string]any { + return map[string]any{ + "uid": uid, + "metadata": map[string]any{ + "uid": uid, + "resourceVersion": fmt.Sprint(rv), + }, + "status": map[string]any{"hostname": hostname}, + } +} + +func hostnameOf(doc map[string]any) string { + if doc == nil { + return "" + } + return doc["status"].(map[string]any)["hostname"].(string) +} + +func newOrderingBatcher(client SearchClient, store DocStateStore) *Batcher { + return NewBatcher(client, BatchConfig{BatchSize: 100, FlushInterval: time.Hour, DocState: store}) +} + +// TestBatcher_OlderUpdate_SameWindow_DoesNotDisplaceNewer covers a redelivered +// or late event for an object arriving after a newer one in the same window: +// the buffer keeps the newer state. +func TestBatcher_OlderUpdate_SameWindow_DoesNotDisplaceNewer(t *testing.T) { + client := newRVSearchClient() + batcher := newOrderingBatcher(client, nil) + + m1 := &boundMsg{seq: 1} + m2 := &boundMsg{seq: 2} + var jm1, jm2 jetstream.Msg = m1, m2 + batcher.QueueUpsert("proxies", rvDoc("p1", 7, "poet-jump"), &jm1) + batcher.QueueUpsert("proxies", rvDoc("p1", 5, ""), &jm2) + + batcher.flush() + + requireAcked(t, m1, m2) + assert.Equal(t, []string{"add proxies/p1@7"}, client.Log()) + assert.Equal(t, "poet-jump", hostnameOf(client.Doc("proxies", "p1"))) +} + +// TestBatcher_OlderUpdate_LaterBatch_KeepsNewerState covers the ordering bug +// across flush windows: an object's create flushed after its status update +// used to replace the document with the create response. The older write is +// recognised from the recorded state and the newer state is what lands. +func TestBatcher_OlderUpdate_LaterBatch_KeepsNewerState(t *testing.T) { + client := newRVSearchClient() + batcher := newOrderingBatcher(client, nil) + + m1 := &boundMsg{seq: 1} + var jm1 jetstream.Msg = m1 + batcher.QueueUpsert("proxies", rvDoc("p1", 7, "poet-jump"), &jm1) + batcher.flush() + requireAcked(t, m1) + + m2 := &boundMsg{seq: 2} + var jm2 jetstream.Msg = m2 + batcher.QueueUpsert("proxies", rvDoc("p1", 5, ""), &jm2) + batcher.flush() + requireAcked(t, m2) + + assert.Equal(t, uint64(7), resourceVersionOf(client.Doc("proxies", "p1"))) + assert.Equal(t, "poet-jump", hostnameOf(client.Doc("proxies", "p1"))) + assert.Equal(t, []string{"add proxies/p1@7", "add proxies/p1@7"}, client.Log(), + "the later batch must re-apply the recorded newer state, never its own older one") +} + +// TestBatcher_Delete_ThenLateWrites_StaysDeleted covers an object deleted +// before its create or update flushes: the document must not come back, +// whether the late write shares the window with the delete or follows it. +func TestBatcher_Delete_ThenLateWrites_StaysDeleted(t *testing.T) { + client := newRVSearchClient() + batcher := newOrderingBatcher(client, nil) + + m1 := &boundMsg{seq: 1} + m2 := &boundMsg{seq: 2} + var jm1, jm2 jetstream.Msg = m1, m2 + batcher.Submit(&jm1, nil, []DeleteOp{{IndexUID: "proxies", DocID: "p1", ObjectDeleted: true}}) + batcher.QueueUpsert("proxies", rvDoc("p1", 9, "late-update"), &jm2) + batcher.flush() + requireAcked(t, m1, m2) + assert.Nil(t, client.Doc("proxies", "p1"), "an update in the delete's window must not resurrect the document") + + m3 := &boundMsg{seq: 3} + var jm3 jetstream.Msg = m3 + batcher.QueueUpsert("proxies", rvDoc("p1", 3, "late-create"), &jm3) + batcher.flush() + requireAcked(t, m3) + + assert.Nil(t, client.Doc("proxies", "p1"), "a create flushed after the delete must not resurrect the document") + assert.Equal(t, []string{"delete proxies/p1", "delete proxies/p1"}, client.Log()) +} + +// TestBatcher_PolicyRemoval_DoesNotOutliveNewerWrite covers the other kind of +// delete: an object that stopped matching a policy is removed from that index, +// but it still exists, so a newer write that matches again must bring it back. +func TestBatcher_PolicyRemoval_DoesNotOutliveNewerWrite(t *testing.T) { + client := newRVSearchClient() + batcher := newOrderingBatcher(client, nil) + + m1 := &boundMsg{seq: 1} + var jm1 jetstream.Msg = m1 + batcher.Submit(&jm1, nil, []DeleteOp{{IndexUID: "proxies", DocID: "p1", ResourceVersion: 4}}) + batcher.flush() + requireAcked(t, m1) + + m2 := &boundMsg{seq: 2} + var jm2 jetstream.Msg = m2 + batcher.QueueUpsert("proxies", rvDoc("p1", 5, "matches-again"), &jm2) + batcher.flush() + requireAcked(t, m2) + + assert.Equal(t, "matches-again", hostnameOf(client.Doc("proxies", "p1"))) +} + +// TestBatcher_SameObject_SplitAcrossBatchers_NewestWins covers the production +// failure: one object's events land on different replicas, and the replica +// holding the older event flushes last. Two batchers share one state store +// and one backend, as replicas share a KV bucket and Meilisearch. +func TestBatcher_SameObject_SplitAcrossBatchers_NewestWins(t *testing.T) { + client := newRVSearchClient() + store := NewMemoryDocStateStore() + replicaA := newOrderingBatcher(client, store) + replicaB := newOrderingBatcher(client, store) + + mA := &boundMsg{seq: 1} + mB := &boundMsg{seq: 2} + var jmA, jmB jetstream.Msg = mA, mB + replicaA.QueueUpsert("proxies", rvDoc("p1", 10, ""), &jmA) + replicaB.QueueUpsert("proxies", rvDoc("p1", 20, "poet-jump"), &jmB) + + replicaB.flush() + requireAcked(t, mB) + replicaA.flush() + requireAcked(t, mA) + + assert.Equal(t, uint64(20), resourceVersionOf(client.Doc("proxies", "p1"))) + assert.Equal(t, "poet-jump", hostnameOf(client.Doc("proxies", "p1"))) +} + +// TestBatcher_SameObject_ConcurrentFlushes_RepairsOverwrite covers the race +// the recorded state alone cannot settle: both replicas claim their write +// before either enqueues, and the older write is enqueued second. Meilisearch +// would apply it last, so the replica that enqueued it must enqueue the newer +// state again behind its own task. +func TestBatcher_SameObject_ConcurrentFlushes_RepairsOverwrite(t *testing.T) { + client := newRVSearchClient() + store := NewMemoryDocStateStore() + replicaA := newOrderingBatcher(client, store) + replicaB := newOrderingBatcher(client, store) + + // Hold replica A between claiming its older write and enqueueing it. + olderClaimed := make(chan struct{}) + releaseOlder := make(chan struct{}) + var holdOnce sync.Once + client.onAdd = func(_ string, docs []any) { + if resourceVersionOf(docs[0]) == 10 { + holdOnce.Do(func() { + close(olderClaimed) + <-releaseOlder + }) + } + } + + mA := &boundMsg{seq: 1} + mB := &boundMsg{seq: 2} + var jmA, jmB jetstream.Msg = mA, mB + replicaA.QueueUpsert("proxies", rvDoc("p1", 10, ""), &jmA) + replicaB.QueueUpsert("proxies", rvDoc("p1", 20, "poet-jump"), &jmB) + + flushA := make(chan struct{}) + go func() { + defer close(flushA) + replicaA.flush() + }() + + select { + case <-olderClaimed: + case <-time.After(time.Second): + t.Fatal("replica A never reached its enqueue") + } + + // Replica B claims and enqueues the newer write while A is held. + replicaB.flush() + requireAcked(t, mB) + require.Equal(t, []string{"add proxies/p1@20"}, client.Log()) + + close(releaseOlder) + select { + case <-flushA: + case <-time.After(time.Second): + t.Fatal("replica A's flush did not finish") + } + requireAcked(t, mA) + + assert.Equal(t, []string{"add proxies/p1@20", "add proxies/p1@10", "add proxies/p1@20"}, client.Log(), + "the older write's flush must enqueue the newer state again behind its own task") + assert.Equal(t, "poet-jump", hostnameOf(client.Doc("proxies", "p1"))) +} + +func TestDocRecord_Supersedes(t *testing.T) { + indexed := func(rv uint64) docRecord { return docRecord{RV: rv, State: docIndexed} } + unindexed := func(rv uint64) docRecord { return docRecord{RV: rv, State: docUnindexed} } + deleted := func(rv uint64) docRecord { return docRecord{RV: rv, State: docDeleted} } + + assert.True(t, indexed(2).supersedes(indexed(1))) + assert.False(t, indexed(1).supersedes(indexed(2))) + assert.True(t, indexed(2).supersedes(indexed(2)), "a repeat of the same version replaces it") + assert.True(t, unindexed(2).supersedes(indexed(1))) + assert.True(t, indexed(3).supersedes(unindexed(2)), "a policy removal is ordered like any write") + assert.True(t, deleted(0).supersedes(indexed(100)), "a deleted object wins regardless of version") + assert.False(t, indexed(100).supersedes(deleted(0))) + assert.False(t, deleted(5).supersedes(deleted(1)), "a repeat delete changes nothing") +} + +func TestResourceVersionOf(t *testing.T) { + assert.Equal(t, uint64(1769506541), resourceVersionOf(rvDoc("p1", 1769506541, ""))) + assert.Equal(t, uint64(0), resourceVersionOf(map[string]any{"uid": "p1"})) + assert.Equal(t, uint64(0), resourceVersionOf(map[string]any{"metadata": map[string]any{"resourceVersion": "abc"}})) + assert.Equal(t, uint64(0), resourceVersionOf("not a document")) +} diff --git a/internal/indexer/consumer.go b/internal/indexer/consumer.go index 5232b7a..85f62a2 100644 --- a/internal/indexer/consumer.go +++ b/internal/indexer/consumer.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "net/url" "sync" "github.com/nats-io/nats.go/jetstream" @@ -26,6 +27,7 @@ type Indexer struct { type auditEvent struct { AuditID string `json:"auditID"` Verb string `json:"verb"` + RequestURI string `json:"requestURI"` Annotations map[string]string `json:"annotations"` User struct { Extra map[string][]string `json:"extra"` @@ -117,6 +119,17 @@ func (i *Indexer) Start(ctx context.Context) error { return } + // A dry-run request is audited like a real one, with a response object + // that carries a uid, but it never persisted anything: indexing a + // dry-run create would publish an object that does not exist, and a + // dry-run delete would remove one that does. + if isDryRun(&event) { + klog.V(4).Infof("Skipping dry-run %s of %s %s/%s (auditID: %s)", + event.Verb, event.ObjectRef.Resource, event.ObjectRef.Namespace, event.ObjectRef.Name, event.AuditID) + msg.Ack() + return + } + // Handle deletions separately if event.Verb == deleteVerb { i.handleDelete(msg, &event) @@ -150,6 +163,17 @@ func (i *Indexer) Start(ctx context.Context) error { return } + // Every persisted write returns the object with its resourceVersion. + // Without one there is nothing to order the write by, and nothing was + // written for it to describe. + if obj.GetResourceVersion() == "" { + klog.V(2).Infof("Skipping %s of %s %s/%s with no resourceVersion in its response (auditID: %s)", + event.Verb, event.ObjectRef.Resource, event.ObjectRef.Namespace, event.ObjectRef.Name, event.AuditID) + msg.Ack() + return + } + resourceVersion := parseResourceVersion(obj.GetResourceVersion()) + // In single-tenant mode, skip non-platform // events entirely so that no policy can accidentally queue them. tenantName, tenantType := extractTenantFromAuditEvent(&event) @@ -202,7 +226,11 @@ func (i *Indexer) Start(ctx context.Context) error { // it is deleted. Policies for other kinds never indexed it. if (event.Verb == "update" || event.Verb == "patch") && targetsResource(cp, gvk.Group, gvk.Kind) { if cp.Policy.Status.IndexName != "" { - deletes = append(deletes, DeleteOp{IndexUID: cp.Policy.Status.IndexName, DocID: resourceUID}) + deletes = append(deletes, DeleteOp{ + IndexUID: cp.Policy.Status.IndexName, + DocID: resourceUID, + ResourceVersion: resourceVersion, + }) } } } @@ -258,7 +286,12 @@ func (i *Indexer) handleDelete(msg jetstream.Msg, event *auditEvent) { continue } - deletes = append(deletes, DeleteOp{IndexUID: cp.Policy.Status.IndexName, DocID: docID}) + deletes = append(deletes, DeleteOp{ + IndexUID: cp.Policy.Status.IndexName, + DocID: docID, + ResourceVersion: deletedResourceVersion(event), + ObjectDeleted: true, + }) } // No policy has an index yet, so there is nothing to delete from. @@ -279,6 +312,29 @@ func targetsResource(cp *policyevaluation.CachedPolicy, group, kind string) bool return target.Group == group && (kind == "" || target.Kind == kind) } +// isDryRun reports whether the event's request asked the API server not to +// persist anything. The audit event records that only in the request URI. +func isDryRun(event *auditEvent) bool { + if event.RequestURI == "" { + return false + } + u, err := url.Parse(event.RequestURI) + if err != nil { + return false + } + return u.Query().Has("dryRun") +} + +// deletedResourceVersion is the resourceVersion of the object an event +// deletes, when the response carries the object rather than a Status. The +// deletion wins over other writes regardless, so zero only means unknown. +func deletedResourceVersion(event *auditEvent) uint64 { + if event.ResponseObject == nil { + return 0 + } + return parseResourceVersion((&unstructured.Unstructured{Object: event.ResponseObject}).GetResourceVersion()) +} + // deletedResourceType returns the API group and kind of the resource an event // deletes. The kind comes from the response object when it is the resource // itself, as it is for terminating resources. A delete-verb response is often diff --git a/internal/indexer/consumer_dryrun_test.go b/internal/indexer/consumer_dryrun_test.go new file mode 100644 index 0000000..0818cd5 --- /dev/null +++ b/internal/indexer/consumer_dryrun_test.go @@ -0,0 +1,153 @@ +package indexer + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + internalcel "go.miloapis.net/search/internal/cel" + policyevaluation "go.miloapis.net/search/internal/policy/evaluation" + "go.miloapis.net/search/pkg/apis/search/v1alpha1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// consumeOne runs a single audit event through an indexer whose one policy +// indexes every HTTPProxy, and returns the batcher's buffered operations and +// whether the message was acked without being buffered. +func consumeOne(t *testing.T, event map[string]any) (pending map[string]pendingOp, ackedImmediately bool) { + t.Helper() + + policy := &v1alpha1.ResourceIndexPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "proxies"}, + Spec: v1alpha1.ResourceIndexPolicySpec{ + TargetResource: v1alpha1.TargetResource{Group: "networking.datumapis.com", Kind: "HTTPProxy", Version: "v1alpha"}, + Conditions: []v1alpha1.PolicyCondition{{Name: "all", Expression: "true"}}, + }, + Status: v1alpha1.ResourceIndexPolicyStatus{IndexName: "proxies-index"}, + } + env, err := internalcel.NewEnv() + require.NoError(t, err) + policyCache := &PolicyCache{policies: make(map[string]*policyevaluation.CachedPolicy), celEnv: env} + policyCache.upsertPolicy(policy) + + batcher := NewBatcher(new(MockSearchClient), BatchConfig{BatchSize: 1000, FlushInterval: time.Hour}) + + eventBytes, err := json.Marshal(event) + require.NoError(t, err) + msg := &MockJetStreamMsg{seq: 1} + msg.On("Data").Return(eventBytes) + msg.On("Ack").Return(nil).Maybe() + + mockContext := new(MockConsumeContext) + mockContext.On("Stop").Return() + mockConsumer := new(MockConsumer) + mockConsumer.On("Consume", mock.Anything).Return(mockContext, []jetstream.Msg{msg}, nil) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + require.NoError(t, NewIndexer(mockConsumer, policyCache, batcher, false).Start(ctx)) + + batcher.mu.Lock() + defer batcher.mu.Unlock() + pending = make(map[string]pendingOp, len(batcher.pending)) + for k, v := range batcher.pending { + pending[k] = v + } + for _, call := range msg.Calls { + if call.Method == "Ack" { + ackedImmediately = true + } + } + return pending, ackedImmediately +} + +func proxyEvent(verb, requestURI string, meta map[string]any) map[string]any { + return map[string]any{ + "verb": verb, + "auditID": "dry", + "requestURI": requestURI, + "objectRef": map[string]string{ + "apiGroup": "networking.datumapis.com", + "resource": "httpproxies", + "namespace": "default", + "name": "alb-validate", + }, + "responseObject": map[string]any{ + "apiVersion": "networking.datumapis.com/v1alpha", + "kind": "HTTPProxy", + "metadata": meta, + }, + } +} + +// A dry-run create is audited with a response that carries a uid but no +// resourceVersion, because nothing was persisted. It must not produce a +// document for an object that never existed. +func TestIndexer_DryRunCreate_IsIgnored(t *testing.T) { + pending, acked := consumeOne(t, proxyEvent("create", + "/apis/networking.datumapis.com/v1alpha/namespaces/default/httpproxies?dryRun=All&fieldManager=kubectl-create", + map[string]any{"name": "alb-validate", "uid": "proxy-dry"}, + )) + + require.Empty(t, pending) + require.True(t, acked) +} + +// A dry-run delete returns the object as if it had been removed; acting on it +// would drop a live object from search. +func TestIndexer_DryRunDelete_IsIgnored(t *testing.T) { + pending, acked := consumeOne(t, proxyEvent("delete", + "/apis/networking.datumapis.com/v1alpha/namespaces/default/httpproxies/alb-validate?dryRun=All", + map[string]any{"name": "alb-validate", "uid": "proxy-live", "resourceVersion": "5"}, + )) + + require.Empty(t, pending) + require.True(t, acked) +} + +// A write whose response lacks a resourceVersion cannot be ordered against the +// object's other writes and describes nothing persisted, so it is skipped +// even without a dryRun marker. +func TestIndexer_WriteWithoutResourceVersion_IsIgnored(t *testing.T) { + pending, acked := consumeOne(t, proxyEvent("create", + "/apis/networking.datumapis.com/v1alpha/namespaces/default/httpproxies", + map[string]any{"name": "alb-validate", "uid": "proxy-norv"}, + )) + + require.Empty(t, pending) + require.True(t, acked) +} + +// The same create with a resourceVersion is indexed, and the buffered +// operation carries that version for ordering. +func TestIndexer_Create_IsOrderedByResourceVersion(t *testing.T) { + pending, acked := consumeOne(t, proxyEvent("create", + "/apis/networking.datumapis.com/v1alpha/namespaces/default/httpproxies?fieldManager=kubectl-create", + map[string]any{"name": "alb-validate", "uid": "proxy-real", "resourceVersion": "1769505307"}, + )) + + require.False(t, acked) + op, ok := pending["proxies-index/proxy-real"] + require.True(t, ok) + require.Equal(t, docIndexed, op.state) + require.Equal(t, uint64(1769505307), op.rv) +} + +// A real delete marks the object gone so that a create or update flushed +// after it cannot bring the document back. +func TestIndexer_Delete_MarksObjectGone(t *testing.T) { + pending, acked := consumeOne(t, proxyEvent("delete", + "/apis/networking.datumapis.com/v1alpha/namespaces/default/httpproxies/alb-validate", + map[string]any{"name": "alb-validate", "uid": "proxy-gone", "resourceVersion": "9"}, + )) + + require.False(t, acked) + op, ok := pending["proxies-index/proxy-gone"] + require.True(t, ok) + require.Equal(t, docDeleted, op.state) + require.Equal(t, uint64(9), op.rv) +} diff --git a/internal/indexer/consumer_scope_test.go b/internal/indexer/consumer_scope_test.go index 701dd0e..43f4cb9 100644 --- a/internal/indexer/consumer_scope_test.go +++ b/internal/indexer/consumer_scope_test.go @@ -81,7 +81,7 @@ func consumeForPending(t *testing.T, event map[string]any) (deleteKeys []string, batcher.mu.Lock() defer batcher.mu.Unlock() for key, op := range batcher.pending { - require.True(t, op.isDelete, "no event here should queue an upsert, got one for %s", key) + require.False(t, op.isUpsert(), "no event here should queue an upsert, got one for %s", key) deleteKeys = append(deleteKeys, key) } sort.Strings(deleteKeys) @@ -95,7 +95,7 @@ func consumeForPending(t *testing.T, event map[string]any) (deleteKeys []string, } func updateEvent(apiVersion, kind, uid string, extraMeta map[string]any) map[string]any { - meta := map[string]any{"name": "obj", "uid": uid} + meta := map[string]any{"name": "obj", "uid": uid, "resourceVersion": "1"} for k, v := range extraMeta { meta[k] = v } diff --git a/internal/indexer/consumer_test.go b/internal/indexer/consumer_test.go index d2e3ccf..d91eb71 100644 --- a/internal/indexer/consumer_test.go +++ b/internal/indexer/consumer_test.go @@ -116,8 +116,9 @@ func TestIndexer_Start_ConsumeFlow(t *testing.T) { "apiVersion": "v1", "kind": "Pod", "metadata": map[string]interface{}{ - "name": "mypod", - "uid": "pod-uid-1", + "name": "mypod", + "uid": "pod-uid-1", + "resourceVersion": "1", }, }, } @@ -405,8 +406,9 @@ func TestIndexer_Consume_UpdateWithoutDeletionTimestamp_Upserts(t *testing.T) { "apiVersion": "v1", "kind": "Pod", "metadata": map[string]interface{}{ - "name": "mypod", - "uid": "pod-uid-live", + "name": "mypod", + "uid": "pod-uid-live", + "resourceVersion": "1", }, }, } diff --git a/internal/indexer/docstate.go b/internal/indexer/docstate.go new file mode 100644 index 0000000..d8057d7 --- /dev/null +++ b/internal/indexer/docstate.go @@ -0,0 +1,210 @@ +package indexer + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strconv" + "sync" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/klog/v2" +) + +// docState is what an index should hold for one object. +type docState string + +const ( + // docIndexed means the object's document belongs in the index. + docIndexed docState = "indexed" + // docUnindexed means the object exists but no longer matches the policy + // behind the index, so its document is removed. A newer write can bring it + // back. + docUnindexed docState = "unindexed" + // docDeleted means the object itself is gone. It wins over every other + // state regardless of resourceVersion, because an object's uid is never + // reused and nothing written after its deletion can be current. + docDeleted docState = "deleted" +) + +// docRecord is the shared record for one document in one index. The first +// three fields are the newest state any replica has accepted for it; the Task +// fields describe the latest Meilisearch task known to carry a state for it. +// Meilisearch applies an index's tasks in task-uid order, so the document +// matches the newest state exactly when the task fields agree with the state +// fields. +type docRecord struct { + RV uint64 `json:"rv"` + State docState `json:"state"` + Doc json.RawMessage `json:"doc,omitempty"` + + Task int64 `json:"task,omitempty"` + TaskRV uint64 `json:"taskRv,omitempty"` + TaskState docState `json:"taskState,omitempty"` +} + +// supersedes reports whether r is at least as new as prev and may replace it. +// Equal resourceVersions replace so that a repeat of the same write, or a +// policy change re-evaluating the same object, lands as the latest. +func (r docRecord) supersedes(prev docRecord) bool { + if prev.State == docDeleted { + return false + } + if r.State == docDeleted { + return true + } + return r.RV >= prev.RV +} + +// applied reports whether the latest task recorded for r carries r's state, +// which means the document in Meilisearch will match it once that task runs. +func (r docRecord) applied() bool { + return r.Task != 0 && r.TaskRV == r.RV && r.TaskState == r.State +} + +// errDocStateConflict is returned by a store when the record changed since it +// was read; the caller re-reads and decides again. +var errDocStateConflict = errors.New("document state changed since it was read") + +// DocStateStore keeps one docRecord per index/document key with +// compare-and-set semantics. Revision 0 on put creates the key and fails if it +// already exists; any other revision must match the one returned by the read +// the caller is acting on. +type DocStateStore interface { + Get(ctx context.Context, key string) (rec docRecord, revision uint64, found bool, err error) + Put(ctx context.Context, key string, rec docRecord, revision uint64) (uint64, error) +} + +// memoryDocStateStore is an in-process store. It gives a single batcher +// monotonic writes and lets tests share one arbiter between batchers, but it +// is not shared between replicas and keeps every document it has seen, so +// production wires a NATS KV bucket instead. +type memoryDocStateStore struct { + mu sync.Mutex + recs map[string]memoryDocEntry +} + +type memoryDocEntry struct { + rec docRecord + rev uint64 +} + +// NewMemoryDocStateStore returns a DocStateStore held in process memory. +func NewMemoryDocStateStore() DocStateStore { + return &memoryDocStateStore{recs: make(map[string]memoryDocEntry)} +} + +func (s *memoryDocStateStore) Get(_ context.Context, key string) (docRecord, uint64, bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + e, ok := s.recs[key] + if !ok { + return docRecord{}, 0, false, nil + } + return e.rec, e.rev, true, nil +} + +func (s *memoryDocStateStore) Put(_ context.Context, key string, rec docRecord, revision uint64) (uint64, error) { + s.mu.Lock() + defer s.mu.Unlock() + e, ok := s.recs[key] + if (!ok && revision != 0) || (ok && e.rev != revision) { + return 0, errDocStateConflict + } + e.rec = rec + e.rev++ + s.recs[key] = e + return e.rev, nil +} + +// natsDocStateStore backs DocStateStore with a JetStream KV bucket, which is +// shared by every indexer replica and linearizable per key. +type natsDocStateStore struct { + kv jetstream.KeyValue +} + +// NewNATSDocStateStore returns a DocStateStore on the given KV bucket. +func NewNATSDocStateStore(kv jetstream.KeyValue) DocStateStore { + return &natsDocStateStore{kv: kv} +} + +func (s *natsDocStateStore) Get(ctx context.Context, key string) (docRecord, uint64, bool, error) { + entry, err := s.kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return docRecord{}, 0, false, nil + } + if err != nil { + return docRecord{}, 0, false, fmt.Errorf("get document state %s: %w", key, err) + } + var rec docRecord + if err := json.Unmarshal(entry.Value(), &rec); err != nil { + return docRecord{}, 0, false, fmt.Errorf("decode document state %s: %w", key, err) + } + return rec, entry.Revision(), true, nil +} + +func (s *natsDocStateStore) Put(ctx context.Context, key string, rec docRecord, revision uint64) (uint64, error) { + data, err := json.Marshal(rec) + if err != nil { + return 0, fmt.Errorf("encode document state %s: %w", key, err) + } + + var rev uint64 + if revision == 0 { + rev, err = s.kv.Create(ctx, key, data) + } else { + rev, err = s.kv.Update(ctx, key, data, revision) + } + if err == nil { + return rev, nil + } + if errors.Is(err, jetstream.ErrKeyExists) { + return 0, errDocStateConflict + } + + // A document beyond the server's payload limit is still ordered by its + // resourceVersion; only the copy that lets another flush re-enqueue it is + // given up, which claimFromRecord reports when it matters. + if errors.Is(err, nats.ErrMaxPayload) && len(rec.Doc) > 0 { + klog.Warningf("Document state %s: document exceeds the NATS payload limit, recording its resourceVersion without a copy", key) + rec.Doc = nil + return s.Put(ctx, key, rec, revision) + } + + // The client does not type every wrong-revision rejection, so a changed + // revision is detected by reading the key back. + if entry, getErr := s.kv.Get(ctx, key); getErr == nil && entry.Revision() != revision { + return 0, errDocStateConflict + } + return 0, fmt.Errorf("put document state %s: %w", key, err) +} + +// docStateKey names a document's record in the store. +func docStateKey(indexUID, docID string) string { + return indexUID + "." + docID +} + +// resourceVersionOf reads a document's resourceVersion as an integer, which +// is how etcd issues them. A document without one, or with one that does not +// parse, sorts before every versioned write. +func resourceVersionOf(doc any) uint64 { + m, ok := doc.(map[string]any) + if !ok { + return 0 + } + return parseResourceVersion((&unstructured.Unstructured{Object: m}).GetResourceVersion()) +} + +func parseResourceVersion(rv string) uint64 { + if rv == "" { + return 0 + } + n, err := strconv.ParseUint(rv, 10, 64) + if err != nil { + return 0 + } + return n +} diff --git a/internal/indexer/ordering.go b/internal/indexer/ordering.go new file mode 100644 index 0000000..0484834 --- /dev/null +++ b/internal/indexer/ordering.go @@ -0,0 +1,288 @@ +package indexer + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sync" + + "github.com/meilisearch/meilisearch-go" + searchmetrics "go.miloapis.net/search/internal/metrics" + "k8s.io/klog/v2" +) + +// Writes to one document must land in resourceVersion order, yet the events +// for one object are spread across replicas that each batch for a flush +// interval, so an older event's batch can be enqueued after a newer one's. A +// Meilisearch AddDocuments replaces the whole document, which froze objects at +// whichever state flushed last. +// +// The document state store is the arbiter. Reading the current document from +// Meilisearch is not: its writes are queued tasks that take longer to apply +// than a flush interval under load, so a read would miss exactly the newer +// write it needs to see. Each flush therefore: +// +// 1. claims every operation, which replaces an operation the store shows to +// be superseded with the newer state recorded there; +// 2. enqueues the claimed states, learning the task uid that carries them; +// 3. records that task against each document. Meilisearch applies tasks in +// uid order, so when the store already names a newer state than the one +// this task carries, that newer state was enqueued earlier and is about to +// be overwritten; the flush enqueues it again so it lands last. +// +// Only the flush whose task is the latest recorded performs the repair; a +// flush that sees a higher task uid in the store leaves the outcome to that +// task's flush. A flush that fails between steps is nacked, and the +// redelivery runs the protocol again from the newest recorded state. + +const ( + // maxRepairRounds bounds how many times a flush re-enqueues superseded + // documents before giving up and nacking the batch for redelivery. + maxRepairRounds = 3 + + // maxClaimAttempts bounds the compare-and-set retries on one key. + maxClaimAttempts = 10 + + // docStateParallelism is how many store calls one flush has in flight. + docStateParallelism = 32 +) + +// docClaim is one document's state as a flush will apply it. +type docClaim struct { + indexUID string + docID string + rec docRecord + doc any +} + +// flushGroup is one Meilisearch task's worth of claims: every claim of one +// index that is either added or removed. +type flushGroup struct { + indexUID string + remove bool + claims []*docClaim + + // task is the earliest uid of the tasks the group was enqueued as, set + // once the enqueue returns. Zero means the enqueue failed or the client + // reported no tasks. + task int64 +} + +func (g *flushGroup) enqueue(client SearchClient) ([]*meilisearch.Task, error) { + if g.remove { + ids := make([]string, 0, len(g.claims)) + for _, c := range g.claims { + ids = append(ids, c.docID) + } + return client.DeleteDocumentsAsync(g.indexUID, ids) + } + docs := make([]any, 0, len(g.claims)) + for _, c := range g.claims { + docs = append(docs, c.doc) + } + return client.AddDocumentsAsync(g.indexUID, docs) +} + +// groupClaims splits claims by index and by whether the document is being +// added or removed. +func groupClaims(claims []*docClaim) []*flushGroup { + byKey := make(map[string]*flushGroup) + var groups []*flushGroup + for _, c := range claims { + if c == nil { + continue + } + remove := c.rec.State != docIndexed + key := c.indexUID + "/add" + if remove { + key = c.indexUID + "/remove" + } + g, ok := byKey[key] + if !ok { + g = &flushGroup{indexUID: c.indexUID, remove: remove} + byKey[key] = g + groups = append(groups, g) + } + g.claims = append(g.claims, c) + } + return groups +} + +// claimBatch runs the claim step for every operation. A nil entry in the +// result is an operation the store superseded with a state it cannot supply, +// which is left to the flush that recorded that state. +func (b *Batcher) claimBatch(ctx context.Context, ops []pendingOp) ([]*docClaim, error) { + claims := make([]*docClaim, len(ops)) + err := forEachParallel(len(ops), docStateParallelism, func(i int) error { + c, err := b.claimOne(ctx, ops[i]) + if err != nil { + return err + } + claims[i] = c + return nil + }) + return claims, err +} + +// claimOne makes op's state the newest in the store if it is, and returns the +// state the flush should apply: op's own, or the store's when that is newer. +func (b *Batcher) claimOne(ctx context.Context, op pendingOp) (*docClaim, error) { + key := pendingKey(op.indexUID, op.docID) + mine := op.record() + if op.isUpsert() { + raw, err := json.Marshal(op.doc) + if err != nil { + return nil, fmt.Errorf("encode document %s: %w", key, err) + } + mine.Doc = raw + } + + for attempt := 0; attempt < maxClaimAttempts; attempt++ { + stored, rev, found, err := b.docState.Get(ctx, key) + if err != nil { + return nil, err + } + + if found && !mine.supersedes(stored) { + searchmetrics.IndexerStaleWritesSuperseded.Inc() + klog.V(2).Infof("Document %s: %s at resourceVersion %d is superseded by %s at %d already recorded, applying the recorded state", + key, mine.State, mine.RV, stored.State, stored.RV) + return claimFromRecord(op.indexUID, op.docID, stored) + } + + desired := mine + if found { + desired.Task, desired.TaskRV, desired.TaskState = stored.Task, stored.TaskRV, stored.TaskState + } + if _, err := b.docState.Put(ctx, key, desired, rev); err != nil { + if errors.Is(err, errDocStateConflict) { + continue + } + return nil, err + } + return &docClaim{indexUID: op.indexUID, docID: op.docID, rec: desired, doc: op.doc}, nil + } + return nil, fmt.Errorf("document %s: state changed under %d claim attempts", key, maxClaimAttempts) +} + +// claimFromRecord builds the claim that applies a stored state. It is nil when +// the store names an indexed state but holds no document for it, which only +// happens when the document was too large to keep; the write that recorded it +// has enqueued it already. +func claimFromRecord(indexUID, docID string, rec docRecord) (*docClaim, error) { + c := &docClaim{indexUID: indexUID, docID: docID, rec: rec} + if rec.State != docIndexed { + return c, nil + } + if len(rec.Doc) == 0 { + klog.Warningf("Document %s/%s: recorded state at resourceVersion %d has no document to apply, leaving it to the flush that recorded it", indexUID, docID, rec.RV) + return nil, nil + } + var doc map[string]any + if err := json.Unmarshal(rec.Doc, &doc); err != nil { + return nil, fmt.Errorf("decode recorded document %s/%s: %w", indexUID, docID, err) + } + c.doc = doc + return c, nil +} + +// recordBatch runs the record step for every claim in every enqueued group +// and returns the claims that must be enqueued again. +func (b *Batcher) recordBatch(ctx context.Context, groups []*flushGroup) ([]*docClaim, error) { + type job struct { + claim *docClaim + task int64 + } + var jobs []job + for _, g := range groups { + if g.task == 0 { + continue + } + for _, c := range g.claims { + jobs = append(jobs, job{claim: c, task: g.task}) + } + } + + repairs := make([]*docClaim, len(jobs)) + err := forEachParallel(len(jobs), docStateParallelism, func(i int) error { + r, err := b.recordOne(ctx, jobs[i].claim, jobs[i].task) + if err != nil { + return err + } + repairs[i] = r + return nil + }) + + var out []*docClaim + for _, r := range repairs { + if r != nil { + out = append(out, r) + } + } + return out, err +} + +// recordOne notes that task carries c's state, and returns the newer state to +// enqueue again if the store shows task is about to overwrite it. +func (b *Batcher) recordOne(ctx context.Context, c *docClaim, task int64) (*docClaim, error) { + key := pendingKey(c.indexUID, c.docID) + + for attempt := 0; attempt < maxClaimAttempts; attempt++ { + stored, rev, found, err := b.docState.Get(ctx, key) + if err != nil { + return nil, err + } + if !found { + stored = docRecord{RV: c.rec.RV, State: c.rec.State, Doc: c.rec.Doc} + rev = 0 + } + if task <= stored.Task { + return nil, nil + } + + stored.Task, stored.TaskRV, stored.TaskState = task, c.rec.RV, c.rec.State + if _, err := b.docState.Put(ctx, key, stored, rev); err != nil { + if errors.Is(err, errDocStateConflict) { + continue + } + return nil, err + } + if stored.applied() { + return nil, nil + } + + klog.V(2).Infof("Document %s: task %d carries %s at resourceVersion %d but %s at %d was enqueued before it, enqueueing the newer state again", + key, task, c.rec.State, c.rec.RV, stored.State, stored.RV) + return claimFromRecord(c.indexUID, c.docID, stored) + } + return nil, fmt.Errorf("document %s: state changed under %d record attempts", key, maxClaimAttempts) +} + +// forEachParallel runs fn for every index with at most workers in flight and +// returns every error raised. +func forEachParallel(n, workers int, fn func(i int) error) error { + if n == 0 { + return nil + } + if workers > n { + workers = n + } + + var wg sync.WaitGroup + var errs flushErrors + slots := make(chan struct{}, workers) + for i := 0; i < n; i++ { + slots <- struct{}{} + wg.Add(1) + go func(i int) { + defer wg.Done() + defer func() { <-slots }() + if err := fn(i); err != nil { + errs.add(err) + } + }(i) + } + wg.Wait() + return errs.join() +} diff --git a/internal/indexer/reindex_consumer.go b/internal/indexer/reindex_consumer.go index b1ddcdd..84d403d 100644 --- a/internal/indexer/reindex_consumer.go +++ b/internal/indexer/reindex_consumer.go @@ -90,7 +90,12 @@ func (r *ReindexConsumer) processTargetedEvent(msg jetstream.Msg, event ReindexE if obj.GetDeletionTimestamp() != nil { klog.V(4).Infof("ReindexConsumer: resource %s/%s is terminating, deleting from index %s (id=%s)", obj.GetNamespace(), obj.GetName(), event.IndexName, event.ID) - r.batcher.QueueDelete(event.IndexName, resourceUID, &msg) + r.batcher.Submit(&msg, nil, []DeleteOp{{ + IndexUID: event.IndexName, + DocID: resourceUID, + ResourceVersion: parseResourceVersion(obj.GetResourceVersion()), + ObjectDeleted: true, + }}) return } @@ -143,6 +148,10 @@ func (r *ReindexConsumer) processTargetedEvent(msg jetstream.Msg, event ReindexE klog.V(4).Infof("ReindexConsumer: policy %s did not match resource %s/%s (id=%s), deleting from index", event.PolicyName, obj.GetNamespace(), obj.GetName(), event.ID) - r.batcher.QueueDelete(event.IndexName, resourceUID, &msg) + r.batcher.Submit(&msg, nil, []DeleteOp{{ + IndexUID: event.IndexName, + DocID: resourceUID, + ResourceVersion: parseResourceVersion(obj.GetResourceVersion()), + }}) } } diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index e79363a..a86a7f9 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -126,6 +126,32 @@ var ( }, []string{"type"}, ) + + // IndexerStaleWritesSuperseded counts operations the document state store + // showed to be older than a state already recorded by an earlier flush, + // possibly on another replica. The recorded state is applied instead. + IndexerStaleWritesSuperseded = metrics.NewCounter( + &metrics.CounterOpts{ + Namespace: namespace, + Subsystem: indexerSubsystem, + Name: "stale_writes_superseded_total", + Help: "Total number of indexer operations replaced by a newer recorded state for the same document", + StabilityLevel: metrics.ALPHA, + }, + ) + + // IndexerOrderRepairs counts documents a flush enqueued a second time + // because a newer state had been enqueued before the flush's own task and + // would otherwise have been overwritten by it. + IndexerOrderRepairs = metrics.NewCounter( + &metrics.CounterOpts{ + Namespace: namespace, + Subsystem: indexerSubsystem, + Name: "order_repairs_total", + Help: "Total number of documents re-enqueued so their newest state lands after an older task", + StabilityLevel: metrics.ALPHA, + }, + ) ) // init registers all custom metrics with the legacy registry @@ -141,5 +167,7 @@ func init() { IndexerFlushSlotWait, IndexerAckProgress, IndexerMessagesNacked, + IndexerStaleWritesSuperseded, + IndexerOrderRepairs, ) } diff --git a/test/e2e/search-flow-multi-tenant-audit/chainsaw-test.yaml b/test/e2e/search-flow-multi-tenant-audit/chainsaw-test.yaml index e016888..f4730dd 100644 --- a/test/e2e/search-flow-multi-tenant-audit/chainsaw-test.yaml +++ b/test/e2e/search-flow-multi-tenant-audit/chainsaw-test.yaml @@ -100,13 +100,13 @@ spec: # Top-level annotations field carries platform.miloapis.com/scope.type and # scope.name so the indexer's extractTenantFromAuditEvent() will resolve # the tenant as type="Project", name="e2e-audit-test-project". - EVENT_A='{"auditID":"e2e-audit-project-event-001","verb":"create","annotations":{"platform.miloapis.com/scope.type":"Project","platform.miloapis.com/scope.name":"e2e-audit-test-project"},"objectRef":{"apiGroup":"rbac.authorization.k8s.io","apiVersion":"v1","resource":"rolebindings","name":"e2e-audit-project-role","namespace":"default","uid":"e2e-audit-project-uid-00000001"},"responseObject":{"apiVersion":"rbac.authorization.k8s.io/v1","kind":"RoleBinding","metadata":{"name":"e2e-audit-project-role","namespace":"default","uid":"e2e-audit-project-uid-00000001","labels":{"e2e-multi-tenant-audit":"true"},"annotations":{"e2e.search.test/service":"audit-project-service-mt-rn2"}},"roleRef":{"apiGroup":"rbac.authorization.k8s.io","kind":"Role","name":"view"},"subjects":[]},"user":{"extra":{"iam.miloapis.com/parent-type":["project"],"iam.miloapis.com/parent-name":["e2e-audit-test-project"]}}}' + EVENT_A='{"auditID":"e2e-audit-project-event-001","verb":"create","annotations":{"platform.miloapis.com/scope.type":"Project","platform.miloapis.com/scope.name":"e2e-audit-test-project"},"objectRef":{"apiGroup":"rbac.authorization.k8s.io","apiVersion":"v1","resource":"rolebindings","name":"e2e-audit-project-role","namespace":"default","uid":"e2e-audit-project-uid-00000001"},"responseObject":{"apiVersion":"rbac.authorization.k8s.io/v1","kind":"RoleBinding","metadata":{"name":"e2e-audit-project-role","namespace":"default","uid":"e2e-audit-project-uid-00000001","resourceVersion":"1","labels":{"e2e-multi-tenant-audit":"true"},"annotations":{"e2e.search.test/service":"audit-project-service-mt-rn2"}},"roleRef":{"apiGroup":"rbac.authorization.k8s.io","kind":"Role","name":"view"},"subjects":[]},"user":{"extra":{"iam.miloapis.com/parent-type":["project"],"iam.miloapis.com/parent-name":["e2e-audit-test-project"]}}}' # Event B: platform-tenant resource. # No top-level annotations field — exercises the fallback path in # extractTenantFromAuditEvent() that returns "platform"/"platform" # when top-level annotations are absent. - EVENT_B='{"auditID":"e2e-audit-platform-event-001","verb":"create","objectRef":{"apiGroup":"rbac.authorization.k8s.io","apiVersion":"v1","resource":"rolebindings","name":"e2e-audit-platform-role","namespace":"default","uid":"e2e-audit-platform-uid-00000001"},"responseObject":{"apiVersion":"rbac.authorization.k8s.io/v1","kind":"RoleBinding","metadata":{"name":"e2e-audit-platform-role","namespace":"default","uid":"e2e-audit-platform-uid-00000001","labels":{"e2e-multi-tenant-audit":"true"},"annotations":{"e2e.search.test/service":"audit-platform-service-mt-rn2"}},"roleRef":{"apiGroup":"rbac.authorization.k8s.io","kind":"Role","name":"view"},"subjects":[]}}' + EVENT_B='{"auditID":"e2e-audit-platform-event-001","verb":"create","objectRef":{"apiGroup":"rbac.authorization.k8s.io","apiVersion":"v1","resource":"rolebindings","name":"e2e-audit-platform-role","namespace":"default","uid":"e2e-audit-platform-uid-00000001"},"responseObject":{"apiVersion":"rbac.authorization.k8s.io/v1","kind":"RoleBinding","metadata":{"name":"e2e-audit-platform-role","namespace":"default","uid":"e2e-audit-platform-uid-00000001","resourceVersion":"1","labels":{"e2e-multi-tenant-audit":"true"},"annotations":{"e2e.search.test/service":"audit-platform-service-mt-rn2"}},"roleRef":{"apiGroup":"rbac.authorization.k8s.io","kind":"Role","name":"view"},"subjects":[]}}' # Store event payloads in a ConfigMap so they can be mounted into the # Job container without shell escaping issues.