Skip to content
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.

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

Add discarded samples per labelset metric #6464

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open
Changes from 1 commit
Commits
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
Next Next commit
discarded samples per labelset metrics
Signed-off-by: Ben Ye <benye@amazon.com>
yeya24 committed Dec 27, 2024
commit 75d98988a42d4fb0b02185dfd32c6f215d959f08
38 changes: 32 additions & 6 deletions pkg/distributor/distributor.go
Original file line number Diff line number Diff line change
@@ -36,6 +36,7 @@ import (
"github.com/cortexproject/cortex/pkg/tenant"
"github.com/cortexproject/cortex/pkg/util"
"github.com/cortexproject/cortex/pkg/util/extract"
"github.com/cortexproject/cortex/pkg/util/labelset"
"github.com/cortexproject/cortex/pkg/util/limiter"
util_log "github.com/cortexproject/cortex/pkg/util/log"
util_math "github.com/cortexproject/cortex/pkg/util/math"
@@ -130,7 +131,7 @@ type Distributor struct {
asyncExecutor util.AsyncExecutor

// Map to track label sets from user.
labelSetTracker *labelSetTracker
labelSetTracker *labelset.LabelSetTracker
}

// Config contains the configuration required to
@@ -388,7 +389,7 @@ func New(cfg Config, clientConfig ingester_client.Config, limits *validation.Ove
asyncExecutor: util.NewNoOpExecutor(),
}

d.labelSetTracker = newLabelSetTracker(d.receivedSamplesPerLabelSet)
d.labelSetTracker = labelset.NewLabelSetTracker()

if cfg.NumPushWorkers > 0 {
util_log.WarnExperimentalUse("Distributor: using goroutine worker pool")
@@ -810,7 +811,18 @@ func (d *Distributor) updateLabelSetMetrics() {
}
}

d.labelSetTracker.updateMetrics(activeUserSet)
d.labelSetTracker.UpdateMetrics(activeUserSet, func(user, labelSetStr string, removeUser bool) {
if removeUser {
if err := util.DeleteMatchingLabels(d.receivedSamplesPerLabelSet, map[string]string{"user": user}); err != nil {
level.Warn(d.log).Log("msg", "failed to remove cortex_distributor_received_samples_per_labelset_total metric for user", "user", user, "err", err)
}
return
}
d.receivedSamplesPerLabelSet.DeleteLabelValues(user, sampleMetricTypeFloat, labelSetStr)
d.receivedSamplesPerLabelSet.DeleteLabelValues(user, sampleMetricTypeHistogram, labelSetStr)
})
// Update label set metrics in validate metrics.
d.validateMetrics.UpdateLabelSet(activeUserSet, d.log)
}

func (d *Distributor) cleanStaleIngesterMetrics() {
@@ -913,6 +925,12 @@ func (d *Distributor) prepareMetadataKeys(req *cortexpb.WriteRequest, limits *va
return metadataKeys, validatedMetadata, firstPartialErr
}

type samplesLabelSetEntry struct {
floatSamples int64
histogramSamples int64
labels labels.Labels
}

func (d *Distributor) prepareSeriesKeys(ctx context.Context, req *cortexpb.WriteRequest, userID string, limits *validation.Limits, removeReplica bool) ([]uint32, []cortexpb.PreallocTimeseries, int, int, int, error, error) {
pSpan, _ := opentracing.StartSpanFromContext(ctx, "prepareSeriesKeys")
defer pSpan.Finish()
@@ -954,7 +972,7 @@ func (d *Distributor) prepareSeriesKeys(ctx context.Context, req *cortexpb.Write
d.dedupedSamples.WithLabelValues(userID, cluster).Add(float64(len(ts.Samples) + len(ts.Histograms)))
}
if errors.Is(err, ha.TooManyReplicaGroupsError{}) {
d.validateMetrics.DiscardedSamples.WithLabelValues(validation.TooManyHAClusters, userID).Add(float64(len(ts.Samples) + len(ts.Histograms)))
d.validateMetrics.UpdateSamplesDiscardedForSeries(userID, validation.TooManyHAClusters, limitsPerLabelSet, cortexpb.FromLabelAdaptersToLabels(ts.Labels), len(ts.Samples)+len(ts.Histograms))
}

continue
@@ -1048,7 +1066,7 @@ func (d *Distributor) prepareSeriesKeys(ctx context.Context, req *cortexpb.Write
matchedLabelSetLimits := validation.LimitsPerLabelSetsForSeries(limitsPerLabelSet, cortexpb.FromLabelAdaptersToLabels(validatedSeries.Labels))
if len(matchedLabelSetLimits) > 0 && labelSetCounters == nil {
// TODO: use pool.
labelSetCounters = make(map[uint64]*samplesLabelSetEntry, len(matchedLabelSetLimits))
labelSetCounters = make(map[uint64]*samplesLabelSetEntry, len(limitsPerLabelSet))
}
for _, l := range matchedLabelSetLimits {
if c, exists := labelSetCounters[l.Hash]; exists {
@@ -1070,8 +1088,16 @@ func (d *Distributor) prepareSeriesKeys(ctx context.Context, req *cortexpb.Write
validatedExemplars += len(ts.Exemplars)
}
for h, counter := range labelSetCounters {
d.labelSetTracker.increaseSamplesLabelSet(userID, h, counter.labels, counter.floatSamples, counter.histogramSamples)
d.labelSetTracker.Track(userID, h, counter.labels)
labelSetStr := counter.labels.String()
if counter.floatSamples > 0 {
d.receivedSamplesPerLabelSet.WithLabelValues(userID, sampleMetricTypeFloat, labelSetStr).Add(float64(counter.floatSamples))
}
if counter.histogramSamples > 0 {
d.receivedSamplesPerLabelSet.WithLabelValues(userID, sampleMetricTypeHistogram, labelSetStr).Add(float64(counter.histogramSamples))
}
}

return seriesKeys, validatedTimeseries, validatedFloatSamples, validatedHistogramSamples, validatedExemplars, firstPartialErr, nil
}

95 changes: 0 additions & 95 deletions pkg/distributor/metrics.go

This file was deleted.

103 changes: 0 additions & 103 deletions pkg/distributor/metrics_test.go

This file was deleted.

Loading