如何在Python gRPC服务器中记录请求和响应活动日志？

问题说明 #

• 完成gRPC Python基础教程学习后，服务可正常响应请求，但命令行无任何请求相关日志输出
• 期望日志效果对齐Django开发服务器、Python内置http.server，可直观看到请求接收、响应状态、请求耗时等信息，参考日志格式如下：

Serving HTTP on 127.0.0.1 port 8000 (http://127.0.0.1:8000/) ...
127.0.0.1 - - [27/Jun/2022 13:06:08] "GET / HTTP/1.1" 200 -
127.0.0.1 - - [27/Jun/2022 13:06:08] code 404, message File not found

• 尝试通过启动参数GRPC_VERBOSITY=DEBUG python greeter_server.py开启日志，仅输出大量框架底层无关信息，将日志级别调整为INFO后依然无请求相关日志输出。

grpc.server的interceptors参数官方说明：可选的ServerInterceptor对象列表，可在传入的RPC请求交付给对应处理器之前对其进行监听，也可按需修改请求；拦截器将按照指定顺序获得控制权，该API目前为实验性API。

实现方法 #

GRPC_VERBOSITY环境变量仅控制gRPC底层C核心的运行日志，不会输出业务请求访问记录，需要通过grpc.server提供的interceptors参数自定义服务端拦截器实现访问日志打印，步骤如下：

1. 编写自定义日志拦截器

from grpc import ServerInterceptor, StatusCode
from datetime import datetime
import time

class AccessLogInterceptor(ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        req_start = time.time()
        rpc_method = handler_call_details.method
        # 获取客户端IP
        client_ip = "unknown"
        for meta_key, meta_val in handler_call_details.invocation_metadata():
            if meta_key == "x-forwarded-for":
                client_ip = meta_val.split(",")[0].strip()
                break
            if meta_key == "peer":
                client_ip = meta_val.split(":")[0]

        # 执行实际RPC逻辑
        rpc_handler = continuation(handler_call_details)

        def wrap_behavior(behavior):
            def wrapped(request, context):
                try:
                    resp = behavior(request, context)
                    cost_ms = round((time.time() - req_start)*1000, 2)
                    status_code = context.code() if context.code() else StatusCode.OK
                    print(f'{client_ip} - - [{datetime.now().strftime("%d/%b/%Y %H:%M:%S")}] "{rpc_method}" {status_code.value[0]} - {cost_ms}ms')
                    return resp
                except Exception as e:
                    cost_ms = round((time.time() - req_start)*1000, 2)
                    print(f'{client_ip} - - [{datetime.now().strftime("%d/%b/%Y %H:%M:%S")}] "{rpc_method}" 500 - {cost_ms}ms, error: {str(e)}')
                    raise e
            return wrapped

        if rpc_handler:
            # 适配四种gRPC调用模式：一元、服务端流、客户端流、双向流
            if rpc_handler.unary_unary:
                rpc_handler.unary_unary = wrap_behavior(rpc_handler.unary_unary)
            if rpc_handler.unary_stream:
                rpc_handler.unary_stream = wrap_behavior(rpc_handler.unary_stream)
            if rpc_handler.stream_unary:
                rpc_handler.stream_unary = wrap_behavior(rpc_handler.stream_unary)
            if rpc_handler.stream_stream:
                rpc_handler.stream_stream = wrap_behavior(rpc_handler.stream_stream)
        return rpc_handler

2. 服务启动时注入拦截器

import grpc
from concurrent import futures
# 引入自身项目的pb模块、服务实现类
# import helloworld_pb2_grpc
# from service_impl import GreeterServicer

def run_server():
    listen_port = "50051"
    server = grpc.server(
        futures.ThreadPoolExecutor(max_workers=10),
        interceptors=[AccessLogInterceptor()]
    )
    # 注册自身服务
    # helloworld_pb2_grpc.add_GreeterServicer_to_server(GreeterServicer(), server)
    server.add_insecure_port(f"0.0.0.0:{listen_port}")
    server.start()
    print(f"Serving gRPC on 0.0.0.0 port {listen_port} ...")
    server.wait_for_termination()

if __name__ == "__main__":
    run_server()

补充说明 #

• 上述拦截器兼容所有gRPC调用模式，无需为单个接口单独配置
• 日志输出字段可按需调整，支持新增请求参数、TraceID、响应大小等自定义字段
• 若服务部署在Nginx等反向代理后，需要根据代理转发的真实请求头调整客户端IP获取逻辑

内容的提问来源于stack exchange，提问作者tread