Python基于gRPC实现Envoy限流服务时如何构造并返回RateLimitResponse

• 异步gRPC方法未声明为协程，且未接收上下文参数，调用时会出现参数不匹配异常
• unary_unary_rpc_method_handler 传入的是类的未绑定方法RL.ShouldRateLimit，而非实例的绑定方法，运行时会丢失self上下文
• 未添加server.wait_for_termination()阻塞逻辑，服务启动后会直接退出
• RateLimitResponse仅实例化空对象未填充必填字段，会导致Envoy端行为不符合预期

根据对应版本proto定义，最少需要填充overall_code枚举字段，支持的枚举值包括：

• RateLimitResponse.OK：表示当前请求不限流
• RateLimitResponse.OVER_LIMIT：表示当前请求已触发限流
• RateLimitResponse.UNKNOWN：默认值，Envoy遇到该值默认会放行，但建议显式指定

如需返回限流配额、重置时间等扩展信息，可填充statuses数组（v3版本推荐使用）或旧版status单个字段，示例：

from google.protobuf.duration_pb2 import Duration
from envoy.service.ratelimit.v3.rls_pb2 import RateLimitResponse

# 构造不限流响应
ok_resp = RateLimitResponse(
    overall_code=RateLimitResponse.OK
)

# 构造限流响应：10秒后重置，当前窗口每分钟允许100次请求
over_limit_resp = RateLimitResponse(
    overall_code=RateLimitResponse.OVER_LIMIT,
    statuses=[
        RateLimitResponse.QuotaStatus(
            status=RateLimitResponse.RateLimit(
                unit=RateLimitResponse.RateLimit.MINUTE,
                requests_per_unit=100,
                reset_duration=Duration(seconds=10)
            )
        )
    ]
)

import logging
import asyncio
import grpc
from google.protobuf.duration_pb2 import Duration
from envoy.service.ratelimit.v3.rls_pb2 import RateLimitResponse, RateLimitRequest


class RL:
    # gRPC异步方法必须是协程，且显式接收context参数
    async def ShouldRateLimit(self, request: RateLimitRequest, context: grpc.aio.ServicerContext) -> RateLimitResponse:
        # 此处可添加自定义限流逻辑，比如解析request中的descriptors、查询存储等
        return RateLimitResponse(
            overall_code=RateLimitResponse.OK
        )

def add_handler(servicer: RL, server: grpc.aio.Server):
    rpc_method_handlers = {
        'ShouldRateLimit': grpc.unary_unary_rpc_method_handler(
            # 传入实例绑定的方法，避免上下文丢失
            servicer.ShouldRateLimit,
            request_deserializer=RateLimitRequest.FromString,
            response_serializer=RateLimitResponse.SerializeToString,
        )
    }
    generic_handler = grpc.method_handlers_generic_handler(
            'envoy.service.ratelimit.v3.RateLimitService',
            rpc_method_handlers
    )
    server.add_generic_rpc_handlers((generic_handler,))


async def serve():
    server = grpc.aio.server()
    add_handler(RL(), server)
    listen_addr = '[::]:5051'
    server.add_insecure_port(listen_addr)
    logging.info(f'Starting server on {listen_addr}')
    await server.start()
    # 阻塞等待服务终止，避免进程直接退出
    await server.wait_for_termination()

if __name__ == '__main__':
    logging.basicConfig(level=logging.DEBUG)
    asyncio.run(serve())

• 直接将构造完成的protobuf对象作为协程返回值即可，序列化逻辑由handler自动处理，无需手动调用SerializeToString
• 所有枚举值必须使用编译生成的类属性，禁止直接传入数字常量，避免版本不匹配导致的解析错误
• 如需返回gRPC层错误（比如内部异常），可通过context.set_code(grpc.StatusCode.INTERNAL)设置状态码，无需手动抛异常

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