Merge branch 'main' into fix/fc-saturation-staleness

Author: LukeAVanDrie

pkg/epp/flowcontrol/controller/internal/processor.go

@@ -91,6 +91,11 @@ type Processor struct {
 	wg             sync.WaitGroup
 	isShuttingDown atomic.Bool
 	shutdownOnce   sync.Once
+
+	// dropCounts accumulates non-dispatched request outcomes between periodic summary flushes.
+	// Written by both the main Run goroutine (enqueue, shutdown) and the runCleanupSweep goroutine (sweep),
+	// so each slot is an atomic to avoid a data race.
+	dropCounts [types.NumQueueOutcomes]atomic.Uint64
 }
 
 // NewProcessor creates a new Processor instance.
@@ -263,6 +268,7 @@ func (p *Processor) enqueue(item *FlowItem) {
 	if finalState := outcome; finalState != nil {
 		p.logger.V(logutil.TRACE).Info("Item finalized externally before processing, discarding.",
 			"outcome", finalState.Outcome, "err", finalState.Err, "flowKey", key, "requestID", req.ID())
+		p.recordDrop(finalState.Outcome)
 		return
 	}
 
@@ -272,6 +278,7 @@ func (p *Processor) enqueue(item *FlowItem) {
 		finalErr := fmt.Errorf("configuration error: failed to get queue for flow key %s: %w", key, err)
 		p.logger.Error(finalErr, "Rejecting request, queue lookup failed", "flowKey", key, "requestID", req.ID())
 		item.FinalizeWithOutcome(types.QueueOutcomeRejectedOther, fmt.Errorf("%w: %w", types.ErrRejected, finalErr))
+		p.recordDrop(types.QueueOutcomeRejectedOther)
 		return
 	}
 
@@ -280,6 +287,7 @@ func (p *Processor) enqueue(item *FlowItem) {
 		finalErr := fmt.Errorf("configuration error: failed to get priority band for priority %d: %w", key.Priority, err)
 		p.logger.Error(finalErr, "Rejecting request, priority band lookup failed", "flowKey", key, "requestID", req.ID())
 		item.FinalizeWithOutcome(types.QueueOutcomeRejectedOther, fmt.Errorf("%w: %w", types.ErrRejected, finalErr))
+		p.recordDrop(types.QueueOutcomeRejectedOther)
 		return
 	}
 
@@ -290,9 +298,12 @@ func (p *Processor) enqueue(item *FlowItem) {
 		// that state reflects genuine unavailability (surfaced as 503), not backpressure against a contended pool (429).
 		if p.poolEmpty {
 			p.logger.V(logutil.DEBUG).Info("Rejecting request, queue at capacity with no endpoints",
-				"flowKey", key, "requestID", req.ID(), "reqByteSize", req.ByteSize())
+				"flowKey", key, "requestID", req.ID(), "reqByteSize", req.ByteSize(),
+				"totalLen", stats.TotalLen, "totalCapacityRequests", stats.TotalCapacityRequests,
+				"totalByteSize", stats.TotalByteSize, "totalCapacityBytes", stats.TotalCapacityBytes)
 			item.FinalizeWithOutcome(types.QueueOutcomeRejectedNoEndpoints, fmt.Errorf("%w: %w",
 				types.ErrRejected, types.ErrNoEndpoints))
+			p.recordDrop(types.QueueOutcomeRejectedNoEndpoints)
 			return
 		}
 		p.logger.V(logutil.DEBUG).Info("Rejecting request, queue at capacity",
@@ -301,6 +312,7 @@ func (p *Processor) enqueue(item *FlowItem) {
 			"totalByteSize", stats.TotalByteSize, "totalCapacityBytes", stats.TotalCapacityBytes)
 		item.FinalizeWithOutcome(types.QueueOutcomeRejectedCapacity, fmt.Errorf("%w: %w",
 			types.ErrRejected, types.ErrQueueAtCapacity))
+		p.recordDrop(types.QueueOutcomeRejectedCapacity)
 		return
 	}
 
@@ -311,6 +323,7 @@ func (p *Processor) enqueue(item *FlowItem) {
 		p.logger.Error(finalErr, "Rejecting request, queue add failed",
 			"flowKey", key, "requestID", req.ID())
 		item.FinalizeWithOutcome(types.QueueOutcomeRejectedOther, fmt.Errorf("%w: %w", types.ErrRejected, finalErr))
+		p.recordDrop(types.QueueOutcomeRejectedOther)
 		return
 	}
 	p.logger.V(logutil.TRACE).Info("Item enqueued.",
@@ -446,7 +459,7 @@ func (p *Processor) dispatchItem(itemAcc flowcontrol.QueueItemAccessor) error {
 		// This happens benignly if the item was already removed by the cleanup sweep loop.
 		// We log it at a low level for visibility but return nil so the dispatch cycle proceeds.
 		p.logger.V(logutil.DEBUG).Info("Failed to remove item during dispatch (likely already finalized and swept).",
-			"flowKey", key, "requestID", req.ID(), "error", err)
+			"flowKey", key, "requestID", req.ID(), "err", err)
 		return nil
 	}
 
@@ -473,6 +486,7 @@ func (p *Processor) runCleanupSweep(ctx context.Context) {
 			return
 		case <-ticker.C():
 			p.sweepFinalizedItems()
+			p.flushDropSummary()
 		}
 	}
 }
