delete old prometheus entries from report

Author: KalmanMeth

docs/api.md

@@ -15,6 +15,7 @@ Following is the supported API format for prometheus encode:
                  buckets: histogram buckets
          port: port number to expose "/metrics" endpoint
          prefix: prefix added to each metric name
+         expirytime: seconds of no-flow to wait before deleting prometheus data item
 </pre>
 ## Ingest collector API
 Following is the supported API format for the netflow collector:

pkg/api/encode_prom.go

@@ -18,9 +18,10 @@
 package api
 
 type PromEncode struct {
-	Metrics PromMetricsItems `yaml:"metrics" doc:"list of prometheus metric definitions, each includes:"`
-	Port    int              `yaml:"port" doc:"port number to expose \"/metrics\" endpoint"`
-	Prefix  string           `yaml:"prefix" doc:"prefix added to each metric name"`
+	Metrics    PromMetricsItems `yaml:"metrics" doc:"list of prometheus metric definitions, each includes:"`
+	Port       int              `yaml:"port" doc:"port number to expose \"/metrics\" endpoint"`
+	Prefix     string           `yaml:"prefix" doc:"prefix added to each metric name"`
+	ExpiryTime int              `yaml:"expirytime" doc:"seconds of no-flow to wait before deleting prometheus data item"`
 }
 
 type PromEncodeOperationEnum struct {

pkg/pipeline/encode/encode_prom.go

@@ -18,6 +18,7 @@
 package encode
 
 import (
+	"container/list"
 	"encoding/json"
 	"fmt"
 	"github.com/netobserv/flowlogs2metrics/pkg/api"
@@ -28,46 +29,54 @@ import (
 	"net/http"
 	"os"
 	"strconv"
+	"sync"
+	"time"
 )
 
-// TODO when do we clean up old entries?
+const defaultExpiryTime = 120
 
-type gaugeInfo struct {
-	input     string
-	tags      []string
-	promGauge *prometheus.GaugeVec
-}
-
-type counterInfo struct {
+type genericMetricInfo struct {
 	input       string
 	tags        []string
+	metricType  string
+	promGauge   *prometheus.GaugeVec
 	promCounter *prometheus.CounterVec
+	promHist    *prometheus.HistogramVec
 }
 
-type histInfo struct {
-	input    string
-	tags     []string
-	promHist *prometheus.HistogramVec
+type entrySignature struct {
+	Name   string
+	Labels map[string]string
 }
 
-type counterEntryInfo struct {
-	counterName  string
-	counterValue float64
-	labels       map[string]string
+type entryInfo struct {
+	eInfo entrySignature
+	value float64
 }
 
-type gaugeEntryInfo struct {
-	gaugeName  string
-	gaugeValue float64
-	labels     map[string]string
+type metricCacheEntry struct {
+	label       prometheus.Labels
+	timeStamp   int64
+	e           *list.Element
+	key         string
+	metricType  string
+	promGauge   *prometheus.GaugeVec
+	promCounter *prometheus.CounterVec
+	promHist    *prometheus.HistogramVec
 }
 
+type metricCache map[string]*metricCacheEntry
+
 type encodeProm struct {
+	mu         sync.Mutex
 	port       string
 	prefix     string
-	counters   map[string]counterInfo
-	gauges     map[string]gaugeInfo
-	histograms map[string]histInfo
+	counters   map[string]genericMetricInfo
+	gauges     map[string]genericMetricInfo
+	histograms map[string]genericMetricInfo
+	expiryTime int64
+	mList      *list.List
+	mCache     metricCache
 }
 
 func (e *encodeProm) EncodeCounter(metric config.GenericMap) []interface{} {
@@ -88,19 +97,22 @@ func (e *encodeProm) EncodeCounter(metric config.GenericMap) []interface{} {
 		for _, t := range counterInfo.tags {
 			entryLabels[t] = fmt.Sprintf("%v", metric[t])
 		}
-		entry := counterEntryInfo{
-			counterName:  e.prefix + counterName,
-			counterValue: valueFloat,
-			labels:       entryLabels,
+		entry := entryInfo{
+			eInfo: entrySignature{
+				Name:   e.prefix + counterName,
+				Labels: entryLabels,
+			},
+			value: valueFloat,
 		}
 		out = append(out, entry)
 		// push the metric to prometheus
-		// TODO - fix the types here
 		if counterInfo.promCounter != nil {
 			counterInfo.promCounter.With(entryLabels).Add(valueFloat)
 		}
+		cEntry := e.saveEntryInCache(entry, entryLabels)
+		cEntry.metricType = api.PromEncodeOperationName("Counter")
+		cEntry.promCounter = counterInfo.promCounter
 	}
-
 	return out
 }
 func (e *encodeProm) EncodeGauge(metric config.GenericMap) []interface{} {
@@ -121,34 +133,116 @@ func (e *encodeProm) EncodeGauge(metric config.GenericMap) []interface{} {
 		for _, t := range gaugeInfo.tags {
 			entryLabels[t] = fmt.Sprintf("%v", metric[t])
 		}
-		entry := gaugeEntryInfo{
-			gaugeName:  e.prefix + gaugeName,
-			gaugeValue: valueFloat,
-			labels:     entryLabels,
+		entry := entryInfo{
+			eInfo: entrySignature{
+				Name:   e.prefix + gaugeName,
+				Labels: entryLabels,
+			},
+			value: valueFloat,
 		}
 		out = append(out, entry)
 		// push the metric to prometheus
-		// TODO - fix the types here
 		if gaugeInfo.promGauge != nil {
 			gaugeInfo.promGauge.With(entryLabels).Set(valueFloat)
 		}
-	}
 
+		cEntry := e.saveEntryInCache(entry, entryLabels)
+		cEntry.metricType = api.PromEncodeOperationName("Gauge")
+		cEntry.promGauge = gaugeInfo.promGauge
+	}
 	return out
 }
 
+func (e *encodeProm) saveEntryInCache(entry entryInfo, entryLabels map[string]string) *metricCacheEntry {
+	// save item in cache; use eInfo as key to the cache
+	var cEntry *metricCacheEntry
+	secs := time.Now().Unix()
+	eInfoBytes, _ := json.Marshal(&entry.eInfo)
+	eInfoString := string(eInfoBytes)
+	cEntry, ok := e.mCache[eInfoString]
+	if ok {
+		// item already exists in cache; update the element and move to end of list
+		cEntry.timeStamp = secs
+		// move to end of list
+		e.mList.MoveToBack(cEntry.e)
+	} else {
+		// create new entry for cache
+		cEntry = &metricCacheEntry{
+			label:     entryLabels,
+			timeStamp: secs,
+			key:       eInfoString,
+		}
+		// place at end of list
+		log.Debugf("adding entry = %v", cEntry)
+		cEntry.e = e.mList.PushBack(cEntry)
+		e.mCache[eInfoString] = cEntry
+		log.Debugf("mlist = %v", e.mList)
+	}
+	return cEntry
+}
+
 // Encode encodes a flow before being stored
 func (e *encodeProm) Encode(metrics []config.GenericMap) []interface{} {
+	log.Debugf("entering encodeProm Encode")
+	e.mu.Lock()
+	defer e.mu.Unlock()
 	out := make([]interface{}, 0)
 	for _, metric := range metrics {
 		gaugeOut := e.EncodeGauge(metric)
 		out = append(out, gaugeOut...)
 		counterOut := e.EncodeCounter(metric)
 		out = append(out, counterOut...)
 	}
+	log.Debugf("cache = %v", e.mCache)
+	log.Debugf("list = %v", e.mList)
 	return out
 }
 
+func (e *encodeProm) cleanupExpiredEntriesLoop() {
+	for {
+		e.cleanupExpiredEntries()
+		time.Sleep(time.Duration(e.expiryTime) * time.Second)
+	}
+}
+
+// cleanupExpiredEntries - any entry that has expired should be removed from the prometheus reporting and cache
+func (e *encodeProm) cleanupExpiredEntries() {
+	log.Debugf("entering cleanupExpiredEntries")
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	log.Debugf("cache = %v", e.mCache)
+	log.Debugf("list = %v", e.mList)
+	secs := time.Now().Unix()
+	expireTime := secs - e.expiryTime
+	// go through the list until we reach recently used connections
+	for {
+		entry := e.mList.Front()
+		if entry == nil {
+			return
+		}
+		c := entry.Value.(*metricCacheEntry)
+		log.Debugf("timeStamp = %d, expireTime = %d", c.timeStamp, expireTime)
+		log.Debugf("c = %v", c)
+		if c.timeStamp > expireTime {
+			// no more expired items
+			return
+		}
+
+		// clean up the entry
+		log.Debugf("secs = %d, deleting %s", secs, c.label)
+		switch c.metricType {
+		case api.PromEncodeOperationName("Gauge"):
+			c.promGauge.Delete(c.label)
+		case api.PromEncodeOperationName("Counter"):
+			c.promCounter.Delete(c.label)
+		case api.PromEncodeOperationName("Histogram"):
+			c.promHist.Delete(c.label)
+		}
+		delete(e.mCache, c.key)
+		e.mList.Remove(entry)
+	}
+}
+
 // startPrometheusInterface listens for prometheus resource usage requests
 func startPrometheusInterface(w *encodeProm) {
 	log.Debugf("entering startPrometheusInterface")
@@ -176,15 +270,20 @@ func NewEncodeProm() (Encoder, error) {
 
 	portNum := jsonEncodeProm.Port
 	promPrefix := jsonEncodeProm.Prefix
+	expiryTime := int64(jsonEncodeProm.ExpiryTime)
+	if expiryTime == 0 {
+		expiryTime = defaultExpiryTime
+	}
+	log.Debugf("expiryTime = %d", expiryTime)
 
-	counters := make(map[string]counterInfo)
-	gauges := make(map[string]gaugeInfo)
-	histograms := make(map[string]histInfo)
+	counters := make(map[string]genericMetricInfo)
+	gauges := make(map[string]genericMetricInfo)
+	histograms := make(map[string]genericMetricInfo)
 	for _, metricInfo := range jsonEncodeProm.Metrics {
 		fullMetricName := promPrefix + metricInfo.Name
 		labels := metricInfo.Labels
 		log.Debugf("fullMetricName = %v", fullMetricName)
-		log.Debugf("labels = %v", labels)
+		log.Debugf("Labels = %v", labels)
 		switch metricInfo.Type {
 		case api.PromEncodeOperationName("Counter"):
 			counter := prometheus.NewCounterVec(prometheus.CounterOpts{Name: fullMetricName, Help: ""}, labels)
@@ -193,15 +292,25 @@ func NewEncodeProm() (Encoder, error) {
 				log.Errorf("error during prometheus.Register: %v", err)
 				return nil, err
 			}
-			counters[metricInfo.Name] = counterInfo{metricInfo.ValueKey, labels, counter}
+			counters[metricInfo.Name] = genericMetricInfo{
+				input:       metricInfo.ValueKey,
+				tags:        labels,
+				metricType:  api.PromEncodeOperationName("Counter"),
+				promCounter: counter,
+			}
 		case api.PromEncodeOperationName("Gauge"):
 			gauge := prometheus.NewGaugeVec(prometheus.GaugeOpts{Name: fullMetricName, Help: ""}, labels)
 			err := prometheus.Register(gauge)
 			if err != nil {
 				log.Errorf("error during prometheus.Register: %v", err)
 				return nil, err
 			}
-			gauges[metricInfo.Name] = gaugeInfo{metricInfo.ValueKey, labels, gauge}
+			gauges[metricInfo.Name] = genericMetricInfo{
+				input:      metricInfo.ValueKey,
+				tags:       labels,
+				metricType: api.PromEncodeOperationName("Gauge"),
+				promGauge:  gauge,
+			}
 		case api.PromEncodeOperationName("Histogram"):
 			log.Debugf("buckets = %v", metricInfo.Buckets)
 			hist := prometheus.NewHistogramVec(prometheus.HistogramOpts{Name: fullMetricName, Help: "", Buckets: metricInfo.Buckets}, labels)
@@ -210,7 +319,12 @@ func NewEncodeProm() (Encoder, error) {
 				log.Errorf("error during prometheus.Register: %v", err)
 				return nil, err
 			}
-			histograms[metricInfo.Name] = histInfo{metricInfo.ValueKey, labels, hist}
+			histograms[metricInfo.Name] = genericMetricInfo{
+				input:      metricInfo.ValueKey,
+				tags:       labels,
+				metricType: api.PromEncodeOperationName("Histogram"),
+				promHist:   hist,
+			}
 		case "default":
 			log.Errorf("invalid metric type = %v, skipping", metricInfo.Type)
 		}
@@ -222,7 +336,11 @@ func NewEncodeProm() (Encoder, error) {
 		counters:   counters,
 		gauges:     gauges,
 		histograms: histograms,
+		expiryTime: expiryTime,
+		mList:      list.New(),
+		mCache:     make(metricCache),
 	}
 	go startPrometheusInterface(w)
+	go w.cleanupExpiredEntriesLoop()
 	return w, nil
 }