receiver/opencensus: multiplex application/grpc+proto for gRPC based receiver

The registered grpc-gateway HTTP/2 content-type header field
was "application/grpc" but some clients in the wild send
"application/grpc+proto" and as per the gRPC spec, it is a descriptive
content length that should be handled by Proto (un)marshalers.

This change removed the use of

    (*grpc.Server).Stop()

which painfully took 20+s even without the "-race" go start option.
Sure, that method adds some sort of internal shutdown mechanism but
it is very painful to wait for 20+s for a server with zero connections
to stop yet all we need to do is stop the actual net.Listener by

     (net.Listener).Stop()

This change also adds a test which should work in a perfect world.
However, we don't have control over writes on the wire with the gRPC
connection hence our expectations for results to assert against are
blissfully optimistic and will fail most of the time as those writes
take sometime. The test has been included but skipped until we can
figure out what's going on. In the meantime though, users should be
able to use the OpenCensus receiver without trouble. Also internally
in the receivers, we've got data bundling which is controlled
separately despite options.

Fixes #365
Fixes #366

Author: odeke-em

receiver/opencensus/opencensus.go

@@ -189,10 +189,16 @@ func (ocr *Receiver) Stop() error {
 
 	var err = errAlreadyStopped
 	ocr.stopOnce.Do(func() {
-		// TODO: (@odeke-em) should we instead do (*grpc.Server).GracefulStop?
-		ocr.serverGRPC.Stop()
 		_ = ocr.serverHTTP.Close()
+
 		_ = ocr.ln.Close()
+
+		// TODO: @(odeke-em) investigate what utility invoking (*grpc.Server).Stop()
+		// gives us yet we invoke (net.Listener).Close().
+		// Sure (*grpc.Server).Stop() enables proper shutdown but imposes
+		// a painful and artificial wait time that goes into 20+seconds yet most of our
+		// tests and code should be reactive in less than even 1second.
+		// ocr.serverGRPC.Stop()
 	})
 	return err
 }
