[processor/logdedup] Add include_fields option (open-telemetry#37219)

#### Link to tracking issue

Fixes open-telemetry#36965

<!--Describe what testing was performed and which tests were added.-->
#### Testing

Added tests that validate the new `include_fields` option as well as
integration tests with ConsumeLogs.

<!--Describe the documentation added.-->
#### Documentation

Added documentation for the new `include_fields` option as well as a
README example.

---------

Co-authored-by: Mike Goldsmith <[email protected]>

Author: 2 people

.chloggen/logdedupprocessor-deduplication-key.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: logdedupprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add 'include_fields' option to deduplicate log records via body or attribute fields.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [36965]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

processor/logdedupprocessor/README.md

@@ -30,8 +30,9 @@ This processor is used to deduplicate logs by detecting identical logs over a ra
 | interval            | duration | `10s`       | The interval at which logs are aggregated. The counter will reset after each interval.                                                                                                                                                                                                                                                                                                                                                                  |
 | conditions          | []string | `[]`        | A slice of [OTTL] expressions used to evaluate which log records are deduped.  All paths in the [log context] are available to reference. All [converters] are available to use.                                                                                                                                                                                                                                                                        |
 | log_count_attribute | string   | `log_count` | The name of the count attribute of deduplicated logs that will be added to the emitted aggregated log.                                                                                                                                                                                                                                                                                                                                                  |
+| include_fields                | []string | `[]`        | Fields to include in duplication matching. Fields can be from the log `body` or `attributes`.  Nested fields must be `.` delimited. If a field contains a `.` it can be escaped by using a `\`.  This option is **mutually exclusive** with `exclude_fields`. See [example config](#example-config-with-deduplication-key).
 | timezone            | string   | `UTC`       | The timezone of the `first_observed_timestamp` and `last_observed_timestamp` timestamps on the emitted aggregated log. The available locations depend on the local IANA Time Zone database. [This page](https://en.wikipedia.org/wiki/List_of_tz_database_time_zones) contains many examples, such as `America/New_York`.                                                                                                                               |
-| exclude_fields      | []string | `[]`        | Fields to exclude from duplication matching. Fields can be excluded from the log `body` or `attributes`. These fields will not be present in the emitted aggregated log. Nested fields must be `.` delimited. If a field contains a `.` it can be escaped by using a `\` see [example config](#example-config-with-excluded-fields).<br><br>**Note**: The entire `body` cannot be excluded. If the body is a map then fields within it can be excluded. |
+| exclude_fields      | []string | `[]`        | Fields to exclude from duplication matching. Fields can be excluded from the log `body` or `attributes`. These fields will not be present in the emitted aggregated log. Nested fields must be `.` delimited. This option is `mutually exclusive` with `include_fields`. If a field contains a `.` it can be escaped by using a `\` see [example config](#example-config-with-excluded-fields).<br><br>**Note**: The entire `body` cannot be excluded. If the body is a map then fields within it can be excluded. |
 
 [OTTL]: https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/v0.109.0/pkg/ottl#readme
 [converters]: https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/v0.109.0/pkg/ottl/ottlfuncs/README.md#converters
@@ -87,6 +88,31 @@ service:
             exporters: [googlecloud]
 ```
 
+### Example Config with Include Fields
+This example demonstrates a configuration where deduplication is applied to telemetry based on specified fields. Only logs with the same values for the fields defined in the `include_fields` parameter are deduplicated:
+
+```yaml
+receivers:
+    filelog:
+        include: [./example/*.log]
+processors:
+    logdedup:
+        include_fields:
+          - attributes.id
+          - attributes.name
+        interval: 60s
+        log_count_attribute: dedup_count
+        timezone: 'America/Los_Angeles'
+exporters:
+    googlecloud:
+
+service:
+    pipelines:
+        logs:
+            receivers: [filelog]
+            processors: [logdedup]
+            exporters: [googlecloud]
+```
 
 ### Example Config with Conditions
 The following config is an example configuration that only performs the deduping process on telemetry where Attribute `ID` equals `1` OR where Resource Attribute `service.name` equals `my-service`:

processor/logdedupprocessor/config.go

@@ -36,6 +36,7 @@ var (
 	errInvalidLogCountAttribute = errors.New("log_count_attribute must be set")
 	errInvalidInterval          = errors.New("interval must be greater than 0")
 	errCannotExcludeBody        = errors.New("cannot exclude the entire body")
+	errCannotIncludeBody        = errors.New("cannot include the entire body")
 )
 
 // Config is the config of the processor.
@@ -44,6 +45,7 @@ type Config struct {
 	Interval          time.Duration `mapstructure:"interval"`
 	Timezone          string        `mapstructure:"timezone"`
 	ExcludeFields     []string      `mapstructure:"exclude_fields"`
+	IncludeFields     []string      `mapstructure:"include_fields"`
 	Conditions        []string      `mapstructure:"conditions"`
 }
 
@@ -54,6 +56,7 @@ func createDefaultConfig() component.Config {
 		Interval:          defaultInterval,
 		Timezone:          defaultTimezone,
 		ExcludeFields:     []string{},
+		IncludeFields:     []string{},
 		Conditions:        []string{},
 	}
 }
