How to log requests and responses with interceptor in v2?

[alex-vasenin-peggy]

What are you trying to achieve?

I would like to have an ability to (conditionally) see contents of all messages that are sent or received via GRPCClient v2 for debug purposes. In our app we have a command line argument to enable debugging interaction with backend. Sources of ClientInterceptor has an example of LoggingInterceptor, but it only logs metadata, but not the messages.

What have you tried so far?

I tried investigation whether ClientContext, StreamingClientRequest and StreamingClientResponse has the data I need. It seems like there is no way to get sent message from the StreamingClientRequest.producer. It might be possible to get the message StreamingClientResponse.bodyParts, but I can find an example how to do it properly (and if it's safe to do at all in interceptor). Any advice on how to approach this?

[glbrntt]

Here's an example interceptor you can use as a starting point. I've tried to explain what each step is doing, hopefully it makes sense. There are a few TODOs where errors aren't logged but should be. I hope it helps!

import GRPCCore
import Logging

struct LoggingInterceptor: ClientInterceptor {
  private let logger: Logger

  init(logger: Logger) {
    self.logger = logger
  }

  func intercept<Input, Output>(
    request: StreamingClientRequest<Input>,
    context: ClientContext,
    next: (
      _ request: StreamingClientRequest<Input>,
      _ context: ClientContext
    ) async throws -> StreamingClientResponse<Output>
  ) async throws -> StreamingClientResponse<Output> where Input: Sendable, Output: Sendable {
    // Log the request head, i.e. the RPC being called and its metadata.
    self.logger.debug("New request for \(context.descriptor), metadata=\(request.metadata)")

    // Prepare a *new* request object which wraps the one which was passed in. The request
    // has a 'producer' closure which, when called, produces messages to write to the server.
    // The pattern here is to create a new writer object which logs each message and then
    // calls through to the 'real' underlying writer.
    let loggingRequest = StreamingClientRequest(of: Input.self, metadata: request.metadata) { writer in
      // Wrap the provided logger in a new writer.
      let loggingWriter = LoggingWriter(writer, logger: self.logger)
      // Pass this through to the original message producer.
      try await request.producer(RPCWriter(wrapping: loggingWriter))
    }

    // Pass the logging request through to the next interceptor.
    let response = try await next(loggingRequest, context)

    // Check if the server accepted the response (i.e. processed the RPC or rejected it
    // without sending back any messages).
    switch response.accepted {
    case .success(let contents):
      // Server accepted the request and attempted to process it (note: it does *not* mean
      // the RPC completed successfully).
      self.logger.debug("Request accpeted")
      // The way this works is by wrapping the 'contents.bodyParts' async sequence in one
      // which logs messages as well.
      return StreamingClientResponse(
        accepted: .success(
          .init(
            metadata: contents.metadata,
            bodyParts: RPCAsyncSequence(
              wrapping: LoggingAsyncSequence(
                contents.bodyParts,
                logger: self.logger
              )
            )
          )
        )
      )

    case .failure(let error):
      // Server rejected the request and didn't respond with any messages.
      self.logger.debug("Request rejected code=\(error.code), message=\(error.message), metadata=\(error.metadata)")
      return StreamingClientResponse(error: error)
    }
  }
}

struct LoggingWriter<Element: Sendable>: RPCWriterProtocol {
  private var base: RPCWriter<Element>
  private var logger: Logger

  init(_ base: RPCWriter<Element>, logger: Logger) {
    self.base = base
    self.logger = logger
  }

  func write(_ element: Element) async throws {
    self.logger.debug("Writing message=\(element)")
    // TODO: log errors.
    try await self.base.write(element)
  }

  func write(contentsOf elements: some Sequence<Self.Element>) async throws {
    for element in elements {
      try await self.write(element)
    }
  }
}

struct LoggingAsyncSequence<Base: AsyncSequence, Message: Sendable>: AsyncSequence where Base.Element == StreamingClientResponse<Message>.Contents.BodyPart, Base: Sendable {
  typealias Element = StreamingClientResponse<Message>.Contents.BodyPart

  private var base: Base
  private var logger: Logger

  init(_ base: Base, logger: Logger) {
    self.base = base
    self.logger = logger
  }

  struct AsyncIterator: AsyncIteratorProtocol {
    typealias Element = StreamingClientResponse<Message>.Contents.BodyPart

    private var base: Base.AsyncIterator
    private var logger: Logger

    init(base: Base.AsyncIterator, logger: Logger) {
      self.base = base
      self.logger = logger
    }

    mutating func next() async throws -> StreamingClientResponse<Message>.Contents.BodyPart? {
      // TODO: log errors. Particularly 'RPCErrorConvertible' which will contain a status code
      // and message.
      let part = try await self.base.next()

      switch part {
      case .message(let message):
        self.logger.debug("Received message=\(message)")
      case .trailingMetadata(let metadata):
        self.logger.debug("Received trailing metadata=\(metadata)")
      case .none:
        ()
      }

      return part
    }
  }

  func makeAsyncIterator() -> AsyncIterator {
    AsyncIterator(base: self.base.makeAsyncIterator(), logger: self.logger)
  }
}

As an addendum to this, we are aware that this is harder than it could be. I mean that on two fronts:

1. writing the interceptor to look at every message isn't straightforward, and
2. getting logs about the lifecycle of an RPC.

We would definitely like to make these easier to do.

[alex-vasenin-peggy]

Oh, wow! That's a lot of code, thanks for very extensive answer! I will need some time to figure it all out, but that's a good starting point. It certainly deserves a place in examples.

[alex-vasenin-peggy]

@glbrntt Based on your very detailed answer I managed to implement debug logging of all requests and responses that is transparent to callers. One question that I wasn't be able to answer easily is this one: if the server didn't accept the request for whatever reason I don't have an easy way to log the request message. In order to do that, I think I should cache either the request itself or the data I want to log, so if it fails - I can log it. Is this correct or I'm missing something obvious?

[glbrntt]

That's correct but bear in mind that for streaming requests you could end up storing many messages.