@@ -231,7 +237,10 @@ func (ocr *Receiver) startServer() error {
 
 			// Start the gRPC and HTTP/JSON (grpc-gateway) servers on the same port.
 			m := cmux.New(ocr.ln)
-			grpcL := m.Match(cmux.HTTP2HeaderField("content-type", "application/grpc"))
+			grpcL := m.Match(
+				cmux.HTTP2HeaderField("content-type", "application/grpc"),
+				cmux.HTTP2HeaderField("content-type", "application/grpc+proto"))
+
 			httpL := m.Match(cmux.Any())
 			go func() {
 				errChan <- ocr.serverGRPC.Serve(grpcL)

receiver/opencensus/opencensus_test.go

@@ -24,14 +24,17 @@ import (
 	"testing"
 	"time"
 
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
+
 	"github.com/census-instrumentation/opencensus-service/receiver/opencensus"
+	"github.com/census-instrumentation/opencensus-service/receiver/opencensus/octrace"
 	"github.com/census-instrumentation/opencensus-service/receiver/testhelper"
 
 	commonpb "github.com/census-instrumentation/opencensus-proto/gen-go/agent/common/v1"
 	agenttracepb "github.com/census-instrumentation/opencensus-proto/gen-go/agent/trace/v1"
 	tracepb "github.com/census-instrumentation/opencensus-proto/gen-go/trace/v1"
 	"github.com/census-instrumentation/opencensus-service/internal"
-	"github.com/census-instrumentation/opencensus-service/receiver/opencensus/octrace"
 )
 
 func TestGrpcGateway_endToEnd(t *testing.T) {
@@ -153,7 +156,7 @@ func TestGrpcGatewayCors_endToEnd(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Failed to create trace receiver: %v", err)
 	}
-	defer ocr.StopTraceReception(context.Background())
+	defer ocr.Stop()
 
 	sink := new(testhelper.ConcurrentSpanSink)
 	go func() {
@@ -174,6 +177,101 @@ func TestGrpcGatewayCors_endToEnd(t *testing.T) {
 	verifyCorsResp(t, url, "disallowed-origin.com", 200, false)
 }
 
+// As per Issue https://github.com/census-instrumentation/opencensus-service/issues/366
+// the agent's mux should be able to accept all Proto affiliated content-types and not
+// redirect them to the web-grpc-gateway endpoint.
+func TestAcceptAllGRPCProtoAffiliatedContentTypes(t *testing.T) {
+	t.Skip("Currently a flaky test as we need a way to flush all written traces")
+
+	addr := ":35991"
+	ocr, err := opencensus.New(addr, opencensus.WithTraceReceiverOptions(octrace.WithSpanBufferCount(1)))
+	if err != nil {
+		t.Fatalf("Failed to create trace receiver: %v", err)
+	}
+
+	cbts := new(testhelper.ConcurrentSpanSink)
+	if err := ocr.StartTraceReception(context.Background(), cbts); err != nil {
+		t.Fatalf("Failed to start the trace receiver: %v", err)
+	}
+	defer ocr.Stop()
+
+	// Now start the client with the various Proto affiliated gRPC Content-SubTypes as per:
+	//      https://godoc.org/google.golang.org/grpc#CallContentSubtype
+	protoAffiliatedContentSubTypes := []string{"", "proto"}
+	for _, subContentType := range protoAffiliatedContentSubTypes {
+		if err := runContentTypeTests(addr, asSubContentType, subContentType); err != nil {
+			t.Errorf("%q subContentType failed to send proto: %v", subContentType, err)
+		}
+	}
+
+	// Now start the client with the various Proto affiliated gRPC Content-Types,
+	// as we encountered in https://github.com/census-instrumentation/opencensus-service/issues/366
+	protoAffiliatedContentTypes := []string{"application/grpc", "application/grpc+proto"}
+	for _, contentType := range protoAffiliatedContentTypes {
+		if err := runContentTypeTests(addr, asContentType, contentType); err != nil {
+			t.Errorf("%q Content-type failed to send proto: %v", contentType, err)
+		}
+	}
+
+	// Before we exit we have to verify that we got exactly 4 TraceService requests.
+	wantLen := len(protoAffiliatedContentSubTypes) + len(protoAffiliatedContentTypes)
+	gotReqs := cbts.AllTraces()
+	if len(gotReqs) != wantLen {
+		t.Errorf("Receiver ExportTraceServiceRequest length mismatch:: Got %d Want %d", len(gotReqs), wantLen)
+	}
+}
+
+const (
+	asSubContentType = true
+	asContentType    = false
+)
+
+func runContentTypeTests(addr string, contentTypeDesignation bool, contentType string) error {
+	opts := []grpc.DialOption{
+		grpc.WithInsecure(),
+		grpc.WithBlock(),
+		grpc.WithDisableRetry(),
+	}
+
+	if contentTypeDesignation == asContentType {
+		opts = append(opts, grpc.WithDefaultCallOptions(
+			grpc.Header(&metadata.MD{"Content-Type": []string{contentType}})))
+	} else {
+		opts = append(opts, grpc.WithDefaultCallOptions(grpc.CallContentSubtype(contentType)))
+	}
+
+	cc, err := grpc.Dial(addr, opts...)
+	if err != nil {
+		return fmt.Errorf("Creating grpc.ClientConn: %v", err)
+	}
+	defer cc.Close()
+
+	// First step is to send the Node.
+	acc := agenttracepb.NewTraceServiceClient(cc)
+
+	stream, err := acc.Export(context.Background())
+	if err != nil {
+		return fmt.Errorf("Initializing the export stream: %v", err)
+	}
+
+	msg := &agenttracepb.ExportTraceServiceRequest{
+		Node: &commonpb.Node{
+			Attributes: map[string]string{
+				"sub-type": contentType,
+			},
+		},
+		Spans: []*tracepb.Span{
+			{
+				TraceId: []byte{
+					0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08,
+					0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x0E, 0x0F, 0X10,
+				},
+			},
+		},
+	}
+	return stream.Send(msg)
+}
+
 func verifyCorsResp(t *testing.T, url string, origin string, wantStatus int, wantAllowed bool) {
 	req, err := http.NewRequest("OPTIONS", url, nil)
 	if err != nil {