Middleware chain e interceptors no gRPC

1. Introdução aos Interceptors no gRPC

Interceptors são funções de middleware que interceptam chamadas gRPC antes que elas atinjam o handler final. Eles permitem executar lógica adicional — como logging, autenticação, métricas e tracing — de forma modular e reutilizável.

No gRPC, existem dois tipos principais de interceptors:

  • Interceptors unários: interceptam chamadas do tipo request-response (um único pedido, uma única resposta).
  • Interceptors de stream: interceptam chamadas que envolvem fluxos contínuos de dados (server streaming, client streaming ou bidirecional).

Diferente do middleware HTTP tradicional, que opera sobre requisições HTTP, os interceptors gRPC operam sobre chamadas RPC, com acesso direto ao contexto, metadados e ao fluxo de mensagens serializadas pelo Protocol Buffers.

2. Interceptors Unários: Implementação e Uso

A assinatura de um interceptor unário do lado do servidor é:

type UnaryServerInterceptor func(
    ctx context.Context,
    req interface{},
    info *grpc.UnaryServerInfo,
    handler grpc.UnaryHandler,
) (interface{}, error)

Exemplo prático: interceptor de logging que registra requisições e respostas:

func LoggingInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
    log.Printf("Requisição recebida: método=%s, payload=%v", info.FullMethod, req)

    resp, err := handler(ctx, req)

    if err != nil {
        log.Printf("Erro na resposta: método=%s, erro=%v", info.FullMethod, err)
    } else {
        log.Printf("Resposta enviada: método=%s, payload=%v", info.FullMethod, resp)
    }

    return resp, err
}

Para registrar o interceptor no servidor:

s := grpc.NewServer(
    grpc.UnaryInterceptor(LoggingInterceptor),
)

Do lado do cliente, o UnaryClientInterceptor segue padrão similar, permitindo interceptar chamadas de saída.

3. Interceptors de Stream: Trabalhando com Fluxos Contínuos

Interceptors de stream possuem assinatura diferente:

type StreamServerInterceptor func(
    srv interface{},
    ss grpc.ServerStream,
    info *grpc.StreamServerInfo,
    handler grpc.StreamHandler,
) error

Para interceptar o fluxo, precisamos wrappear o ServerStream:

type wrappedStream struct {
    grpc.ServerStream
}

func (w *wrappedStream) RecvMsg(m interface{}) error {
    err := w.ServerStream.RecvMsg(m)
    if err == nil {
        log.Printf("Mensagem recebida no stream: %v", m)
    }
    return err
}

func (w *wrappedStream) SendMsg(m interface{}) error {
    log.Printf("Mensagem enviada no stream: %v", m)
    return w.ServerStream.SendMsg(m)
}

func StreamLoggingInterceptor(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
    return handler(srv, &wrappedStream{ss})
}

Esse padrão é essencial para monitorar streams bidirecionais, onde cada mensagem enviada ou recebida pode ser interceptada.

4. Middleware Chain: Encadeamento de Interceptors

Encadear interceptors significa executar múltiplos interceptors em sequência. A ordem importa: interceptores são executados como uma pilha (LIFO).

Implementação manual de uma chain simples:

func chainUnaryInterceptors(interceptors ...grpc.UnaryServerInterceptor) grpc.UnaryServerInterceptor {
    n := len(interceptors)
    return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
        chain := handler
        for i := n - 1; i >= 0; i-- {
            chain = wrapInterceptor(interceptors[i], info, chain)
        }
        return chain(ctx, req)
    }
}

func wrapInterceptor(interceptor grpc.UnaryServerInterceptor, info *grpc.UnaryServerInfo, next grpc.UnaryHandler) grpc.UnaryHandler {
    return func(ctx context.Context, req interface{}) (interface{}, error) {
        return interceptor(ctx, req, info, next)
    }
}

Para simplificar, utilize a biblioteca go-grpc-middleware:

go get github.com/grpc-ecosystem/go-grpc-middleware

import "github.com/grpc-ecosystem/go-grpc-middleware"

s := grpc.NewServer(
    grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(
        LoggingInterceptor,
        AuthInterceptor,
        MetricsInterceptor,
    )),
)

5. Interceptors para Autenticação e Autorização

Validação de tokens JWT em interceptors unários:

func AuthInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
    md, ok := metadata.FromIncomingContext(ctx)
    if !ok {
        return nil, status.Errorf(codes.Unauthenticated, "metadados ausentes")
    }

    token := md["authorization"]
    if len(token) == 0 {
        return nil, status.Errorf(codes.Unauthenticated, "token ausente")
    }

    claims, err := validateJWT(token[0])
    if err != nil {
        return nil, status.Errorf(codes.Unauthenticated, "token inválido: %v", err)
    }

    // Propaga claims no contexto
    ctx = context.WithValue(ctx, "claims", claims)
    return handler(ctx, req)
}

Para streams, a verificação deve ocorrer antes de iniciar o stream, mas também pode ser feita a cada mensagem recebida.

6. Interceptors para Observabilidade (Tracing e Métricas)

Integração com OpenTelemetry para tracing distribuído:

import (
    "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
)

s := grpc.NewServer(
    grpc.UnaryInterceptor(otelgrpc.UnaryServerInterceptor()),
    grpc.StreamInterceptor(otelgrpc.StreamServerInterceptor()),
)

Para métricas com Prometheus:

import "github.com/grpc-ecosystem/go-grpc-prometheus"

s := grpc.NewServer(
    grpc.UnaryInterceptor(grpc_prometheus.UnaryServerInterceptor),
    grpc.StreamInterceptor(grpc_prometheus.StreamServerInterceptor),
)

Logging estruturado com correlação de spans pode ser feito extraindo o trace ID do contexto e incluindo-o nos logs.

7. Interceptors do Lado do Cliente (Client-Side)

Interceptors client-side permitem implementar retry, circuit breaker e timeout:

func RetryInterceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
    var err error
    for i := 0; i < 3; i++ {
        err = invoker(ctx, method, req, reply, cc, opts...)
        if err == nil {
            return nil
        }
        if !shouldRetry(err) {
            break
        }
        time.Sleep(time.Duration(i+1) * 100 * time.Millisecond)
    }
    return err
}

Propagação de deadline:

ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()

8. Boas Práticas e Armadilhas Comuns

  • Performance: Evite alocações desnecessárias dentro dos interceptors. Reutilize buffers e estruturas quando possível.
  • Ordem: Coloque interceptors de segurança (auth) antes dos de observabilidade para evitar logar requisições não autorizadas.
  • Testes: Use mocks para testar interceptors isoladamente:
func TestLoggingInterceptor(t *testing.T) {
    handler := func(ctx context.Context, req interface{}) (interface{}, error) {
        return &pb.Response{Message: "ok"}, nil
    }

    resp, err := LoggingInterceptor(context.Background(), &pb.Request{Name: "test"}, &grpc.UnaryServerInfo{FullMethod: "/test"}, handler)
    assert.NoError(t, err)
    assert.NotNil(t, resp)
}
  • Contexto: Nunca ignore o contexto — ele carrega metadados, deadlines e valores de tracing.
  • Panics: Use recover dentro de interceptors para evitar que panics derrubem o servidor.

Referências