Add opentelemetry support

go.mod

@@ -11,29 +11,39 @@ require (
 	github.com/edwarnicke/serialize v1.0.7
 	github.com/fsnotify/fsnotify v1.4.9
 	github.com/ghodss/yaml v1.0.0
-	github.com/golang/protobuf v1.4.3
-	github.com/google/go-cmp v0.5.2
+	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/hashicorp/go-uuid v1.0.1 // indirect
 	github.com/nats-io/nats-streaming-server v0.17.0
 	github.com/nats-io/stan.go v0.6.0
 	github.com/networkservicemesh/api v1.0.1-0.20210811070028-10403c0f20c8
 	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/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-client-go v2.25.0+incompatible
 	github.com/uber/jaeger-lib v2.4.0+incompatible // indirect
+	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.22.0
+	go.opentelemetry.io/otel v1.0.0-RC2
+	go.opentelemetry.io/otel/exporters/otlp/otlpmetric v0.22.0
+	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v0.22.0
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.0.0-RC2
+	go.opentelemetry.io/otel/metric v0.22.0
+	go.opentelemetry.io/otel/sdk v1.0.0-RC2
+	go.opentelemetry.io/otel/sdk/metric v0.22.0
+	go.opentelemetry.io/otel/trace v1.0.0-RC2
 	go.uber.org/atomic v1.7.0
 	go.uber.org/goleak v1.1.10
 	golang.org/x/crypto v0.0.0-20210513164829-c07d793c2f9a // indirect
 	golang.org/x/net v0.0.0-20210525063256-abc453219eb5 // indirect
 	golang.org/x/sys v0.0.0-20210603125802-9665404d3644 // indirect
 	golang.org/x/tools v0.1.2 // indirect
 	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/grpc v1.39.0
+	google.golang.org/protobuf v1.27.1
+	gopkg.in/yaml.v2 v2.2.3
 )

pkg/networkservice/chains/nsmgr/server.go

