Resolve #72 - gRPC Interceptor (#621)

* Move interceptor to plugin

* Add basic net.peer info

* Ensure that grpc status match span status

See: https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/data-rpc.md#status

* Set rpc.service attribute

* Add StreamClientInterceptor and StreamServerInterceptor

* Fix: golint errors

* Apply automated go.mod changes from make

* Implement suggestions to improve readability

Author: reicheltp

example/grpc/api/hello-service.pb.go

@@ -17,6 +17,12 @@ package api;
 
 service HelloService {
   rpc SayHello (HelloRequest) returns (HelloResponse);
+
+  rpc SayHelloServerStream (HelloRequest) returns (stream HelloResponse);
+
+  rpc SayHelloClientStream (stream HelloRequest) returns (HelloResponse);
+
+  rpc SayHelloBidiStream (stream HelloRequest) returns (stream HelloResponse);
 }
 
 message HelloRequest {

example/grpc/api/hello-service.proto

@@ -16,39 +16,167 @@ package main
 
 import (
 	"context"
+	"io"
 	"log"
 	"time"
 
+	"go.opentelemetry.io/otel/api/global"
 	"go.opentelemetry.io/otel/example/grpc/api"
 	"go.opentelemetry.io/otel/example/grpc/config"
+	"go.opentelemetry.io/otel/plugin/grpctrace"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/metadata"
-
-	"go.opentelemetry.io/otel/example/grpc/middleware/tracing"
 )
 
 func main() {
 	config.Init()
 
 	var conn *grpc.ClientConn
-	conn, err := grpc.Dial(":7777", grpc.WithInsecure(), grpc.WithUnaryInterceptor(tracing.UnaryClientInterceptor))
+	conn, err := grpc.Dial(":7777", grpc.WithInsecure(),
+		grpc.WithUnaryInterceptor(grpctrace.UnaryClientInterceptor(global.Tracer(""))),
+		grpc.WithStreamInterceptor(grpctrace.StreamClientInterceptor(global.Tracer(""))),
+	)
+
 	if err != nil {
 		log.Fatalf("did not connect: %s", err)
 	}
 	defer func() { _ = conn.Close() }()
 
 	c := api.NewHelloServiceClient(conn)
 
+	callSayHello(c)
+	callSayHelloClientStream(c)
+	callSayHelloServerStream(c)
+	callSayHelloBidiStream(c)
+
+	time.Sleep(10 * time.Millisecond)
+}
+
+func callSayHello(c api.HelloServiceClient) {
 	md := metadata.Pairs(
 		"timestamp", time.Now().Format(time.StampNano),
 		"client-id", "web-api-client-us-east-1",
 		"user-id", "some-test-user-id",
 	)
+
 	ctx := metadata.NewOutgoingContext(context.Background(), md)
 	response, err := c.SayHello(ctx, &api.HelloRequest{Greeting: "World"})
 	if err != nil {
 		log.Fatalf("Error when calling SayHello: %s", err)
 	}
 	log.Printf("Response from server: %s", response.Reply)
 }
+
+func callSayHelloClientStream(c api.HelloServiceClient) {
+	md := metadata.Pairs(
+		"timestamp", time.Now().Format(time.StampNano),
+		"client-id", "web-api-client-us-east-1",
+		"user-id", "some-test-user-id",
+	)
+
+	ctx := metadata.NewOutgoingContext(context.Background(), md)
+	stream, err := c.SayHelloClientStream(ctx)
+	if err != nil {
+		log.Fatalf("Error when opening SayHelloClientStream: %s", err)
+	}
+
+	for i := 0; i < 5; i++ {
+		err := stream.Send(&api.HelloRequest{Greeting: "World"})
+
+		time.Sleep(time.Duration(i*50) * time.Millisecond)
+
+		if err != nil {
+			log.Fatalf("Error when sending to SayHelloClientStream: %s", err)
+		}
+	}
+
+	response, err := stream.CloseAndRecv()
+	if err != nil {
+		log.Fatalf("Error when closing SayHelloClientStream: %s", err)
+	}
+
+	log.Printf("Response from server: %s", response.Reply)
+}
+
+func callSayHelloServerStream(c api.HelloServiceClient) {
+	md := metadata.Pairs(
+		"timestamp", time.Now().Format(time.StampNano),
+		"client-id", "web-api-client-us-east-1",
+		"user-id", "some-test-user-id",
+	)
+
+	ctx := metadata.NewOutgoingContext(context.Background(), md)
+	stream, err := c.SayHelloServerStream(ctx, &api.HelloRequest{Greeting: "World"})
+	if err != nil {
+		log.Fatalf("Error when opening SayHelloServerStream: %s", err)
+	}
+
+	for {
+		response, err := stream.Recv()
+		if err == io.EOF {
+			break
+		} else if err != nil {
+			log.Fatalf("Error when receiving from SayHelloServerStream: %s", err)
+		}
+
+		log.Printf("Response from server: %s", response.Reply)
+		time.Sleep(50 * time.Millisecond)
+	}
+}
+
+func callSayHelloBidiStream(c api.HelloServiceClient) {
+	md := metadata.Pairs(
+		"timestamp", time.Now().Format(time.StampNano),
+		"client-id", "web-api-client-us-east-1",
+		"user-id", "some-test-user-id",
+	)
+
+	ctx := metadata.NewOutgoingContext(context.Background(), md)
+	stream, err := c.SayHelloBidiStream(ctx)
+	if err != nil {
+		log.Fatalf("Error when opening SayHelloBidiStream: %s", err)
+	}
+
+	serverClosed := make(chan struct{})
+	clientClosed := make(chan struct{})
+
+	go func() {
+		for i := 0; i < 5; i++ {
+			err := stream.Send(&api.HelloRequest{Greeting: "World"})
+
+			if err != nil {
+				log.Fatalf("Error when sending to SayHelloBidiStream: %s", err)
+			}
+
+			time.Sleep(50 * time.Millisecond)
+		}
+
+		err := stream.CloseSend()
+		if err != nil {
+			log.Fatalf("Error when closing SayHelloBidiStream: %s", err)
+		}
+
+		clientClosed <- struct{}{}
+	}()
+
+	go func() {
+		for {
+			response, err := stream.Recv()
+			if err == io.EOF {
+				break
+			} else if err != nil {
+				log.Fatalf("Error when receiving from SayHelloBidiStream: %s", err)
+			}
+
+			log.Printf("Response from server: %s", response.Reply)
+			time.Sleep(50 * time.Millisecond)
+		}
+
+		serverClosed <- struct{}{}
+	}()
+
+	// Wait until client and server both closed the connection.
+	<-clientClosed
+	<-serverClosed
+}

example/grpc/client/main.go

@@ -7,5 +7,6 @@ replace go.opentelemetry.io/otel => ../..
 require (
 	github.com/golang/protobuf v1.3.2
 	go.opentelemetry.io/otel v0.4.2
+	golang.org/x/net v0.0.0-20190311183353-d8887717615a
 	google.golang.org/grpc v1.27.1
 )

example/grpc/go.mod

@@ -16,15 +16,18 @@ package main
 
 import (
 	"context"
+	"fmt"
+	"io"
 	"log"
 	"net"
+	"time"
 
+	"go.opentelemetry.io/otel/api/global"
 	"go.opentelemetry.io/otel/example/grpc/api"
 	"go.opentelemetry.io/otel/example/grpc/config"
+	"go.opentelemetry.io/otel/plugin/grpctrace"
 
 	"google.golang.org/grpc"
-
-	"go.opentelemetry.io/otel/example/grpc/middleware/tracing"
 )
 
 const (
@@ -33,15 +36,78 @@ const (
 
 // server is used to implement api.HelloServiceServer
 type server struct {
-	api.UnimplementedHelloServiceServer
+	api.HelloServiceServer
 }
 
 // SayHello implements api.HelloServiceServer
 func (s *server) SayHello(ctx context.Context, in *api.HelloRequest) (*api.HelloResponse, error) {
-	log.Printf("Received: %v", in.GetGreeting())
+	log.Printf("Received: %v\n", in.GetGreeting())
+	time.Sleep(50 * time.Millisecond)
+
 	return &api.HelloResponse{Reply: "Hello " + in.Greeting}, nil
 }
 
+func (s *server) SayHelloServerStream(in *api.HelloRequest, out api.HelloService_SayHelloServerStreamServer) error {
+	log.Printf("Received: %v\n", in.GetGreeting())
+
+	for i := 0; i < 5; i++ {
+		err := out.Send(&api.HelloResponse{Reply: "Hello " + in.Greeting})
+		if err != nil {
+			return err
+		}
+
+		time.Sleep(time.Duration(i*50) * time.Millisecond)
+	}
+
+	return nil
+}
+
+func (s *server) SayHelloClientStream(stream api.HelloService_SayHelloClientStreamServer) error {
+	i := 0
+
+	for {
+		in, err := stream.Recv()
+
+		if err == io.EOF {
+			break
+		} else if err != nil {
+			log.Printf("Non EOF error: %v\n", err)
+			return err
+		}
+
+		log.Printf("Received: %v\n", in.GetGreeting())
+		i++
+	}
+
+	time.Sleep(50 * time.Millisecond)
+
+	return stream.SendAndClose(&api.HelloResponse{Reply: fmt.Sprintf("Hello (%v times)", i)})
+}
+
+func (s *server) SayHelloBidiStream(stream api.HelloService_SayHelloBidiStreamServer) error {
+	for {
+		in, err := stream.Recv()
+
+		if err == io.EOF {
+			break
+		} else if err != nil {
+			log.Printf("Non EOF error: %v\n", err)
+			return err
+		}
+
+		time.Sleep(50 * time.Millisecond)
+
+		log.Printf("Received: %v\n", in.GetGreeting())
+		err = stream.Send(&api.HelloResponse{Reply: "Hello " + in.Greeting})
+
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
 func main() {
 	config.Init()
 
@@ -50,7 +116,10 @@ func main() {
 		log.Fatalf("failed to listen: %v", err)
 	}
 
-	s := grpc.NewServer(grpc.UnaryInterceptor(tracing.UnaryServerInterceptor))
+	s := grpc.NewServer(
+		grpc.UnaryInterceptor(grpctrace.UnaryServerInterceptor(global.Tracer(""))),
+		grpc.StreamInterceptor(grpctrace.StreamServerInterceptor(global.Tracer(""))),
+	)
 
 	api.RegisterHelloServiceServer(s, &server{})
 	if err := s.Serve(lis); err != nil {

example/grpc/middleware/tracing/tracing.go

@@ -6,6 +6,7 @@ require (
 	github.com/DataDog/sketches-go v0.0.0-20190923095040-43f19ad77ff7
 	github.com/benbjohnson/clock v1.0.0
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/golang/protobuf v1.3.2
 	github.com/google/go-cmp v0.4.0
 	github.com/google/gofuzz v1.0.0 // indirect
 	github.com/kr/pretty v0.1.0 // indirect