Add option to force uniform output (#196)

This adds the `WithRPCSystem` option, which forces the output to either
use Connect or gRPC semantic conventions. When the option is not
provided, which conventions are used depends on the actual wire protocol
of the request being measured (same as before this change).

This also removes the "grpc_web" value for the "rpc.system" attribute
since that is not specified as a valid value anywhere in [OpenTelemetry
RPC semantic
conventions](https://opentelemetry.io/docs/specs/semconv/rpc/). The gRPC
semantic conventions suggest that it should always be "grpc" --
"grpc-web" is just a transport-level dialect and should not change
telemetry.

Author: jhump

attributes.go

@@ -62,12 +62,11 @@ func addProcedureAttributes(attrs []attribute.KeyValue, procedure string) []attr
 	return attrs
 }
 
-func addRequestAttributes(attrs []attribute.KeyValue, spec connect.Spec, peer connect.Peer) []attribute.KeyValue {
+func addRequestAttributes(protocol string, attrs []attribute.KeyValue, spec connect.Spec, peer connect.Peer) []attribute.KeyValue {
 	if addr := peer.Addr; addr != "" {
 		attrs = addAddressAttributes(attrs, addr)
 	}
 	name := strings.TrimLeft(spec.Procedure, "/")
-	protocol := protocolToSemConv(peer.Protocol)
 	attrs = append(attrs, semconv.RPCSystemKey.String(protocol))
 	attrs = addProcedureAttributes(attrs, name)
 	return attrs
@@ -90,7 +89,7 @@ func statusCodeAttribute(protocol string, serverErr error) (attribute.KeyValue,
 	// Following the respective specifications, use integers and "status_code" for
 	// gRPC codes in contrast to strings and "error_code" for Connect codes.
 	switch protocol {
-	case grpcProtocol, grpcwebProtocol:
+	case grpcProtocol:
 		codeKey := attribute.Key("rpc." + protocol + ".status_code")
 		if serverErr != nil {
 			return codeKey.Int64(int64(connect.CodeOf(serverErr))), true

interceptor.go

@@ -92,9 +92,9 @@ func (i *Interceptor) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc {
 		attributeFilter := i.config.filterAttribute.filter
 		isClient := request.Spec().IsClient
 		name := strings.TrimLeft(request.Spec().Procedure, "/")
-		protocol := protocolToSemConv(request.Peer().Protocol)
+		protocol := protocolToSemConv(request.Peer().Protocol, i.config.rpcSystem)
 		attributes := make([]attribute.KeyValue, 0, 6+len(i.config.requestHeaderKeys)) // 5 max request attrs + status code attr + headers
-		attributes = attributeFilter(request.Spec(), addRequestAttributes(attributes, request.Spec(), request.Peer())...)
+		attributes = attributeFilter(request.Spec(), addRequestAttributes(protocol, attributes, request.Spec(), request.Peer())...)
 		instrumentation := i.getInstruments(isClient)
 		carrier := propagation.HeaderCarrier(request.Header())
 		spanKind := trace.SpanKindClient
@@ -206,15 +206,16 @@ func (i *Interceptor) WrapStreamingClient(next connect.StreamingClientFunc) conn
 		// inject the newly created span into the carrier
 		carrier := propagation.HeaderCarrier(conn.RequestHeader())
 		i.config.propagator.Inject(ctx, carrier)
+		protocol := protocolToSemConv(conn.Peer().Protocol, i.config.rpcSystem)
 		state := newStreamingState(
+			protocol,
 			spec,
 			conn.Peer(),
 			i.config.filterAttribute,
 			i.config.omitTraceEvents,
 			instrumentation.responseSize,
 			instrumentation.requestSize,
 		)
-		protocol := protocolToSemConv(conn.Peer().Protocol)
 		var requestOnce sync.Once
 		setRequestAttributes := func() {
 			if span.IsRecording() {
@@ -282,8 +283,9 @@ func (i *Interceptor) WrapStreamingHandler(next connect.StreamingHandlerFunc) co
 			}
 		}
 		name := strings.TrimLeft(conn.Spec().Procedure, "/")
-		protocol := protocolToSemConv(conn.Peer().Protocol)
+		protocol := protocolToSemConv(conn.Peer().Protocol, i.config.rpcSystem)
 		state := newStreamingState(
+			protocol,
 			conn.Spec(),
 			conn.Peer(),
 			i.config.filterAttribute,
@@ -358,11 +360,13 @@ func (i *Interceptor) getInstruments(isClient bool) *instruments {
 }
 
 // protocolToSemConv converts the protocol string to the OpenTelemetry format.
-func protocolToSemConv(protocol string) string {
+func protocolToSemConv(protocol string, system RPCSystem) string {
+	if system != nil {
+		// If an explicit system was configured, that overrides the wire protocol.
+		return system.protocol()
+	}
 	switch protocol {
-	case grpcwebString:
-		return grpcwebProtocol
-	case grpcProtocol:
+	case grpcwebString, grpcString:
 		return grpcProtocol
 	case connectString:
 		return connectProtocol