Add opentelemetry support

go.mod

@@ -3,37 +3,45 @@ module github.com/networkservicemesh/sdk
 go 1.16
 
 require (
-	github.com/HdrHistogram/hdrhistogram-go v1.0.1 // indirect
 	github.com/RoaringBitmap/roaring v0.9.4
 	github.com/antonfisher/nested-logrus-formatter v1.3.1
-	github.com/benbjohnson/clock v1.1.0
+	github.com/benbjohnson/clock v1.3.0
 	github.com/edwarnicke/exechelper v1.0.2
 	github.com/edwarnicke/grpcfd v0.1.1
 	github.com/edwarnicke/serialize v1.0.7
 	github.com/fsnotify/fsnotify v1.4.9
 	github.com/ghodss/yaml v1.0.0
 	github.com/golang-jwt/jwt/v4 v4.1.0
-	github.com/golang/protobuf v1.4.3
-	github.com/google/go-cmp v0.5.4
+	github.com/golang/protobuf v1.5.2
+	github.com/google/go-cmp v0.5.6
 	github.com/google/uuid v1.1.2
 	github.com/grpc-ecosystem/grpc-opentracing v0.0.0-20180507213350-8e809c8a8645
 	github.com/nats-io/nats-streaming-server v0.22.1
 	github.com/nats-io/stan.go v0.10.0
 	github.com/networkservicemesh/api v1.0.1-0.20211110183123-3038992da61a
 	github.com/open-policy-agent/opa v0.16.1
-	github.com/opentracing/opentracing-go v1.1.0
+	github.com/opentracing/opentracing-go v1.2.0
 	github.com/pkg/errors v0.9.1
-	github.com/sirupsen/logrus v1.7.0
+	github.com/sirupsen/logrus v1.8.1
 	github.com/spiffe/go-spiffe/v2 v2.0.0-alpha.4.0.20200528145730-dc11d0c74e85
 	github.com/stretchr/testify v1.7.0
-	github.com/uber/jaeger-client-go v2.21.1+incompatible
-	github.com/uber/jaeger-lib v2.4.0+incompatible // indirect
+	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.27.0
+	go.opentelemetry.io/otel v1.3.0
+	go.opentelemetry.io/otel/exporters/jaeger v1.3.0
+	go.opentelemetry.io/otel/exporters/otlp/otlpmetric v0.26.0
+	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v0.26.0
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.3.0
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.3.0
+	go.opentelemetry.io/otel/exporters/zipkin v1.3.0
+	go.opentelemetry.io/otel/metric v0.26.0
+	go.opentelemetry.io/otel/sdk v1.3.0
+	go.opentelemetry.io/otel/sdk/metric v0.26.0
+	go.opentelemetry.io/otel/trace v1.3.0
 	go.uber.org/atomic v1.7.0
-	go.uber.org/goleak v1.1.10
-	golang.org/x/net v0.0.0-20210525063256-abc453219eb5 // indirect
-	golang.org/x/tools v0.1.2 // indirect
+	go.uber.org/goleak v1.1.12
 	gonum.org/v1/gonum v0.6.2
-	google.golang.org/grpc v1.35.0
-	google.golang.org/protobuf v1.25.0
-	gopkg.in/yaml.v2 v2.2.2
+	google.golang.org/genproto v0.0.0-20211129164237-f09f9a12af12 // indirect
+	google.golang.org/grpc v1.42.0
+	google.golang.org/protobuf v1.27.1
+	gopkg.in/yaml.v2 v2.4.0
 )

pkg/networkservice/chains/nsmgr/scale_test.go

