Skip to content

feat: use map instead of slice in metrics aggregator #11815

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 7 commits into from
Jan 29, 2024
Merged

feat: use map instead of slice in metrics aggregator #11815

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
b936232
feat: use map instead slice in metrics aggregator
mtojek Jan 25, 2024
c90a41f
fix
mtojek Jan 25, 2024
f407b47
add store size gauge
mtojek Jan 25, 2024
12fcdab
Use struct key
mtojek Jan 25, 2024
628e4ec
Address PR comments
mtojek Jan 25, 2024
bf48edc
Address Spike's feedback
mtojek Jan 26, 2024
4d0d889
fix test
mtojek Jan 26, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
131 changes: 90 additions & 41 deletions coderd/prometheusmetrics/aggregator.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ package prometheusmetrics

import (
"context"
"strings"
"time"

"github.com/prometheus/client_golang/prometheus"
Expand All @@ -24,20 +25,23 @@ const (
loggerName = "prometheusmetrics"

sizeCollectCh = 10
sizeUpdateCh = 1024
sizeUpdateCh = 4096

defaultMetricsCleanupInterval = 2 * time.Minute
)

var MetricLabelValueEncoder = strings.NewReplacer("|", "\\|", ",", "\\,", "=", "\\=")

type MetricsAggregator struct {
queue []annotatedMetric
store map[metricKey]annotatedMetric

log slog.Logger
metricsCleanupInterval time.Duration

collectCh chan (chan []prometheus.Metric)
updateCh chan updateRequest

storeSizeGauge prometheus.Gauge
updateHistogram prometheus.Histogram
cleanupHistogram prometheus.Histogram
}
Expand All @@ -64,15 +68,58 @@ type annotatedMetric struct {
expiryDate time.Time
}

type metricKey struct {
username string
workspaceName string
agentName string
templateName string

metricName string
labelsStr string
}

func (m1 metricKey) Equal(m2 metricKey) bool {
return m1.metricName == m2.metricName &&
m1.labelsStr == m2.labelsStr &&
m1.username == m2.username &&
m1.workspaceName == m2.workspaceName &&
m1.agentName == m2.agentName &&
m1.templateName == m2.templateName
}

func hashKey(req *updateRequest, m *agentproto.Stats_Metric) metricKey {
var sb strings.Builder
for _, label := range m.GetLabels() {
if label.Value == "" {
continue
}

_, _ = sb.WriteString(label.Name)
_ = sb.WriteByte('=')
_, _ = sb.WriteString(MetricLabelValueEncoder.Replace(label.Value))
_ = sb.WriteByte(',')
}
labels := strings.TrimRight(sb.String(), ",")

return metricKey{
username: req.username,
workspaceName: req.workspaceName,
agentName: req.agentName,
templateName: req.templateName,
metricName: m.Name,
labelsStr: labels,
}
}

var _ prometheus.Collector = new(MetricsAggregator)

func (am *annotatedMetric) is(req updateRequest, m *agentproto.Stats_Metric) bool {
return am.username == req.username &&
return am.Name == m.Name &&
agentproto.LabelsEqual(am.Labels, m.Labels) &&
am.username == req.username &&
am.workspaceName == req.workspaceName &&
am.agentName == req.agentName &&
am.templateName == req.templateName &&
am.Name == m.Name &&
agentproto.LabelsEqual(am.Labels, m.Labels)
am.templateName == req.templateName
}

func (am *annotatedMetric) asPrometheus() (prometheus.Metric, error) {
Expand Down Expand Up @@ -101,14 +148,25 @@ func NewMetricsAggregator(logger slog.Logger, registerer prometheus.Registerer,
metricsCleanupInterval = duration
}

storeSizeGauge := prometheus.NewGauge(prometheus.GaugeOpts{
Namespace: "coderd",
Subsystem: "prometheusmetrics",
Name: "metrics_aggregator_store_size",
Help: "The number of metrics stored in the aggregator",
})
err := registerer.Register(storeSizeGauge)
if err != nil {
return nil, err
}

updateHistogram := prometheus.NewHistogram(prometheus.HistogramOpts{
Namespace: "coderd",
Subsystem: "prometheusmetrics",
Name: "metrics_aggregator_execution_update_seconds",
Help: "Histogram for duration of metrics aggregator update in seconds.",
Buckets: []float64{0.001, 0.005, 0.010, 0.025, 0.050, 0.100, 0.500, 1, 5, 10, 30},
})
err := registerer.Register(updateHistogram)
err = registerer.Register(updateHistogram)
if err != nil {
return nil, err
}
Expand All @@ -129,9 +187,12 @@ func NewMetricsAggregator(logger slog.Logger, registerer prometheus.Registerer,
log: logger.Named(loggerName),
metricsCleanupInterval: metricsCleanupInterval,

store: map[metricKey]annotatedMetric{},

collectCh: make(chan (chan []prometheus.Metric), sizeCollectCh),
updateCh: make(chan updateRequest, sizeUpdateCh),

storeSizeGauge: storeSizeGauge,
updateHistogram: updateHistogram,
cleanupHistogram: cleanupHistogram,
}, nil
Expand All @@ -152,32 +213,32 @@ func (ma *MetricsAggregator) Run(ctx context.Context) func() {
ma.log.Debug(ctx, "update metrics")

timer := prometheus.NewTimer(ma.updateHistogram)
UpdateLoop:
for _, m := range req.metrics {
for i, q := range ma.queue {
if q.is(req, m) {
ma.queue[i].Stats_Metric.Value = m.Value
ma.queue[i].expiryDate = req.timestamp.Add(ma.metricsCleanupInterval)
continue UpdateLoop
key := hashKey(&req, m)

if val, ok := ma.store[key]; ok {
val.Stats_Metric.Value = m.Value
val.expiryDate = req.timestamp.Add(ma.metricsCleanupInterval)
ma.store[key] = val
} else {
ma.store[key] = annotatedMetric{
Stats_Metric: m,
username: req.username,
workspaceName: req.workspaceName,
agentName: req.agentName,
templateName: req.templateName,
expiryDate: req.timestamp.Add(ma.metricsCleanupInterval),
}
}

ma.queue = append(ma.queue, annotatedMetric{
Stats_Metric: m,
username: req.username,
workspaceName: req.workspaceName,
agentName: req.agentName,
templateName: req.templateName,
expiryDate: req.timestamp.Add(ma.metricsCleanupInterval),
})
}

timer.ObserveDuration()

ma.storeSizeGauge.Set(float64(len(ma.store)))
case outputCh := <-ma.collectCh:
ma.log.Debug(ctx, "collect metrics")

output := make([]prometheus.Metric, 0, len(ma.queue))
for _, m := range ma.queue {
output := make([]prometheus.Metric, 0, len(ma.store))
for _, m := range ma.store {
promMetric, err := m.asPrometheus()
if err != nil {
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))
Expand All @@ -191,29 +252,17 @@ func (ma *MetricsAggregator) Run(ctx context.Context) func() {
ma.log.Debug(ctx, "clean expired metrics")

timer := prometheus.NewTimer(ma.cleanupHistogram)

now := time.Now()

var hasExpiredMetrics bool
for _, m := range ma.queue {
if now.After(m.expiryDate) {
hasExpiredMetrics = true
break
}
}

if hasExpiredMetrics {
fresh := make([]annotatedMetric, 0, len(ma.queue))
for _, m := range ma.queue {
if m.expiryDate.After(now) {
fresh = append(fresh, m)
}
for key, val := range ma.store {
if now.After(val.expiryDate) {
delete(ma.store, key)
}
ma.queue = fresh
}

timer.ObserveDuration()
cleanupTicker.Reset(ma.metricsCleanupInterval)
ma.storeSizeGauge.Set(float64(len(ma.store)))

case <-ctx.Done():
ma.log.Debug(ctx, "metrics aggregator is stopped")
Expand Down
83 changes: 81 additions & 2 deletions coderd/prometheusmetrics/aggregator_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package prometheusmetrics_test
import (
"context"
"sort"
"strings"
"sync/atomic"
"testing"
"time"
Expand Down Expand Up @@ -70,6 +71,24 @@ func TestUpdateMetrics_MetricsDoNotExpire(t *testing.T) {
{Name: "hello", Value: "world"},
}},
{Name: "d_gauge_four", Type: agentproto.Stats_Metric_GAUGE, Value: 6},
{Name: "e_gauge_four", Type: agentproto.Stats_Metric_GAUGE, Value: 15, Labels: []*agentproto.Stats_Metric_Label{
{Name: "foobar", Value: "Foo,ba=z"},
{Name: "hello", Value: "wo,,r=d"},
}},
{Name: "f_gauge_four", Type: agentproto.Stats_Metric_GAUGE, Value: 6, Labels: []*agentproto.Stats_Metric_Label{
{Name: "foobar", Value: "foobaz"},
{Name: "empty", Value: ""},
}},
}

given3 := []*agentproto.Stats_Metric{
{Name: "e_gauge_four", Type: agentproto.Stats_Metric_GAUGE, Value: 17, Labels: []*agentproto.Stats_Metric_Label{
{Name: "cat", Value: "do,=g"},
{Name: "hello", Value: "wo,,rld"},
}},
{Name: "f_gauge_four", Type: agentproto.Stats_Metric_GAUGE, Value: 8, Labels: []*agentproto.Stats_Metric_Label{
{Name: "foobar", Value: "foobaz"},
}},
}

commonLabels := []*agentproto.Stats_Metric_Label{
Expand All @@ -80,15 +99,14 @@ func TestUpdateMetrics_MetricsDoNotExpire(t *testing.T) {
}
expected := []*agentproto.Stats_Metric{
{Name: "a_counter_one", Type: agentproto.Stats_Metric_COUNTER, Value: 1, Labels: commonLabels},
{Name: "b_counter_two", Type: agentproto.Stats_Metric_COUNTER, Value: 4, Labels: commonLabels},
{Name: "b_counter_two", Type: agentproto.Stats_Metric_COUNTER, Value: -9, Labels: []*agentproto.Stats_Metric_Label{
{Name: "agent_name", Value: testAgentName},
{Name: "lizz", Value: "rizz"},
{Name: "username", Value: testUsername},
{Name: "workspace_name", Value: testWorkspaceName},
{Name: "template_name", Value: testTemplateName},
}},
{Name: "c_gauge_three", Type: agentproto.Stats_Metric_GAUGE, Value: 5, Labels: commonLabels},
{Name: "b_counter_two", Type: agentproto.Stats_Metric_COUNTER, Value: 4, Labels: commonLabels},
{Name: "c_gauge_three", Type: agentproto.Stats_Metric_GAUGE, Value: 2, Labels: []*agentproto.Stats_Metric_Label{
{Name: "agent_name", Value: testAgentName},
{Name: "foobar", Value: "Foobaz"},
Expand All @@ -97,12 +115,37 @@ func TestUpdateMetrics_MetricsDoNotExpire(t *testing.T) {
{Name: "workspace_name", Value: testWorkspaceName},
{Name: "template_name", Value: testTemplateName},
}},
{Name: "c_gauge_three", Type: agentproto.Stats_Metric_GAUGE, Value: 5, Labels: commonLabels},
{Name: "d_gauge_four", Type: agentproto.Stats_Metric_GAUGE, Value: 6, Labels: commonLabels},
{Name: "e_gauge_four", Type: agentproto.Stats_Metric_GAUGE, Value: 17, Labels: []*agentproto.Stats_Metric_Label{
{Name: "agent_name", Value: testAgentName},
{Name: "cat", Value: "do,=g"},
{Name: "hello", Value: "wo,,rld"},
{Name: "username", Value: testUsername},
{Name: "workspace_name", Value: testWorkspaceName},
{Name: "template_name", Value: testTemplateName},
}},
{Name: "e_gauge_four", Type: agentproto.Stats_Metric_GAUGE, Value: 15, Labels: []*agentproto.Stats_Metric_Label{
{Name: "agent_name", Value: testAgentName},
{Name: "foobar", Value: "Foo,ba=z"},
{Name: "hello", Value: "wo,,r=d"},
{Name: "username", Value: testUsername},
{Name: "workspace_name", Value: testWorkspaceName},
{Name: "template_name", Value: testTemplateName},
}},
{Name: "f_gauge_four", Type: agentproto.Stats_Metric_GAUGE, Value: 8, Labels: []*agentproto.Stats_Metric_Label{
{Name: "agent_name", Value: testAgentName},
{Name: "foobar", Value: "foobaz"},
{Name: "username", Value: testUsername},
{Name: "workspace_name", Value: testWorkspaceName},
{Name: "template_name", Value: testTemplateName},
}},
}

// when
metricsAggregator.Update(ctx, testLabels, given1)
metricsAggregator.Update(ctx, testLabels, given2)
metricsAggregator.Update(ctx, testLabels, given3)

// then
require.Eventually(t, func() bool {
Expand Down Expand Up @@ -130,6 +173,12 @@ func verifyCollectedMetrics(t *testing.T, expected []*agentproto.Stats_Metric, a
return false
}

sort.Slice(actual, func(i, j int) bool {
m1 := prometheusMetricToString(t, actual[i])
m2 := prometheusMetricToString(t, actual[j])
return m1 < m2
})

for i, e := range expected {
desc := actual[i].Desc()
assert.Contains(t, desc.String(), e.Name)
Expand All @@ -156,9 +205,39 @@ func verifyCollectedMetrics(t *testing.T, expected []*agentproto.Stats_Metric, a
return true
}

func prometheusMetricToString(t *testing.T, m prometheus.Metric) string {
var sb strings.Builder

desc := m.Desc()
_, _ = sb.WriteString(desc.String())
_ = sb.WriteByte('|')

var d dto.Metric
err := m.Write(&d)
require.NoError(t, err)
dtoLabels := asMetricAgentLabels(d.GetLabel())
sort.Slice(dtoLabels, func(i, j int) bool {
return dtoLabels[i].Name < dtoLabels[j].Name
})

for _, dtoLabel := range dtoLabels {
if dtoLabel.Value == "" {
continue
}
_, _ = sb.WriteString(dtoLabel.Name)
_ = sb.WriteByte('=')
_, _ = sb.WriteString(prometheusmetrics.MetricLabelValueEncoder.Replace(dtoLabel.Value))
}
return strings.TrimRight(sb.String(), ",")
}

func asMetricAgentLabels(dtoLabels []*dto.LabelPair) []*agentproto.Stats_Metric_Label {
metricLabels := make([]*agentproto.Stats_Metric_Label, 0, len(dtoLabels))
for _, dtoLabel := range dtoLabels {
if dtoLabel.GetValue() == "" {
continue
}

metricLabels = append(metricLabels, &agentproto.Stats_Metric_Label{
Name: dtoLabel.GetName(),
Value: dtoLabel.GetValue(),
Expand Down