[processor/logdedup] feat: add ottl condition to logdedup processor

[colelaven]

Description:

Adds OTTL Condition field to Deduplicate Logs Processor

Link to tracking Issue: Closes #35440

Testing:

• Tested functionality with BindPlane
• Added unit tests for the condition logic

Documentation: Added documentation to the logdedup processor README about the condition field and an example configuration with a condition.

[linux-foundation-easycla]

The committers listed above are authorized under a signed CLA.

• ✅ login: colelaven / name: Cole Laven (3906b2c, b58545c, 671e47c, e7c0b52, 0464a9f, d9f79e6, 1a6b92e, a7909f0, a760a64, 50bef8c, c33fcac)

[BinaryFissionGames]

Since we are doing this in this repo, we should be able to make use of this internal package:

opentelemetry-collector-contrib/internal/filter/filterottl/filter.go

Line 87 in 0f2fea6

func NewBoolExprForLog(conditions []string, functions map[string]ottl.Factory[ottllog.TransformContext], errorMode ottl.ErrorMode, set component.TelemetrySettings) (expr.BoolExpr[ottllog.TransformContext], error) {

Can you investigate and see if we can switch over to using this package?

[colelaven]

Great find, looks like it works just fine. That way we don't use the bindplane-agent/expr package 👍

Thanks Brandon

[BinaryFissionGames]

I don't think we need this empty condition string here, an empty string won't compile so the processor won't run if you specify an empty string as the condition.

[BinaryFissionGames]

I think we should log if this errors

[djaglowski]

Intuitively, as a user I think I would expect a condition to control whether or not a log is deduplicated, not whether it is kept at all. In other words, I would have expected that logs that do not match the condition pass through the processor without being affected.

The actual meaning seems to be that the condition acts as a filter for the logs. The problem is, we already have ways to filter data, so I'm unconvinced this functionality belongs in this component. The user can always filter a log stream before sending it through this component.

[colelaven]

Ahhh good catch, I had forgotten to change the description when I edited the readme.

I just updated it to be accurate with what the condition does.

The condition is actually used to determine what logs are deduped. Any logs that don't match the condition are just exported without processing.

[djaglowski]

Instead of computing this boolean for every single log record that flows through the processor, we can just compute it once in the factory and save it on the receiver (rather than saving the string).

[djaglowski]

Better yet, just break this into two functions and save the appropriate function onto the reciever:

func dedupMatches(logRecord plog.LogRecord) bool {
	logCtx := ottllog.NewTransformContext(logRecord, scope, resource, sl, rl)
	logMatch, err := p.condition.Eval(ctx, logCtx)
	if err != nil {
		p.logger.Error("error matching condition", zap.Error(err))
		return false
	}
	return logMatch
}

func dedupAll(logRecord plog.LogRecord) bool {
	p.remover.RemoveFields(logRecord)
	p.aggregator.Add(resource, scope, logRecord)
	return true
}

type logDedupProcessor struct {
	...
	matchFunc func(plog.LogRecord) bool
}

// in factory
matchFunc := dedupAll
if c.Condition != defaultCondition {
	matchFunc = dedupMatches
}
...
return &logDedupProcessor{
	...
	matchFunc:       matchFunc,
}

// here, in the receiver
logs.RemoveIf(p.matchFunc)

[colelaven]

Good ideas, thanks for the suggestions. Just committed a refactor.

[djaglowski]

When parameter lists cause very long lines we can break them up for clarity or move them into a function.

Suggested change

	condition, err := filterottl.NewBoolExprForLog([]string{processorCfg.Condition}, filterottl.StandardLogFuncs(), ottl.PropagateError, settings.TelemetrySettings)
	condition, err := filterottl.NewBoolExprForLog(
	[]string{processorCfg.Condition},
	filterottl.StandardLogFuncs(),
	ottl.PropagateError,
	settings.TelemetrySettings,
	)

[djaglowski]

The matchFunc signature ended up being very verbose. Maybe we should just bypass RemoveIf altogether when appropriate.

if p.condition == nil {
	// Don't worry about RemoveIf
	p.remover.RemoveFields(logRecord)
	p.aggregator.Add(resource, scope, logRecord)
	continue
}

// now worry about RemoveIf
logs.RemoveIf(func(logRecord plog.LogRecord) bool {
	logCtx := ottllog.NewTransformContext(logRecord, scope, resource, sl, rl)
	logMatch, err := p.condition.Eval(ctx, logCtx)
	if err != nil {
		p.logger.Error("error matching condition", zap.Error(err))
		return false
	}
	if logMatch {
		p.remover.RemoveFields(logRecord)
		p.aggregator.Add(resource, scope, logRecord)
	}
	return logMatch
})

You might also consider moving the two common lines into a function to ensure we don't update one and not the other later.