Add OTel attributes to server tracing interceptor and rename

Author: gjcairo

Sources/GRPCInterceptors/HookedWriter.swift

@@ -18,27 +18,22 @@ internal import Tracing
 
 struct HookedWriter<Element: Sendable>: RPCWriterProtocol {
   private let writer: any RPCWriterProtocol<Element>
-  private let beforeEachWrite: @Sendable () -> Void
   private let afterEachWrite: @Sendable () -> Void
 
   init(
     wrapping other: some RPCWriterProtocol<Element>,
-    beforeEachWrite: @Sendable @escaping () -> Void,
     afterEachWrite: @Sendable @escaping () -> Void
   ) {
     self.writer = other
-    self.beforeEachWrite = beforeEachWrite
     self.afterEachWrite = afterEachWrite
   }
 
   func write(_ element: Element) async throws {
-    self.beforeEachWrite()
     try await self.writer.write(element)
     self.afterEachWrite()
   }
 
   func write(contentsOf elements: some Sequence<Element>) async throws {
-    self.beforeEachWrite()
     try await self.writer.write(contentsOf: elements)
     self.afterEachWrite()
   }

Sources/GRPCInterceptors/Tracing/ClientOTelTracingInterceptor.swift

@@ -1,5 +1,5 @@
 /*
- * Copyright 2024, gRPC Authors All rights reserved.
+ * Copyright 2024-2025, gRPC Authors All rights reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,9 +21,14 @@ internal 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 emitEventOnEachWrite: Bool
@@ -82,7 +87,11 @@ public struct ClientOTelTracingInterceptor: ClientInterceptor {
       context: serviceContext,
       ofKind: .client
     ) { span in
-      self.setOTelSpanAttributes(into: span, context: context)
+      span.setOTelClientSpanGRPCAttributes(
+        context: context,
+        serverHostname: self.serverHostname,
+        networkTransportMethod: self.networkTransportMethod
+      )
 
       span.addEvent("Request started")
 
@@ -92,7 +101,6 @@ public struct ClientOTelTracingInterceptor: ClientInterceptor {
           let messageSentCounter = Atomic(1)
           let eventEmittingWriter = HookedWriter(
             wrapping: writer,
-            beforeEachWrite: {},
             afterEachWrite: {
               var event = SpanEvent(name: "rpc.message")
               event.attributes.rpc.messageType = "SENT"
@@ -158,31 +166,6 @@ public struct ClientOTelTracingInterceptor: ClientInterceptor {
       return response
     }
   }
-
-  private func setOTelSpanAttributes(into span: any Span, context: ClientContext) {
-    span.attributes.rpc.system = "grpc"
-    span.attributes.rpc.service = context.descriptor.service.fullyQualifiedService
-    span.attributes.rpc.method = context.descriptor.method
-    span.attributes.rpc.serverAddress = self.serverHostname
-    span.attributes.rpc.networkTransport = self.networkTransportMethod
-
-    let peer = context.remotePeer
-    // We expect this address to be of either of these two formats:
-    // - <type>:<host>:<port> for ipv4 and ipv6 addresses
-    // - unix:<uds-pathname> for UNIX domain sockets
-    let components = peer.split(separator: ":")
-    if components.count == 2 {
-      // This is the UDS case
-      span.attributes.rpc.networkType = String(components[0])
-      span.attributes.rpc.networkPeerAddress = String(components[1])
-    } else if components.count == 3 {
-      // This is the ipv4 or ipv6 case
-      span.attributes.rpc.networkType = String(components[0])
-      span.attributes.rpc.networkPeerAddress = String(components[1])
-      span.attributes.rpc.networkPeerPort = Int(components[2])
-      span.attributes.rpc.serverPort = Int(components[2])
-    }
-  }
 }
 
 /// An injector responsible for injecting the required instrumentation keys from the `ServiceContext` into

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

@@ -16,23 +16,41 @@
 
 public import GRPCCore
 internal import Tracing
+internal import Synchronization
 
 /// 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 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", "udp"). This will be the value for the
+  ///  `network.transport` attribute in spans.
+  ///  - 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(
+    serverHostname: String,
+    networkTransportMethod: String,
+    emitEventOnEachWrite: Bool = false
+  ) {
     self.extractor = ServerRequestExtractor()
     self.emitEventOnEachWrite = emitEventOnEachWrite
+    self.serverHostname = serverHostname
+    self.networkTransportMethod = networkTransportMethod
   }
 
   /// This interceptor will extract whatever `ServiceContext` key-value pairs have been inserted into the
@@ -41,6 +59,11 @@ 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,
@@ -65,36 +88,52 @@ 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
+        span.addEvent("Received request")
 
+        var request = request
         if self.emitEventOnEachWrite {
+          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.rpc.messageType = "RECEIVED"
+              event.attributes.rpc.messageID =
+              messageReceivedCounter
+                .wrappingAdd(1, ordering: .sequentiallyConsistent)
+                .oldValue
+              span.addEvent(event)
               return element
             }
           )
         }
 
         var response = try await next(request, context)
 
-        span.addEvent("Received request end")
+        span.addEvent("Finished processing request")
 
         switch response.accepted {
         case .success(var success):
           let wrappedProducer = success.producer
 
           if self.emitEventOnEachWrite {
             success.producer = { writer in
+              let messageSentCounter = Atomic(1)
               let eventEmittingWriter = HookedWriter(
                 wrapping: writer,
-                beforeEachWrite: {
-                  span.addEvent("Sending response part")
-                },
                 afterEachWrite: {
-                  span.addEvent("Sent response part")
+                  var event = SpanEvent(name: "rpc.message")
+                  event.attributes.rpc.messageType = "SENT"
+                  event.attributes.rpc.messageID =
+                  messageSentCounter
+                    .wrappingAdd(1, ordering: .sequentiallyConsistent)
+                    .oldValue
+                  span.addEvent(event)
                 }
               )
 
@@ -115,8 +154,11 @@ public struct ServerTracingInterceptor: ServerInterceptor {
 
           response = .init(accepted: .success(success))
 
-        case .failure:
+        case .failure(let error):
+          span.attributes.rpc.grpcStatusCode = error.code.rawValue
+          span.setStatus(SpanStatus(code: .error))
           span.addEvent("Sent error response")
+          span.recordError(error)
         }
 
         return response

Sources/GRPCInterceptors/Tracing/SpanAttributes+RPCAttributes.swift

@@ -14,7 +14,8 @@
  * limitations under the License.
  */
 
-import Tracing
+internal import Tracing
+internal import GRPCCore
 
 @dynamicMemberLookup
 package struct RPCAttributes: SpanAttributeNamespace {
@@ -58,3 +59,149 @@ extension SpanAttributes {
     }
   }
 }
