New component: Enrichment Processor

[jsvd]

The purpose and use-cases of the new component

This issue is a follow up to a presentation regarding enhancing enrichment capabilities of the Collector on the Collector SIG meeting (July 23rd).
The feedback was to create an issue to get the discussion going, so here it is:

The OpenTelemetry Collector currently supports limited enrichment types, mostly focusing on self-contained parsing and contextual metadata. To improve the versatility of the Collector in comparison to other data collectors and transformation tools, we should expand its capabilities to include other enrichment types.

The original document “Enrichment in OTel Collector” introduces a taxonomy of enrichment types and its support in the Collector:
• Type 1: Self-Contained Parsing & Derivation (supported)
• Type 2: Reference Data Lookup (Static or Semi-Static) (very limited support)
• Type 3: Dynamic External Enrichment (Live Lookups) (not supported)
• Type 4: Contextual Metadata Enrichment (supported)
• Type 5: Cross-Event Correlation & Aggregation (not supported)
• Type 6: Analytical & ML-Based Enrichment (not supported)

Of this list, looking at similar tools (comparison can be seen in the original document), type 2 and type 3 are the strongest candidates to include in the Collector to facilitate migration of workloads to the Collector from other tools.

From this problem statement we could consider introducing a Lookup Processor to aimed at handling both static reference data lookups (Type 2) and dynamic external enrichments (Type 3).

The processor would support:

• Local lookups: Using static or semi-static data sources such as CSV, JSON, or inline key/value pairs.
• Remote lookups: Dynamic enrichment from APIs, DNS, databases, or cache systems like Redis or Memcached.

Example configuration for the component

processors:
  lookupprocessor/json:
    source: json
    path: "/tmp/file.json"
    json_path: .process.name
    source_attribute: process.name
    target_attribute: aws.log.group.names
    timeout: 25ms
    refresh_interval: 1h

  lookupprocessor/http:
    source: http
    url: "https://my.url/query"
    params:
      org: resource.attributes["foo"]
    target_attribute: org
    timeout: 1s
    cache:
      size: 100
      ttl: 60 # seconds

Implementing such a processor requires significant considerations to abide by the long term vision, namely:
• Progressive implementation, starting with basic local and then remote lookup capabilities.
• Modular structure facilitating easy addition of new lookup sources.
• Built-in caching and timeout mechanisms for performance optimization.
• Comprehensive and useful observability metrics (success/failure rates, latency percentiles).

This is not an exhaustive list of concerns and neither does it provide solutions, just an acknowledgment that these will have to be addressed.

Telemetry data types supported

Logs, Metrics, Traces, Profiles

Code Owner(s)

@jsvd

Sponsor (optional)

No response

Additional context

Similar ideas and alternative solutions have been suggested in the past:

• New component: GRPC processor/connector #20888
• New component: Alert Manager Receiver and Exporter  #18526
• Generic resource detector  #29627
• New component: Enrich Attributes based on key matching from yaml or csv definition #40526
• Ability to enrich telemetry with additional resource metadata from an inventory datasource #40936
• New component: DNS lookup processor #34398

Tip

_{React with 👍 to help prioritize this issue. Please use comments to provide useful context, avoiding +1 or me too, to help us triage it. Learn more here.}

[dehaansa]

I definitely support this concept, but do think the implementation details are important. I think in the interest of maximizing flexibility, what makes the most sense to me is to create a new extension type, and then a generic processor to utilize them.

• Allows for greatest flexibility in lookup logic. If a specific distribution needs their own lookup logic, they only need to implement an extension and include it in their distribution, it doesn't require a component fork or upstream contribution.
• Minimizes the configuration complexity required within the lookup processor.
• Allows for reuse of lookup logic. While the generic processor is a great base use case, there are certainly independent components that may want to utilize lookup on tighter guardrails or limit the kinds of lookups that can be used.
• Distributes codeownership of lookup logic, so it doesn't need to all be owned by the processor author(s).

