add worker_cache_entries

Author: VinozzZ

collect/cache/cache.go

@@ -44,29 +44,39 @@ var _ Cache = (*DefaultInMemCache)(nil)
 // order) so it is important to have a cache larger than trace throughput *
 // longest trace.
 type DefaultInMemCache struct {
-	Logger logger.Logger
+	Metrics metrics.Metrics
+	Logger  logger.Logger
 
 	pq    *kpq.KeyedPriorityQueue[string, time.Time]
 	cache map[string]*types.Trace
 }
 
 const DefaultInMemCacheCapacity = 10000
 
+var collectCacheMetrics = []metrics.Metadata{
+	{Name: "worker_cache_entries", Type: metrics.Histogram, Unit: metrics.Dimensionless, Description: "The number of traces currently stored in the trace cache per worker"},
+}
+
 func NewInMemCache(
 	met metrics.Metrics,
 	logger logger.Logger,
 ) *DefaultInMemCache {
 	logger.Debug().Logf("Starting DefaultInMemCache")
 	defer func() { logger.Debug().Logf("Finished starting DefaultInMemCache") }()
 
+	for _, metadata := range collectCacheMetrics {
+		met.Register(metadata)
+	}
+
 	cmp := func(v1, v2 time.Time) bool {
 		return v1.Before(v2)
 	}
 
 	return &DefaultInMemCache{
-		Logger: logger,
-		pq:     kpq.NewKeyedPriorityQueue[string](cmp),
-		cache:  make(map[string]*types.Trace),
+		Metrics: met,
+		Logger:  logger,
+		pq:      kpq.NewKeyedPriorityQueue[string](cmp),
+		cache:   make(map[string]*types.Trace),
 	}
 }
 
@@ -104,6 +114,8 @@ func (d *DefaultInMemCache) GetCacheCapacity() int {
 // It removes and returns them.
 // If a filter is provided, it will be called with each trace to determine if it should be skipped.
 func (d *DefaultInMemCache) TakeExpiredTraces(now time.Time, max int, filter func(*types.Trace) bool) []*types.Trace {
+	d.Metrics.Histogram("worker_cache_entries", float64(len(d.cache)))
+
 	var expired, skipped []*types.Trace
 	for !d.pq.IsEmpty() && (max <= 0 || len(expired) < max) {
 		// pop the the next trace from the queue
@@ -146,6 +158,8 @@ func (d *DefaultInMemCache) TakeExpiredTraces(now time.Time, max int, filter fun
 // RemoveTraces accepts a set of trace IDs and removes any matching ones from
 // the insertion list. This is used in the case of a cache overrun.
 func (d *DefaultInMemCache) RemoveTraces(toDelete generics.Set[string]) {
+	d.Metrics.Histogram("worker_cache_entries", float64(len(d.cache)))
+
 	for _, traceID := range toDelete.Members() {
 		delete(d.cache, traceID)
 		d.pq.Remove(traceID)

collect/collect.go

@@ -130,7 +130,7 @@ var inMemCollectorMetrics = []metrics.Metadata{
 	{Name: "collector_incoming_queue_length", Type: metrics.Gauge, Unit: metrics.Dimensionless, Description: "number of spans in the incoming queue"},
 	{Name: "collector_peer_queue", Type: metrics.Histogram, Unit: metrics.Dimensionless, Description: "number of spans currently in the peer queue"},
 	{Name: "collector_cache_size", Type: metrics.Gauge, Unit: metrics.Dimensionless, Description: "number of traces currently stored in the trace cache"},
-	{Name: "collect_cache_entries", Type: metrics.Histogram, Unit: metrics.Dimensionless, Description: "The number of traces currently stored in the cache"},
+	{Name: "collect_cache_entries", Type: metrics.Histogram, Unit: metrics.Dimensionless, Description: "Total number of traces currently stored in the cache from all workers"},
 	{Name: "memory_heap_allocation", Type: metrics.Gauge, Unit: metrics.Bytes, Description: "current heap allocation"},
 	{Name: "span_received", Type: metrics.Counter, Unit: metrics.Dimensionless, Description: "number of spans received by the collector"},
 	{Name: "span_processed", Type: metrics.Counter, Unit: metrics.Dimensionless, Description: "number of spans processed by the collector"},