Example on how to intercept the context _after_ the RPC is processed

[troyswanson]

I'm trying to write a very basic interceptor that will log the method name, status code, and duration of the RPC method, but I'm having a difficulty understanding how that would be done with the documentation and the examples that are currently available.

Any help with this would be appreciated.

[dan-hipschman]

Hi Troy,

Something like this should do it:

class ServerLoggingInterceptor(ServerInterceptor):
    def intercept(
        self,
        method: Callable,
        request: Any,
        context: grpc.ServicerContext,
        method_name: str,
    ) -> Any:
        parse_result = parse_method_name(method_name)
        service_name = parse_result.service
        endpoint_name = parse_result.method

        status = "succeeded"
        time_start = time.time()
        try:
            return method(self, request, context)
        except:
            status = "failed"
            raise
        finally:
            time_end = time.time()
            total_time_ms = (time_end - time_start) * 1000
            logger.info(f"{service_name}/{endpoint_name} took {total_time_ms}ms and {status}")

If you want more info on errors you can extract it from the exception.

[troyswanson]

@dan-hipschman-od Thank you!

Slightly modified:

from typing import Callable, Any, cast
from time import time

from grpc import ServicerContext, StatusCode
from grpc_interceptor import ServerInterceptor, parse_method_name

class AccessLoggerInterceptor(ServerInterceptor):
    def __init__(self, logger):
        self.logger = logger

    def intercept(
        self,
        method: Callable,
        request: Any,
        context: ServicerContext,
        method_name: str,
    ) -> Any:

        parsed_method = parse_method_name(method_name)
        time_start = time()

        try:
            return method(request, context)

        finally:
            time_end = time()
            duration = round((time_end - time_start) * 1000, 5)
            status_code = cast(StatusCode, context.code()) or StatusCode.OK
            self.logger.info(
                "AccessLogger",
                package=parsed_method.package,
                service=parsed_method.service,
                method=parsed_method.method,
                duration=duration,
                status_code=status_code.name,
            )

This works with the ExceptionToStatusInteceptor that comes with this package. The order of how the interceptors are listed when creating the server is important.

interceptors = [AccessLoggerInterceptor(logger), ExceptionToStatusInterceptor()]

[troyswanson]

This actually doesn't work properly with uncaught exceptions that are handled deeper by the grpc package...

[troyswanson]

After some more work, pairing this with ExceptionToStatusInterceptor and passing grpc.StatusCode.UNKNOWN to the status_on_unknown_exception argument correctly handles generic exceptions that are raised.

Example:

interceptors = [
    AccessLoggerInterceptor(logger),
    ExceptionToStatusInterceptor(grpc.StatusCode.UNKNOWN),
]