Depending on how the conversation progresses I may be interested in sponsoring, and contributing.

[jsvd]

The original idea of having the component + its sources in the same location was taken from the resourcedetectionprocessor, with a few arguments in favor:

1. reduces the spread of elements needed for the processor to work across the collector code base. Lookup sources will require functionality (e.g. caching, error handling, metric instrumentation, circuit breakers) that should be shared and it'd be easier to access it from the single processor code base
2. ownership of sources can be attributed to others that are not the maintainers of the processor, e.g. https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/7589c18/.github/CODEOWNERS#L205-L206*

This said, the ability to customize the set of lookup sources in a collector distro is significant and the extension suggestion solves it perfectly.
I am not sure what would be the use cases for reusing the lookup sources logic in other components outside the generic lookup processor but I could see it happening for the local file based lookups?

Lastly, thanks for the very quick feedback and taking interest in it, let's keep talking!

[jsvd]

Expanding on the extension idea, I could foresee the implementation look something like this:

Lookup processor:

Accept a single extension, perform one or more lookups, configure an error mode.

type Config struct {
	Extension component.ID `mapstructure:"extension"`
	Lookups []LookupConfig `mapstructure:"lookups"`
         // define if non-missing key errors should terminate the pipeline
	ErrorMode lookup.ErrorMode `mapstructure:"error_mode"`
}

type LookupConfig struct {
	Field string `mapstructure:"field"`
	TargetField string `mapstructure:"target_field"`
         // define if missing keys should terminate the pipeline
	Optional bool `mapstructure:"optional"`
}

Extensions:

located under "pkg/lookup" create the necessary interfaces for source extensions:

pkg/lookup/source.go:

type Source interface {
	Lookup(ctx context.Context, key string) (value any, found bool, err error)
	Type() string
	Start(ctx context.Context, host component.Host) error
	Shutdown(ctx context.Context) error
}

type LookupExtension interface {
	Source
}

pkg/lookup/errors.go:

type ErrorMode string

const (
	ErrorModeIgnore ErrorMode = "ignore"
	ErrorModePropagate ErrorMode = "propagate"
)

We can also have orthogonal but optional behaviors such as caching:

pkg/lookup/cache.go:

type cacheEntry[V any] struct {
	key        string
	value      V
	expiration time.Time
}

type Cache[V any] struct {
	mu      sync.RWMutex
	maxSize int
	ttl     time.Duration
	entries map[string]*list.Element
	lru     *list.List
}

func NewCache[V any](maxSize int, ttl time.Duration) *Cache[V] { }
func (c *Cache[V]) Get(key string) (V, bool) { }
func (c *Cache[V]) Set(key string, value V) { .. }
func (c *Cache[V]) Clear() { .. }

Example extension (memcache)

In extension/lookupsource/memcachedlookupextension/extension.go:

const TypeStr = "memcached"

type memcachedExtension struct {
	*lookup.BaseSource
	config *Config
	client memcachedClient
}

func NewMemcachedExtension(config *Config, settings component.TelemetrySettings) (*memcachedExtension, error) {
	// BaseSource with caching
	base, err := lookup.NewBaseSource(TypeStr, settings, &config.Cache)
	if err != nil {
		return nil, err
	}

	return &memcachedExtension{
		BaseSource: base,
		config:     config,
	}, nil
}

User experience

The configuration would look like:

extensions:
  memcachedlookup:
    servers: ["localhost:11211"]
    cache:
      enabled: true
      size: 100
      ttl: 30s

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317

processors:
  lookup/memcached:
    source:
      extension: memcachedlookup
    attributes:
      - key: cached.user
        from_attribute: user_id
        mode: upsert
      - key: cached.data
        from_attribute: lookup_key
        mode: upsert
    error_mode: ignore

exporters:
  debug:
    verbosity: detailed

service:
  extensions: [memcachedlookup]
  pipelines:
    traces:
      receivers: [otlp]
      processors: [lookup/memcached]
      exporters: [debug]