mirror of
https://github.com/hanzoai/telemetry.git
synced 2026-10-30 20:01:49 +00:00
155fb0264e
#### Description Feature: The enablement of the sending_queue in the configuration of the otplhttp exporter does not preserve the metadata information of the initial context. This pr introduces a new configuration called `metadata_keys` that will be used in the exporter config #### Testing - Added partionetr_test - Built my local otlp exporter Configuration of otlphttp: ```yaml otlphttp: endpoint: https://endpoint.invalid/_otlp timeout: {{ .http_request_timeout | quote }} sending_queue: enabled: true wait_for_result: false block_on_overflow: true # false = data loss on queue full sizer: requests queue_size: {{ .queuedRequests }} batch: partition: metadata_keys: - project-id - tenant-id ``` See "ProjectID":"local" that is populated below in my logs ```yaml {"log.level":"info","@timestamp":"2025-11-06T15:51:49.955Z","message":"ecpRoutingRoundTripper: after URL modification","resource":{"cloud.availability_zone":"local","k8s.namespace.name":"motel-index-collector","k8s.node.name":"hotel-worker2","k8s.pod.name":"motel-index-collector-local-58484b67c8-l4vrt","k8s.pod.uid":"1a8f6fad-91b6-413a-a929-6d004ca489b3","orchestrator.cluster.name":"default","orchestrator.deploymentslice":"","orchestrator.environment":"default","service.instance.id":"ecf43bce-5aac-444c-b244-8871bbf2d69a","service.name":"motel-index-collector","service.version":"git"},"otelcol.component.id":"ecproutingmiddleware","otelcol.component.kind":"extension","final_url_host":"local.es.svc.cluster.local:9200","final_url_scheme":"http","final_url":"[http://local.es.svc.cluster.local:9200/_otlp/v1/metrics","final_host":"","ProjectID":"local","header_count":6,"ecs.version":"1.6.0"}](http://local.es.svc.cluster.local:9200/_otlp/v1/metrics%22,%22final_host%22:%22%22,%22ProjectID%22:%22local%22,%22header_count%22:6,%22ecs.version%22:%221.6.0%22%7D) ``` This PR is a feature for exporterhelper, so is not just otlphttp-specific. --------- Signed-off-by: Andreas Gkizas <[email protected]> Co-authored-by: Andrew Wilkins <[email protected]>
75 lines
1.9 KiB
Go
75 lines
1.9 KiB
Go
// Copyright The OpenTelemetry Authors
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
|
|
package queuebatch // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
|
|
"go.opentelemetry.io/collector/client"
|
|
"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
|
|
)
|
|
|
|
// metadataKeysPartitioner partitions requests based on client metadata keys.
|
|
type metadataKeysPartitioner struct {
|
|
keys []string
|
|
}
|
|
|
|
// NewMetadataKeysPartitioner creates a new partitioner that partitions requests
|
|
// based on the specified metadata keys. If keys is empty, returns nil.
|
|
func NewMetadataKeysPartitioner(keys []string) Partitioner[request.Request] {
|
|
if len(keys) == 0 {
|
|
return nil
|
|
}
|
|
return &metadataKeysPartitioner{keys: keys}
|
|
}
|
|
|
|
// GetKey returns a partition key based on the metadata keys configured.
|
|
func (p *metadataKeysPartitioner) GetKey(
|
|
ctx context.Context,
|
|
_ request.Request,
|
|
) string {
|
|
var kb bytes.Buffer
|
|
meta := client.FromContext(ctx).Metadata
|
|
|
|
var afterFirst bool
|
|
for _, k := range p.keys {
|
|
if values := meta.Get(k); len(values) != 0 {
|
|
if afterFirst {
|
|
kb.WriteByte(0)
|
|
}
|
|
kb.WriteString(k)
|
|
afterFirst = true
|
|
for _, val := range values {
|
|
kb.WriteByte(0)
|
|
kb.WriteString(val)
|
|
}
|
|
}
|
|
}
|
|
return kb.String()
|
|
}
|
|
|
|
// NewMetadataKeysMergeCtx creates a merge function for contexts that merges
|
|
// metadata based on the specified keys. If keys is empty, returns nil.
|
|
func NewMetadataKeysMergeCtx(keys []string) func(context.Context, context.Context) context.Context {
|
|
if len(keys) == 0 {
|
|
return nil
|
|
}
|
|
return func(ctx1, _ context.Context) context.Context {
|
|
m1 := client.FromContext(ctx1).Metadata
|
|
|
|
m := make(map[string][]string, len(keys))
|
|
for _, key := range keys {
|
|
v1 := m1.Get(key)
|
|
if len(v1) > 0 {
|
|
m[key] = v1
|
|
}
|
|
}
|
|
return client.NewContext(
|
|
context.Background(),
|
|
client.Info{Metadata: client.NewMetadata(m)},
|
|
)
|
|
}
|
|
}
|