Fix a few incorrect source locations in logs (#933)

Motivation:

swift-log 1.3.0 introduced a new concepct: the source of a log. This
identifies the creator of the emitted log, typically the module or
system that emitted the log.

The default value is currently (meant to be) the module name. This is
found by parsing the '#file' to find the directory the file is
contained in. However, this isn't always the module.

Modifications:

- Capture the log source in tests
- Validate the source in test tear down
- Fix up incorrect sources

Result:

- gRPC more uniformly emits a useful log source

Sources/GRPC/CallHandlers/BidirectionalStreamingCallHandler.swift

@@ -78,7 +78,7 @@ public class BidirectionalStreamingCallHandler<
 
   override internal func processMessage(_ message: RequestPayload) {
     guard let eventObserver = self.eventObserver else {
-      self.logger.warning("eventObserver is nil; ignoring message")
+      self.logger.warning("eventObserver is nil; ignoring message", source: "GRPC")
       return
     }
     eventObserver.whenSuccess { observer in
@@ -88,7 +88,7 @@ public class BidirectionalStreamingCallHandler<
 
   override internal func endOfStreamReceived() throws {
     guard let eventObserver = self.eventObserver else {
-      self.logger.warning("eventObserver is nil; ignoring end-of-stream")
+      self.logger.warning("eventObserver is nil; ignoring end-of-stream", source: "GRPC")
       return
     }
     eventObserver.whenSuccess { observer in

Sources/GRPC/CallHandlers/ClientStreamingCallHandler.swift

@@ -78,7 +78,7 @@ public final class ClientStreamingCallHandler<
 
   override internal func processMessage(_ message: RequestPayload) {
     guard let eventObserver = self.eventObserver else {
-      self.logger.warning("eventObserver is nil; ignoring message")
+      self.logger.warning("eventObserver is nil; ignoring message", source: "GRPC")
       return
     }
     eventObserver.whenSuccess { observer in
@@ -88,7 +88,7 @@ public final class ClientStreamingCallHandler<
 
   override internal func endOfStreamReceived() throws {
     guard let eventObserver = self.eventObserver else {
-      self.logger.warning("eventObserver is nil; ignoring end-of-stream")
+      self.logger.warning("eventObserver is nil; ignoring end-of-stream", source: "GRPC")
       return
     }
     eventObserver.whenSuccess { observer in

Sources/GRPC/CallHandlers/ServerStreamingCallHandler.swift

@@ -68,8 +68,10 @@ public final class ServerStreamingCallHandler<
   override internal func processMessage(_ message: RequestPayload) throws {
     guard let eventObserver = self.eventObserver,
       let callContext = self.callContext else {
-      self.logger
-        .error("processMessage(_:) called before the call started or after the call completed")
+      self.logger.error(
+        "processMessage(_:) called before the call started or after the call completed",
+        source: "GRPC"
+      )
       throw GRPCError.StreamCardinalityViolation.request.captureContext()
     }
 

Sources/GRPC/CallHandlers/UnaryCallHandler.swift

@@ -68,8 +68,10 @@ public final class UnaryCallHandler<
   override internal func processMessage(_ message: RequestPayload) throws {
     guard let eventObserver = self.eventObserver,
       let context = self.callContext else {
-      self.logger
-        .error("processMessage(_:) called before the call started or after the call completed")
+      self.logger.error(
+        "processMessage(_:) called before the call started or after the call completed",
+        source: "GRPC"
+      )
       throw GRPCError.StreamCardinalityViolation.request.captureContext()
     }
 

Sources/GRPC/CallHandlers/_BaseCallHandler.swift

@@ -115,7 +115,8 @@ extension _BaseCallHandler: ChannelInboundHandler {
       } catch {
         self.logger.error(
           "error caught while user handler was processing message",
-          metadata: [MetadataKey.error: "\(error)"]
+          metadata: [MetadataKey.error: "\(error)"],
+          source: "GRPC"
         )
         self.errorCaught(context: context, error: error)
       }
@@ -126,7 +127,8 @@ extension _BaseCallHandler: ChannelInboundHandler {
       } catch {
         self.logger.error(
           "error caught on receiving end of stream",
-          metadata: [MetadataKey.error: "\(error)"]
+          metadata: [MetadataKey.error: "\(error)"],
+          source: "GRPC"
         )
         self.errorCaught(context: context, error: error)
       }

Sources/GRPC/ClientCalls/ClientCallTransport.swift

@@ -169,7 +169,7 @@ internal class ChannelTransport<Request, Response> {
         switch result {
         case let .success(mux):
           mux.createStreamChannel(promise: streamPromise) { stream in
-            logger.trace("created http/2 stream")
+            logger.trace("created http/2 stream", source: "GRPC")
 
             return stream.pipeline.addHandlers([
               _GRPCClientChannelHandler(callType: callType, logger: logger),
@@ -262,7 +262,7 @@ extension ChannelTransport: ClientCallOutbound {
   ///
   /// Does not have to be called from the event loop.
   internal func cancel(promise: EventLoopPromise<Void>?) {
-    self.logger.info("rpc cancellation requested")
+    self.logger.info("rpc cancellation requested", source: "GRPC")
 
     if self.eventLoop.inEventLoop {
       self.handleError(GRPCError.RPCCancelledByClient().captureContext(), promise: promise)
@@ -371,7 +371,7 @@ extension ChannelTransport {
       self.logger.debug("buffering request part", metadata: [
         "request_part": "\(part.name)",
         "call_state": "\(self.describeCallState())",
-      ])
+      ], source: "GRPC")
       self.requestBuffer.append(BufferedRequest(message: part, promise: promise))
       if flush {
         self.requestBuffer.mark()
@@ -379,7 +379,11 @@ extension ChannelTransport {
 
     // We have an active stream, just pass the write and promise through.
     case let .active(stream):
-      self.logger.debug("writing request part", metadata: ["request_part": "\(part.name)"])
+      self.logger.debug(
+        "writing request part",
+        metadata: ["request_part": "\(part.name)"],
+        source: "GRPC"
+      )
       stream.write(part, promise: promise)
       if flush {
         stream.flush()
@@ -390,7 +394,7 @@ extension ChannelTransport {
       self.logger.debug("dropping request part", metadata: [
         "request_part": "\(part.name)",
         "call_state": "\(self.describeCallState())",
-      ])
+      ], source: "GRPC")
       promise?.fail(ChannelError.ioOnClosedChannel)
     }
   }
@@ -521,7 +525,11 @@ extension ChannelTransport: ClientCallInbound {
       preconditionFailure("Received response part in 'buffering' state")
 
     case .active:
-      self.logger.debug("received response part", metadata: ["response_part": "\(part.name)"])
+      self.logger.debug(
+        "received response part",
+        metadata: ["response_part": "\(part.name)"],
+        source: "GRPC"
+      )
 
       switch part {
       case let .initialMetadata(metadata):
@@ -557,7 +565,7 @@ extension ChannelTransport: ClientCallInbound {
       self.logger.debug("dropping response part", metadata: [
         "response_part": "\(part.name)",
         "call_state": "\(self.describeCallState())",
-      ])
+      ], source: "GRPC")
     }
   }
 
@@ -579,7 +587,8 @@ extension ChannelTransport: ClientCallInbound {
 
         self.logger.debug(
           "unbuffering request part",
-          metadata: ["request_part": "\(request.message.name)"]
+          metadata: ["request_part": "\(request.message.name)"],
+          source: "GRPC"
         )
         stream.write(request.message, promise: request.promise)
         if shouldFlush {
@@ -587,7 +596,7 @@ extension ChannelTransport: ClientCallInbound {
         }
       }
 
-      self.logger.debug("request buffer drained")
+      self.logger.debug("request buffer drained", source: "GRPC")
       self.state = .active(stream)
 
     case .active:
@@ -620,7 +629,7 @@ extension ChannelTransport {
   private func startTimer() {
     assert(self.stopwatch == nil)
     self.stopwatch = Stopwatch()
-    self.logger.debug("starting rpc")
+    self.logger.debug("starting rpc", source: "GRPC")
   }
 
   private func stopTimer(status: GRPCStatus) {
@@ -632,7 +641,7 @@ extension ChannelTransport {
         "duration_ms": "\(millis)",
         "status_code": "\(status.code.rawValue)",
         "status_message": "\(status.message ?? "nil")",
-      ])
+      ], source: "GRPC")
       self.stopwatch = nil
     }
   }

Sources/GRPC/ClientErrorDelegate.swift

@@ -49,6 +49,7 @@ public class LoggingClientErrorDelegate: ClientErrorDelegate {
     logger.error(
       "grpc client error",
       metadata: [MetadataKey.error: "\(error)"],
+      source: "GRPC",
       file: "\(file)",
       function: "<unknown>",
       line: UInt(line)

Tests/GRPCTests/CapturingLogHandler.swift

@@ -47,6 +47,7 @@ internal struct CapturedLog {
   var level: Logger.Level
   var message: Logger.Message
   var metadata: Logger.Metadata
+  var source: String
   var file: String
   var function: String
   var line: UInt
@@ -70,6 +71,7 @@ internal struct CapturingLogHandler: LogHandler {
     level: Logger.Level,
     message: Logger.Message,
     metadata: Logger.Metadata?,
+    source: String,
     file: String,
     function: String,
     line: UInt
@@ -87,6 +89,7 @@ internal struct CapturingLogHandler: LogHandler {
       level: level,
       message: message,
       metadata: merged,
+      source: source,
       file: file,
       function: function,
       line: line,

Tests/GRPCTests/GRPCTestCase.swift

@@ -36,8 +36,20 @@ class GRPCTestCase: XCTestCase {
   }
 
   override func tearDown() {
+    let logs = self.capturedLogs()
+
+    // The default source emitted by swift-log is the directory containing the '#file' in which the
+    // log was emitted. It's meant to represent the system which emitted the log, typically the
+    // module name. In most cases it's right but in a few places, i.e. where the source lives in
+    // child directories below 'GRPC', it isn't. We'll use this as a sanity check.
+    //
+    // See also: https://github.com/apple/swift-log/issues/145
+    for log in logs {
+      XCTAssertEqual(log.source, "GRPC", "Incorrect log source in \(log.file) on line \(log.line)")
+    }
+
     if GRPCTestCase.alwaysLog || (self.testRun.map { $0.totalFailureCount > 0 } ?? false) {
-      self.printCapturedLogs()
+      self.printCapturedLogs(logs)
     }
 
     super.tearDown()
@@ -86,9 +98,7 @@ class GRPCTestCase: XCTestCase {
   }
 
   /// Prints all captured logs.
-  private func printCapturedLogs() {
-    let logs = self.capturedLogs()
-
+  private func printCapturedLogs(_ logs: [CapturedLog]) {
     let formatter = DateFormatter()
     // We don't care about the date.
     formatter.dateFormat = "HH:mm:ss.SSS"