Ftr:otel trace

common/constant/key.go

@@ -93,6 +93,8 @@ const (
  TpsLimitFilterKey = "tps"
  TracingFilterKey = "tracing"
  XdsCircuitBreakerKey = "xds_circuit_reaker"
+ OTELServerTraceKey = "otelServerTrace"
+ OTELClientTraceKey = "otelClientTrace"
 )
 
 const (

filter/otel/trace/attachment.go

@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package trace
+
+import (
+ "context"
+)
+
+import (
+ "go.opentelemetry.io/otel/baggage"
+
+ "go.opentelemetry.io/otel/propagation"
+
+ "go.opentelemetry.io/otel/trace"
+)
+
+type metadataSupplier struct {
+ metadata map[string]interface{}
+}
+
+var _ propagation.TextMapCarrier = &metadataSupplier{}
+
+func (s *metadataSupplier) Get(key string) string {
+ if s.metadata == nil {
+ return ""
+ }
+ item, ok := s.metadata[key].([]string)
+ if !ok {
+ return ""
+ }
+ if len(item) == 0 {
+ return ""
+ }
+ return item[0]
+}
+
+func (s *metadataSupplier) Set(key string, value string) {
+ if s.metadata == nil {
+ s.metadata = map[string]interface{}{}
+ }
+ s.metadata[key] = value
+}
+
+func (s *metadataSupplier) Keys() []string {
+ out := make([]string, 0, len(s.metadata))
+ for key := range s.metadata {
+ out = append(out, key)
+ }
+ return out
+}
+
+// Inject injects correlation context and span context into the dubbo
+// metadata object. This function is meant to be used on outgoing
+// requests.
+func Inject(ctx context.Context, metadata map[string]interface{}, propagators propagation.TextMapPropagator) {
+ propagators.Inject(ctx, &metadataSupplier{
+ metadata: metadata,
+ })
+}
+
+// Extract returns the correlation context and span context that
+// another service encoded in the dubbo metadata object with Inject.
+// This function is meant to be used on incoming requests.
+func Extract(ctx context.Context, metadata map[string]interface{}, propagators propagation.TextMapPropagator) (baggage.Baggage, trace.SpanContext) {
+ ctx = propagators.Extract(ctx, &metadataSupplier{
+ metadata: metadata,
+ })
+ return baggage.FromContext(ctx), trace.SpanContextFromContext(ctx)
+}

filter/otel/trace/attachment_test.go

@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package trace
+
+import (
+ "reflect"
+ "testing"
+)
+
+func Test_metadataSupplier_Keys(t *testing.T) {
+ tests := []struct {
+ name string
+ metadata map[string]interface{}
+ want []string
+ }{
+ {
+ name: "test",
+ metadata: map[string]interface{}{
+ "key1": nil,
+ },
+ want: []string{"key1"},
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ s := &metadataSupplier{
+ metadata: tt.metadata,
+ }
+ if got := s.Keys(); !reflect.DeepEqual(got, tt.want) {
+ t.Errorf("Keys() = %v, want %v", got, tt.want)
+ }
+ })
+ }
+}
+
+func Test_metadataSupplier_Set(t *testing.T) {
+ tests := []struct {
+ name string
+ metadata map[string]interface{}
+ key string
+ value string
+ }{
+ {
+ name: "test",
+ metadata: nil,
+ key: "key",
+ value: "value",
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ s := &metadataSupplier{
+ metadata: tt.metadata,
+ }
+ s.Set(tt.key, tt.value)
+ })
+ }
+}
+
+func Test_metadataSupplier_Get(t *testing.T) {
+ tests := []struct {
+ name string
+ metadata map[string]interface{}
+ key string
+ want string
+ }{
+ {
+ name: "nil metadata",
+ metadata: nil,
+ key: "key",
+ want: "",
+ },
+ {
+ name: "not exist",
+ metadata: map[string]interface{}{
+ "k": nil,
+ },
+ key: "key",
+ want: "",
+ },
+ {
+ name: "nil slice",
+ metadata: map[string]interface{}{
+ "key": nil,
+ },
+ key: "key",
+ want: "",
+ },
+ {
+ name: "empty slice",
+ metadata: map[string]interface{}{
+ "key": []string{},
+ },
+ key: "key",
+ want: "",
+ },
+ {
+ name: "test",
+ metadata: map[string]interface{}{
+ "key": []string{"test"},
+ },
+ key: "key",
+ want: "test",
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ s := &metadataSupplier{
+ metadata: tt.metadata,
+ }
+ if got := s.Get(tt.key); got != tt.want {
+ t.Errorf("Get() = %v, want %v", got, tt.want)
+ }
+ })
+ }
+}

