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.

Lookups should be done according to the scope that they're configured for depending on the location of the attribute used for the lookups. The configuration below shows examples of resource and log level lookups.

Example configuration for the component

processors:
  # YAML source - map service.name to display name
  lookup/yaml:
    source:
      type: yaml
      path: ./mappings.yaml
    attributes:
      - key: service.display_name
        from_attribute: service.name
        default: "Unknown Service"
        action: upsert
        context: resource

  # HTTP source - lookup user name from API
  lookup/http:
    source:
      type: http
      url: "http://localhost:8080/users/{key}"
      method: GET
      timeout: 5s
      response_field: "name"
      cache:
        enabled: true
        size: 1000
        ttl: 5m
    attributes:
      - key: user.name
        from_attribute: user.id
        context: log
        action: upsert

  # DNS source - reverse DNS lookup for client IP
  lookup/dns:
    source:
      type: dns
      record_type: PTR
      timeout: 2s
      server: "8.8.8.8:53"
      cache:
        enabled: true
        size: 1000
        ttl: 5m
    attributes:
      - key: client.hostname
        from_attribute: client.ip
        default: "unknown"
        action: upsert
        context: log

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 first given that enrichment is a more common logging use case requirement.
This could then be extended to Metrics, Traces, and Profiles.

Code Owner(s)

@jsvd, @VihasMakwana, @dehaansa

Sponsor (optional)

@dehaansa

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

Trade Offs Considered

New component vs resourcedetectionprocessor

Conclusion: new component

While there are similarities to the resourcedetectionprocessor, performing lookups to a remote source is not in scope for resource detection which, according to its README, is to detect resource information from the host, while this processor would look to data that is not necessarily either present in the host or about the host.
Also these remote lookups introduce enough performance concerns that are better addressed with a separate implementation such as supporting configurable caching.

Extensibility method - extensions vs NewFactoryWithOptions

Conclusion: NewFactoryWithOptions

This processor should support third-party lookup sources. This could either be implemented by collector extensions or with an extensible factory using NewFactoryWithOptions (like filterprocessor and transformprocessors).

For now the decision is to use NewFactoryWithOptions not not extensions, as extensions in the collector should be reusable by any component, and there are no current use cases requiring lookups during a component's execution, like the storageextension, that can't be replaced by using a lookupprocessor before or after said component.

Signal Support

Conclusion: Logs first, others after

The decision is to support Log signal first, the signal that most often needs such enrichment, and then extend to others as the processor matures.

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]