feat: make drop decision queue size configurable

collect/cache/cuckoo.go

@@ -39,7 +39,7 @@ type CuckooTraceChecker struct {
 
 const (
 	// This is how many items can be in the Add Queue before we start blocking on Add.
-	AddQueueDepth = 1000
+	defaultAddQueueDepth = 1000
 	// This is how long we'll sleep between possible lock cycles.
 	AddQueueSleepTime = 100 * time.Microsecond
 )
@@ -50,13 +50,16 @@ var cuckooTraceCheckerMetrics = []metrics.Metadata{
 	{Name: CurrentLoadFactor, Type: metrics.Gauge, Unit: metrics.Percent, Description: "the fraction of slots occupied in the current cuckoo filter"},
 }
 
-func NewCuckooTraceChecker(capacity uint, m metrics.Metrics) *CuckooTraceChecker {
+func NewCuckooTraceChecker(capacity uint, addQueueDepth uint, m metrics.Metrics) *CuckooTraceChecker {
+	if addQueueDepth == 0 {
+		addQueueDepth = defaultAddQueueDepth
+	}
 	c := &CuckooTraceChecker{
 		capacity: capacity,
 		current:  cuckoo.NewFilter(capacity),
 		future:   nil,
 		met:      m,
-		addch:    make(chan string, AddQueueDepth),
+		addch:    make(chan string, defaultAddQueueDepth),
 	}
 	for _, metric := range cuckooTraceCheckerMetrics {
 		m.Register(metric)

config/file_config.go

@@ -402,6 +402,7 @@ type GRPCServerParameters struct {
 type SampleCacheConfig struct {
 	KeptSize          uint     `yaml:"KeptSize" default:"10_000"`
 	DroppedSize       uint     `yaml:"DroppedSize" default:"1_000_000"`
+	DroppedQueueSize  uint     `yaml:"DroppedQueueSize" default: "1000"`
 	SizeCheckInterval Duration `yaml:"SizeCheckInterval" default:"10s"`
 }
 

config/metadata/configMeta.yaml

@@ -1774,6 +1774,16 @@ groups:
           Changing its size with live reload sets a future limit, but does not
           have an immediate effect.
 
+      - name: DroppedQueueSize
+        type: int
+        valuetype: nondefault
+        default: 1000
+        reload: true
+        summary: is the maximum number of in-flight drop decision allowed before adding to the dropped trace cache.
+        description: >
+          The dropped decision queue is used to buffer drop decisions before they are stored in the decision cache.
+          If this queue fills up, then subsequent drop decisions will be discarded.
+
       - name: SizeCheckInterval
         v1group: SampleCacheConfig/SampleCache
         v1name: SizeCheckInterval

route/route.go

@@ -128,6 +128,8 @@ func (r *Router) SetVersion(ver string) {
 var routerMetrics = []metrics.Metadata{
 	{Name: "_router_proxied", Type: metrics.Counter, Unit: metrics.Dimensionless, Description: "the number of events proxied to another refinery"},
 	{Name: "_router_event", Type: metrics.Counter, Unit: metrics.Dimensionless, Description: "the number of events received"},
+	{Name: "_router_batch", Type: metrics.Counter, Unit: metrics.Dimensionless, Description: "the number of batches of events received"},
+	{Name: "_router_otlp", Type: metrics.Counter, Unit: metrics.Dimensionless, Description: "the number of batches of otlp requests received"},
 	{Name: "_router_span", Type: metrics.Counter, Unit: metrics.Dimensionless, Description: "the number of spans received"},
 	{Name: "_router_dropped", Type: metrics.Counter, Unit: metrics.Dimensionless, Description: "the number of events dropped because the channel was full"},
 	{Name: "_router_nonspan", Type: metrics.Counter, Unit: metrics.Dimensionless, Description: "the number of non-span events received"},
@@ -535,6 +537,7 @@ func (router *Router) processOTLPRequest(
 	batches []huskyotlp.Batch,
 	apiKey string,
 	incomingUserAgent string) error {
+	router.Metrics.Increment(router.incomingOrPeer + "_router_otlp")
 
 	var requestID types.RequestIDContextKey
 	apiHost := router.Config.GetHoneycombAPI()
@@ -608,7 +611,6 @@ func (r *Router) processEvent(ev *types.Event, reqID interface{}) error {
 		if processed {
 			if !kept {
 				return nil
-
 			}
 
 			// If the span was kept, we want to generate a probe that we'll forward