filter/otel/trace/doc.go

@@ -0,0 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Package trace instruments dubbogo with open-telemetry
+// (https://github.com/open-telemetry/opentelemetry-go).
+package trace

filter/otel/trace/filter.go

@@ -0,0 +1,144 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package trace
+
+import (
+ "context"
+)
+
+import (
+ "go.opentelemetry.io/otel"
+ "go.opentelemetry.io/otel/baggage"
+ "go.opentelemetry.io/otel/codes"
+ "go.opentelemetry.io/otel/propagation"
+ semconv "go.opentelemetry.io/otel/semconv/v1.7.0"
+ "go.opentelemetry.io/otel/trace"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/common/extension"
+ "dubbo.apache.org/dubbo-go/v3/filter"
+ "dubbo.apache.org/dubbo-go/v3/protocol"
+)
+
+const (
+ instrumentationName = "dubbo.apache.org/dubbo-go/v3/oteldubbo"
+)
+
+func init() {
+ extension.SetFilter(constant.OTELServerTraceKey, func() filter.Filter {
+ return &otelServerFilter{
+ Propagators: otel.GetTextMapPropagator(),
+ TracerProvider: otel.GetTracerProvider(),
+ }
+ })
+ extension.SetFilter(constant.OTELClientTraceKey, func() filter.Filter {
+ return &otelClientFilter{
+ Propagators: otel.GetTextMapPropagator(),
+ TracerProvider: otel.GetTracerProvider(),
+ }
+ })
+}
+
+type otelServerFilter struct {
+ Propagators propagation.TextMapPropagator
+ TracerProvider trace.TracerProvider
+}
+
+func (f *otelServerFilter) OnResponse(ctx context.Context, result protocol.Result, invoker protocol.Invoker, protocol protocol.Invocation) protocol.Result {
+ return result
+}
+
+func (f *otelServerFilter) Invoke(ctx context.Context, invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result {
+ attachments := invocation.Attachments()
+ bags, spanCtx := Extract(ctx, attachments, f.Propagators)
+ ctx = baggage.ContextWithBaggage(ctx, bags)
+
+ tracer := f.TracerProvider.Tracer(
+ instrumentationName,
+ trace.WithInstrumentationVersion(SemVersion()),
+ )
+
+ ctx, span := tracer.Start(
+ trace.ContextWithRemoteSpanContext(ctx, spanCtx),
+ invocation.ActualMethodName(),
+ trace.WithSpanKind(trace.SpanKindServer),
+ trace.WithAttributes(
+ RPCSystemDubbo,
+ semconv.RPCServiceKey.String(invoker.GetURL().ServiceKey()),
+ semconv.RPCMethodKey.String(invocation.MethodName()),
+ ),
+ )
+ defer span.End()
+
+ result := invoker.Invoke(ctx, invocation)
+
+ if result.Error() != nil {
+ span.SetStatus(codes.Error, result.Error().Error())
+ } else {
+ span.SetStatus(codes.Ok, codes.Ok.String())
+ }
+ return result
+}
+
+type otelClientFilter struct {
+ Propagators propagation.TextMapPropagator
+ TracerProvider trace.TracerProvider
+}
+
+func (f *otelClientFilter) OnResponse(ctx context.Context, result protocol.Result, invoker protocol.Invoker, protocol protocol.Invocation) protocol.Result {
+ return result
+}
+
+func (f *otelClientFilter) Invoke(ctx context.Context, invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result {
+ tracer := f.TracerProvider.Tracer(
+ instrumentationName,
+ trace.WithInstrumentationVersion(SemVersion()),
+ )
+
+ var span trace.Span
+ ctx, span = tracer.Start(
+ ctx,
+ invocation.ActualMethodName(),
+ trace.WithSpanKind(trace.SpanKindClient),
+ trace.WithAttributes(
+ RPCSystemDubbo,
+ semconv.RPCServiceKey.String(invoker.GetURL().ServiceKey()),
+ semconv.RPCMethodKey.String(invocation.MethodName()),
+ ),
+ )
+ defer span.End()
+
+ attachments := invocation.Attachments()
+ if attachments == nil {
+ attachments = map[string]interface{}{}
+ }
+ Inject(ctx, attachments, f.Propagators)
+ for k, v := range attachments {
+ invocation.SetAttachment(k, v)
+ }
+ result := invoker.Invoke(ctx, invocation)
+
+ if result.Error() != nil {
+ span.SetStatus(codes.Error, result.Error().Error())
+ } else {
+ span.SetStatus(codes.Ok, codes.Ok.String())
+ }
+ return result
+}