refactor: Add retention hours to `discarded` metrics #15875

DylanGuedes · 2025-01-21T23:39:25Z

What this PR does / why we need it:
We have no distinction between the type of data that gets discarded. On this PR I'm adding retention_hours to our discarded metrics so that we can better look at what data was discarded.

I'm also using a more performant form of retention strigifying when loading the retention string. Benchmark results:

goos: linux
goarch: amd64
pkg: github.com/grafana/loki/v3/pkg/util
cpu: AMD Ryzen 9 7900X 12-Core Processor            
BenchmarkRetentionHours/current/24h-24     	35803432	        34.14 ns/op	       2 B/op	       1 allocs/op
BenchmarkRetentionHours/current/7d-24      	33798613	        35.33 ns/op	       3 B/op	       1 allocs/op
BenchmarkRetentionHours/current/30d-24     	23425309	        50.95 ns/op	      16 B/op	       2 allocs/op
BenchmarkRetentionHours/current/365d-24    	23175356	        51.52 ns/op	      16 B/op	       2 allocs/op
BenchmarkRetentionHours/optimized/24h-24   	412592362	         2.916 ns/op	       0 B/op	       0 allocs/op
BenchmarkRetentionHours/optimized/7d-24    	89319780	        14.32 ns/op	       3 B/op	       1 allocs/op
BenchmarkRetentionHours/optimized/30d-24   	88080440	        14.60 ns/op	       3 B/op	       1 allocs/op
BenchmarkRetentionHours/optimized/365d-24  	75923084	        15.57 ns/op	       4 B/op	       1 allocs/op
PASS
ok  	github.com/grafana/loki/v3/pkg/util	11.191s

Which issue(s) this PR fixes:
Fixes #

Special notes for your reviewer:

Checklist

Reviewed the CONTRIBUTING.md guide (required)
Documentation added
Tests updated
Title matches the required conventional commits format, see here
- Note that Promtail is considered to be feature complete, and future development for logs collection will be in Grafana Alloy. As such, feat PRs are unlikely to be accepted unless a case can be made for the feature actually being a bug fix to existing behavior.
Changes that require user attention or interaction to upgrade are documented in docs/sources/setup/upgrade/_index.md
If the change is deprecating or removing a configuration option, update the deprecated-config.yaml and deleted-config.yaml files respectively in the tools/deprecated-config-checker directory. Example PR

DylanGuedes · 2025-01-21T23:40:53Z

pkg/validation/limits.go

