Add a client debugging initializer and PCAP debugging example (#908)

Motivation:

Sometimes it's useful to have access to the channel so that additional
handlers can be added for debugging. This currently isn't possible for
the client.

Modifications:

- Add a `debugChannelInitializer` to the client configuration and
  builder.
- Add an example using the NIOExtras PCAP handler. Note: it's
  out-of-source so we don't have to directly depend on NIOExtras.

Result:

- Users can get .pcaps from their clients

Examples/PCAPExample/.gitignore

@@ -0,0 +1,7 @@
+.DS_Store
+/.build
+/Packages
+/*.xcodeproj
+xcuserdata/
+*.pcap
+Package.resolved

Examples/PCAPExample/Package.swift

@@ -0,0 +1,30 @@
+// swift-tools-version:5.0
+/*
+ * Copyright 2020, 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.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import PackageDescription
+
+import PackageDescription
+
+let package = Package(
+  name: "PCAPExample",
+  dependencies: [
+    .package(path: "../../"),
+    .package(url: "https://github.com/apple/swift-nio-extras", from: "1.4.0")
+  ],
+  targets: [
+    .target(name: "PCAPExample", dependencies: ["GRPC", "NIOExtras"]),
+  ]
+)

Examples/PCAPExample/README.md

@@ -0,0 +1,46 @@
+# PCAP Debugging Example
+
+This example demonstrates how to use the `NIOWritePCAPHandler` from
+[NIOExtras][swift-nio-extras] with gRPC Swift.
+
+The example configures a client to use the `NIOWritePCAPHandler` with a file
+sink so that all network traffic captured by the handler is written to a
+`.pcap` file. The client makes a single bidirectional streaming RPC to an Echo
+server provided by gRPC Swift.
+
+The captured network traffic can be inspected by opening the `.pcap` with
+tools like [Wireshark][wireshark] or `tcpdump`.
+
+## Running the Example
+
+The example relies on the Echo server from the gRPC Swift repository. To start
+the server, from the **root of the gRPC repository** run:
+
+```sh
+$ swift run Echo server 0
+```
+
+Note the port printed by the server. In a separate shell, in **this** directory
+run:
+
+```sh
+$ swift run PCAPExample localhost <SERVER_PORT>
+```
+
+Some logs should be emitted similar to below, including the path of the
+*.pcap* file:
+
+```sh
+2020-07-24T10:48:50+0100 info gRPC PCAP Demo : Creating fileSink for path './channel-ObjectIdentifier(0x00007f8a25604c40).pcap'
+2020-07-24T10:48:50+0100 info gRPC PCAP Demo : ✅ Successfully created fileSink for path './channel-ObjectIdentifier(0x00007f8a25604c40).pcap'
+...
+2020-07-24T10:48:50+0100 info gRPC PCAP Demo : ✅ RPC completed successfully
+...
+2020-07-24T10:48:50+0100 info gRPC PCAP Demo : Done!
+```
+
+The *.pcap* file can be opened with either: [Wireshark][wireshark] or `tcpdump
+-r <PCAP_FILE>`.
+
+[swift-nio-extras]: https://github.com/apple/swift-nio-extras
+[wireshark]: https://wireshark.org

Examples/PCAPExample/Sources/PCAPExample/echo.grpc.swift

@@ -0,0 +1 @@
+../../../../Sources/Examples/Echo/Model/echo.grpc.swift

Examples/PCAPExample/Sources/PCAPExample/echo.pb.swift

@@ -0,0 +1 @@
+../../../../Sources/Examples/Echo/Model/echo.pb.swift

Examples/PCAPExample/Sources/PCAPExample/main.swift

@@ -0,0 +1,142 @@
+/*
+ * Copyright 2020, 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.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import Dispatch
+import GRPC
+import NIO
+import NIOExtras
+import Logging
+
+// Parse the command line args.
+var args = CommandLine.arguments
+guard args.count == 3, let port = Int(args[2]) else {
+  let usage = """
+  Usage: \(args[0]) SERVER_HOST SERVER_PORT
+
+  Note: you can start a server from the root of the gRPC Swift directory by running:
+
+    $ swift run Echo server 0
+  """
+  print(usage)
+  exit(1)
+}
+let host = args[1]
+
+// Create a logger.
+let logger = Logger(label: "gRPC PCAP Demo")
+
+// Closing file sinks is blocking, it therefore can't be done on an EventLoop.
+let fileSinkCloseQueue = DispatchQueue(label: "io.grpc")
+let fileSinkCloseGroup = DispatchGroup()
+defer {
+  // Make sure we wait for all file sinks to be closed before we exit.
+  fileSinkCloseGroup.wait()
+  logger.info("Done!")
+}
+
+/// Adds a `NIOWritePCAPHandler` to the given channel.
+///
+/// A file sink will also be created to write the PCAP to `./channel-{ID}.pcap` where `{ID}` is
+/// an identifier created from the given `channel`. The file sink will be closed when the channel
+/// closes and will notify the `fileSinkCloseGroup` when it has been closed.
+///
+/// - Parameter channel: The channel to add the PCAP handler to.
+/// - Returns: An `EventLoopFuture` indicating whether the PCAP handler was successfully added.
+func addPCAPHandler(toChannel channel: Channel) -> EventLoopFuture<Void> {
+  // The debug initializer can be called multiple times. We'll use the object ID of the channel
+  // to disambiguate between the files.
+  let channelID = ObjectIdentifier(channel)
+  let path = "./channel-\(channelID).pcap"
+
+  logger.info("Creating fileSink for path '\(path)'")
+
+  do {
+    // Create a file sink.
+    let fileSink = try NIOWritePCAPHandler.SynchronizedFileSink.fileSinkWritingToFile(path: path) { error in
+      logger.error("💥 Failed to write with error '\(error)' for path '\(path)'")
+    }
+
+    logger.info("✅ Successfully created fileSink for path '\(path)'")
+
+    // We need to close the file sink when we're done. It can't be closed from the event loop so
+    // we'll use a dispatch queue instead.
+    fileSinkCloseGroup.enter()
+    channel.closeFuture.whenComplete { _ in
+      fileSinkCloseQueue.async {
+        do {
+          try fileSink.syncClose()
+        } catch {
+          logger.error("💥 Failed to close fileSink with error '\(error)' for path '\(path)'")
+        }
+      }
+      fileSinkCloseGroup.leave()
+    }
+
+    // Add the handler to the pipeline.
+    let handler = NIOWritePCAPHandler(mode: .client, fileSink: fileSink.write(buffer:))
+    // We're not using TLS in this example so ".first" is the right place.
+    return channel.pipeline.addHandler(handler, position: .first)
+  } catch {
+    logger.error("💥 Failed to create fileSink with error '\(error)' for path '\(path)'")
+    return channel.eventLoop.makeFailedFuture(error)
+  }
+}
+
+// Create an `EventLoopGroup`.
+let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+defer {
+  try! group.syncShutdownGracefully()
+}
+
+// Create a channel.
+let channel = ClientConnection.insecure(group: group)
+  // Set the debug initializer: it will add a handler to each created channel to write a PCAP when
+  // the channel is closed.
+  .withDebugChannelInitializer(addPCAPHandler(toChannel:))
+  // We're connecting to our own server here; we'll disable connection re-establishment.
+  .withConnectionReestablishment(enabled: false)
+  // Connect!
+  .connect(host: host, port: port)
+
+// Create a client.
+let echo = Echo_EchoClient(channel: channel)
+
+// Start an RPC.
+let update = echo.update { response in
+  logger.info("Received response '\(response.text)'")
+}
+
+// Send some requests.
+for text in ["foo", "bar", "baz", "thud", "grunt", "gorp"] {
+  update.sendMessage(.with { $0.text = text }).whenSuccess {
+    logger.info("Sent request '\(text)'")
+  }
+}
+// Close the request stream.
+update.sendEnd(promise: nil)
+
+// Once the RPC finishes close the connection.
+let closed = update.status.flatMap { status -> EventLoopFuture<Void> in
+  if status.isOk {
+    logger.info("✅ RPC completed successfully")
+  } else {
+    logger.error("💥 RPC failed with status '\(status)'")
+  }
+  logger.info("Closing channel")
+  return channel.close()
+}
+
+// Wait for the channel to be closed.
+try closed.wait()

Examples/README.md

@@ -9,6 +9,11 @@ Samples that call Google gRPC APIs are in the [Google](Google) directory.
 [EchoWeb](EchoWeb) is an example which uses a JavaScript client to communicate
 with a gRPC Server using the gRPC-Web protocol.
 
+## PCAP
+
+[PCAPExample](PCAPExample) demonstrates how to configure a client to use a
+debug channel initializer which writes *.pcap* files.
+
 ## Other Examples
 
 See also:

Sources/GRPC/ClientConnection.swift

@@ -518,6 +518,12 @@ extension ClientConnection {
     /// Defaults to a no-op logger.
     public var backgroundActivityLogger: Logger
 
+    /// A channel initializer which will be run after gRPC has initialized each channel. This may be
+    /// used to add additional handlers to the pipeline and is intended for debugging.
+    ///
+    /// - Warning: The initializer closure may be invoked *multiple times*.
+    public var debugChannelInitializer: ((Channel) -> EventLoopFuture<Void>)?
+
     /// Create a `Configuration` with some pre-defined defaults. Prefer using
     /// `ClientConnection.secure(group:)` to build a connection secured with TLS or
     /// `ClientConnection.insecure(group:)` to build a plaintext connection.
@@ -538,6 +544,8 @@ extension ClientConnection {
     /// - Parameter httpTargetWindowSize: The HTTP/2 flow control target window size.
     /// - Parameter backgroundActivityLogger: A logger for background information (such as
     ///     connectivity state). Defaults to a no-op logger.
+    /// - Parameter debugChannelInitializer: A channel initializer will be called after gRPC has
+    ///     initialized the channel. Defaults to `nil`.
     public init(
       target: ConnectionTarget,
       eventLoopGroup: EventLoopGroup,
@@ -550,7 +558,8 @@ extension ClientConnection {
       connectionIdleTimeout: TimeAmount = .minutes(5),
       callStartBehavior: CallStartBehavior = .waitsForConnectivity,
       httpTargetWindowSize: Int = 65535,
-      backgroundActivityLogger: Logger = Logger(label: "io.grpc", factory: { _ in SwiftLogNoOpLogHandler() })
+      backgroundActivityLogger: Logger = Logger(label: "io.grpc", factory: { _ in SwiftLogNoOpLogHandler() }),
+      debugChannelInitializer: ((Channel) -> EventLoopFuture<Void>)? = nil
     ) {
       self.target = target
       self.eventLoopGroup = eventLoopGroup
@@ -564,6 +573,7 @@ extension ClientConnection {
       self.callStartBehavior = callStartBehavior
       self.httpTargetWindowSize = httpTargetWindowSize
       self.backgroundActivityLogger = backgroundActivityLogger
+      self.debugChannelInitializer = debugChannelInitializer
     }
   }
 }

Sources/GRPC/ConnectionManager.swift

@@ -603,7 +603,7 @@ extension ConnectionManager {
       .channelOption(ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR), value: 1)
       .channelOption(ChannelOptions.socket(IPPROTO_TCP, TCP_NODELAY), value: 1)
       .channelInitializer { channel in
-        channel.configureGRPCClient(
+        let initialized = channel.configureGRPCClient(
           httpTargetWindowSize: configuration.httpTargetWindowSize,
           tlsConfiguration: configuration.tls?.configuration,
           tlsServerHostname: serverHostname,
@@ -613,6 +613,15 @@ extension ConnectionManager {
           errorDelegate: configuration.errorDelegate,
           logger: self.logger
         )
+
+        // Run the debug initializer, if there is one.
+        if let debugInitializer = configuration.debugChannelInitializer {
+          return initialized.flatMap {
+            debugInitializer(channel)
+          }
+        } else {
+          return initialized
+        }
       }
 
     if let connectTimeout = connectTimeout {

Sources/GRPC/GRPCChannel/GRPCChannelBuilder.swift

@@ -250,6 +250,20 @@ extension ClientConnection.Builder {
   }
 }
 
+extension ClientConnection.Builder {
+  /// A channel initializer which will be run after gRPC has initialized each channel. This may be
+  /// used to add additional handlers to the pipeline and is intended for debugging.
+  ///
+  /// - Warning: The initializer closure may be invoked *multiple times*.
+  @discardableResult
+  public func withDebugChannelInitializer(
+    _ debugChannelInitializer: @escaping (Channel) -> EventLoopFuture<Void>
+  ) -> Self {
+    self.configuration.debugChannelInitializer = debugChannelInitializer
+    return self
+  }
+}
+
 fileprivate extension Double {
   static func seconds(from amount: TimeAmount) -> Double {
     return Double(amount.nanoseconds) / 1_000_000_000

Tests/GRPCTests/ClientDebugChannelInitializerTests.swift

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2020, 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.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import GRPC
+import EchoModel
+import EchoImplementation
+import NIO
+import NIOConcurrencyHelpers
+import XCTest
+
+class ClientDebugChannelInitializerTests: GRPCTestCase {
+  func testDebugChannelInitializerIsCalled() throws {
+    let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+    defer {
+      XCTAssertNoThrow(try group.syncShutdownGracefully())
+    }
+
+    let server = try Server.insecure(group: group)
+      .withServiceProviders([EchoProvider()])
+      .bind(host: "localhost", port: 0)
+      .wait()
+    defer {
+      XCTAssertNoThrow(try server.close().wait())
+    }
+
+    let debugInitializerCalled = group.next().makePromise(of: Void.self)
+    let connection = ClientConnection.insecure(group: group)
+      .withBackgroundActivityLogger(self.clientLogger)
+      .withDebugChannelInitializer { channel in
+        debugInitializerCalled.succeed(())
+        return channel.eventLoop.makeSucceededFuture(())
+      }
+      .connect(host: "localhost", port: server.channel.localAddress!.port!)
+    defer {
+      XCTAssertNoThrow(try connection.close().wait())
+    }
+
+    let echo = Echo_EchoClient(channel: connection)
+    // Make an RPC to trigger channel creation.
+    let get = echo.get(.with { $0.text = "Hello!" })
+    XCTAssertTrue(try get.status.map { $0.isOk }.wait())
+
+    // Check the initializer was called.
+    XCTAssertNoThrow(try debugInitializerCalled.futureResult.wait())
+  }
+}

Tests/GRPCTests/XCTestManifests.swift

@@ -80,6 +80,15 @@ extension ClientConnectionBackoffTests {
     ]
 }
 
+extension ClientDebugChannelInitializerTests {
+    // DO NOT MODIFY: This is autogenerated, use:
+    //   `swift test --generate-linuxmain`
+    // to regenerate.
+    static let __allTests__ClientDebugChannelInitializerTests = [
+        ("testDebugChannelInitializerIsCalled", testDebugChannelInitializerIsCalled),
+    ]
+}
+
 extension ClientTLSFailureTests {
     // DO NOT MODIFY: This is autogenerated, use:
     //   `swift test --generate-linuxmain`
@@ -912,6 +921,7 @@ public func __allTests() -> [XCTestCaseEntry] {
         testCase(ClientCancellingTests.__allTests__ClientCancellingTests),
         testCase(ClientClosedChannelTests.__allTests__ClientClosedChannelTests),
         testCase(ClientConnectionBackoffTests.__allTests__ClientConnectionBackoffTests),
+        testCase(ClientDebugChannelInitializerTests.__allTests__ClientDebugChannelInitializerTests),
         testCase(ClientTLSFailureTests.__allTests__ClientTLSFailureTests),
         testCase(ClientTLSHostnameOverrideTests.__allTests__ClientTLSHostnameOverrideTests),
         testCase(ClientThrowingWhenServerReturningErrorTests.__allTests__ClientThrowingWhenServerReturningErrorTests),