@@ -486,6 +500,11 @@ func (p *Processor) sweepFinalizedItems() {
 		}
 		removedItems := managedQ.Cleanup(predicate)
 		if len(removedItems) > 0 {
+			for _, itemAcc := range removedItems {
+				if fi, ok := itemAcc.(*FlowItem); ok && fi != nil && fi.FinalState() != nil {
+					p.recordDrop(fi.FinalState().Outcome)
+				}
+			}
 			logger.V(logutil.TRACE).Info("Swept finalized items and released capacity.",
 				"count", len(removedItems))
 		}
@@ -510,13 +529,15 @@ func (p *Processor) shutdown() {
 				// Finalize buffered items.
 				item.FinalizeWithOutcome(types.QueueOutcomeRejectedOther,
 					fmt.Errorf("%w: %w", types.ErrRejected, types.ErrFlowControllerNotRunning))
+				p.recordDrop(types.QueueOutcomeRejectedOther)
 			default:
 				break DrainLoop
 			}
 		}
 		// We do not close enqueueChan because external goroutines (Controller) send on it.
 		// The channel will be garbage collected when the processor terminates.
 		p.evictAll()
+		p.flushDropSummary()
 	})
 }
 
@@ -539,11 +560,36 @@ func (p *Processor) evictAll() {
 			// Finalization is idempotent; safe to call even if already finalized externally.
 			// The per-request log is emitted by EnqueueAndWait when it unblocks.
 			item.FinalizeWithOutcome(outcome, errShutdown)
+			p.recordDrop(item.FinalState().Outcome)
 		}
 	}
 	p.processAllQueuesConcurrently("evictAll", processFn)
 }
 
+func (p *Processor) recordDrop(outcome types.QueueOutcome) {
+	if outcome == types.QueueOutcomeDispatched || outcome == types.QueueOutcomeNotYetFinalized {
+		return
+	}
+	p.dropCounts[outcome].Add(1)
+}
+
+func (p *Processor) flushDropSummary() {
+	var total uint64
+	counts := make(map[string]uint64)
+	for i := range p.dropCounts {
+		if c := p.dropCounts[i].Swap(0); c > 0 {
+			total += c
+			counts[types.QueueOutcome(i).String()] = c
+		}
+	}
+	if total > 0 {
+		p.logger.V(logutil.DEFAULT).Info("Flow control request drop summary",
+			"poolName", p.poolName,
+			"totalDropped", total,
+			"counts", counts)
+	}
+}
+
 // processAllQueuesConcurrently iterates over all queues in all priority bands and executes the given
 // `processFn` for each queue using a dynamically sized worker pool.
 func (p *Processor) processAllQueuesConcurrently(

pkg/epp/flowcontrol/controller/internal/processor_test.go

@@ -1191,6 +1191,10 @@ func TestProcessor(t *testing.T) {
 				assert.Equal(t, types.QueueOutcomeEvictedContextCancelled, finalState.Outcome,
 					"Outcome should be EvictedContextCancelled")
 				assert.ErrorIs(t, finalState.Err, types.ErrContextCancelled, "Error should be ErrContextCancelled")
+
+				// Verify the sweep path recorded the drop.
+				assert.Equal(t, uint64(1), h.processor.dropCounts[types.QueueOutcomeEvictedContextCancelled].Load(),
+					"Drop should be recorded for EvictedContextCancelled")
 			})
 
 			t.Run("should not sweep items not finalized", func(t *testing.T) {
@@ -1447,3 +1451,124 @@ func TestProcessor(t *testing.T) {
 		})
 	})
 }