@@ -73,7 +76,19 @@ func (c Config) Validate() error {
 		return fmt.Errorf("timezone is invalid: %w", err)
 	}
 
-	return c.validateExcludeFields()
+	if len(c.ExcludeFields) > 0 && len(c.IncludeFields) > 0 {
+		return errors.New("cannot define both exclude_fields and include_fields")
+	}
+
+	if err = c.validateExcludeFields(); err != nil {
+		return err
+	}
+
+	if err = c.validateIncludeFields(); err != nil {
+		return err
+	}
+
+	return nil
 }
 
 // validateExcludeFields validates that all the exclude fields
@@ -102,3 +117,30 @@ func (c Config) validateExcludeFields() error {
 
 	return nil
 }
+
+// validateIncludeFields validates that all the exclude fields
+func (c Config) validateIncludeFields() error {
+	knownFields := make(map[string]struct{})
+
+	for _, field := range c.IncludeFields {
+		// Special check to make sure the entire body is not included
+		if field == bodyField {
+			return errCannotIncludeBody
+		}
+
+		// Split and ensure the field starts with `body` or `attributes`
+		parts := strings.Split(field, fieldDelimiter)
+		if parts[0] != bodyField && parts[0] != attributeField {
+			return fmt.Errorf("an include_fields must start with %s or %s", bodyField, attributeField)
+		}
+
+		// If a field is valid make sure we haven't already seen it
+		if _, ok := knownFields[field]; ok {
+			return fmt.Errorf("duplicate include_fields %s", field)
+		}
+
+		knownFields[field] = struct{}{}
+	}
+
+	return nil
+}

processor/logdedupprocessor/config_test.go

@@ -84,6 +84,36 @@ func TestValidateConfig(t *testing.T) {
 			},
 			expectedErr: errors.New("duplicate exclude_field"),
 		},
