Metrics expiry

coder · mtojek · Apr 27, 2023 · Apr 24, 2023 · Apr 24, 2023 · Apr 24, 2023
commit 8df9eeac51cff84e37dd7691cfea2a2ae2cf9808
diff --git a/cli/server.go b/cli/server.go
@@ -724,7 +724,7 @@ func (r *RootCmd) Server(newAPI func(context.Context, *coderd.Options) (*coderd.
 					}
 					defer closeAgentStatsFunc()
 
-					metricsAggregator := prometheusmetrics.NewMetricsAggregator(logger)
+					metricsAggregator := prometheusmetrics.NewMetricsAggregator(logger, 0)
 					cancelMetricsAggregator := metricsAggregator.Run(ctx)
 					defer cancelMetricsAggregator()
 

diff --git a/coderd/prometheusmetrics/aggregator.go b/coderd/prometheusmetrics/aggregator.go
@@ -2,6 +2,7 @@ package prometheusmetrics
 
 import (
 	"context"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"golang.org/x/xerrors"
@@ -22,12 +23,15 @@ const (
 const (
 	sizeCollectCh = 10
 	sizeUpdateCh  = 1024
+
+	defaultMetricsCleanupInterval = 2 * time.Minute
 )
 
 type MetricsAggregator struct {
 	queue []annotatedMetric
 
-	log slog.Logger
+	log                    slog.Logger
+	metricsCleanupInterval time.Duration
 
 	collectCh chan (chan<- prometheus.Metric)
 	updateCh  chan updateRequest
@@ -39,6 +43,8 @@ type updateRequest struct {
 	agentName     string
 
 	metrics []agentsdk.AgentMetric
+
+	timestamp time.Time
 }
 
 type annotatedMetric struct {
@@ -47,13 +53,20 @@ type annotatedMetric struct {
 	username      string
 	workspaceName string
 	agentName     string
+
+	expiryDate time.Time
 }
 
 var _ prometheus.Collector = new(MetricsAggregator)
 
-func NewMetricsAggregator(logger slog.Logger) *MetricsAggregator {
+func NewMetricsAggregator(logger slog.Logger, duration time.Duration) *MetricsAggregator {
+	metricsCleanupInterval := defaultMetricsCleanupInterval
+	if duration > 0 {
+		metricsCleanupInterval = duration
+	}
 	return &MetricsAggregator{
-		log: logger,
+		log:                    logger,
+		metricsCleanupInterval: metricsCleanupInterval,
 
 		collectCh: make(chan (chan<- prometheus.Metric), sizeCollectCh),
 		updateCh:  make(chan updateRequest, sizeUpdateCh),
@@ -64,17 +77,22 @@ func (ma *MetricsAggregator) Run(ctx context.Context) func() {
 	ctx, cancelFunc := context.WithCancel(ctx)
 	done := make(chan struct{})
 
+	cleanupTicker := time.NewTicker(ma.metricsCleanupInterval)
 	go func() {
 		defer close(done)
+		defer cleanupTicker.Stop()
 
 		for {
 			select {
 			case req := <-ma.updateCh:
+				ma.log.Debug(ctx, "metrics aggregator: update metrics")
+
 			UpdateLoop:
 				for _, m := range req.metrics {
 					for i, q := range ma.queue {
 						if q.username == req.username && q.workspaceName == req.workspaceName && q.agentName == req.agentName && q.Name == m.Name {
 							ma.queue[i].AgentMetric.Value = m.Value
+							ma.queue[i].expiryDate = req.timestamp.Add(ma.metricsCleanupInterval)
 							continue UpdateLoop
 						}
 					}
@@ -85,20 +103,51 @@ func (ma *MetricsAggregator) Run(ctx context.Context) func() {
 						agentName:     req.agentName,
 
 						AgentMetric: m,
+
+						expiryDate: req.timestamp.Add(ma.metricsCleanupInterval),
 					})
 				}
 			case inputCh := <-ma.collectCh:
+				ma.log.Debug(ctx, "metrics aggregator: collect metrics")
+
 				for _, m := range ma.queue {
 					desc := prometheus.NewDesc(m.Name, metricHelpForAgent, agentMetricsLabels, nil)
 					valueType, err := asPrometheusValueType(m.Type)
 					if err != nil {
-						ma.log.Error(ctx, "can't convert Prometheus value type", slog.F("value_type", m.Type), slog.Error(err))
+						ma.log.Error(ctx, "can't convert Prometheus value type", slog.F("name", m.Name), slog.F("type", m.Type), slog.F("value", m.Value), slog.Error(err))
 						continue
 					}
 					constMetric := prometheus.MustNewConstMetric(desc, valueType, m.Value, m.username, m.workspaceName, m.agentName)
 					inputCh <- constMetric
 				}
 				close(inputCh)
+			case <-cleanupTicker.C:
+				ma.log.Debug(ctx, "metrics aggregator: clean expired metrics")
+
+				now := time.Now()
+
+				var hasExpiredMetrics bool
+				for _, m := range ma.queue {
+					if m.expiryDate.After(now) {
+						hasExpiredMetrics = true
+						break
+					}
+				}
+
+				if !hasExpiredMetrics {
+					continue
+				}
+
+				var j int
+				fresh := make([]annotatedMetric, len(ma.queue))
+				for _, m := range ma.queue {
+					if m.expiryDate.After(now) {
+						fresh[j] = m
+						j++
+					}
+				}
+				fresh = fresh[:j]
+				ma.queue = fresh
 			case <-ctx.Done():
 				ma.log.Debug(ctx, "metrics aggregator: is stopped")
 				return
@@ -140,9 +189,11 @@ func (ma *MetricsAggregator) Update(ctx context.Context, username, workspaceName
 		workspaceName: workspaceName,
 		agentName:     agentName,
 		metrics:       metrics,
+
+		timestamp: time.Now(),
 	}:
 	case <-ctx.Done():
-		ma.log.Debug(ctx, "metrics aggregator: update is canceled")
+		ma.log.Debug(ctx, "metrics aggregator: update request is canceled")
 	default:
 		ma.log.Error(ctx, "metrics aggregator: update queue is full")
 	}