diff --git a/pkg/handler/processor/process/process.go b/pkg/handler/processor/process/process.go index e522cf0618..79b9ce39b9 100644 --- a/pkg/handler/processor/process/process.go +++ b/pkg/handler/processor/process/process.go @@ -24,7 +24,12 @@ import ( "io" "path/filepath" "strings" + "sync" + "gocloud.dev/gcerrors" + "gocloud.dev/pubsub" + + "github.com/nats-io/nats.go/jetstream" "go.uber.org/zap" uuid "github.com/gofrs/uuid" @@ -44,9 +49,9 @@ import ( "github.com/guacsec/guac/pkg/handler/processor/scorecard" "github.com/guacsec/guac/pkg/handler/processor/spdx" "github.com/guacsec/guac/pkg/logging" + "github.com/guacsec/guac/pkg/metrics" jsoniter "github.com/json-iterator/go" "github.com/klauspost/compress/zstd" - "gocloud.dev/pubsub" ) var ( @@ -54,6 +59,21 @@ var ( json = jsoniter.ConfigCompatibleWithStandardLibrary ) +// maxBlobNotFoundRetries caps how many times a single pubsub message will be +// retried when the blob store reports the object is missing. After this many +// consecutive NotFound responses for the same message, it is acked (dropped) +// so it stops being redelivered indefinitely. Guards against orphaned events +// whose underlying blob has been deleted or never landed. +const maxBlobNotFoundRetries = 3 + +// metricsCollectorName is the namespace used for processor metrics. +const metricsCollectorName = "guac_processor" + +// blobNotFoundDroppedCounter is the metric name (sans namespace) for the +// counter tracking pubsub messages dropped because their referenced blob is +// persistently missing. +const blobNotFoundDroppedCounter = "blob_notfound_dropped_total" + func init() { _ = RegisterDocumentProcessor(&ite6.ITE6Processor{}, processor.DocumentITE6Generic) _ = RegisterDocumentProcessor(&ite6.ITE6Processor{}, processor.DocumentITE6SLSA) @@ -80,6 +100,72 @@ func RegisterDocumentProcessor(p processor.DocumentProcessor, d processor.Docume return nil } +// deliveryCountSource returns a stable label describing where the delivery +// count was sourced from, for log/metric clarity. +func deliveryCountSource(fromBroker bool) string { + if fromBroker { + return "broker" + } + return "in_process" +} + +// notFoundTracker counts consecutive NotFound responses per blob key for +// pubsub backends that don't expose a native delivery count (e.g. mempubsub +// in tests). NATS JetStream tracks NumDelivered server-side, so we prefer +// that path when available. +type notFoundTracker struct { + mu sync.Mutex + attempts map[string]int +} + +func newNotFoundTracker() *notFoundTracker { + return ¬FoundTracker{attempts: make(map[string]int)} +} + +func (t *notFoundTracker) increment(key string) int { + t.mu.Lock() + defer t.mu.Unlock() + t.attempts[key]++ + return t.attempts[key] +} + +func (t *notFoundTracker) clear(key string) { + t.mu.Lock() + defer t.mu.Unlock() + delete(t.attempts, key) +} + +// deliveryAttempt returns the broker-reported delivery count for msg if the +// underlying transport exposes one (currently NATS JetStream). The boolean +// indicates whether a server-side count was available; callers fall back to +// in-process tracking when it is not. +func deliveryAttempt(msg *pubsub.Message) (int, bool) { + var jsMsg jetstream.Msg + if !msg.As(&jsMsg) { + return 0, false + } + md, err := jsMsg.Metadata() + if err != nil || md == nil { + return 0, false + } + return int(md.NumDelivered), true +} + +// registerProcessorMetrics registers (or recovers an existing) counter for +// dropped-message accounting. Returns nil for the counter if registration +// fails so the caller can degrade gracefully without aborting ingest. +func registerProcessorMetrics(ctx context.Context, logger *zap.SugaredLogger) metrics.Counter { + collector := metrics.FromContext(ctx, metricsCollectorName) + counter, err := collector.RegisterCounter(ctx, blobNotFoundDroppedCounter, "processor_id") + if err != nil { + // Most commonly this fires in tests where the counter was registered + // by a prior Subscribe; the metric is still usable via AddCounter. + logger.Debugf("blob-notfound counter registration: %v", err) + return nil + } + return counter +} + // Subscribe receives the CD event and decodes the event to obtain the blob store key. // The key is used to retrieve the "document" from the blob store to be processed and ingested. func Subscribe(ctx context.Context, em collector.Emitter, blobStore *blob.BlobStore, emPubSub *emitter.EmitterPubSub) error { @@ -98,6 +184,11 @@ func Subscribe(ctx context.Context, em collector.Emitter, blobStore *blob.BlobSt retries := 0 + // In-process fallback for backends without a native delivery count. + tracker := newNotFoundTracker() + droppedCounter := registerProcessorMetrics(ctx, logger) + metricsCollector := metrics.FromContext(ctx, metricsCollectorName) + // should still continue if there are errors since problem is with individual documents processFunc := func(d *pubsub.Message) error { retries++ @@ -116,10 +207,44 @@ func Subscribe(ctx context.Context, em collector.Emitter, blobStore *blob.BlobSt documentBytes, err := blobStore.Read(ctx, blobStoreKey) if err != nil { + if gcerrors.Code(err) == gcerrors.NotFound { + attempts, fromBroker := deliveryAttempt(d) + if !fromBroker { + attempts = tracker.increment(blobStoreKey) + } + + childLogger.Errorw("failed read document to blob store", + "processor_id", uuidString, + "attempt", attempts, + "max_attempts", maxBlobNotFoundRetries, + "delivery_count_source", deliveryCountSource(fromBroker), + "error", err.Error(), + ) + + if attempts >= maxBlobNotFoundRetries { + childLogger.Warnw("blob persistently missing — dropping pubsub message to stop redelivery", + "processor_id", uuidString, + "attempts", attempts, + "blob_key", blobStoreKey, + "delivery_count_source", deliveryCountSource(fromBroker), + ) + if droppedCounter != nil { + droppedCounter.Inc() + } else if err := metricsCollector.AddCounter(ctx, blobNotFoundDroppedCounter, 1, uuidString); err != nil { + childLogger.Debugf("[processor: %s] AddCounter blob-notfound: %v", uuidString, err) + } + tracker.clear(blobStoreKey) + d.Ack() + } + return nil + } childLogger.Errorf("[processor: %s] failed read document to blob store: %v", uuidString, err) return nil } + // successful read — clear any prior NotFound bookkeeping for this key + tracker.clear(blobStoreKey) + doc := processor.Document{} if err = json.Unmarshal(documentBytes, &doc); err != nil { childLogger.Errorf("[processor: %s] failed unmarshal the document bytes: %v", uuidString, err) diff --git a/pkg/handler/processor/process/process_test.go b/pkg/handler/processor/process/process_test.go index 0195afb40c..a7ba16742f 100644 --- a/pkg/handler/processor/process/process_test.go +++ b/pkg/handler/processor/process/process_test.go @@ -19,12 +19,13 @@ import ( "bytes" "context" "fmt" - "go.uber.org/zap" - "go.uber.org/zap/zapcore" "strings" "testing" "time" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "github.com/guacsec/guac/internal/testing/dochelper" nats_test "github.com/guacsec/guac/internal/testing/nats" "github.com/guacsec/guac/internal/testing/simpledoc" @@ -852,3 +853,35 @@ func testPublish(ctx context.Context, d *processor.Document, blobStore *blob.Blo logger.Debugf("doc published: %+v", d.SourceInformation.Source) return nil } + +func TestNotFoundTracker(t *testing.T) { + tracker := newNotFoundTracker() + + if got := tracker.increment("a"); got != 1 { + t.Fatalf("first increment for 'a' = %d, want 1", got) + } + if got := tracker.increment("a"); got != 2 { + t.Fatalf("second increment for 'a' = %d, want 2", got) + } + if got := tracker.increment("b"); got != 1 { + t.Fatalf("first increment for 'b' = %d, want 1", got) + } + + tracker.clear("a") + if got := tracker.increment("a"); got != 1 { + t.Fatalf("increment after clear for 'a' = %d, want 1", got) + } + // 'b' should be untouched by clearing 'a' + if got := tracker.increment("b"); got != 2 { + t.Fatalf("increment for 'b' after clearing 'a' = %d, want 2", got) + } +} + +func TestDeliveryCountSource(t *testing.T) { + if got := deliveryCountSource(true); got != "broker" { + t.Errorf("deliveryCountSource(true) = %q, want %q", got, "broker") + } + if got := deliveryCountSource(false); got != "in_process" { + t.Errorf("deliveryCountSource(false) = %q, want %q", got, "in_process") + } +}