+	selectedRetention := o.getOverridesForUser(userID).RetentionPeriod
+	highestPriority := -1
+
+	if len(ls) > 0 {


this check is to help using this same function for when you want to automatically gets the Tenant retention but have no labels available

I think it'd be more idiomatic to return early if ls is empty and keep this loop in the top level.

salvacorts · 2025-01-22T14:28:24Z

pkg/distributor/validator.go

 	if len(ls) == 0 {
-		validation.DiscardedSamples.WithLabelValues(validation.MissingLabels, ctx.userID).Inc()
+		validation.DiscardedSamples.WithLabelValues(validation.MissingLabels, ctx.userID, retentionHours).Inc()


I see we only added the retentionHours in the PR but, shouldn't this be updateMetrics as well?

I agree. We can fix that in a separate PR to involve more people in the discussion.

pkg/distributor/distributor.go

salvacorts · 2025-01-22T14:37:16Z

pkg/ingester/instance.go

 	if err != nil {
-		return i.onStreamCreationError(ctx, pushReqStream, err, labels)
+		return i.onStreamCreationError(ctx, pushReqStream, err, labels, retentionHours)


nit: This err handler looks a bit off-place. What about getting the retentionHours after getting the labels?

labels, err := syntax.ParseLabels(pushReqStream.Labels) if err != nil { ... } retentionHours := i.limiter.limits.RetentionHours(i.instanceID, labels) if record != nil { err = i.streamCountLimiter.AssertNewStreamAllowed(i.instanceID) return i.onStreamCreationError(ctx, pushReqStream, err, labels, retentionHours) }

good call, fixed it

Not quite my suggestion but it's fine since it's a nit

are you sure? I'm doing it immediately after parsing the labels.

salvacorts · 2025-01-22T14:41:01Z

pkg/validation/limits.go

+	selectedRetention := o.getOverridesForUser(userID).RetentionPeriod
+	highestPriority := -1
+
+	if len(ls) > 0 {


I think it'd be more idiomatic to return early if ls is empty and keep this loop in the top level.

salvacorts · 2025-01-22T14:41:37Z

pkg/validation/limits.go

@@ -959,6 +970,27 @@ func (o *Overrides) StreamRetention(userID string) []StreamRetention {
 	return o.getOverridesForUser(userID).StreamRetention
 }

+// RetentionHours returns the retention period for a given user.
+func (o *Overrides) RetentionHours(userID string, ls labels.Labels) string {
+	streamRetentions := o.getOverridesForUser(userID).StreamRetention


nit: rather call the public StreamRetention() than the implementation.

salvacorts · 2025-01-22T14:46:15Z

pkg/validation/limits.go

@@ -959,6 +970,27 @@ func (o *Overrides) StreamRetention(userID string) []StreamRetention {
 	return o.getOverridesForUser(userID).StreamRetention
 }

+// RetentionHours returns the retention period for a given user.
+func (o *Overrides) RetentionHours(userID string, ls labels.Labels) string {


I think we should use this one instead which is the one the compactor actually uses

loki/pkg/compactor/retention/expiration.go

Line 134 in 4c88be0

func (tr *TenantsRetention) RetentionPeriodFor(userID string, lbs labels.Labels) time.Duration {

salvacorts · 2025-01-22T14:48:51Z

pkg/distributor/distributor.go

 			if err != nil {
 				d.writeFailuresManager.Log(tenantID, err)
 				validationErrors.Add(err)
-				validation.DiscardedSamples.WithLabelValues(validation.InvalidLabels, tenantID).Add(float64(len(stream.Entries)))
+				validation.DiscardedSamples.WithLabelValues(validation.InvalidLabels, tenantID, tenantRetentionHours).Add(float64(len(stream.Entries)))


I think we may be counting these twice since we already update the discarded metrics inside parseStreamLabels.

yep, you are right @salvacorts

…na/loki into retention-as-label-when-discarding

salvacorts

LGTM! Left some more nits. I'm gonna ask Vlad to review it as well since he spotted the issue first.

salvacorts · 2025-01-23T10:53:11Z

pkg/distributor/distributor.go

@@ -493,7 +492,8 @@ func (d *Distributor) Push(ctx context.Context, req *logproto.PushRequest) (*log
 		}
 	}

-	tenantRetentionHours := d.validator.Limits.RetentionHours(tenantID, nil)
+	tenantRetentionHours := util.RetentionHours(d.tenantsRetention.RetentionPeriodFor(tenantID, nil))


nit: instead of calling util.RetentionHours what about having tenantsRetention.RetentionHoursFor() string that calls util.RetentionHours for the result of tenantsRetention.RetentionPeriodFor.

salvacorts · 2025-01-23T10:54:57Z

pkg/distributor/distributor.go

@@ -524,7 +524,7 @@ func (d *Distributor) Push(ctx context.Context, req *logproto.PushRequest) (*log
 				continue
 			}

-			retentionHours := d.validator.Limits.RetentionHours(tenantID, lbs)
+			retentionHours := util.RetentionHours(d.tenantsRetention.RetentionPeriodFor(tenantID, lbs))


nit for a followup PR since we were resolving the retention for the streams before the changes. I think we could be able to speed up pushes by caching this.

I think optimizing that would do more harm than good (based on the benchmarks). Doing the new retention evaluation on pushes didn't make a difference but adding the cache will likely do, either by the extra memory usage or by the extra complexity of adding of caching more stuff.

salvacorts · 2025-01-23T10:57:33Z

pkg/distributor/distributor.go

@@ -538,7 +541,7 @@ func (d *Distributor) Push(ctx context.Context, req *logproto.PushRequest) (*log
 			prevTs := stream.Entries[0].Timestamp

 			for _, entry := range stream.Entries {
-				if err := d.validator.ValidateEntry(ctx, validationContext, lbs, entry); err != nil {
+				if err := d.validator.ValidateEntry(ctx, validationContext, lbs, entry, retentionHours); err != nil {


nit: maybe validation context should have the validationMetrics?

done, wdyt?

salvacorts · 2025-01-23T10:58:29Z

pkg/ingester/instance.go

 	if err != nil {
-		return i.onStreamCreationError(ctx, pushReqStream, err, labels)
+		return i.onStreamCreationError(ctx, pushReqStream, err, labels, retentionHours)


Not quite my suggestion but it's fine since it's a nit

salvacorts · 2025-01-23T11:03:35Z

pkg/ingester/limiter.go

@@ -33,6 +33,10 @@ type Limits interface {
 	PerStreamRateLimit(userID string) validation.RateLimit
 	ShardStreams(userID string) shardstreams.Config
 	IngestionPartitionsTenantShardSize(userID string) int
+	RetentionPeriod(userID string) time.Duration


You can add retention.Limits here instead

type Limits interface { ... ShardStreams(userID string) shardstreams.Config IngestionPartitionsTenantShardSize(userID string) int retention.Limits }

hah good catch 😄

vlad-diachenko

Looks good. however, there are some comments on distributor's side.

vlad-diachenko · 2025-01-24T07:25:41Z

pkg/distributor/distributor.go

@@ -494,6 +492,8 @@ func (d *Distributor) Push(ctx context.Context, req *logproto.PushRequest) (*log
 		}
 	}

+	tenantRetentionHours := d.tenantsRetention.RetentionHoursFor(tenantID, nil)


I assume that in this case, we use the global tenant's retention period, rather than the retention period for the particular stream.
If so, we can not do it because the metric that shows ingested bytes count always use the retention period from the stream, and we need to follow it exactly

I'll check but, we're only using tenantRetentionHours when we fail to parse the stream labels, how can the ingested_bytes know the stream retention period if it couldn't parse the stream labels?

if it completely fails parsing the labels, it returns an error :

loki/pkg/loghttp/push/push.go

Line 245 in ebc84ca

if err != nil {

but if the labels are parsed, then the retention hours is calculated based on what we got

loki/pkg/loghttp/push/push.go

Line 282 in ebc84ca

retentionPeriod = tenantsRetention.RetentionPeriodFor(userID, lbs)

even if the length of the labels breaks the validation....

vlad-diachenko · 2025-01-24T07:29:22Z

pkg/distributor/distributor.go

 			if err != nil {
 				d.writeFailuresManager.Log(tenantID, err)
 				validationErrors.Add(err)
-				validation.DiscardedSamples.WithLabelValues(validation.InvalidLabels, tenantID).Add(float64(len(stream.Entries)))
+				validation.DiscardedSamples.WithLabelValues(validation.InvalidLabels, tenantID, tenantRetentionHours).Add(float64(len(stream.Entries)))


yep, you are right @salvacorts

vlad-diachenko · 2025-01-24T07:36:35Z

pkg/distributor/distributor.go

-	if !d.ingestionRateLimiter.AllowN(now, tenantID, validatedLineSize) {
-		d.trackDiscardedData(ctx, req, validationContext, tenantID, validatedLineCount, validatedLineSize, validation.RateLimited)
+	if !d.ingestionRateLimiter.AllowN(now, tenantID, validationContext.validationMetrics.lineSize) {
+		d.trackDiscardedData(ctx, req, validationContext, tenantID, validationContext.validationMetrics, validation.RateLimited)


nit: I would ask @trevorwhitney if it's necessary to pass retention_hours to usageTacker for the discarded bytes metric. because they also expose this label in ingested bytes metric and might have a similar issue

pull-request-size bot added the size/L label Jan 21, 2025

DylanGuedes commented Jan 21, 2025

View reviewed changes

DylanGuedes force-pushed the retention-as-label-when-discarding branch from 87db09e to f269d13 Compare January 21, 2025 23:48

DylanGuedes changed the title ~~Refactor: Add retention hours to discarded metrics~~ refactor: Add retention hours to discarded metrics Jan 22, 2025

Refactor discarded metrics to report retention_hours

62b8cfc

DylanGuedes force-pushed the retention-as-label-when-discarding branch from f269d13 to 62b8cfc Compare January 22, 2025 00:13

Add benchmark.

e09aba2

DylanGuedes marked this pull request as ready for review January 22, 2025 12:26

DylanGuedes requested a review from a team as a code owner January 22, 2025 12:26

Merge branch 'main' into retention-as-label-when-discarding

2e81059

salvacorts reviewed Jan 22, 2025

View reviewed changes

DylanGuedes added 6 commits January 22, 2025 19:41

Use public method instead.

57976bc

Refactor

86e1cc8

Merge branch 'retention-as-label-when-discarding' of github.com:grafa…

a936fdf

…na/loki into retention-as-label-when-discarding

Reuse tenantRetention on ingesters

4d25e22

Add missing file

045d3bf

Lint fix

7e288ab

salvacorts reviewed Jan 23, 2025

View reviewed changes

Address review suggestions

12ec27e

vlad-diachenko reviewed Jan 24, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

refactor: Add retention hours to `discarded` metrics #15875

refactor: Add retention hours to `discarded` metrics #15875

DylanGuedes commented Jan 21, 2025 •

edited

Loading

DylanGuedes Jan 21, 2025

salvacorts Jan 22, 2025

DylanGuedes Jan 22, 2025

salvacorts Jan 22, 2025

DylanGuedes Jan 22, 2025

salvacorts Jan 22, 2025

DylanGuedes Jan 22, 2025

salvacorts Jan 23, 2025

DylanGuedes Jan 23, 2025

salvacorts Jan 22, 2025

salvacorts Jan 22, 2025

DylanGuedes Jan 22, 2025

salvacorts Jan 22, 2025

salvacorts Jan 22, 2025

vlad-diachenko Jan 24, 2025

salvacorts left a comment

salvacorts Jan 23, 2025

DylanGuedes Jan 23, 2025

salvacorts Jan 23, 2025

DylanGuedes Jan 23, 2025

salvacorts Jan 23, 2025

DylanGuedes Jan 23, 2025

salvacorts Jan 23, 2025

salvacorts Jan 23, 2025

DylanGuedes Jan 23, 2025

vlad-diachenko left a comment

vlad-diachenko Jan 24, 2025

DylanGuedes Jan 24, 2025

vlad-diachenko Jan 27, 2025

vlad-diachenko Jan 24, 2025

vlad-diachenko Jan 24, 2025

refactor: Add retention hours to discarded metrics #15875

Are you sure you want to change the base?

refactor: Add retention hours to discarded metrics #15875

Conversation

DylanGuedes commented Jan 21, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

salvacorts left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vlad-diachenko left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

refactor: Add retention hours to `discarded` metrics #15875

refactor: Add retention hours to `discarded` metrics #15875

DylanGuedes commented Jan 21, 2025 •

edited

Loading