+		{
+			desc: "invalid include_fields using entire body",
+			cfg: &Config{
+				LogCountAttribute: defaultLogCountAttribute,
+				Interval:          defaultInterval,
+				Timezone:          defaultTimezone,
+				IncludeFields:     []string{bodyField},
+			},
+			expectedErr: errors.New("cannot include the entire body"),
+		},
+		{
+			desc: "invalid include_fields not starting with body or attributes",
+			cfg: &Config{
+				LogCountAttribute: defaultLogCountAttribute,
+				Interval:          defaultInterval,
+				Timezone:          defaultTimezone,
+				IncludeFields:     []string{"not.valid"},
+			},
+			expectedErr: errors.New("an include_fields must start with body or attributes"),
+		},
+		{
+			desc: "empty include_fields is the default behavior",
+			cfg: &Config{
+				LogCountAttribute: defaultLogCountAttribute,
+				Interval:          defaultInterval,
+				Timezone:          defaultTimezone,
+				IncludeFields:     []string{},
+			},
+			expectedErr: nil,
+		},
 		{
 			desc: "valid config",
 			cfg: &Config{
@@ -95,6 +125,29 @@ func TestValidateConfig(t *testing.T) {
 			},
 			expectedErr: nil,
 		},
+		{
+			desc: "valid config include_fields",
+			cfg: &Config{
+				LogCountAttribute: defaultLogCountAttribute,
+				Interval:          defaultInterval,
+				Timezone:          defaultTimezone,
+				Conditions:        []string{},
+				IncludeFields:     []string{"body.thing", "attributes.otherthing"},
+			},
+			expectedErr: nil,
+		},
+		{
+			desc: "invalid config defines both exclude_fields and include_fields",
+			cfg: &Config{
+				LogCountAttribute: defaultLogCountAttribute,
+				Interval:          defaultInterval,
+				Timezone:          defaultTimezone,
+				Conditions:        []string{},
+				ExcludeFields:     []string{"body.thing", "attributes.otherthing"},
+				IncludeFields:     []string{"body.thing", "attributes.otherthing"},
+			},
+			expectedErr: errors.New("cannot define both exclude_fields and include_fields"),
+		},
 	}
 
 	for _, tc := range testCases {

processor/logdedupprocessor/counter.go

@@ -29,15 +29,17 @@ type logAggregator struct {
 	logCountAttribute string
 	timezone          *time.Location
 	telemetryBuilder  *metadata.TelemetryBuilder
+	dedupFields       []string
 }
 
 // newLogAggregator creates a new LogCounter.
-func newLogAggregator(logCountAttribute string, timezone *time.Location, telemetryBuilder *metadata.TelemetryBuilder) *logAggregator {
+func newLogAggregator(logCountAttribute string, timezone *time.Location, telemetryBuilder *metadata.TelemetryBuilder, dedupFields []string) *logAggregator {
 	return &logAggregator{
 		resources:         make(map[uint64]*resourceAggregator),
 		logCountAttribute: logCountAttribute,
 		timezone:          timezone,
 		telemetryBuilder:  telemetryBuilder,
+		dedupFields:       dedupFields,
 	}
 }
 
@@ -83,7 +85,7 @@ func (l *logAggregator) Add(resource pcommon.Resource, scope pcommon.Instrumenta
 	key := getResourceKey(resource)
 	resourceAggregator, ok := l.resources[key]
 	if !ok {
-		resourceAggregator = newResourceAggregator(resource)
+		resourceAggregator = newResourceAggregator(resource, l.dedupFields)
 		l.resources[key] = resourceAggregator
 	}
 	resourceAggregator.Add(scope, logRecord)
@@ -98,13 +100,15 @@ func (l *logAggregator) Reset() {
 type resourceAggregator struct {
 	resource      pcommon.Resource
 	scopeCounters map[uint64]*scopeAggregator
+	dedupFields   []string
 }
 
 // newResourceAggregator creates a new ResourceCounter.
-func newResourceAggregator(resource pcommon.Resource) *resourceAggregator {
+func newResourceAggregator(resource pcommon.Resource, dedupFields []string) *resourceAggregator {
 	return &resourceAggregator{
 		resource:      resource,
 		scopeCounters: make(map[uint64]*scopeAggregator),
+		dedupFields:   dedupFields,
 	}
 }
 
@@ -113,7 +117,7 @@ func (r *resourceAggregator) Add(scope pcommon.InstrumentationScope, logRecord p
 	key := getScopeKey(scope)
 	scopeAggregator, ok := r.scopeCounters[key]
 	if !ok {
-		scopeAggregator = newScopeAggregator(scope)
+		scopeAggregator = newScopeAggregator(scope, r.dedupFields)
 		r.scopeCounters[key] = scopeAggregator
 	}
 	scopeAggregator.Add(logRecord)
@@ -123,19 +127,21 @@ func (r *resourceAggregator) Add(scope pcommon.InstrumentationScope, logRecord p
 type scopeAggregator struct {
 	scope       pcommon.InstrumentationScope
 	logCounters map[uint64]*logCounter
+	dedupFields []string
 }
 
 // newScopeAggregator creates a new ScopeCounter.
-func newScopeAggregator(scope pcommon.InstrumentationScope) *scopeAggregator {
+func newScopeAggregator(scope pcommon.InstrumentationScope, dedupFields []string) *scopeAggregator {
 	return &scopeAggregator{
 		scope:       scope,
 		logCounters: make(map[uint64]*logCounter),
+		dedupFields: dedupFields,
 	}
 }
 
 // Add increments the counter that the logRecord matches.
 func (s *scopeAggregator) Add(logRecord plog.LogRecord) {
-	key := getLogKey(logRecord)
+	key := getLogKey(logRecord, s.dedupFields)
 	lc, ok := s.logCounters[key]
 	if !ok {
 		lc = newLogCounter(logRecord)
@@ -184,12 +190,62 @@ func getScopeKey(scope pcommon.InstrumentationScope) uint64 {
 	)
 }
 
-// getLogKey creates a unique hash for the log record to use as a map key
-func getLogKey(logRecord plog.LogRecord) uint64 {
+// getLogKey creates a unique hash for the log record to use as a map key.
+// If dedupFields is non-empty, it is used to determine the fields whose values are hashed.
+func getLogKey(logRecord plog.LogRecord, dedupFields []string) uint64 {
+	if len(dedupFields) > 0 {
+		var opts []pdatautil.HashOption
+
+		for _, field := range dedupFields {
+			parts := splitField(field)
+			var m pcommon.Map
+			switch parts[0] {
+			case bodyField:
+				if logRecord.Body().Type() == pcommon.ValueTypeMap {
+					m = logRecord.Body().Map()
+				}
+			case attributeField:
+				m = logRecord.Attributes()
+			}
+
+			value, ok := getKeyValue(m, parts[1:])
+			if ok {
+				opts = append(opts, pdatautil.WithString(value.AsString()))
+			}
+		}
+
+		if len(opts) > 0 {
+			return pdatautil.Hash64(opts...)
+		}
+	}
+
 	return pdatautil.Hash64(
 		pdatautil.WithMap(logRecord.Attributes()),
 		pdatautil.WithValue(logRecord.Body()),
 		pdatautil.WithString(logRecord.SeverityNumber().String()),
 		pdatautil.WithString(logRecord.SeverityText()),
 	)
 }
+
+func getKeyValue(valueMap pcommon.Map, keyParts []string) (pcommon.Value, bool) {
+	nextKeyPart, remainingParts := keyParts[0], keyParts[1:]
+
+	// Look for the value associated with the next key part.
+	// If we don't find it then return
+	value, ok := valueMap.Get(nextKeyPart)
+	if !ok {
+		return pcommon.NewValueEmpty(), false
+	}
+
+	// No more key parts that means we have found the value
+	if len(remainingParts) == 0 {
+		return valueMap.Get(nextKeyPart)
+	}
+
+	// If the value is a map then recurse through with the remaining parts
+	if value.Type() == pcommon.ValueTypeMap {
+		return getKeyValue(value.Map(), remainingParts)
+	}
+
+	return pcommon.NewValueEmpty(), false
+}