[processor/redaction] add support for redacting metrics and logs attributes

.chloggen/redaction-add-metrics-and-logs.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: redactionprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support for logs and metrics
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [34479]
+
+# (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: []

processor/redactionprocessor/README.md

@@ -3,7 +3,7 @@
 <!-- status autogenerated section -->
 | Status        |           |
 | ------------- |-----------|
-| Stability     | [beta]: traces   |
+| Stability     | [beta]: traces, logs, metrics   |
 | Distributions | [contrib] |
 | Issues        | [![Open issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aopen%20label%3Aprocessor%2Fredaction%20&label=open&color=orange&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aopen+is%3Aissue+label%3Aprocessor%2Fredaction) [![Closed issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aclosed%20label%3Aprocessor%2Fredaction%20&label=closed&color=blue&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aclosed+is%3Aissue+label%3Aprocessor%2Fredaction) |
 | [Code Owners](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/CONTRIBUTING.md#becoming-a-code-owner)    | [@dmitryax](https://www.github.com/dmitryax), [@mx-psi](https://www.github.com/mx-psi), [@TylerHelmuth](https://www.github.com/TylerHelmuth) |
@@ -13,9 +13,9 @@
 [contrib]: https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib
 <!-- end autogenerated section -->
 
-This processor deletes span attributes that don't match a list of allowed span
-attributes. It also masks span attribute values that match a blocked value
-list. Span attributes that aren't on the allowed list are removed before any
+This processor deletes span/log/metric attributes that don't match a list of allowed
+attributes. It also masks attribute values that match a blocked value
+list. Attributes that aren't on the allowed list are removed before any
 value checks are done.
 
 ## Use Cases
@@ -57,9 +57,9 @@ processors:
     # allowed_keys list. The list of blocked_values is applied regardless. If
     # you just want to block values, set this to true.
     allow_all_keys: false
-    # allowed_keys is a list of span attribute keys that are kept on the span and
+    # allowed_keys is a list of span/log/metric attribute keys that are kept on the span/log/metric and
     # processed. The list is designed to fail closed. If allowed_keys is empty,
-    # no span attributes are allowed and all span attributes are removed. To
+    # no attributes are allowed and all span attributes are removed. To
     # allow all keys, set allow_all_keys to true.
     allowed_keys:
       - description
@@ -76,7 +76,7 @@ processors:
       - "4[0-9]{12}(?:[0-9]{3})?" ## Visa credit card number
       - "(5[1-5][0-9]{14})"       ## MasterCard number
     # summary controls the verbosity level of the diagnostic attributes that
-    # the processor adds to the spans when it redacts or masks other
+    # the processor adds to the spans/logs/metrics when it redacts or masks other
     # attributes. In some contexts a list of redacted attributes leaks
     # information, while it is valuable when integrating and testing a new
     # configuration. Possible values:
@@ -93,16 +93,16 @@ Ignored attributes are processed first so they're always allowed and never
 blocked. This field should only be used where you know the data is always
 safe to send to the telemetry system.
 
-Only span attributes included on the list of allowed keys list are retained.
-If `allowed_keys` is empty, then no span attributes are allowed. All span
+Only span/log/metric attributes included on the list of allowed keys list are retained.
+If `allowed_keys` is empty, then no attributes are allowed. All
 attributes are removed in that case. To keep all span attributes, you should
 explicitly set `allow_all_keys` to true.
 
 `blocked_values` applies to the values of the allowed keys. If the value of an
 allowed key matches the regular expression for a blocked value, the matching
 part of the value is then masked with a fixed length of asterisks.
 
-For example, if `notes` is on the list of allowed keys, then the `notes` span
+For example, if `notes` is on the list of allowed keys, then the `notes`
 attribute is retained. However, if there is a value such as a credit card
 number in the `notes` field that matched a regular expression on the list of
 blocked values, then that value is masked.

processor/redactionprocessor/factory.go

@@ -23,6 +23,8 @@ func NewFactory() processor.Factory {
 		metadata.Type,
 		createDefaultConfig,
 		processor.WithTraces(createTracesProcessor, metadata.TracesStability),
+		processor.WithLogs(createLogsProcessor, metadata.LogsStability),
+		processor.WithMetrics(createMetricsProcessor, metadata.MetricsStability),
 	)
 }
 
@@ -53,3 +55,47 @@ func createTracesProcessor(
 		redaction.processTraces,
 		processorhelper.WithCapabilities(consumer.Capabilities{MutatesData: true}))
 }
+
+// createLogsProcessor creates an instance of redaction for processing logs
+func createLogsProcessor(
+	ctx context.Context,
+	set processor.Settings,
+	cfg component.Config,
+	next consumer.Logs) (processor.Logs, error) {
+	oCfg := cfg.(*Config)
+
+	red, err := newRedaction(ctx, oCfg, set.Logger)
+	if err != nil {
+		return nil, fmt.Errorf("error creating a redaction processor: %w", err)
+	}
+
+	return processorhelper.NewLogsProcessor(
+		ctx,
+		set,
+		cfg,
+		next,
+		red.processLogs,
+		processorhelper.WithCapabilities(consumer.Capabilities{MutatesData: true}))
+}
+
+// createMetricsProcessor creates an instance of redaction for processing metrics
+func createMetricsProcessor(
+	ctx context.Context,
+	set processor.Settings,
+	cfg component.Config,
+	next consumer.Metrics) (processor.Metrics, error) {
+	oCfg := cfg.(*Config)
+
+	red, err := newRedaction(ctx, oCfg, set.Logger)
+	if err != nil {
+		return nil, fmt.Errorf("error creating a redaction processor: %w", err)
+	}
+
+	return processorhelper.NewMetricsProcessor(
+		ctx,
+		set,
+		cfg,
+		next,
+		red.processMetrics,
+		processorhelper.WithCapabilities(consumer.Capabilities{MutatesData: true}))
+}

processor/redactionprocessor/factory_test.go

@@ -26,3 +26,21 @@ func TestCreateTestProcessor(t *testing.T) {
 	assert.NotNil(t, tp)
 	assert.Equal(t, true, tp.Capabilities().MutatesData)
 }
+
+func TestCreateTestLogsProcessor(t *testing.T) {
+	cfg := &Config{}
+
+	tp, err := createLogsProcessor(context.Background(), processortest.NewNopSettings(), cfg, consumertest.NewNop())
+	assert.NoError(t, err)
+	assert.NotNil(t, tp)
+	assert.Equal(t, true, tp.Capabilities().MutatesData)
+}
+
+func TestCreateTestMetricsProcessor(t *testing.T) {
+	cfg := &Config{}
+
+	tp, err := createMetricsProcessor(context.Background(), processortest.NewNopSettings(), cfg, consumertest.NewNop())
+	assert.NoError(t, err)
+	assert.NotNil(t, tp)
+	assert.Equal(t, true, tp.Capabilities().MutatesData)
+}

processor/redactionprocessor/metadata.yaml

@@ -3,7 +3,7 @@ type: redaction
 status:
   class: processor
   stability:
-    beta: [traces]
+    beta: [traces,logs,metrics]
   distributions: [contrib]
   codeowners:
     active: [dmitryax, mx-psi, TylerHelmuth]

processor/redactionprocessor/processor.go

@@ -11,6 +11,8 @@ import (
 	"strings"
 
 	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.uber.org/zap"
 )
@@ -59,6 +61,22 @@ func (s *redaction) processTraces(ctx context.Context, batch ptrace.Traces) (ptr
 	return batch, nil
 }
 
+func (s *redaction) processLogs(ctx context.Context, logs plog.Logs) (plog.Logs, error) {
+	for i := 0; i < logs.ResourceLogs().Len(); i++ {
+		rl := logs.ResourceLogs().At(i)
+		s.processResourceLog(ctx, rl)
+	}
+	return logs, nil
+}
+
+func (s *redaction) processMetrics(ctx context.Context, metrics pmetric.Metrics) (pmetric.Metrics, error) {
+	for i := 0; i < metrics.ResourceMetrics().Len(); i++ {
+		rm := metrics.ResourceMetrics().At(i)
+		s.processResourceMetric(ctx, rm)
+	}
+	return metrics, nil
+}
+
 // processResourceSpan processes the RS and all of its spans and then returns the last
 // view metric context. The context can be used for tests
 func (s *redaction) processResourceSpan(ctx context.Context, rs ptrace.ResourceSpans) {
@@ -79,6 +97,37 @@ func (s *redaction) processResourceSpan(ctx context.Context, rs ptrace.ResourceS
 	}
 }
 
+// processResourceLog processes the log resource and all of its logs and then returns the last
+// view metric context. The context can be used for tests
+func (s *redaction) processResourceLog(ctx context.Context, rl plog.ResourceLogs) {
+	rsAttrs := rl.Resource().Attributes()
+
+	s.processAttrs(ctx, rsAttrs)
+
+	for j := 0; j < rl.ScopeLogs().Len(); j++ {
+		ils := rl.ScopeLogs().At(j)
+		for k := 0; k < rl.ScopeLogs().Len(); k++ {
+			log := ils.LogRecords().At(k)
+			s.processAttrs(ctx, log.Attributes())
+		}
+	}
+}
+
+func (s *redaction) processResourceMetric(ctx context.Context, rm pmetric.ResourceMetrics) {
+	rsAttrs := rm.Resource().Attributes()
+
+	s.processAttrs(ctx, rsAttrs)
+
+	for j := 0; j < rm.ScopeMetrics().Len(); j++ {
+		ils := rm.ScopeMetrics().At(j)
+		s.processAttrs(ctx, ils.Scope().Attributes())
+		for k := 0; k < ils.Metrics().Len(); k++ {
+			metric := ils.Metrics().At(k)
+			s.processAttrs(ctx, metric.Metadata())
+		}
+	}
+}
+
 // processAttrs redacts the attributes of a resource span or a span
 func (s *redaction) processAttrs(_ context.Context, attributes pcommon.Map) {
 	// TODO: Use the context for recording metrics