@@ -43,6 +43,7 @@ import (
 	"github.com/networkservicemesh/sdk/pkg/networkservice/common/interpose"
 	"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/common/roundrobin"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/core/adapters"
 	"github.com/networkservicemesh/sdk/pkg/registry"
@@ -191,6 +192,7 @@ func NewServer(ctx context.Context, tokenGenerator token.GeneratorFunc, options
 			roundrobin.NewServer(),
 			excludedprefixes.NewServer(ctx),
 			recvfd.NewServer(), // Receive any files passed
+			metrics.NewServer(),
 			interpose.NewServer(&interposeRegistryServer),
 			filtermechanisms.NewServer(&urlsRegistryServer),
 			heal.NewServer(ctx,

pkg/networkservice/common/heal/server.go

@@ -24,7 +24,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/golang/protobuf/ptypes"
 	"github.com/golang/protobuf/ptypes/empty"
 
 	"github.com/networkservicemesh/api/pkg/api/networkservice"
@@ -221,10 +220,10 @@ func (f *healServer) restoreConnection(
 
 	// Make sure we have a valid expireTime to work with
 	expires := request.GetConnection().GetNextPathSegment().GetExpires()
-	expireTime, err := ptypes.Timestamp(expires)
-	if err != nil {
+	if !expires.IsValid() {
 		return
 	}
+	expireTime := expires.AsTime()
 
 	deadline := clockTime.Now().Add(f.restoreTimeout)
 	if deadline.After(expireTime) {

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/refresh/client.go

@@ -24,7 +24,6 @@ import (
 	"context"
 	"time"
 
-	"github.com/golang/protobuf/ptypes"
 	"github.com/golang/protobuf/ptypes/empty"
 	"github.com/pkg/errors"
 	"google.golang.org/grpc"
@@ -86,10 +85,10 @@ func (t *refreshClient) startTimer(ctx context.Context, connectionID string, req
 	clockTime := clock.FromContext(ctx)
 
 	nextClient := next.Client(ctx)
-	expireTime, err := ptypes.Timestamp(request.GetConnection().GetCurrentPathSegment().GetExpires())
-	if err != nil {
+	if !request.GetConnection().GetCurrentPathSegment().GetExpires().IsValid() {
 		return
 	}
+	expireTime := request.GetConnection().GetCurrentPathSegment().GetExpires().AsTime()
 
 	// A heuristic to reduce the number of redundant requests in a chain
 	// made of refreshing clients with the same expiration time: let outer

pkg/networkservice/common/updatepath/server.go

@@ -23,12 +23,9 @@ import (
 	"context"
 	"time"
 
-	"github.com/golang/protobuf/ptypes"
 	"github.com/golang/protobuf/ptypes/empty"
-	"github.com/golang/protobuf/ptypes/timestamp"
-	"github.com/pkg/errors"
-
 	"github.com/networkservicemesh/api/pkg/api/networkservice"
+	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/networkservicemesh/sdk/pkg/networkservice/core/next"
 	"github.com/networkservicemesh/sdk/pkg/tools/log"
@@ -66,13 +63,7 @@ func (i *updatePathServer) 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)
-			}
-
-			prev.Expires = expires
+			prev.Expires = timestamppb.New(expireTime.Local())
 			prev.Token = tok
 		}
 	}

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_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/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/tools/jaeger/jaeger.go

@@ -24,38 +24,14 @@ import (
 	"fmt"
 	"io"
 	"os"
-	"strconv"
-
-	"github.com/networkservicemesh/sdk/pkg/tools/log"
 
 	"github.com/opentracing/opentracing-go"
 	"github.com/uber/jaeger-client-go"
 	"github.com/uber/jaeger-client-go/config"
-)
 
-const (
-	opentracingEnv     = "TRACER_ENABLED"
-	opentracingDefault = true
+	"github.com/networkservicemesh/sdk/pkg/tools/log"
 )
 
-// IsOpentracingEnabled returns true if opentracing enabled
-func IsOpentracingEnabled() bool {
-	val, err := readEnvBool(opentracingEnv, opentracingDefault)
-	if err == nil {
-		return val
-	}
-	return opentracingDefault
-}
-
-func readEnvBool(env string, value bool) (bool, error) {
-	str := os.Getenv(env)
-	if str == "" {
-		return value, nil
-	}
-
-	return strconv.ParseBool(str)
-}
-
 type emptyCloser struct {
 }
 
@@ -66,7 +42,7 @@ func (*emptyCloser) Close() error {
 
 // InitJaeger -  returns an instance of Jaeger Tracer that samples 100% of traces and logs all spans to stdout.
 func InitJaeger(ctx context.Context, service string) io.Closer {
-	if !IsOpentracingEnabled() {
+	if !log.IsOpentracingEnabled() {
 		return &emptyCloser{}
 	}
 	if opentracing.IsGlobalTracerRegistered() {

pkg/tools/log/logger.go

@@ -19,15 +19,41 @@ package log
 
 import (
 	"context"
+	"os"
 )
 
 type contextKeyType string
 
 const (
 	logKey       contextKeyType = "Logger"
 	logFieldsKey contextKeyType = "LoggerFields"
+
+	telemetryEnv  = "TELEMETRY"
+	telemetryOT   = "opentracing"
+	telemetryOTel = "opentelemetry"
+
+	// Opentracing enabled by default
+	telemetryDefault = telemetryOTel
 )
 
+// IsOpentracingEnabled returns true if opentracing enabled
+func IsOpentracingEnabled() bool {
+	return telemetryOT == getTelemetryEnv()
+}
+
+// IsOpentelemetryEnabled returns true if opentelemetry enabled
+func IsOpentelemetryEnabled() bool {
+	return telemetryOTel == getTelemetryEnv()
+}
+
+func getTelemetryEnv() string {
+	val := os.Getenv(telemetryEnv)
+	if val == "" {
+		return telemetryDefault
+	}
+	return val
+}
+
 var (
 	isTracingEnabled = false
 )