feat(opentelemetry source): allow to add headers to metadata for OpenTelemetry metrics and traces

[ozanichkovsky]

Summary

Adds header enrichment support to OpenTelemetry source metrics and traces.

Unlike logs, which support either legacy way of adding headers directly to the event or to the event metadata, this implementation adds headers to metric/trace event metadata only and ignores log_namespace settings on the source.

One of the use cases would be a tenant separation by a header.

Vector configuration

api:
  enabled: true
  address: "127.0.0.1:8686"

sources:
  opentelemetry:
    type: "opentelemetry"
    grpc:
      address: 0.0.0.0:4317
    http:
      address: 0.0.0.0:4318
      headers:
        - "X-Tenant-ID"
    use_otlp_decoding: false
    log_namespace: true

transforms:
  debug:
    type: remap
    inputs:
      - "opentelemetry.metrics"
      - "opentelemetry.logs"
      - "opentelemetry.traces" 
    source: |
      log(%opentelemetry.headers."X-Tenant-ID")

sinks:
  console_output:
    type: "console"
    inputs:
      - "debug"
    encoding:
      codec: "native_json"

How did you test this PR?

The code is covered with unit tests and also was tested with different OTLP generators implemented in go, like this one for metrics:

package main

import (
	"context"
	"flag"
	"fmt"
	"log"
	"os"
	"time"

	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp"
	"go.opentelemetry.io/otel/metric"
	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
)

func main() {
	var (
		endpoint = flag.String("endpoint", "http://localhost:4318", "OTLP/HTTP base endpoint, e.g. http://localhost:4318")
		header   = flag.String("header", "12345678", "Value for X-Tenant-ID")
		value    = flag.Int64("value", 1, "Counter increment value")
	)
	flag.Parse()

	ctx := context.Background()

	// Create OTLP metric HTTP exporter with custom headers.
	// For OTLP/HTTP, metrics usually go to <base>/v1/metrics.
	exporter, err := otlpmetrichttp.New(ctx,
		otlpmetrichttp.WithEndpointURL(*endpoint),
		otlpmetrichttp.WithURLPath("/v1/metrics"),
		otlpmetrichttp.WithHeaders(map[string]string{
			"X-Tenant-ID": *header,
			// Example:
			// "Authorization": "Bearer " + *header,
		}),
	)
	if err != nil {
		log.Fatalf("create exporter: %v", err)
	}

	reader := sdkmetric.NewPeriodicReader(exporter,
		sdkmetric.WithInterval(2*time.Second),
	)

	mp := sdkmetric.NewMeterProvider(
		sdkmetric.WithReader(reader),
	)

	defer func() {
		shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		if err := mp.Shutdown(shutdownCtx); err != nil {
			fmt.Fprintf(os.Stderr, "shutdown error: %v\n", err)
		}
	}()

	meter := mp.Meter("example-counter-cli")

	counter, err := meter.Int64Counter(
		"example_counter_total",
		metric.WithDescription("A simple demo counter sent via OTLP/HTTP"),
	)
	if err != nil {
		log.Fatalf("create counter: %v", err)
	}

	counter.Add(ctx, *value,
		metric.WithAttributes(
			attribute.String("service.name", "counter-cli"),
			attribute.String("env", "dev"),
		),
	)

	// Force an export now so the process can exit immediately.
	flushCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	if err := mp.ForceFlush(flushCtx); err != nil {
		log.Fatalf("force flush: %v", err)
	}

	fmt.Println("metric sent")
}

Change Type

• Bug fix
• New feature
• Dependencies
• Non-functional (chore, refactoring, docs)
• Performance

Is this a breaking change?

• Yes
• No

Does this PR include user facing changes?

• Yes. Please add a changelog fragment based on our guidelines.
• No. A maintainer will apply the no-changelog label to this PR.

References

Closes: #24619

Notes

• Please read our Vector contributor resources.
• Do not hesitate to use @vectordotdev/vector to reach out to us regarding this PR.
• Some CI checks run only after we manually approve them.
  • We recommend adding a pre-push hook, please see this template.
  • Alternatively, we recommend running the following locally before pushing to the remote branch:
    • make fmt
    • make check-clippy (if there are failures it's possible some of them can be fixed with make clippy-fix)
    • make test
• After a review is requested, please avoid force pushes to help us review incrementally.
  • Feel free to push as many commits as you want. They will be squashed into one before merging.
  • For example, you can run git merge origin master and git push.
• If this PR introduces changes Vector dependencies (modifies Cargo.lock), please
  run make build-licenses to regenerate the license inventory and commit the changes (if any). More details here.

[ozanichkovsky]

@vectordotdev/vector please review

[pront]

@codex review

[chatgpt-codex-connector]

Codex Review: Didn't find any major issues. Delightful!

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[urseberry]

I approved on behalf of documentation. Engineering will review also.

[ozanichkovsky]

@vectordotdev/vector I think I implemented everything I wanted to. How does it look in your opinion?

[pront]

Thanks @ozanichkovsky