+
+// TestProcessor_DropSummary verifies the periodic drop-count accounting introduced in #2101.
+func TestProcessor_DropSummary(t *testing.T) {
+	t.Parallel()
+
+	// Verify that the dropCounts array is large enough to hold every QueueOutcome value.
+	// If new outcomes are added in the future, this check will catch them at compile time.
+	var _ = [types.NumQueueOutcomes]struct{}{}
+
+	t.Run("capacity rejection increments counter", func(t *testing.T) {
+		t.Parallel()
+		h := newTestHarness(t, testCleanupTick)
+		h.addQueue(testFlow)
+
+		// Force capacity-full: band has 1 slot already used, CapacityRequests=1.
+		h.StatsFunc = func() contracts.AggregateStats {
+			return contracts.AggregateStats{
+				TotalCapacityBytes: 1e9,
+				PerPriorityBandStats: map[int]contracts.PriorityBandStats{
+					testFlow.Priority: {CapacityRequests: 1, Len: 1},
+				},
+			}
+		}
+
+		item := h.newTestItem("req-cap", testFlow, testTTL)
+		h.processor.enqueue(item)
+
+		outcome, _ := h.waitForFinalization(item)
+		require.Equal(t, types.QueueOutcomeRejectedCapacity, outcome)
+
+		count := h.processor.dropCounts[types.QueueOutcomeRejectedCapacity].Load()
+		assert.Equal(t, uint64(1), count, "RejectedCapacity counter should be 1 after one capacity rejection")
+	})
+
+	t.Run("counters reset after flush", func(t *testing.T) {
+		t.Parallel()
+		h := newTestHarness(t, testCleanupTick)
+		h.addQueue(testFlow)
+
+		// Force capacity-full.
+		h.StatsFunc = func() contracts.AggregateStats {
+			return contracts.AggregateStats{
+				TotalCapacityBytes: 1e9,
+				PerPriorityBandStats: map[int]contracts.PriorityBandStats{
+					testFlow.Priority: {CapacityRequests: 1, Len: 1},
+				},
+			}
+		}
+
+		item := h.newTestItem("req-flush", testFlow, testTTL)
+		h.processor.enqueue(item)
+		h.waitForFinalization(item) //nolint:errcheck
+
+		require.Equal(t, uint64(1), h.processor.dropCounts[types.QueueOutcomeRejectedCapacity].Load())
+
+		// Flushing should zero out all counters.
+		h.processor.flushDropSummary()
+		assert.Equal(t, uint64(0), h.processor.dropCounts[types.QueueOutcomeRejectedCapacity].Load(),
+			"Counter should be 0 after flush")
+	})
+
+	t.Run("multiple outcome types are counted independently", func(t *testing.T) {
+		t.Parallel()
+		h := newTestHarness(t, testCleanupTick)
+		h.addQueue(testFlow)
+
+		// Reject via capacity: band has 1 slot already used, CapacityRequests=1.
+		h.StatsFunc = func() contracts.AggregateStats {
+			return contracts.AggregateStats{
+				TotalCapacityBytes: 1e9,
+				PerPriorityBandStats: map[int]contracts.PriorityBandStats{
+					testFlow.Priority: {CapacityRequests: 1, Len: 1},
+				},
+			}
+		}
+		for i := range 2 {
+			item := h.newTestItem(fmt.Sprintf("req-multi-cap-%d", i), testFlow, testTTL)
+			h.processor.enqueue(item)
+			h.waitForFinalization(item) //nolint:errcheck
+		}
+
+		// Reject via configuration error (RejectedOther): pass capacity, fail priority band lookup.
+		h.StatsFunc = func() contracts.AggregateStats {
+			return contracts.AggregateStats{
+				TotalCapacityBytes: 1e9,
+				PerPriorityBandStats: map[int]contracts.PriorityBandStats{
+					testFlow.Priority: {CapacityRequests: 100, Len: 0},
+				},
+			}
+		}
+		h.PriorityBandAccessorFunc = func(priority int) (flowcontrol.PriorityBandAccessor, error) {
+			return nil, errors.New("forced priority band failure")
+		}
+		item := h.newTestItem("req-multi-other", testFlow, testTTL)
+		h.processor.enqueue(item)
+		h.waitForFinalization(item) //nolint:errcheck
+
+		assert.Equal(t, uint64(2), h.processor.dropCounts[types.QueueOutcomeRejectedCapacity].Load(),
+			"RejectedCapacity counter should be 2")
+		assert.Equal(t, uint64(1), h.processor.dropCounts[types.QueueOutcomeRejectedOther].Load(),
+			"RejectedOther counter should be 1")
+	})
+
+	t.Run("shutdown eviction counts unfinalized queued items exactly once", func(t *testing.T) {
+		t.Parallel()
+		h := newTestHarness(t, testCleanupTick)
+		mockQueue := h.addQueue(testFlow)
+
+		item := h.newTestItem("req-evict-shutdown", testFlow, testTTL)
+		require.NoError(t, mockQueue.Add(item))
+		require.Nil(t, item.FinalState(), "item must not be finalized before evictAll")
+
+		// evictAll runs on shutdown to drain all queues.
+		h.processor.evictAll()
+
+		h.waitForFinalization(item) //nolint:errcheck
+
+		assert.Equal(t, uint64(1), h.processor.dropCounts[types.QueueOutcomeEvictedOther].Load(),
+			"evictAll should count the unfinalized queued item exactly once")
+	})
+}

