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]>
421 lines
11 KiB
Go
421 lines
11 KiB
Go
// Copyright The OpenTelemetry Authors
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
|
|
package queuebatch // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"go.opentelemetry.io/collector/client"
|
|
"go.opentelemetry.io/collector/exporter/exporterhelper/internal/requesttest"
|
|
)
|
|
|
|
func TestMetadataKeysPartitioner_MergeCtx(t *testing.T) {
|
|
t.Run("merge contexts with same metadata key values", func(t *testing.T) {
|
|
mergeCtx := NewMetadataKeysMergeCtx([]string{"key1", "key2"})
|
|
|
|
ctx1 := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value1"},
|
|
"key2": {"value2"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
ctx2 := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value1"},
|
|
"key2": {"value2"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
mergedCtx := mergeCtx(ctx1, ctx2)
|
|
require.NotNil(t, mergedCtx)
|
|
|
|
mergedMeta := client.FromContext(mergedCtx).Metadata
|
|
assert.Equal(t, []string{"value1"}, mergedMeta.Get("key1"))
|
|
assert.Equal(t, []string{"value2"}, mergedMeta.Get("key2"))
|
|
})
|
|
|
|
t.Run("merge contexts with same metadata key values and additional keys", func(t *testing.T) {
|
|
mergeCtx := NewMetadataKeysMergeCtx([]string{"key1"})
|
|
|
|
ctx1 := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value1"},
|
|
"other": {"other1"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
ctx2 := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value1"},
|
|
"other": {"other2"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
mergedCtx := mergeCtx(ctx1, ctx2)
|
|
require.NotNil(t, mergedCtx)
|
|
|
|
mergedMeta := client.FromContext(mergedCtx).Metadata
|
|
assert.Equal(t, []string{"value1"}, mergedMeta.Get("key1"))
|
|
// Other keys should not be in merged metadata since they're not in partitioner keys
|
|
assert.Empty(t, mergedMeta.Get("other"))
|
|
})
|
|
|
|
t.Run("merge contexts with empty metadata", func(t *testing.T) {
|
|
mergeCtx := NewMetadataKeysMergeCtx([]string{"key1"})
|
|
|
|
ctx1 := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{}),
|
|
},
|
|
)
|
|
|
|
ctx2 := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{}),
|
|
},
|
|
)
|
|
|
|
mergedCtx := mergeCtx(ctx1, ctx2)
|
|
require.NotNil(t, mergedCtx)
|
|
|
|
mergedMeta := client.FromContext(mergedCtx).Metadata
|
|
assert.Empty(t, mergedMeta.Get("key1"))
|
|
})
|
|
|
|
t.Run("merge when one context has metadata and other is empty", func(t *testing.T) {
|
|
mergeCtx := NewMetadataKeysMergeCtx([]string{"key1"})
|
|
|
|
ctx1 := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value1"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
ctx2 := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{}),
|
|
},
|
|
)
|
|
|
|
mergedCtx := mergeCtx(ctx1, ctx2)
|
|
require.NotNil(t, mergedCtx)
|
|
mergedMeta := client.FromContext(mergedCtx).Metadata
|
|
assert.Equal(t, []string{"value1"}, mergedMeta.Get("key1"))
|
|
})
|
|
|
|
t.Run("merge when contexts have different metadata key values", func(t *testing.T) {
|
|
mergeCtx := NewMetadataKeysMergeCtx([]string{"key1"})
|
|
|
|
ctx1 := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value1"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
ctx2 := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value2"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
mergedCtx := mergeCtx(ctx1, ctx2)
|
|
require.NotNil(t, mergedCtx)
|
|
mergedMeta := client.FromContext(mergedCtx).Metadata
|
|
assert.Equal(t, []string{"value1"}, mergedMeta.Get("key1"))
|
|
})
|
|
|
|
t.Run("merge when contexts have different metadata key values for multiple keys", func(t *testing.T) {
|
|
mergeCtx := NewMetadataKeysMergeCtx([]string{"key1", "key2"})
|
|
|
|
ctx1 := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value1"},
|
|
"key2": {"value2"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
ctx2 := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value1"},
|
|
"key2": {"different"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
mergedCtx := mergeCtx(ctx1, ctx2)
|
|
require.NotNil(t, mergedCtx)
|
|
mergedMeta := client.FromContext(mergedCtx).Metadata
|
|
assert.Equal(t, []string{"value1"}, mergedMeta.Get("key1"))
|
|
assert.Equal(t, []string{"value2"}, mergedMeta.Get("key2"))
|
|
})
|
|
|
|
t.Run("merge contexts with multiple values for same key", func(t *testing.T) {
|
|
mergeCtx := NewMetadataKeysMergeCtx([]string{"key1"})
|
|
|
|
ctx1 := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value1", "value2"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
ctx2 := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value1", "value2"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
mergedCtx := mergeCtx(ctx1, ctx2)
|
|
require.NotNil(t, mergedCtx)
|
|
|
|
mergedMeta := client.FromContext(mergedCtx).Metadata
|
|
assert.Equal(t, []string{"value1", "value2"}, mergedMeta.Get("key1"))
|
|
})
|
|
|
|
t.Run("merge when contexts have different multiple values for same key", func(t *testing.T) {
|
|
mergeCtx := NewMetadataKeysMergeCtx([]string{"key1"})
|
|
|
|
ctx1 := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value1", "value2"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
ctx2 := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value1", "value3"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
mergedCtx := mergeCtx(ctx1, ctx2)
|
|
require.NotNil(t, mergedCtx)
|
|
mergedMeta := client.FromContext(mergedCtx).Metadata
|
|
assert.Equal(t, []string{"value1", "value2"}, mergedMeta.Get("key1"))
|
|
})
|
|
}
|
|
|
|
func TestMetadataKeysPartitioner_GetKey(t *testing.T) {
|
|
t.Run("single key with single value", func(t *testing.T) {
|
|
partitioner := NewMetadataKeysPartitioner([]string{"key1"})
|
|
|
|
ctx := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value1"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
key := partitioner.GetKey(ctx, &requesttest.FakeRequest{})
|
|
expected := "key1\x00value1"
|
|
assert.Equal(t, expected, key)
|
|
})
|
|
|
|
t.Run("single key with multiple values", func(t *testing.T) {
|
|
partitioner := NewMetadataKeysPartitioner([]string{"key1"})
|
|
|
|
ctx := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value1", "value2", "value3"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
key := partitioner.GetKey(ctx, &requesttest.FakeRequest{})
|
|
// Format: key1\0value1\0value2\0value3
|
|
expected := "key1\x00value1\x00value2\x00value3"
|
|
assert.Equal(t, expected, key)
|
|
})
|
|
|
|
t.Run("multiple keys with values", func(t *testing.T) {
|
|
partitioner := NewMetadataKeysPartitioner([]string{"key1", "key2"})
|
|
|
|
ctx := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value1"},
|
|
"key2": {"value2"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
key := partitioner.GetKey(ctx, &requesttest.FakeRequest{})
|
|
// Format: key1\0value1\0key2\0value2 (separator between keys)
|
|
expected := "key1\x00value1\x00key2\x00value2"
|
|
assert.Equal(t, expected, key)
|
|
})
|
|
|
|
t.Run("multiple keys with multiple values", func(t *testing.T) {
|
|
partitioner := NewMetadataKeysPartitioner([]string{"key1", "key2"})
|
|
|
|
ctx := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value1", "value1b"},
|
|
"key2": {"value2", "value2b"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
key := partitioner.GetKey(ctx, &requesttest.FakeRequest{})
|
|
// Format: key1\0value1\0value1b\0key2\0value2\0value2b
|
|
expected := "key1\x00value1\x00value1b\x00key2\x00value2\x00value2b"
|
|
assert.Equal(t, expected, key)
|
|
})
|
|
|
|
t.Run("keys that don't exist in metadata are skipped", func(t *testing.T) {
|
|
partitioner := NewMetadataKeysPartitioner([]string{"key1", "key2", "key3"})
|
|
|
|
ctx := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value1"},
|
|
// key2 is missing
|
|
"key3": {"value3"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
key := partitioner.GetKey(ctx, &requesttest.FakeRequest{})
|
|
// Format: key1\0value1\0key3\0value3 (key2 is skipped)
|
|
expected := "key1\x00value1\x00key3\x00value3"
|
|
assert.Equal(t, expected, key)
|
|
})
|
|
|
|
t.Run("empty metadata returns empty string", func(t *testing.T) {
|
|
partitioner := NewMetadataKeysPartitioner([]string{"key1", "key2"})
|
|
|
|
ctx := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{}),
|
|
},
|
|
)
|
|
|
|
key := partitioner.GetKey(ctx, &requesttest.FakeRequest{})
|
|
assert.Empty(t, key)
|
|
})
|
|
|
|
t.Run("keys with empty values are skipped", func(t *testing.T) {
|
|
partitioner := NewMetadataKeysPartitioner([]string{"key1", "key2", "key3"})
|
|
|
|
ctx := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value1"},
|
|
"key2": {}, // empty slice
|
|
"key3": {"value3"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
key := partitioner.GetKey(ctx, &requesttest.FakeRequest{})
|
|
// Format: key1\0value1\0key3\0value3 (key2 is skipped)
|
|
expected := "key1\x00value1\x00key3\x00value3"
|
|
assert.Equal(t, expected, key)
|
|
})
|
|
|
|
t.Run("keys in order respect partitioner key order", func(t *testing.T) {
|
|
partitioner := NewMetadataKeysPartitioner([]string{"key3", "key1", "key2"})
|
|
|
|
ctx := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value1"},
|
|
"key2": {"value2"},
|
|
"key3": {"value3"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
key := partitioner.GetKey(ctx, &requesttest.FakeRequest{})
|
|
// Format should follow partitioner order: key3\0value3\0key1\0value1\0key2\0value2
|
|
expected := "key3\x00value3\x00key1\x00value1\x00key2\x00value2"
|
|
assert.Equal(t, expected, key)
|
|
})
|
|
|
|
t.Run("additional metadata keys not in partitioner are ignored", func(t *testing.T) {
|
|
partitioner := NewMetadataKeysPartitioner([]string{"key1"})
|
|
|
|
ctx := client.NewContext(
|
|
context.Background(),
|
|
client.Info{
|
|
Metadata: client.NewMetadata(map[string][]string{
|
|
"key1": {"value1"},
|
|
"other": {"other1"},
|
|
"extra": {"extra1"},
|
|
}),
|
|
},
|
|
)
|
|
|
|
key := partitioner.GetKey(ctx, &requesttest.FakeRequest{})
|
|
// Format: key1\0value1 (other keys are ignored)
|
|
expected := "key1\x00value1"
|
|
assert.Equal(t, expected, key)
|
|
})
|
|
|
|
t.Run("returns nil partitioner when keys are empty", func(t *testing.T) {
|
|
partitioner := NewMetadataKeysPartitioner([]string{})
|
|
assert.Nil(t, partitioner)
|
|
})
|
|
|
|
t.Run("returns nil merge function when keys are empty", func(t *testing.T) {
|
|
mergeCtx := NewMetadataKeysMergeCtx([]string{})
|
|
assert.Nil(t, mergeCtx)
|
|
})
|
|
}
|