+
+extension Span {
+  // See: https://opentelemetry.io/docs/specs/semconv/rpc/rpc-spans/
+  func setOTelClientSpanGRPCAttributes(
+    context: ClientContext,
+    serverHostname: String,
+    networkTransportMethod: String
+  ) {
+    self.attributes.rpc.system = "grpc"
+    self.attributes.rpc.serverAddress = serverHostname
+    self.attributes.rpc.networkTransport = networkTransportMethod
+    self.attributes.rpc.service = context.descriptor.service.fullyQualifiedService
+    self.attributes.rpc.method = context.descriptor.method
+
+    // Set server address information
+    switch PeerAddress(context.remotePeer) {
+    case .ipv4(let address, let port):
+      self.attributes.rpc.networkType = "ipv4"
+      self.attributes.rpc.networkPeerAddress = address
+      self.attributes.rpc.networkPeerPort = port
+      self.attributes.rpc.serverPort = port
+
+    case .ipv6(let address, let port):
+      self.attributes.rpc.networkType = "ipv6"
+      self.attributes.rpc.networkPeerAddress = address
+      self.attributes.rpc.networkPeerPort = port
+      self.attributes.rpc.serverPort = port
+
+    case .unixDomainSocket(let path):
+      self.attributes.rpc.networkType = "unix"
+      self.attributes.rpc.networkPeerAddress = path
+
+    case .other(let address):
+      // We can't nicely format the span attributes to contain the appropriate information here,
+      // so include the whole thing as part of the server address.
+      self.attributes.rpc.serverAddress = address
+    }
+  }
+
+  func setOTelServerSpanGRPCAttributes(
+    context: ServerContext,
+    serverHostname: String,
+    networkTransportMethod: String
+  ) {
+    self.attributes.rpc.system = "grpc"
+    self.attributes.rpc.serverAddress = serverHostname
+    self.attributes.rpc.networkTransport = networkTransportMethod
+    self.attributes.rpc.service = context.descriptor.service.fullyQualifiedService
+    self.attributes.rpc.method = context.descriptor.method
+
+    // Set server address information
+    switch PeerAddress(context.localPeer) {
+    case .ipv4(let address, let port):
+      self.attributes.rpc.networkType = "ipv4"
+      self.attributes.rpc.networkPeerAddress = address
+      self.attributes.rpc.networkPeerPort = port
+      self.attributes.rpc.serverPort = port
+
+    case .ipv6(let address, let port):
+      self.attributes.rpc.networkType = "ipv6"
+      self.attributes.rpc.networkPeerAddress = address
+      self.attributes.rpc.networkPeerPort = port
+      self.attributes.rpc.serverPort = port
+
+    case .unixDomainSocket(let path):
+      self.attributes.rpc.networkType = "unix"
+      self.attributes.rpc.networkPeerAddress = path
+
+    case .other(let address):
+      // We can't nicely format the span attributes to contain the appropriate information here,
+      // so include the whole thing as part of the server address.
+      self.attributes.rpc.serverAddress = address
+    }
+
+    switch PeerAddress(context.remotePeer) {
+    case .ipv4(let address, let port):
+      self.attributes.rpc.clientAddress = address
+      self.attributes.rpc.clientPort = port
+
+    case .ipv6(let address, let port):
+      self.attributes.rpc.clientAddress = address
+      self.attributes.rpc.clientPort = port
+
+    case .unixDomainSocket(let path):
+      self.attributes.rpc.clientAddress = path
+
+    case .other(let address):
+      self.attributes.rpc.clientAddress = address
+    }
+  }
+}
+
+private enum PeerAddress {
+  case ipv4(address: String, port: Int?)
+  case ipv6(address: String, port: Int?)
+  case unixDomainSocket(path: String)
+  case other(String)
+
+  init(_ address: String) {
+    // We expect this address to be of one of these formats:
+    // - ipv4:<host>:<port> for ipv4 addresses
+    // - ipv6:[<host>]:<port> for ipv6 addresses
+    // - unix:<uds-pathname> for UNIX domain sockets
+    let addressComponents = address.split(separator: ":", omittingEmptySubsequences: false)
+
+    guard addressComponents.count > 1 else {
+      // This is some unexpected/unknown format, so we have no way of splitting it up nicely.
+      self = .other(address)
+      return
+    }
+
+    // Check what type the transport is...
+    switch addressComponents[0] {
+    case "ipv4":
+      guard addressComponents.count == 3, let port = Int(addressComponents[2]) else {
+        // This is some unexpected/unknown format, so we have no way of splitting it up nicely.
+        self = .other(address)
+        return
+      }
+      self = .ipv4(address: String(addressComponents[1]), port: port)
+
+    case "ipv6":
+      guard addressComponents.count > 2, let port = Int(addressComponents.last!) else {
+        // This is some unexpected/unknown format, so we have no way of splitting it up nicely.
+        self = .other(address)
+        return
+      }
+      self = .ipv6(
+        address: String(addressComponents[1..<addressComponents.count-1].joined(separator: ":")),
+        port: port
+      )
+
+    case "unix":
+      guard addressComponents.count == 2 else {
+        // This is some unexpected/unknown format, so we have no way of splitting it up nicely.
+        self = .other(address)
+        return
+      }
+      self = .unixDomainSocket(path: String(addressComponents[1]))
+
+    default:
+      // This is some unexpected/unknown format, so we have no way of splitting it up nicely.
+      self = .other(address)
+    }
+  }
+}