feat: flag to enable gauge replacement

Author: mcfedr

cmd/root.go

@@ -26,6 +26,7 @@ func Execute() {
 	rootCmd.PersistentFlags().StringVar(&cfg.ApiListen, "apiListen", ":80", "Listen for API requests on this host/port.")
 	rootCmd.PersistentFlags().StringVar(&cfg.LifecycleListen, "lifecycleListen", ":8888", "Listen for lifecycle requests (health, metrics) on this host/port")
 	rootCmd.PersistentFlags().StringVar(&cfg.CorsDomain, "cors", "*", "The 'Access-Control-Allow-Origin' value to be returned.")
+	rootCmd.PersistentFlags().StringVar(&cfg.GaugeBehavior, "gaugeBehavior", "sum", "How gauges are aggregated.")
 
 	if err := rootCmd.Execute(); err != nil {
 		os.Exit(1)

cmd/start.go

@@ -23,7 +23,7 @@ func startFunc(cmd *cobra.Command, args []string) error {
 		Accounts:   cfg.AuthUsers,
 	}
 
-	routers.RunServers(apiCfg, cfg.ApiListen, cfg.LifecycleListen)
+	routers.RunServers(apiCfg, cfg.ApiListen, cfg.LifecycleListen, cfg.GaugeBehavior)
 
 	return nil
 }

config/config.go

@@ -20,6 +20,7 @@ type Server struct {
 	LifecycleListen string
 	CorsDomain      string
 	AuthUsers       []string
+	GaugeBehavior   string
 }
 
 const (

metrics/aggregate.go

@@ -17,7 +17,8 @@ import (
 
 type metricFamily struct {
 	*dto.MetricFamily
-	lock sync.RWMutex
+	lock    sync.RWMutex
+	options *aggregateOptions
 }
 
 type Aggregate struct {
@@ -28,9 +29,17 @@ type Aggregate struct {
 
 type ignoredLabels []string
 
+type gaugeBehavior string
+
+const (
+	sumBehavior     gaugeBehavior = "sum"
+	replaceBehavior               = "replace"
+)
+
 type aggregateOptions struct {
 	ignoredLabels     ignoredLabels
 	metricTTLDuration *time.Duration
+	gaugeBehavior     gaugeBehavior
 }
 
 type aggregateOptionsFunc func(a *Aggregate)
@@ -47,6 +56,16 @@ func SetTTLMetricTime(duration *time.Duration) aggregateOptionsFunc {
 	}
 }
 
+func SetGaugeBehavior(behavior string) aggregateOptionsFunc {
+	return func(a *Aggregate) {
+		if behavior == replaceBehavior {
+			a.options.gaugeBehavior = replaceBehavior
+		} else {
+			a.options.gaugeBehavior = sumBehavior
+		}
+	}
+}
+
 func NewAggregate(opts ...aggregateOptionsFunc) *Aggregate {
 	a := &Aggregate{
 		families: map[string]*metricFamily{},
@@ -91,7 +110,7 @@ func (a *Aggregate) setFamilyOrGetExistingFamily(familyName string, family *dto.
 	defer a.familiesLock.Unlock()
 	existingFamily, ok := a.families[familyName]
 	if !ok {
-		a.families[familyName] = &metricFamily{MetricFamily: family}
+		a.families[familyName] = &metricFamily{MetricFamily: family, options: &a.options}
 		return nil
 	}
 	return existingFamily

metrics/merge.go

@@ -71,7 +71,7 @@ func mergeBuckets(a, b []*dto.Bucket) []*dto.Bucket {
 	return output
 }
 
-func mergeMetric(ty dto.MetricType, a, b *dto.Metric) *dto.Metric {
+func mergeMetric(ty dto.MetricType, a, b *dto.Metric, options *aggregateOptions) *dto.Metric {
 	switch ty {
 	case dto.MetricType_COUNTER:
 		return &dto.Metric{
@@ -82,9 +82,14 @@ func mergeMetric(ty dto.MetricType, a, b *dto.Metric) *dto.Metric {
 		}
 
 	case dto.MetricType_GAUGE:
-		// No very meaningful way for us to merge gauges.  We'll sum them
-		// and clear out any gauges on scrape, as a best approximation, but
-		// this relies on client pushing with the same interval as we scrape.
+		if options.gaugeBehavior == replaceBehavior {
+			return &dto.Metric{
+				Label: a.Label,
+				Gauge: &dto.Gauge{
+					Value: float64ptr(*b.Gauge.Value),
+				},
+			}
+		}
 		return &dto.Metric{
 			Label: a.Label,
 			Gauge: &dto.Gauge{
@@ -143,7 +148,7 @@ func (mf *metricFamily) mergeFamily(b *dto.MetricFamily) error {
 			newMetric = append(newMetric, b.Metric[j])
 			j++
 		} else {
-			merged := mergeMetric(*mf.Type, mf.Metric[i], b.Metric[j])
+			merged := mergeMetric(*mf.Type, mf.Metric[i], b.Metric[j], mf.options)
 			if merged != nil {
 				newMetric = append(newMetric, merged)
 			}

routers/server.go

@@ -11,11 +11,11 @@ import (
 	"github.com/zapier/prom-aggregation-gateway/metrics"
 )
 
-func RunServers(cfg ApiRouterConfig, apiListen string, lifecycleListen string) {
+func RunServers(cfg ApiRouterConfig, apiListen string, lifecycleListen string, gaugeBehavior string) {
 	sigChannel := make(chan os.Signal, 1)
 	signal.Notify(sigChannel, syscall.SIGTERM, syscall.SIGINT)
 
-	agg := metrics.NewAggregate()
+	agg := metrics.NewAggregate(metrics.SetGaugeBehavior(gaugeBehavior))
 
 	promMetricsConfig := promMetrics.Config{
 		Registry: metrics.PromRegistry,