Expose cache size as an operational metric

docs/operational-metrics.md

@@ -47,6 +47,14 @@ Each table below provides documentation for an exported flowlogs-pipeline operat
 | **Labels** | error, metric, key | 
 
 
+### encode_prom_metrics_reported
+| **Name** | encode_prom_metrics_reported | 
+|:---|:---|
+| **Description** | Total number of prometheus metrics reported by this stage | 
+| **Type** | gauge | 
+| **Labels** | stage | 
+
+
 ### ingest_batch_size_bytes
 | **Name** | ingest_batch_size_bytes | 
 |:---|:---|

pkg/pipeline/encode/encode_prom.go

@@ -56,6 +56,7 @@ type EncodeProm struct {
 	aggHistos        []histoInfo
 	expiryTime       time.Duration
 	mCache           *putils.TimedCache
+	mChacheLenMetric prometheus.Gauge
 	exitChan         <-chan struct{}
 	metricsProcessed prometheus.Counter
 	metricsDropped   prometheus.Counter
@@ -81,6 +82,12 @@ var (
 		operational.TypeCounter,
 		"error", "metric", "key",
 	)
+	mChacheLen = operational.DefineMetric(
+		"encode_prom_metrics_reported",
+		"Total number of prometheus metrics reported by this stage",
+		operational.TypeGauge,
+		"stage",
+	)
 )
 
 // Encode encodes a metric before being stored
@@ -334,13 +341,16 @@ func NewEncodeProm(opMetrics *operational.Metrics, params config.StageParam) (En
 	log.Debugf("histos = %v", histos)
 	log.Debugf("aggHistos = %v", aggHistos)
 
+	mChacheLenMetric := opMetrics.NewGauge(&mChacheLen, params.Name)
+
 	w := &EncodeProm{
 		counters:         counters,
 		gauges:           gauges,
 		histos:           histos,
 		aggHistos:        aggHistos,
 		expiryTime:       expiryTime,
-		mCache:           putils.NewTimedCache(cfg.MaxMetrics),
+		mCache:           putils.NewTimedCache(cfg.MaxMetrics, mChacheLenMetric),
+		mChacheLenMetric: mChacheLenMetric,
 		exitChan:         putils.ExitChannel(),
 		metricsProcessed: opMetrics.NewCounter(&metricsProcessed, params.Name),
 		metricsDropped:   opMetrics.NewCounter(&metricsDropped, params.Name),

pkg/pipeline/extract/aggregate/aggregate_test.go

@@ -37,7 +37,7 @@ func GetMockAggregate() Aggregate {
 			OperationType: "avg",
 			OperationKey:  "value",
 		},
-		cache:      utils.NewTimedCache(0),
+		cache:      utils.NewTimedCache(0, nil),
 		mutex:      &sync.Mutex{},
 		expiryTime: 30 * time.Second,
 	}

pkg/pipeline/extract/aggregate/aggregates.go

@@ -61,7 +61,7 @@ func (aggregates *Aggregates) GetMetrics() []config.GenericMap {
 func (aggregates *Aggregates) AddAggregate(aggregateDefinition api.AggregateDefinition) []Aggregate {
 	aggregate := Aggregate{
 		Definition: aggregateDefinition,
-		cache:      utils.NewTimedCache(0),
+		cache:      utils.NewTimedCache(0, nil),
 		mutex:      &sync.Mutex{},
 		expiryTime: aggregates.expiryTime,
 	}

pkg/pipeline/utils/timed_cache.go

@@ -22,6 +22,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/sirupsen/logrus"
 )
 
@@ -45,10 +46,11 @@ type TimedCacheMap map[string]*cacheEntry
 // If maxEntries is non-zero, this limits the number of entries in the cache to the number specified.
 // If maxEntries is zero, the cache has no size limit.
 type TimedCache struct {
-	mu         sync.RWMutex
-	cacheList  *list.List
-	cacheMap   TimedCacheMap
-	maxEntries int
+	mu             sync.RWMutex
+	cacheList      *list.List
+	cacheMap       TimedCacheMap
+	maxEntries     int
+	cacheLenMetric prometheus.Gauge
 }
 
 func (tc *TimedCache) GetCacheEntry(key string) (interface{}, bool) {
@@ -90,6 +92,9 @@ func (tc *TimedCache) UpdateCacheEntry(key string, entry interface{}) (*cacheEnt
 		// place at end of list
 		cEntry.e = tc.cacheList.PushBack(cEntry)
 		tc.cacheMap[key] = cEntry
+		if tc.cacheLenMetric != nil {
+			tc.cacheLenMetric.Inc()
+		}
 	}
 	return cEntry, true
 }
@@ -140,14 +145,18 @@ func (tc *TimedCache) CleanupExpiredEntries(expiry time.Duration, callback Cache
 		callback(pCacheInfo.SourceEntry)
 		delete(tc.cacheMap, pCacheInfo.key)
 		tc.cacheList.Remove(listEntry)
+		if tc.cacheLenMetric != nil {
+			tc.cacheLenMetric.Dec()
+		}
 	}
 }
 
-func NewTimedCache(maxEntries int) *TimedCache {
+func NewTimedCache(maxEntries int, cacheLenMetric prometheus.Gauge) *TimedCache {
 	l := &TimedCache{
-		cacheList:  list.New(),
-		cacheMap:   make(TimedCacheMap),
-		maxEntries: maxEntries,
+		cacheList:      list.New(),
+		cacheMap:       make(TimedCacheMap),
+		maxEntries:     maxEntries,
+		cacheLenMetric: cacheLenMetric,
 	}
 	return l
 }