CL-Softwares
/
telegraf
8
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: outputs.opentelemetry use headers config in grpc requests (#9587)

This commit is contained in:
alrex 2021-09-02 09:09:05 -07:00 committed by GitHub
parent 7af4c5fa15
commit 5a71f761dc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 10 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
plugins/outputs/opentelemetry/opentelemetry.go
View File

@ -13,6 +13,7 @@ import (
"go.opentelemetry.io/collector/model/otlpgrpc" "go.opentelemetry.io/collector/model/otlpgrpc"
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/credentials" "google.golang.org/grpc/credentials"
"google.golang.org/grpc/metadata"
) )
type OpenTelemetry struct { type OpenTelemetry struct {
@ -160,6 +161,10 @@ func (o *OpenTelemetry) Write(metrics []telegraf.Metric) error {
} }
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(o.Timeout)) ctx, cancel := context.WithTimeout(context.Background(), time.Duration(o.Timeout))
if len(o.Headers) > 0 {
ctx = metadata.NewOutgoingContext(ctx, metadata.New(o.Headers))
}
defer cancel() defer cancel()
_, err := o.metricsServiceClient.Export(ctx, md, o.callOptions...) _, err := o.metricsServiceClient.Export(ctx, md, o.callOptions...)
return err return err

5
plugins/outputs/opentelemetry/opentelemetry_test.go
View File

@ -18,6 +18,7 @@ import (
"go.opentelemetry.io/collector/model/otlpgrpc" "go.opentelemetry.io/collector/model/otlpgrpc"
"go.opentelemetry.io/collector/model/pdata" "go.opentelemetry.io/collector/model/pdata"
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/metadata"
) )
func TestOpenTelemetry(t *testing.T) { func TestOpenTelemetry(t *testing.T) {
@ -43,6 +44,7 @@ func TestOpenTelemetry(t *testing.T) {
plugin := &OpenTelemetry{ plugin := &OpenTelemetry{
ServiceAddress: m.Address(), ServiceAddress: m.Address(),
Timeout: config.Duration(time.Second), Timeout: config.Duration(time.Second),
Headers: map[string]string{"test": "header1"},
metricsConverter: metricsConverter, metricsConverter: metricsConverter,
grpcClientConn: m.GrpcClient(), grpcClientConn: m.GrpcClient(),
metricsServiceClient: otlpgrpc.NewMetricsClient(m.GrpcClient()), metricsServiceClient: otlpgrpc.NewMetricsClient(m.GrpcClient()),
@ -131,5 +133,8 @@ func (m *mockOtelService) Address() string {
func (m *mockOtelService) Export(ctx context.Context, request pdata.Metrics) (otlpgrpc.MetricsResponse, error) { func (m *mockOtelService) Export(ctx context.Context, request pdata.Metrics) (otlpgrpc.MetricsResponse, error) {
m.metrics = request.Clone() m.metrics = request.Clone()
ctxMetadata, ok := metadata.FromIncomingContext(ctx)
assert.Equal(m.t, []string{"header1"}, ctxMetadata.Get("test"))
assert.True(m.t, ok)
return otlpgrpc.MetricsResponse{}, nil return otlpgrpc.MetricsResponse{}, nil
} }