Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Change internal tracing to use otel trace #3567

Merged
merged 1 commit into from
Jul 8, 2021
Merged

Change internal tracing to use otel trace #3567

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 8 additions & 4 deletions config/configgrpc/configgrpc.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ import (
"strings"
"time"

"go.opencensus.io/plugin/ocgrpc"
"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
"google.golang.org/grpc"
"google.golang.org/grpc/balancer/roundrobin"
"google.golang.org/grpc/credentials"
Expand Down Expand Up @@ -226,6 +226,10 @@ func (gcs *GRPCClientSettings) ToDialOptions(ext map[config.ComponentID]componen
opts = append(opts, grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"loadBalancingPolicy":"%s"}`, gcs.BalancerName)))
}

// Enable OpenTelemetry observability plugin.
opts = append(opts, grpc.WithUnaryInterceptor(otelgrpc.UnaryClientInterceptor()))
opts = append(opts, grpc.WithStreamInterceptor(otelgrpc.StreamClientInterceptor()))

Comment on lines +229 to +232
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This enables context propagation, right? So, we will propagate context for grpc-based protocols, like OTLP. Is this something we want?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is similar with what we had before with opencensus using the stats handler.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think any propagators are configured in this PR, so no propagation should happen by default. If a custom build (or later PR for this repo/contrib) were to configure a global propagator then it would be used by these interceptors.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is it not the case that w3c is the default?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, unfortunately it isn't.

The OpenTelemetry API MUST use no-op propagators unless explicitly configured otherwise.

return opts, nil
}

Expand Down Expand Up @@ -316,9 +320,9 @@ func (gss *GRPCServerSettings) ToServerOption(ext map[config.ComponentID]compone
)
}

// Enable OpenCensus observability plugin.
// TODO: Change to OpenTelemetry when collector is changed.
opts = append(opts, grpc.StatsHandler(&ocgrpc.ServerHandler{}))
// Enable OpenTelemetry observability plugin.
opts = append(opts, grpc.UnaryInterceptor(otelgrpc.UnaryServerInterceptor()))
opts = append(opts, grpc.StreamInterceptor(otelgrpc.StreamServerInterceptor()))

return opts, nil
}
Expand Down
10 changes: 5 additions & 5 deletions config/configgrpc/configgrpc_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ func TestDefaultGrpcClientSettings(t *testing.T) {
}
opts, err := gcs.ToDialOptions(map[config.ComponentID]component.Extension{})
assert.NoError(t, err)
assert.Len(t, opts, 1)
assert.Len(t, opts, 3)
}

func TestAllGrpcClientSettings(t *testing.T) {
Expand Down Expand Up @@ -75,14 +75,14 @@ func TestAllGrpcClientSettings(t *testing.T) {

opts, err := gcs.ToDialOptions(ext)
assert.NoError(t, err)
assert.Len(t, opts, 7)
assert.Len(t, opts, 9)
}

func TestDefaultGrpcServerSettings(t *testing.T) {
gss := &GRPCServerSettings{}
opts, err := gss.ToServerOption(map[config.ComponentID]component.Extension{})
assert.NoError(t, err)
assert.Len(t, opts, 1)
assert.Len(t, opts, 2)
}

func TestAllGrpcServerSettingsExceptAuth(t *testing.T) {
Expand Down Expand Up @@ -115,7 +115,7 @@ func TestAllGrpcServerSettingsExceptAuth(t *testing.T) {
}
opts, err := gss.ToServerOption(map[config.ComponentID]component.Extension{})
assert.NoError(t, err)
assert.Len(t, opts, 8)
assert.Len(t, opts, 9)
}

func TestGrpcServerAuthSettings(t *testing.T) {
Expand Down Expand Up @@ -246,7 +246,7 @@ func TestUseSecure(t *testing.T) {
}
dialOpts, err := gcs.ToDialOptions(map[config.ComponentID]component.Extension{})
assert.NoError(t, err)
assert.Equal(t, len(dialOpts), 1)
assert.Len(t, dialOpts, 3)
}

func TestGRPCServerSettingsError(t *testing.T) {
Expand Down
18 changes: 7 additions & 11 deletions exporter/exporterhelper/common_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,8 @@ import (

"github.com/stretchr/testify/require"
"go.opencensus.io/tag"
jpkrohling marked this conversation as resolved.
Show resolved Hide resolved
"go.opencensus.io/trace"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/oteltest"
"go.uber.org/zap"

"go.opentelemetry.io/collector/component"
Expand All @@ -29,20 +30,13 @@ import (
)

var (
okStatus = trace.Status{Code: trace.StatusCodeOK}

defaultExporterCfg = config.NewExporterSettings(config.NewID(typeStr))
exporterTag, _ = tag.NewKey("exporter")
defaultExporterTags = []tag.Tag{
{Key: exporterTag, Value: "test"},
}
)

func TestErrorToStatus(t *testing.T) {
require.Equal(t, okStatus, errToStatus(nil))
require.Equal(t, trace.Status{Code: trace.StatusCodeUnknown, Message: "my_error"}, errToStatus(errors.New("my_error")))
}

func TestBaseExporter(t *testing.T) {
be := newBaseExporter(&defaultExporterCfg, zap.NewNop(), fromOptions())
require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
Expand All @@ -64,9 +58,11 @@ func TestBaseExporterWithOptions(t *testing.T) {
require.Equal(t, want, be.Shutdown(context.Background()))
}

func errToStatus(err error) trace.Status {
func checkStatus(t *testing.T, sd *oteltest.Span, err error) {
if err != nil {
return trace.Status{Code: trace.StatusCodeUnknown, Message: err.Error()}
require.Equal(t, codes.Error, sd.StatusCode(), "SpanData %v", sd)
require.Equal(t, err.Error(), sd.StatusMessage(), "SpanData %v", sd)
} else {
require.Equal(t, codes.Unset, sd.StatusCode(), "SpanData %v", sd)
}
return okStatus
}
48 changes: 26 additions & 22 deletions exporter/exporterhelper/logs_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,10 @@ import (

"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.opencensus.io/trace"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/oteltest"
"go.opentelemetry.io/otel/trace"

"go.opentelemetry.io/collector/component"
"go.opentelemetry.io/collector/component/componenttest"
Expand Down Expand Up @@ -145,18 +148,28 @@ func TestLogsExporter_WithRecordEnqueueFailedMetrics(t *testing.T) {
}

func TestLogsExporter_WithSpan(t *testing.T) {
sr := new(oteltest.SpanRecorder)
tp := oteltest.NewTracerProvider(oteltest.WithSpanRecorder(sr))
otel.SetTracerProvider(tp)
defer otel.SetTracerProvider(trace.NewNoopTracerProvider())

le, err := NewLogsExporter(&fakeLogsExporterConfig, componenttest.NewNopExporterCreateSettings(), newPushLogsData(nil))
require.Nil(t, err)
require.NotNil(t, le)
checkWrapSpanForLogsExporter(t, le, nil, 1)
checkWrapSpanForLogsExporter(t, sr, tp.Tracer("test"), le, nil, 1)
}

func TestLogsExporter_WithSpan_ReturnError(t *testing.T) {
sr := new(oteltest.SpanRecorder)
tp := oteltest.NewTracerProvider(oteltest.WithSpanRecorder(sr))
otel.SetTracerProvider(tp)
defer otel.SetTracerProvider(trace.NewNoopTracerProvider())

want := errors.New("my_error")
le, err := NewLogsExporter(&fakeLogsExporterConfig, componenttest.NewNopExporterCreateSettings(), newPushLogsData(want))
require.Nil(t, err)
require.NotNil(t, le)
checkWrapSpanForLogsExporter(t, le, want, 1)
checkWrapSpanForLogsExporter(t, sr, tp.Tracer("test"), le, want, 1)
}

func TestLogsExporter_WithShutdown(t *testing.T) {
Expand Down Expand Up @@ -207,45 +220,36 @@ func checkRecordedMetricsForLogsExporter(t *testing.T, le component.LogsExporter
}
}

func generateLogsTraffic(t *testing.T, le component.LogsExporter, numRequests int, wantError error) {
func generateLogsTraffic(t *testing.T, tracer trace.Tracer, le component.LogsExporter, numRequests int, wantError error) {
ld := testdata.GenerateLogsOneLogRecord()
ctx, span := trace.StartSpan(context.Background(), fakeLogsParentSpanName, trace.WithSampler(trace.AlwaysSample()))
ctx, span := tracer.Start(context.Background(), fakeLogsParentSpanName)
defer span.End()
for i := 0; i < numRequests; i++ {
require.Equal(t, wantError, le.ConsumeLogs(ctx, ld))
}
}

func checkWrapSpanForLogsExporter(t *testing.T, le component.LogsExporter, wantError error, numLogRecords int64) {
ocSpansSaver := new(testOCTracesExporter)
trace.RegisterExporter(ocSpansSaver)
defer trace.UnregisterExporter(ocSpansSaver)

func checkWrapSpanForLogsExporter(t *testing.T, sr *oteltest.SpanRecorder, tracer trace.Tracer, le component.LogsExporter, wantError error, numLogRecords int64) {
const numRequests = 5
generateLogsTraffic(t, le, numRequests, wantError)
generateLogsTraffic(t, tracer, le, numRequests, wantError)

// Inspection time!
ocSpansSaver.mu.Lock()
defer ocSpansSaver.mu.Unlock()

require.NotEqual(t, 0, len(ocSpansSaver.spanData), "No exported span data")

gotSpanData := ocSpansSaver.spanData
gotSpanData := sr.Completed()
require.Equal(t, numRequests+1, len(gotSpanData))

parentSpan := gotSpanData[numRequests]
require.Equalf(t, fakeLogsParentSpanName, parentSpan.Name, "SpanData %v", parentSpan)
require.Equalf(t, fakeLogsParentSpanName, parentSpan.Name(), "SpanData %v", parentSpan)
for _, sd := range gotSpanData[:numRequests] {
require.Equalf(t, parentSpan.SpanContext.SpanID, sd.ParentSpanID, "Exporter span not a child\nSpanData %v", sd)
require.Equalf(t, errToStatus(wantError), sd.Status, "SpanData %v", sd)
require.Equalf(t, parentSpan.SpanContext().SpanID(), sd.ParentSpanID(), "Exporter span not a child\nSpanData %v", sd)
checkStatus(t, sd, wantError)

sentLogRecords := numLogRecords
var failedToSendLogRecords int64
if wantError != nil {
sentLogRecords = 0
failedToSendLogRecords = numLogRecords
}
require.Equalf(t, sentLogRecords, sd.Attributes[obsmetrics.SentLogRecordsKey], "SpanData %v", sd)
require.Equalf(t, failedToSendLogRecords, sd.Attributes[obsmetrics.FailedToSendLogRecordsKey], "SpanData %v", sd)
require.Equalf(t, attribute.Int64Value(sentLogRecords), sd.Attributes()[obsmetrics.SentLogRecordsKey], "SpanData %v", sd)
require.Equalf(t, attribute.Int64Value(failedToSendLogRecords), sd.Attributes()[obsmetrics.FailedToSendLogRecordsKey], "SpanData %v", sd)
}
}
48 changes: 26 additions & 22 deletions exporter/exporterhelper/metrics_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,10 @@ import (

"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.opencensus.io/trace"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/oteltest"
"go.opentelemetry.io/otel/trace"

"go.opentelemetry.io/collector/component"
"go.opentelemetry.io/collector/component/componenttest"
Expand Down Expand Up @@ -144,18 +147,28 @@ func TestMetricsExporter_WithRecordEnqueueFailedMetrics(t *testing.T) {
}

func TestMetricsExporter_WithSpan(t *testing.T) {
sr := new(oteltest.SpanRecorder)
tp := oteltest.NewTracerProvider(oteltest.WithSpanRecorder(sr))
otel.SetTracerProvider(tp)
defer otel.SetTracerProvider(trace.NewNoopTracerProvider())

me, err := NewMetricsExporter(&fakeMetricsExporterConfig, componenttest.NewNopExporterCreateSettings(), newPushMetricsData(nil))
require.NoError(t, err)
require.NotNil(t, me)
checkWrapSpanForMetricsExporter(t, me, nil, 1)
checkWrapSpanForMetricsExporter(t, sr, tp.Tracer("test"), me, nil, 1)
}

func TestMetricsExporter_WithSpan_ReturnError(t *testing.T) {
sr := new(oteltest.SpanRecorder)
tp := oteltest.NewTracerProvider(oteltest.WithSpanRecorder(sr))
otel.SetTracerProvider(tp)
defer otel.SetTracerProvider(trace.NewNoopTracerProvider())

want := errors.New("my_error")
me, err := NewMetricsExporter(&fakeMetricsExporterConfig, componenttest.NewNopExporterCreateSettings(), newPushMetricsData(want))
require.NoError(t, err)
require.NotNil(t, me)
checkWrapSpanForMetricsExporter(t, me, want, 1)
checkWrapSpanForMetricsExporter(t, sr, tp.Tracer("test"), me, want, 1)
}

func TestMetricsExporter_WithShutdown(t *testing.T) {
Expand Down Expand Up @@ -231,45 +244,36 @@ func checkRecordedMetricsForMetricsExporter(t *testing.T, me component.MetricsEx
}
}

func generateMetricsTraffic(t *testing.T, me component.MetricsExporter, numRequests int, wantError error) {
func generateMetricsTraffic(t *testing.T, tracer trace.Tracer, me component.MetricsExporter, numRequests int, wantError error) {
md := testdata.GenerateMetricsOneMetricOneDataPoint()
ctx, span := trace.StartSpan(context.Background(), fakeMetricsParentSpanName, trace.WithSampler(trace.AlwaysSample()))
ctx, span := tracer.Start(context.Background(), fakeMetricsParentSpanName)
defer span.End()
for i := 0; i < numRequests; i++ {
require.Equal(t, wantError, me.ConsumeMetrics(ctx, md))
}
}

func checkWrapSpanForMetricsExporter(t *testing.T, me component.MetricsExporter, wantError error, numMetricPoints int64) {
ocSpansSaver := new(testOCTracesExporter)
trace.RegisterExporter(ocSpansSaver)
defer trace.UnregisterExporter(ocSpansSaver)

func checkWrapSpanForMetricsExporter(t *testing.T, sr *oteltest.SpanRecorder, tracer trace.Tracer, me component.MetricsExporter, wantError error, numMetricPoints int64) {
const numRequests = 5
generateMetricsTraffic(t, me, numRequests, wantError)
generateMetricsTraffic(t, tracer, me, numRequests, wantError)

// Inspection time!
ocSpansSaver.mu.Lock()
defer ocSpansSaver.mu.Unlock()

require.NotEqual(t, 0, len(ocSpansSaver.spanData), "No exported span data")

gotSpanData := ocSpansSaver.spanData
gotSpanData := sr.Completed()
require.Equal(t, numRequests+1, len(gotSpanData))

parentSpan := gotSpanData[numRequests]
require.Equalf(t, fakeMetricsParentSpanName, parentSpan.Name, "SpanData %v", parentSpan)
require.Equalf(t, fakeMetricsParentSpanName, parentSpan.Name(), "SpanData %v", parentSpan)
for _, sd := range gotSpanData[:numRequests] {
require.Equalf(t, parentSpan.SpanContext.SpanID, sd.ParentSpanID, "Exporter span not a child\nSpanData %v", sd)
require.Equalf(t, errToStatus(wantError), sd.Status, "SpanData %v", sd)
require.Equalf(t, parentSpan.SpanContext().SpanID(), sd.ParentSpanID(), "Exporter span not a child\nSpanData %v", sd)
checkStatus(t, sd, wantError)

sentMetricPoints := numMetricPoints
var failedToSendMetricPoints int64
if wantError != nil {
sentMetricPoints = 0
failedToSendMetricPoints = numMetricPoints
}
require.Equalf(t, sentMetricPoints, sd.Attributes[obsmetrics.SentMetricPointsKey], "SpanData %v", sd)
require.Equalf(t, failedToSendMetricPoints, sd.Attributes[obsmetrics.FailedToSendMetricPointsKey], "SpanData %v", sd)
require.Equalf(t, attribute.Int64Value(sentMetricPoints), sd.Attributes()[obsmetrics.SentMetricPointsKey], "SpanData %v", sd)
require.Equalf(t, attribute.Int64Value(failedToSendMetricPoints), sd.Attributes()[obsmetrics.FailedToSendMetricPointsKey], "SpanData %v", sd)
}
}
Loading