Add garbage collection support to the Prometheus metrics adjuster (#613)

* Add cleanup of job map for jobs that have gone away

* Add timeseries-level gc to the metrics_adjuster

* Fix typo in comment

* Update job and ts GC strategy to a standard mark-an-sweep approach

* Expand comments on GC alternatives

* Update code to use 'defer Unlock' where possible

* Add comments about double-checking synchronized structures to verify that gc() is necessary

Author: dinooliva

receiver/prometheusreceiver/internal/metrics_adjuster.go

@@ -6,33 +6,99 @@ import (
 	"github.com/golang/protobuf/ptypes/wrappers"
 	"go.uber.org/zap"
 	"strings"
+	"sync"
+	"time"
 )
 
+// Notes on garbage collection (gc):
+//
+// Job-level gc:
+// The Prometheus receiver will likely execute in a long running service whose lifetime may exceed
+// the lifetimes of many of the jobs that it is collecting from. In order to keep the JobsMap from
+// leaking memory for entries of no-longer existing jobs, the JobsMap needs to remove entries that
+// haven't been accessed for a long period of time.
+//
+// Timeseries-level gc:
+// Some jobs that the Prometheus receiver is collecting from may export timeseries based on metrics
+// from other jobs (e.g. cAdvisor). In order to keep the timeseriesMap from leaking memory for entries
+// of no-longer existing jobs, the timeseriesMap for each job needs to remove entries that haven't
+// been accessed for a long period of time.
+//
+// The gc strategy uses a standard mark-and-sweep approach - each time a timeseriesMap is accessed,
+// it is marked. Similarly, each time a timeseriesinfo is accessed, it is also marked.
+//
+// At the end of each JobsMap.get(), if the last time the JobsMap was gc'd exceeds the 'gcInterval',
+// the JobsMap is locked and any timeseriesMaps that are unmarked are removed from the JobsMap
+// otherwise the timeseriesMap is gc'd
+//
+// The gc for the timeseriesMap is straightforward - the map is locked and, for each timeseriesinfo
+// in the map, if it has not been marked, it is removed otherwise it is unmarked.
+//
+// Alternative Strategies
+// 1. If the job-level gc doesn't run often enough, or runs too often, a separate go routine can
+//    be spawned at JobMap creation time that gc's at periodic intervals. This approach potentially
+//    adds more contention and latency to each scrape so the current approach is used. Note that
+//    the go routine will need to be cancelled upon StopMetricsReception().
+// 2. If the gc of each timeseriesMap during the gc of the JobsMap causes too much contention,
+//    the gc of timeseriesMaps can be moved to the end of MetricsAdjuster().AdjustMetrics(). This
+//    approach requires adding 'lastGC' Time and (potentially) a gcInterval duration to
+//    timeseriesMap so the current approach is used instead.
+
+// timeseriesinfo contains the information necessary to adjust from the initial point and to detect
+// resets.
 type timeseriesinfo struct {
+	mark     bool
 	initial  *metricspb.TimeSeries
 	previous *metricspb.TimeSeries
 }
 
-type metricsInstanceMap map[string]*timeseriesinfo
-
-func newMetricsInstanceMap() *metricsInstanceMap {
-	mim := metricsInstanceMap(make(map[string]*timeseriesinfo))
-	return &mim
+// timeseriesMap maps from a timeseries instance (metric * label values) to the timeseries info for
+// the instance.
+type timeseriesMap struct {
+	sync.RWMutex
+	mark   bool
+	tsiMap map[string]*timeseriesinfo
 }
 
-func (mim *metricsInstanceMap) get(metric *metricspb.Metric, values []*metricspb.LabelValue) *timeseriesinfo {
+// Get the timeseriesinfo for the timeseries associated with the metric and label values.
+func (tsm *timeseriesMap) get(
+	metric *metricspb.Metric, values []*metricspb.LabelValue) *timeseriesinfo {
 	name := metric.GetMetricDescriptor().GetName()
-	sig := getSignature(name, values)
-	tsi, ok := (*mim)[sig]
+	sig := getTimeseriesSignature(name, values)
+	tsi, ok := tsm.tsiMap[sig]
 	if !ok {
 		tsi = &timeseriesinfo{}
-		(*mim)[sig] = tsi
+		tsm.tsiMap[sig] = tsi
 	}
+	tsm.mark = true
+	tsi.mark = true
 	return tsi
 }
 
-// create a unique signature consisting of a metric's name and label values
-func getSignature(name string, values []*metricspb.LabelValue) string {
+// Remove timeseries that have aged out.
+func (tsm *timeseriesMap) gc() {
+	tsm.Lock()
+	defer tsm.Unlock()
+	// this shouldn't happen under the current gc() strategy
+	if !tsm.mark {
+		return
+	}
+	for ts, tsi := range tsm.tsiMap {
+		if !tsi.mark {
+			delete(tsm.tsiMap, ts)
+		} else {
+			tsi.mark = false
+		}
+	}
+	tsm.mark = false
+}
+
+func newTimeseriesMap() *timeseriesMap {
+	return &timeseriesMap{mark: true, tsiMap: map[string]*timeseriesinfo{}}
+}
+
+// Create a unique timeseries signature consisting of the metric name and label values.
+func getTimeseriesSignature(name string, values []*metricspb.LabelValue) string {
 	labelValues := make([]string, 0, len(values))
 	for _, label := range values {
 		if label.GetValue() != "" {
@@ -42,46 +108,86 @@ func getSignature(name string, values []*metricspb.LabelValue) string {
 	return fmt.Sprintf("%s,%s", name, strings.Join(labelValues, ","))
 }
 
-// JobsMap maps from a job instance to a map of metric instances for the job.
-type JobsMap map[string]*metricsInstanceMap
+// JobsMap maps from a job instance to a map of timeseries instances for the job.
+type JobsMap struct {
+	sync.RWMutex
+	gcInterval time.Duration
+	lastGC     time.Time
+	jobsMap    map[string]*timeseriesMap
+}
 
 // NewJobsMap creates a new (empty) JobsMap.
-func NewJobsMap() *JobsMap {
-	jm := JobsMap(make(map[string]*metricsInstanceMap))
-	return &jm
+func NewJobsMap(gcInterval time.Duration) *JobsMap {
+	return &JobsMap{gcInterval: gcInterval, lastGC: time.Now(), jobsMap: make(map[string]*timeseriesMap)}
 }
 
-func (jm *JobsMap) get(job, instance string) *metricsInstanceMap {
+// Remove jobs and timeseries that have aged out.
+func (jm *JobsMap) gc() {
+	jm.Lock()
+	defer jm.Unlock()
+	// once the structure is locked, confrim that gc() is still necessary
+	if time.Now().Sub(jm.lastGC) > jm.gcInterval {
+		for sig, tsm := range jm.jobsMap {
+			if !tsm.mark {
+				delete(jm.jobsMap, sig)
+			} else {
+				tsm.gc()
+			}
+		}
+		jm.lastGC = time.Now()
+	}
+}
+
+func (jm *JobsMap) maybeGC() {
+	// speculatively check if gc() is necessary, recheck once the structure is locked
+	if time.Now().Sub(jm.lastGC) > jm.gcInterval {
+		go jm.gc()
+	}
+}
+
+func (jm *JobsMap) get(job, instance string) *timeseriesMap {
 	sig := job + ":" + instance
-	metricsMap, ok := (*jm)[sig]
-	if !ok {
-		metricsMap = newMetricsInstanceMap()
-		(*jm)[sig] = metricsMap
+	jm.RLock()
+	tsm, ok := jm.jobsMap[sig]
+	jm.RUnlock()
+	defer jm.maybeGC()
+	if ok {
+		return tsm
+	}
+	jm.Lock()
+	defer jm.Unlock()
+	tsm2, ok2 := jm.jobsMap[sig]
+	if ok2 {
+		return tsm2
 	}
-	return metricsMap
+	tsm2 = newTimeseriesMap()
+	jm.jobsMap[sig] = tsm2
+	return tsm2
 }
 
 // MetricsAdjuster takes a map from a metric instance to the initial point in the metrics instance
 // and provides AdjustMetrics, which takes a sequence of metrics and adjust their values based on
 // the initial points.
 type MetricsAdjuster struct {
-	metricsMap *metricsInstanceMap
-	logger     *zap.SugaredLogger
+	tsm    *timeseriesMap
+	logger *zap.SugaredLogger
 }
 
 // NewMetricsAdjuster is a constructor for MetricsAdjuster.
-func NewMetricsAdjuster(metricsMap *metricsInstanceMap, logger *zap.SugaredLogger) *MetricsAdjuster {
+func NewMetricsAdjuster(tsm *timeseriesMap, logger *zap.SugaredLogger) *MetricsAdjuster {
 	return &MetricsAdjuster{
-		metricsMap: metricsMap,
-		logger:     logger,
+		tsm:    tsm,
+		logger: logger,
 	}
 }
 
-// AdjustMetrics takes a sequence of metrics and adjust their values based on the initial points in the
-// metricsMap. If the metric is the first point in the timeseries, or the timeseries has been reset, it is
-// removed from the sequence and added to the the metricsMap.
+// AdjustMetrics takes a sequence of metrics and adjust their values based on the initial and
+// previous points in the timeseriesMap. If the metric is the first point in the timeseries, or the
+// timeseries has been reset, it is removed from the sequence and added to the the timeseriesMap.
 func (ma *MetricsAdjuster) AdjustMetrics(metrics []*metricspb.Metric) []*metricspb.Metric {
 	var adjusted = make([]*metricspb.Metric, 0, len(metrics))
+	ma.tsm.Lock()
+	defer ma.tsm.Unlock()
 	for _, metric := range metrics {
 		if ma.adjustMetric(metric) {
 			adjusted = append(adjusted, metric)
@@ -90,7 +196,9 @@ func (ma *MetricsAdjuster) AdjustMetrics(metrics []*metricspb.Metric) []*metrics
 	return adjusted
 }
 
-// returns true if at least one of the metric's timeseries was adjusted and false if all of the timeseries are an initial occurence or a reset.
+// Returns true if at least one of the metric's timeseries was adjusted and false if all of the
+// timeseries are an initial occurence or a reset.
+//
 // Types of metrics returned supported by prometheus:
 // - MetricDescriptor_GAUGE_DOUBLE
 // - MetricDescriptor_GAUGE_DISTRIBUTION
@@ -107,17 +215,19 @@ func (ma *MetricsAdjuster) adjustMetric(metric *metricspb.Metric) bool {
 	}
 }
 
-// Returns true if at least one of the metric's timeseries was adjusted and false if all of the timeseries are an initial occurence or a reset.
+// Returns true if at least one of the metric's timeseries was adjusted and false if all of the
+// timeseries are an initial occurence or a reset.
 func (ma *MetricsAdjuster) adjustMetricTimeseries(metric *metricspb.Metric) bool {
 	filtered := make([]*metricspb.TimeSeries, 0, len(metric.GetTimeseries()))
 	for _, current := range metric.GetTimeseries() {
-		tsi := ma.metricsMap.get(metric, current.GetLabelValues())
+		tsi := ma.tsm.get(metric, current.GetLabelValues())
 		if tsi.initial == nil {
 			// initial timeseries
 			tsi.initial = current
 			tsi.previous = current
 		} else {
-			if ma.adjustTimeseries(metric.MetricDescriptor.Type, current, tsi.initial, tsi.previous) {
+			if ma.adjustTimeseries(metric.MetricDescriptor.Type, current, tsi.initial,
+				tsi.previous) {
 				tsi.previous = current
 				filtered = append(filtered, current)
 			} else {
@@ -131,29 +241,38 @@ func (ma *MetricsAdjuster) adjustMetricTimeseries(metric *metricspb.Metric) bool
 	return len(filtered) > 0
 }
 
-// returns true if 'current' was adjusted and false if 'current' is an the initial occurence or a reset of the timeseries.
-func (ma *MetricsAdjuster) adjustTimeseries(metricType metricspb.MetricDescriptor_Type, current, initial, previous *metricspb.TimeSeries) bool {
-	if !ma.adjustPoints(metricType, current.GetPoints(), initial.GetPoints(), previous.GetPoints()) {
+// Returns true if 'current' was adjusted and false if 'current' is an the initial occurence or a
+// reset of the timeseries.
+func (ma *MetricsAdjuster) adjustTimeseries(metricType metricspb.MetricDescriptor_Type,
+	current, initial, previous *metricspb.TimeSeries) bool {
+	if !ma.adjustPoints(
+		metricType, current.GetPoints(), initial.GetPoints(), previous.GetPoints()) {
 		return false
 	}
 	current.StartTimestamp = initial.StartTimestamp
 	return true
 }
 
-func (ma *MetricsAdjuster) adjustPoints(metricType metricspb.MetricDescriptor_Type, current, initial, previous []*metricspb.Point) bool {
+func (ma *MetricsAdjuster) adjustPoints(metricType metricspb.MetricDescriptor_Type,
+	current, initial, previous []*metricspb.Point) bool {
 	if len(current) != 1 || len(initial) != 1 || len(current) != 1 {
-		ma.logger.Infof("len(current): %v, len(initial): %v, len(previous): %v should all be 1", len(current), len(initial), len(previous))
+		ma.logger.Infof(
+			"len(current): %v, len(initial): %v, len(previous): %v should all be 1",
+			len(current), len(initial), len(previous))
 		return true
 	}
 	return ma.adjustPoint(metricType, current[0], initial[0], previous[0])
 }
 
-// Note: There is an important, subtle point here. When a new timeseries or a reset is detected, current and initial are the same object.
-// When initial == previous, the previous value/count/sum are all the initial value. When initial != previous, the previous value/count/sum has
-// been adjusted wrt the initial value so both they must be combined to find the actual previous value/count/sum. This happens because the
-// timeseries are updated in-place - if new copies of the timeseries were created instead, previous could be used directly but this would
-// mean reallocating all of the metrics.
-func (ma *MetricsAdjuster) adjustPoint(metricType metricspb.MetricDescriptor_Type, current, initial, previous *metricspb.Point) bool {
+// Note: There is an important, subtle point here. When a new timeseries or a reset is detected,
+// current and initial are the same object. When initial == previous, the previous value/count/sum
+// are all the initial value. When initial != previous, the previous value/count/sum has been
+// adjusted wrt the initial value so both they must be combined to find the actual previous
+// value/count/sum. This happens because the timeseries are updated in-place - if new copies of the
+// timeseries were created instead, previous could be used directly but this would mean reallocating
+// all of the metrics.
+func (ma *MetricsAdjuster) adjustPoint(metricType metricspb.MetricDescriptor_Type,
+	current, initial, previous *metricspb.Point) bool {
 	switch metricType {
 	case metricspb.MetricDescriptor_CUMULATIVE_DOUBLE:
 		currentValue := current.GetDoubleValue()
@@ -166,7 +285,8 @@ func (ma *MetricsAdjuster) adjustPoint(metricType metricspb.MetricDescriptor_Typ
 			// reset detected
 			return false
 		}
-		current.Value = &metricspb.Point_DoubleValue{DoubleValue: currentValue - initialValue}
+		current.Value =
+			&metricspb.Point_DoubleValue{DoubleValue: currentValue - initialValue}
 	case metricspb.MetricDescriptor_CUMULATIVE_DISTRIBUTION:
 		// note: sum of squared deviation not currently supported
 		currentDist := current.GetDistributionValue()
@@ -200,8 +320,10 @@ func (ma *MetricsAdjuster) adjustPoint(metricType metricspb.MetricDescriptor_Typ
 			// reset detected
 			return false
 		}
-		current.GetSummaryValue().Count = &wrappers.Int64Value{Value: currentCount - initialCount}
-		current.GetSummaryValue().Sum = &wrappers.DoubleValue{Value: currentSum - initialSum}
+		current.GetSummaryValue().Count =
+			&wrappers.Int64Value{Value: currentCount - initialCount}
+		current.GetSummaryValue().Sum =
+			&wrappers.DoubleValue{Value: currentSum - initialSum}
 	default:
 		// this shouldn't happen
 		ma.logger.Infof("adjust unexpect point type %v, skipping ...", metricType.String())
@@ -212,7 +334,8 @@ func (ma *MetricsAdjuster) adjustPoint(metricType metricspb.MetricDescriptor_Typ
 func (ma *MetricsAdjuster) adjustBuckets(current, initial []*metricspb.DistributionValue_Bucket) {
 	if len(current) != len(initial) {
 		// this shouldn't happen
-		ma.logger.Infof("len(current buckets): %v != len(initial buckets): %v", len(current), len(initial))
+		ma.logger.Infof("len(current buckets): %v != len(initial buckets): %v",
+			len(current), len(initial))
 	}
 	for i := 0; i < len(current); i++ {
 		current[i].Count -= initial[i].Count