Conform server tracing interceptor to OTel's conventions (#27)

This PR brings the same changes made to the client in
#25 to the server
interceptor, conforming it to follow recommendations/conventions laid
out in both:
- https://opentelemetry.io/docs/specs/semconv/rpc/rpc-spans
- https://opentelemetry.io/docs/specs/semconv/rpc/grpc/

---------

Co-authored-by: George Barnett <[email protected]>

Author: gjcairo

Sources/GRPCInterceptors/Tracing/ClientOTelTracingInterceptor.swift

@@ -21,9 +21,14 @@ package import Tracing
 /// A client interceptor that injects tracing information into the request.
 ///
 /// The tracing information is taken from the current `ServiceContext`, and injected into the request's
-/// metadata. It will then be picked up by the server-side ``ServerTracingInterceptor``.
+/// metadata. It will then be picked up by the server-side ``ServerOTelTracingInterceptor``.
 ///
 /// For more information, refer to the documentation for `swift-distributed-tracing`.
+///
+/// This interceptor will also inject all required and recommended span and event attributes, and set span status, as defined by
+/// OpenTelemetry's documentation on:
+/// - https://opentelemetry.io/docs/specs/semconv/rpc/rpc-spans
+/// - https://opentelemetry.io/docs/specs/semconv/rpc/grpc/
 public struct ClientOTelTracingInterceptor: ClientInterceptor {
   private let injector: ClientRequestInjector
   private let traceEachMessage: Bool
@@ -37,7 +42,7 @@ public struct ClientOTelTracingInterceptor: ClientInterceptor {
   ///  - networkTransportMethod: The transport in use (e.g. "tcp", "unix"). This will be the value for the
   ///  `network.transport` attribute in spans.
   ///  - traceEachMessage: If `true`, each request part sent and response part received will be recorded as a separate
-  ///  event in a tracing span. Otherwise, only the request/response start and end will be recorded as events.
+  ///  event in a tracing span.
   public init(
     serverHostname: String,
     networkTransportMethod: String,

Sources/GRPCInterceptors/Tracing/ServerTracingInterceptor.swift renamed to Sources/GRPCInterceptors/Tracing/ServerOTelTracingInterceptor.swift

@@ -15,24 +15,42 @@
  */
 
 public import GRPCCore
-internal import Tracing
+internal import Synchronization
+package import Tracing
 
 /// A server interceptor that extracts tracing information from the request.
 ///
 /// The extracted tracing information is made available to user code via the current `ServiceContext`.
+///
 /// For more information, refer to the documentation for `swift-distributed-tracing`.
-public struct ServerTracingInterceptor: ServerInterceptor {
+///
+/// This interceptor will also inject all required and recommended span and event attributes, and set span status, as defined by
+/// OpenTelemetry's documentation on:
+/// - https://opentelemetry.io/docs/specs/semconv/rpc/rpc-spans
+/// - https://opentelemetry.io/docs/specs/semconv/rpc/grpc/
+public struct ServerOTelTracingInterceptor: ServerInterceptor {
   private let extractor: ServerRequestExtractor
-  private let emitEventOnEachWrite: Bool
+  private let traceEachMessage: Bool
+  private var serverHostname: String
+  private var networkTransportMethod: String
 
-  /// Create a new instance of a ``ServerTracingInterceptor``.
+  /// Create a new instance of a ``ServerOTelTracingInterceptor``.
   ///
-  /// - Parameter emitEventOnEachWrite: If `true`, each response part sent and request part
-  /// received will be recorded as a separate event in a tracing span. Otherwise, only the request/response
-  /// start and end will be recorded as events.
-  public init(emitEventOnEachWrite: Bool = false) {
+  /// - Parameters:
+  ///  - severHostname: The hostname of the RPC server. This will be the value for the `server.address` attribute in spans.
+  ///  - networkTransportMethod: The transport in use (e.g. "tcp", "unix"). This will be the value for the
+  ///  `network.transport` attribute in spans.
+  ///  - traceEachMessage: If `true`, each response part sent and request part received will be recorded as a separate
+  ///  event in a tracing span.
+  public init(
+    serverHostname: String,
+    networkTransportMethod: String,
+    traceEachMessage: Bool = true
+  ) {
     self.extractor = ServerRequestExtractor()
-    self.emitEventOnEachWrite = emitEventOnEachWrite
+    self.traceEachMessage = traceEachMessage
+    self.serverHostname = serverHostname
+    self.networkTransportMethod = networkTransportMethod
   }
 
   /// This interceptor will extract whatever `ServiceContext` key-value pairs have been inserted into the
@@ -41,14 +59,34 @@ public struct ServerTracingInterceptor: ServerInterceptor {
   ///
   /// Which key-value pairs are extracted and made available will depend on the specific tracing implementation
   /// that has been configured when bootstrapping `swift-distributed-tracing` in your application.
+  ///
+  /// It will also inject all required and recommended span and event attributes, and set span status, as defined by OpenTelemetry's
+  /// documentation on:
+  /// - https://opentelemetry.io/docs/specs/semconv/rpc/rpc-spans
+  /// - https://opentelemetry.io/docs/specs/semconv/rpc/grpc/
   public func intercept<Input, Output>(
     request: StreamingServerRequest<Input>,
     context: ServerContext,
     next: @Sendable (StreamingServerRequest<Input>, ServerContext) async throws ->
       StreamingServerResponse<Output>
+  ) async throws -> StreamingServerResponse<Output> where Input: Sendable, Output: Sendable {
+    try await self.intercept(
+      tracer: InstrumentationSystem.tracer,
+      request: request,
+      context: context,
+      next: next
+    )
+  }
+
+  /// Same as ``intercept(request:context:next:)``, but allows specifying a `Tracer` for testing purposes.
+  package func intercept<Input, Output>(
+    tracer: any Tracer,
+    request: StreamingServerRequest<Input>,
+    context: ServerContext,
+    next: @Sendable (StreamingServerRequest<Input>, ServerContext) async throws ->
+      StreamingServerResponse<Output>
   ) async throws -> StreamingServerResponse<Output> where Input: Sendable, Output: Sendable {
     var serviceContext = ServiceContext.topLevel
-    let tracer = InstrumentationSystem.tracer
 
     tracer.extract(
       request.metadata,
@@ -65,55 +103,65 @@ public struct ServerTracingInterceptor: ServerInterceptor {
         context: serviceContext,
         ofKind: .server
       ) { span in
-        span.addEvent("Received request start")
+        span.setOTelServerSpanGRPCAttributes(
+          context: context,
+          serverHostname: self.serverHostname,
+          networkTransportMethod: self.networkTransportMethod
+        )
 
         var request = request
-
-        if self.emitEventOnEachWrite {
+        if self.traceEachMessage {
+          let messageReceivedCounter = Atomic(1)
           request.messages = RPCAsyncSequence(
             wrapping: request.messages.map { element in
-              span.addEvent("Received request part")
+              var event = SpanEvent(name: "rpc.message")
+              event.attributes[GRPCTracingKeys.rpcMessageType] = "RECEIVED"
+              event.attributes[GRPCTracingKeys.rpcMessageID] =
+                messageReceivedCounter
+                .wrappingAdd(1, ordering: .sequentiallyConsistent)
+                .oldValue
+              span.addEvent(event)
               return element
             }
           )
         }
 
         var response = try await next(request, context)
 
-        span.addEvent("Received request end")
-
         switch response.accepted {
         case .success(var success):
           let wrappedProducer = success.producer
 
-          if self.emitEventOnEachWrite {
+          if self.traceEachMessage {
             success.producer = { writer in
+              let messageSentCounter = Atomic(1)
               let eventEmittingWriter = HookedWriter(
                 wrapping: writer,
                 afterEachWrite: {
-                  span.addEvent("Sent response part")
+                  var event = SpanEvent(name: "rpc.message")
+                  event.attributes[GRPCTracingKeys.rpcMessageType] = "SENT"
+                  event.attributes[GRPCTracingKeys.rpcMessageID] =
+                    messageSentCounter
+                    .wrappingAdd(1, ordering: .sequentiallyConsistent)
+                    .oldValue
+                  span.addEvent(event)
                 }
               )
 
               let wrappedResult = try await wrappedProducer(
                 RPCWriter(wrapping: eventEmittingWriter)
               )
 
-              span.addEvent("Sent response end")
-              return wrappedResult
-            }
-          } else {
-            success.producer = { writer in
-              let wrappedResult = try await wrappedProducer(writer)
-              span.addEvent("Sent response end")
               return wrappedResult
             }
           }
 
           response = .init(accepted: .success(success))
 
-        case .failure:
-          span.addEvent("Sent error response")
+        case .failure(let error):
+          span.attributes[GRPCTracingKeys.grpcStatusCode] = error.code.rawValue
+          span.setStatus(SpanStatus(code: .error))
+          span.recordError(error)
         }
 
         return response

Sources/GRPCInterceptors/Tracing/SpanAttributes+GRPCTracingKeys.swift

@@ -72,6 +72,58 @@ extension Span {
       ()
     }
   }
+
+  func setOTelServerSpanGRPCAttributes(
+    context: ServerContext,
+    serverHostname: String,
+    networkTransportMethod: String
+  ) {
+    self.attributes[GRPCTracingKeys.rpcSystem] = "grpc"
+    self.attributes[GRPCTracingKeys.serverAddress] = serverHostname
+    self.attributes[GRPCTracingKeys.networkTransport] = networkTransportMethod
+    self.attributes[GRPCTracingKeys.rpcService] = context.descriptor.service.fullyQualifiedService
+    self.attributes[GRPCTracingKeys.rpcMethod] = context.descriptor.method
+
+    // Set server address information
+    switch PeerAddress(context.localPeer) {
+    case .ipv4(let address, let port):
+      self.attributes[GRPCTracingKeys.networkType] = "ipv4"
+      self.attributes[GRPCTracingKeys.networkPeerAddress] = address
+      self.attributes[GRPCTracingKeys.networkPeerPort] = port
+      self.attributes[GRPCTracingKeys.serverPort] = port
+
+    case .ipv6(let address, let port):
+      self.attributes[GRPCTracingKeys.networkType] = "ipv6"
+      self.attributes[GRPCTracingKeys.networkPeerAddress] = address
+      self.attributes[GRPCTracingKeys.networkPeerPort] = port
+      self.attributes[GRPCTracingKeys.serverPort] = port
+
+    case .unixDomainSocket(let path):
+      self.attributes[GRPCTracingKeys.networkPeerAddress] = path
+
+    case .none:
+      // We don't recognise this address format, so don't populate any fields.
+      ()
+    }
+
+    // Set client address information
+    switch PeerAddress(context.remotePeer) {
+    case .ipv4(let address, let port):
+      self.attributes[GRPCTracingKeys.clientAddress] = address
+      self.attributes[GRPCTracingKeys.clientPort] = port
+
+    case .ipv6(let address, let port):
+      self.attributes[GRPCTracingKeys.clientAddress] = address
+      self.attributes[GRPCTracingKeys.clientPort] = port
+
+    case .unixDomainSocket(let path):
+      self.attributes[GRPCTracingKeys.clientAddress] = path
+
+    case .none:
+      // We don't recognise this address format, so don't populate any fields.
+      ()
+    }
+  }
 }
 
 package enum PeerAddress: Equatable {