feat: Add support for blocking a policy to be ingested #16203

DylanGuedes · 2025-02-11T22:51:11Z

What this PR does / why we need it:
Add support for blocking a policy from being ingested.
The policy is defined by stream mapping.

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

salvacorts · 2025-02-12T09:17:47Z

pkg/distributor/distributor.go

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

+			if ok, until := d.ShouldBlockPolicy(lbs, tenantID); ok {


Similarly to how we enforce labels, I think this one should also replace the d.validator.ShouldBlockIngestion down below which enforces the per-tenant block.

loki/pkg/distributor/distributor.go

Lines 652 to 653 in 9f9622d

if block, until, retStatusCode := d.validator.ShouldBlockIngestion(validationContext, now); block {

d.trackDiscardedData(ctx, req, validationContext, tenantID, validationContext.validationMetrics, validation.BlockedIngestion)

salvacorts · 2025-02-12T09:18:17Z

pkg/distributor/distributor.go

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

+			if ok, until := d.ShouldBlockPolicy(lbs, tenantID); ok {


Also, we should pass the already resolved policy here to avoid resolving it twice.

salvacorts · 2025-02-12T09:57:37Z

pkg/validation/limits.go

+	BlockIngestionUntil            dskit_flagext.Time            `yaml:"block_ingestion_until" json:"block_ingestion_until"`
+	BlockIngestionStatusCode       int                           `yaml:"block_ingestion_status_code" json:"block_ingestion_status_code"`
+	BlockIngestionPolicyUntil      map[string]dskit_flagext.Time `yaml:"block_ingestion_policy_until" json:"block_ingestion_policy_until"`
+	BlockIngestionPolicyStatusCode map[string]int                `yaml:"block_ingestion_policy_status_code" json:"block_ingestion_policy_status_code"`


Do we need a per policy status code? I think we can just reuse the per-tenant BlockIngestionStatusCode.
I wouldn't add it unless we need it in the future.

not really. got rid of it, ty

github-actions · 2025-02-12T12:40:15Z

💻 Deploy preview deleted.

…n-policy

salvacorts

LGTM! Left some minor comments

salvacorts · 2025-02-17T08:54:22Z

pkg/distributor/distributor.go

+				// If the status code is 200, return success.
+				// Note that we still log the error and increment the metrics.
+				if statusCode == http.StatusOK {
+					// do not add error to validationErrors.


I'm not clear on what this comment means 🤔

previous and following errors do add the error to the validationErrors slice. We don't want to add it when the statusCode is success otherwise we'll return an error when t he only error that happened was this one.

ah damn I read "do not and error" which made no sense to me 😂 "add" instead of "and" makes 100% sense 🤦

salvacorts · 2025-02-17T08:54:50Z

pkg/distributor/distributor.go

+			if block, statusCode, reason, err := d.validator.ShouldBlockIngestion(validationContext, now, policy); block {
+				d.writeFailuresManager.Log(tenantID, err)
+				discardedBytes := util.EntriesTotalSize(stream.Entries)
+				d.validator.reportDiscardedData(ctx, reason, validationContext, lbs, retentionHours, policy, discardedBytes)


I like this 👏 Maybe worth a followup PR making it consistent across all the function among other error handles.

salvacorts · 2025-02-17T08:58:10Z

pkg/distributor/validator.go

+
+	// Check if this policy is blocked in tenant configs
+	blockUntil := v.Limits.BlockIngestionPolicyUntil(ctx.userID, policy)
+	if blockUntil.IsZero() {


nit: Even though now.Before(blockUntil) would be false when blockUntil.IsZero(), I like the explicit handling for the zero case. May we do the same for the ctx.blockIngestionUntil?

salvacorts · 2025-02-17T11:58:39Z

pkg/distributor/validator.go

@@ -184,19 +184,18 @@ func (v Validator) ValidateLabels(ctx validationContext, ls labels.Labels, strea
 	return nil
 }

-func (v Validator) reportDiscardedData(ctx context.Context, reason string, vCtx validationContext, labels labels.Labels, retentionHours string, policy string, entrySize int) {
+func (v Validator) reportDiscardedData(ctx context.Context, reason string, vCtx validationContext, labels labels.Labels, retentionHours string, policy string, entrySize int, withUsageTracker bool) {


nit: Have reportDiscardedData and reportDiscardedDataWithUsageTracker instead of this bool to make it more explicit

done, wdyt? I also found we had a updateMetrics very simila to reportDiscardedData so I got rid of it

salvacorts · 2025-02-17T12:02:38Z

pkg/distributor/validator.go

-		err := fmt.Errorf(validation.BlockedIngestionErrorMsg, ctx.userID, ctx.blockIngestionUntil.Format(time.RFC3339), ctx.blockIngestionStatusCode)
-		return true, ctx.blockIngestionStatusCode, validation.BlockedIngestion, err
+	if block, code, reason, err := v.shouldBlockGlobalPolicy(ctx, now); block {
+		return block, code, reason, err
 	}

 	if block, until, code := v.ShouldBlockPolicy(ctx, policy, now); block {


ShouldBlockPolicy should be private shouldBlockPolicy

**What this PR does / why we need it**: Add support for blocking a policy from being ingested. The policy is defined by stream mapping. (cherry picked from commit 69089ef)

Add the policy blocking.

9f9622d

DylanGuedes requested a review from a team as a code owner February 11, 2025 22:51

pull-request-size bot added the size/L label Feb 11, 2025

salvacorts reviewed Feb 12, 2025

View reviewed changes

address suggestions

e1df362

github-actions bot added the type/docs Issues related to technical documentation; the Docs Squad uses this label across many repositories label Feb 12, 2025

DylanGuedes added 10 commits February 12, 2025 19:00

update test accordingly

be5ceb6

Merge branch 'main' of github.com:grafana/loki into add-block-based-o…

4218d81

…n-policy

undo merge changes

bfcc228

undo changes

0eaa021

check for blocked ingestion at same place

84a6931

update docs

6d0204e

Fix test

49bd76a

Report things at a single place

70bd1e1

do not check for nil

1b7fdb4

Update docs

119d358

salvacorts approved these changes Feb 17, 2025

View reviewed changes

DylanGuedes added 4 commits February 17, 2025 08:34

handle the zero case isoalted

585b096

Add conditional for the usageTracker reporting

aaeadbf

undo change

d41480f

do it isolated

fd275ad

salvacorts reviewed Feb 17, 2025

View reviewed changes

DylanGuedes added 7 commits February 17, 2025 09:14

refactor how we report metrics

80d72f7

make method private

6ca9ec5

lint

0e908b9

Add test covering metrics reporting

cfbe78b

Remove dead test

23e8d07

lint

2a1a87d

clear test state

626fd51

salvacorts approved these changes Feb 17, 2025

View reviewed changes

DylanGuedes merged commit 69089ef into main Feb 17, 2025
61 checks passed

DylanGuedes deleted the add-block-based-on-policy branch February 17, 2025 16:54

salvacorts added the backport k242 label Feb 20, 2025

This was referenced Feb 20, 2025

feat: Add support for blocking a policy to be ingested (backport k242) #16383

Closed

chore(k243): release 3.5.0 #16416

Open

loki-gh-app bot mentioned this pull request Mar 5, 2025

chore(k244): release 3.5.0 #16562

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: Add support for blocking a policy to be ingested #16203

feat: Add support for blocking a policy to be ingested #16203

DylanGuedes commented Feb 11, 2025

salvacorts Feb 12, 2025

salvacorts Feb 12, 2025

DylanGuedes Feb 13, 2025

salvacorts Feb 12, 2025

DylanGuedes Feb 13, 2025

github-actions bot commented Feb 12, 2025 •

edited

Loading

salvacorts left a comment

salvacorts Feb 17, 2025

DylanGuedes Feb 17, 2025

salvacorts Feb 17, 2025

salvacorts Feb 17, 2025

salvacorts Feb 17, 2025

salvacorts Feb 17, 2025

DylanGuedes Feb 17, 2025

salvacorts Feb 17, 2025

	if block, until, retStatusCode := d.validator.ShouldBlockIngestion(validationContext, now); block {
	d.trackDiscardedData(ctx, req, validationContext, tenantID, validationContext.validationMetrics, validation.BlockedIngestion)

feat: Add support for blocking a policy to be ingested #16203

feat: Add support for blocking a policy to be ingested #16203

Conversation

DylanGuedes commented Feb 11, 2025

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

github-actions bot commented Feb 12, 2025 • edited Loading

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

github-actions bot commented Feb 12, 2025 •

edited

Loading