@@ -41,6 +41,7 @@ func TestCreateEndpointDuringRequest(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
 
 	defer cancel()
+
 	domain := sandbox.NewBuilder(ctx, t).
 		SetNodesCount(2).
 		SetRegistryProxySupplier(nil).

pkg/networkservice/chains/nsmgr/server.go

@@ -40,6 +40,7 @@ import (
 	"github.com/networkservicemesh/sdk/pkg/networkservice/common/excludedprefixes"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/common/mechanisms/recvfd"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/common/mechanisms/sendfd"
+	"github.com/networkservicemesh/sdk/pkg/networkservice/common/metrics"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/core/adapters"
 	"github.com/networkservicemesh/sdk/pkg/registry"
 	registryclientinfo "github.com/networkservicemesh/sdk/pkg/registry/common/clientinfo"
@@ -204,6 +205,7 @@ func NewServer(ctx context.Context, tokenGenerator token.GeneratorFunc, options
 			),
 			excludedprefixes.NewServer(ctx),
 			recvfd.NewServer(), // Receive any files passed
+			metrics.NewServer(),
 			connect.NewServer(
 				client.NewClient(
 					ctx,

pkg/networkservice/chains/nsmgr/single_test.go

@@ -37,7 +37,10 @@ import (
 )
 
 func Test_DNSUsecase(t *testing.T) {
-	t.Cleanup(func() { goleak.VerifyNone(t) })
+	t.Cleanup(func() {
+		time.Sleep(time.Second * 20)
+		goleak.VerifyNone(t)
+	})
 
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
 	defer cancel()

pkg/networkservice/common/metrics/server.go

@@ -0,0 +1,74 @@
+// Copyright (c) 2021 Doc.ai and/or its affiliates.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed 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 metrics provides a chain element that sends metrics to collector
+package metrics
+
+import (
+	"context"
+
+	"github.com/golang/protobuf/ptypes/empty"
+	"github.com/networkservicemesh/api/pkg/api/networkservice"
+
+	"github.com/networkservicemesh/sdk/pkg/networkservice/core/next"
+	"github.com/networkservicemesh/sdk/pkg/tools/opentelemetry/meterhelper"
+)
+
+type metricServer struct {
+	helpers map[string]meterhelper.MeterHelper
+}
+
+// NewServer returns a new metric server chain element
+func NewServer() networkservice.NetworkServiceServer {
+	return &metricServer{
+		helpers: make(map[string]meterhelper.MeterHelper),
+	}
+}
+
+func (t *metricServer) Request(ctx context.Context, request *networkservice.NetworkServiceRequest) (*networkservice.Connection, error) {
+	conn, err := next.Server(ctx).Request(ctx, request)
+	if err != nil {
+		return nil, err
+	}
+
+	t.writeMetrics(ctx, conn.GetPath())
+	return conn, nil
+}
+
+func (t *metricServer) Close(ctx context.Context, conn *networkservice.Connection) (*empty.Empty, error) {
+	_, err := next.Server(ctx).Close(ctx, conn)
+	if err != nil {
+		return nil, err
+	}
+
+	t.writeMetrics(ctx, conn.GetPath())
+	return &empty.Empty{}, nil
+}
+
+func (t *metricServer) writeMetrics(ctx context.Context, path *networkservice.Path) {
+	if path != nil {
+		for _, pathSegment := range path.GetPathSegments() {
+			if pathSegment.Metrics == nil {
+				continue
+			}
+			_, ok := t.helpers[pathSegment.Id]
+			if !ok {
+				t.helpers[pathSegment.Id] = meterhelper.NewMeterHelper(pathSegment.Name, path.GetPathSegments()[0].Id)
+			}
+			t.helpers[pathSegment.Id].WriteMetrics(ctx, pathSegment.Metrics)
+		}
+	}
+}

pkg/networkservice/common/updatepath/server.go

@@ -23,7 +23,6 @@ import (
 	"context"
 
 	"github.com/golang/protobuf/ptypes/empty"
-
 	"github.com/networkservicemesh/api/pkg/api/networkservice"
 
 	"github.com/networkservicemesh/sdk/pkg/networkservice/core/next"

pkg/networkservice/common/updatetoken/common.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2020 Cisco and/or its affiliates.
+// Copyright (c) 2020-2021 Cisco and/or its affiliates.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -19,9 +19,8 @@ package updatetoken
 import (
 	"context"
 
-	"github.com/golang/protobuf/ptypes/timestamp"
+	"google.golang.org/protobuf/types/known/timestamppb"
 
-	"github.com/golang/protobuf/ptypes"
 	"github.com/networkservicemesh/api/pkg/api/networkservice"
 	"github.com/pkg/errors"
 	"google.golang.org/grpc/credentials"
@@ -51,15 +50,8 @@ func updateToken(ctx context.Context, conn *networkservice.Connection, tokenGene
 		return errors.WithStack(err)
 	}
 
-	// Convert the expireTime to proto
-	var expires *timestamp.Timestamp
-	expires, err = ptypes.TimestampProto(expireTime)
-	if err != nil {
-		return errors.WithStack(err)
-	}
-
 	// Update the PathSegment
 	path.GetPathSegments()[path.GetIndex()].Token = tok
-	path.GetPathSegments()[path.GetIndex()].Expires = expires
+	path.GetPathSegments()[path.GetIndex()].Expires = timestamppb.New(expireTime)
 	return nil
 }

pkg/networkservice/common/updatetoken/server.go

@@ -20,11 +20,9 @@ package updatetoken
 import (
 	"context"
 
-	"github.com/golang/protobuf/ptypes"
 	"github.com/golang/protobuf/ptypes/empty"
-	"github.com/golang/protobuf/ptypes/timestamp"
 	"github.com/networkservicemesh/api/pkg/api/networkservice"
-	"github.com/pkg/errors"
+	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/networkservicemesh/sdk/pkg/networkservice/core/next"
 	"github.com/networkservicemesh/sdk/pkg/tools/log"
@@ -50,11 +48,7 @@ func (u *updateTokenServer) Request(ctx context.Context, request *networkservice
 		if err != nil {
 			log.FromContext(ctx).Warnf("an error during getting token from the context: %+v", err)
 		} else {
-			var expires *timestamp.Timestamp
-			expires, err = ptypes.TimestampProto(expireTime.Local())
-			if err != nil {
-				return nil, errors.WithStack(err)
-			}
+			expires := timestamppb.New(expireTime.Local())
 
 			prev.Expires = expires
 			prev.Token = tok
@@ -77,11 +71,7 @@ func (u *updateTokenServer) Close(ctx context.Context, conn *networkservice.Conn
 		if err != nil {
 			log.FromContext(ctx).Warnf("an error during getting token from the context: %+v", err)
 		} else {
-			var expires *timestamp.Timestamp
-			expires, err = ptypes.TimestampProto(expireTime.Local())
-			if err != nil {
-				return nil, errors.WithStack(err)
-			}
+			expires := timestamppb.New(expireTime.Local())
 
 			prev.Expires = expires
 			prev.Token = tok

pkg/networkservice/common/updatetoken/server_test.go

@@ -32,10 +32,10 @@ import (
 	"github.com/networkservicemesh/sdk/pkg/networkservice/common/updatetoken"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/core/next"
 
-	"github.com/golang/protobuf/ptypes"
 	"github.com/golang/protobuf/ptypes/timestamp"
 	"go.uber.org/goleak"
 	"google.golang.org/grpc/credentials"
+	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/networkservicemesh/api/pkg/api/networkservice"
 	"github.com/stretchr/testify/assert"
@@ -55,7 +55,7 @@ type updateTokenServerSuite struct {
 
 func (f *updateTokenServerSuite) SetupSuite() {
 	f.Token, f.Expires, _ = TokenGenerator(nil)
-	f.ExpiresProto, _ = ptypes.TimestampProto(f.Expires)
+	f.ExpiresProto = timestamppb.New(f.Expires)
 }
 
 func (f *updateTokenServerSuite) TestNewServer_EmptyPathInRequest() {

pkg/networkservice/core/trace/client.go

@@ -47,7 +47,7 @@ func NewNetworkServiceClient(traced networkservice.NetworkServiceClient) network
 func (t *beginTraceClient) Request(ctx context.Context, request *networkservice.NetworkServiceRequest, opts ...grpc.CallOption) (*networkservice.Connection, error) {
 	// Create a new logger
 	operation := typeutils.GetFuncName(t.traced, "Request")
-	ctx, finish := withLog(ctx, operation, request.GetConnection().GetId())
+	ctx, finish := withLog(ctx, operation)
 	defer finish()
 
 	logRequest(ctx, request, "request")
@@ -63,7 +63,7 @@ func (t *beginTraceClient) Request(ctx context.Context, request *networkservice.
 func (t *beginTraceClient) Close(ctx context.Context, conn *networkservice.Connection, opts ...grpc.CallOption) (*empty.Empty, error) {
 	// Create a new logger
 	operation := typeutils.GetFuncName(t.traced, "Close")
-	ctx, finish := withLog(ctx, operation, conn.GetId())
+	ctx, finish := withLog(ctx, operation)
 	defer finish()
 
 	logRequest(ctx, conn, "close")

pkg/networkservice/core/trace/context.go

@@ -46,7 +46,7 @@ type traceInfo struct {
 }
 
 // withLog - provides corresponding logger in context
-func withLog(parent context.Context, operation, connectionID string) (c context.Context, f func()) {
+func withLog(parent context.Context, operation string) (c context.Context, f func()) {
 	if parent == nil {
 		panic("cannot create context from nil parent")
 	}
@@ -57,7 +57,7 @@ func withLog(parent context.Context, operation, connectionID string) (c context.
 	if grpcTraceState := grpcutils.TraceFromContext(parent); (grpcTraceState == grpcutils.TraceOn) ||
 		(grpcTraceState == grpcutils.TraceUndefined && log.IsTracingEnabled()) {
 		ctx, sLogger, span, sFinish := spanlogger.FromContext(parent, operation)
-		ctx, lLogger, lFinish := logruslogger.FromSpan(ctx, span, operation, map[string]interface{}{"type": loggedType, "id": connectionID})
+		ctx, lLogger, lFinish := logruslogger.FromSpan(ctx, span, operation)
 		return withTrace(log.WithLog(ctx, sLogger, lLogger)), func() {
 			sFinish()
 			lFinish()

pkg/networkservice/core/trace/server.go

@@ -46,7 +46,7 @@ func NewNetworkServiceServer(traced networkservice.NetworkServiceServer) network
 func (t *beginTraceServer) Request(ctx context.Context, request *networkservice.NetworkServiceRequest) (*networkservice.Connection, error) {
 	// Create a new logger
 	operation := typeutils.GetFuncName(t.traced, "Request")
-	ctx, finish := withLog(ctx, operation, request.GetConnection().GetId())
+	ctx, finish := withLog(ctx, operation)
 	defer finish()
 
 	logRequest(ctx, request, "request")
@@ -62,7 +62,7 @@ func (t *beginTraceServer) Request(ctx context.Context, request *networkservice.
 func (t *beginTraceServer) Close(ctx context.Context, conn *networkservice.Connection) (*empty.Empty, error) {
 	// Create a new logger
 	operation := typeutils.GetFuncName(t.traced, "Close")
-	ctx, finish := withLog(ctx, operation, conn.GetId())
+	ctx, finish := withLog(ctx, operation)
 	defer finish()
 
 	logRequest(ctx, conn, "close")

pkg/registry/chains/proxydns/server_nse_test.go

@@ -22,10 +22,10 @@ import (
 	"testing"
 	"time"
 
-	"github.com/golang/protobuf/ptypes"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/goleak"
 	"google.golang.org/grpc"
+	"google.golang.org/protobuf/types/known/timestamppb"
 
 	registryapi "github.com/networkservicemesh/api/pkg/api/registry"
 
@@ -68,7 +68,7 @@ func TestInterdomainNetworkServiceEndpointRegistry(t *testing.T) {
 		SetDNSDomainName("domain2").
 		Build()
 
-	expirationTime, _ := ptypes.TimestampProto(time.Now().Add(time.Hour))
+	expirationTime := timestamppb.New(time.Now().Add(time.Hour))
 
 	reg, err := domain2.Registry.NetworkServiceEndpointRegistryServer().Register(
 		context.Background(),
@@ -132,7 +132,7 @@ func TestLocalDomain_NetworkServiceEndpointRegistry(t *testing.T) {
 		SetDNSResolver(dnsServer).
 		Build()
 
-	expirationTime, _ := ptypes.TimestampProto(time.Now().Add(time.Hour))
+	expirationTime := timestamppb.New(time.Now().Add(time.Hour))
 
 	reg, err := domain1.Registry.NetworkServiceEndpointRegistryServer().Register(
 		context.Background(),
@@ -208,7 +208,7 @@ func TestInterdomainFloatingNetworkServiceEndpointRegistry(t *testing.T) {
 		SetDNSDomainName("floating.domain").
 		Build()
 
-	expirationTime, _ := ptypes.TimestampProto(time.Now().Add(time.Hour))
+	expirationTime := timestamppb.New(time.Now().Add(time.Hour))
 
 	reg, err := domain2.Registry.NetworkServiceEndpointRegistryServer().Register(
 		context.Background(),

pkg/registry/core/trace/context.go

@@ -27,10 +27,6 @@ import (
 	"github.com/networkservicemesh/sdk/pkg/tools/log/spanlogger"
 )
 
-const (
-	loggedType = "registry"
-)
-
 // withLog - provides corresponding logger in context
 func withLog(parent context.Context, operation string) (c context.Context, f func()) {
 	if parent == nil {
@@ -43,7 +39,7 @@ func withLog(parent context.Context, operation string) (c context.Context, f fun
 	if grpcTraceState := grpcutils.TraceFromContext(parent); (grpcTraceState == grpcutils.TraceOn) ||
 		(grpcTraceState == grpcutils.TraceUndefined && log.IsTracingEnabled()) {
 		ctx, sLogger, span, sFinish := spanlogger.FromContext(parent, operation)
-		ctx, lLogger, lFinish := logruslogger.FromSpan(ctx, span, operation, map[string]interface{}{"type": loggedType})
+		ctx, lLogger, lFinish := logruslogger.FromSpan(ctx, span, operation)
 		return log.WithLog(ctx, sLogger, lLogger), func() {
 			sFinish()
 			lFinish()