pkg/epp/flowcontrol/registry/managedqueue.go

@@ -146,7 +146,17 @@ func (mq *managedQueue) Cleanup(predicate contracts.PredicateFunc) []flowcontrol
 		return nil
 	}
 	mq.propagateStatsDeltaForRemovedItemsLocked(cleanedItems)
-	mq.logger.V(logging.DEBUG).Info("Cleaned up queue", "removedItemCount", len(cleanedItems))
+	if v := mq.logger.V(logging.DEBUG); v.Enabled() {
+		reqIDs := make([]string, 0, len(cleanedItems))
+		for _, item := range cleanedItems {
+			if req := item.OriginalRequest(); req != nil {
+				reqIDs = append(reqIDs, req.ID())
+			}
+		}
+		v.Info("Cleaned up queue", "removedItemCount", len(cleanedItems), "requestIDs", reqIDs)
+	} else {
+		mq.logger.V(logging.DEBUG).Info("Cleaned up queue", "removedItemCount", len(cleanedItems))
+	}
 	return cleanedItems
 }
 
@@ -160,7 +170,17 @@ func (mq *managedQueue) Drain() []flowcontrol.QueueItemAccessor {
 		return nil
 	}
 	mq.propagateStatsDeltaForRemovedItemsLocked(drainedItems)
-	mq.logger.V(logging.DEBUG).Info("Drained queue", "itemCount", len(drainedItems))
+	if v := mq.logger.V(logging.DEBUG); v.Enabled() {
+		reqIDs := make([]string, 0, len(drainedItems))
+		for _, item := range drainedItems {
+			if req := item.OriginalRequest(); req != nil {
+				reqIDs = append(reqIDs, req.ID())
+			}
+		}
+		v.Info("Drained queue", "itemCount", len(drainedItems), "requestIDs", reqIDs)
+	} else {
+		mq.logger.V(logging.DEBUG).Info("Drained queue", "itemCount", len(drainedItems))
+	}
 	return drainedItems
 }
 

pkg/epp/flowcontrol/types/outcomes.go

@@ -71,6 +71,11 @@ const (
 	// The specific underlying cause can be determined from the associated error (e.g., controller shutdown while the item
 	// was queued), which will be wrapped by `ErrEvicted`.
 	QueueOutcomeEvictedOther
+
+	// NumQueueOutcomes is a sentinel that equals the total number of QueueOutcome values.
+	// It is not a valid outcome; it exists to size arrays indexed by QueueOutcome and to allow tests to detect
+	// when new values are added without a corresponding update to dependent code.
+	NumQueueOutcomes
 )
 
 // String returns a human-readable string representation of the QueueOutcome.