Documentation
¶
Overview ¶
Package grpcutil provides a gRPC server factory with a standard interceptor chain and functional options for configuration.
============================================================================ gRPC INTERCEPTOR CHAIN ============================================================================
WHY interceptors:
gRPC interceptors are the equivalent of HTTP middleware. They provide cross-cutting concerns (logging, auth, tracing, recovery) that apply to every RPC without modifying handler code. Without interceptors, every handler would need to: check auth, log request, recover from panics, start trace spans — that's 20+ lines of boilerplate per handler.
CHAIN ORDER: recovery → logging → tracing → auth → handler
┌──────────────────────────────────────────────────────┐ │ Incoming RPC Request │ │ ┌─────────────┐ │ │ │ Recovery │ ← Outermost: catches panics from │ │ │ │ ALL inner interceptors + handler │ │ │ ┌─────────┐│ │ │ │ │ Logging ││ ← Logs method, duration, status │ │ │ │ ││ (even for auth failures) │ │ │ │ ┌──────┐││ │ │ │ │ │Trace │││ ← Creates span, propagates trace_id │ │ │ │ │ │││ │ │ │ │ │┌────┐│││ │ │ │ │ ││Auth│││ ← Validates token, injects claims │ │ │ │ ││ │││ (short-circuits if invalid) │ │ │ │ │└────┘│││ │ │ │ │ └──────┘││ │ │ │ └─────────┘│ │ │ └─────────────┘ │ │ → Handler (your business logic) │ └──────────────────────────────────────────────────────┘
WHY this order:
- Recovery MUST be outermost: if auth panics, we still want to catch it
- Logging before auth: we want to log auth failures (for security audit)
- Tracing before auth: auth spans show up in traces for debugging
- Auth innermost: only authenticated requests reach the handler
HOW UBER DOES IT: go-grpc-middleware v2 provides the same chain pattern. We inline it to keep the dependency footprint minimal.
============================================================================
Index ¶
- func AuthStreamInterceptor(validator TokenValidator, opts ...AuthOption) grpc.StreamServerInterceptor
- func AuthUnaryInterceptor(validator TokenValidator, opts ...AuthOption) grpc.UnaryServerInterceptor
- func LoggingStreamInterceptor(logger *slog.Logger) grpc.StreamServerInterceptor
- func LoggingUnaryInterceptor(logger *slog.Logger) grpc.UnaryServerInterceptor
- func RecoveryStreamInterceptor() grpc.StreamServerInterceptor
- func RecoveryUnaryInterceptor() grpc.UnaryServerInterceptor
- type AuthOption
- type Claims
- type Server
- type ServerOption
- func WithAuthValidator(validator TokenValidator, skipMethods ...string) ServerOption
- func WithLogger(logger *slog.Logger) ServerOption
- func WithReflection() ServerOption
- func WithStreamInterceptors(interceptors ...grpc.StreamServerInterceptor) ServerOption
- func WithUnaryInterceptors(interceptors ...grpc.UnaryServerInterceptor) ServerOption
- type TokenValidator
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func AuthStreamInterceptor ¶
func AuthStreamInterceptor(validator TokenValidator, opts ...AuthOption) grpc.StreamServerInterceptor
AuthStreamInterceptor authenticates streaming RPCs, injecting Claims into the stream's context via a wrappedServerStream. Mirrors AuthUnaryInterceptor.
func AuthUnaryInterceptor ¶
func AuthUnaryInterceptor(validator TokenValidator, opts ...AuthOption) grpc.UnaryServerInterceptor
AuthUnaryInterceptor extracts bearer tokens from gRPC metadata, validates them via the TokenValidator, and injects Claims into the context.
HOW gRPC METADATA WORKS:
gRPC metadata is the equivalent of HTTP headers. Clients set metadata via grpc.WithPerRPCCredentials or metadata.AppendToOutgoingContext. Servers read it via metadata.FromIncomingContext(ctx). The "authorization" key follows HTTP convention: "authorization": "Bearer <token>"
TOKEN FLOW:
Client → sets "authorization" metadata → Server interceptor → extracts token → calls TokenValidator.Validate() → if valid: injects Claims into ctx → handler receives Claims → if invalid: returns Unauthenticated immediately (handler never called)
func LoggingStreamInterceptor ¶
func LoggingStreamInterceptor(logger *slog.Logger) grpc.StreamServerInterceptor
LoggingStreamInterceptor logs each streaming RPC with method, total duration (open → close), and final status code. Duration here is the lifetime of the whole stream, not a single message.
func LoggingUnaryInterceptor ¶
func LoggingUnaryInterceptor(logger *slog.Logger) grpc.UnaryServerInterceptor
LoggingUnaryInterceptor logs every RPC call with method, duration, and status code using structured JSON logging.
WHY: This is the gRPC equivalent of an HTTP access log. Every request/ response is logged with:
- Method: which RPC was called
- Duration: how long it took (for latency analysis)
- Code: gRPC status code (for error rate calculation)
These logs feed into Loki, where you can query:
{service="auth"} | json | grpc_code="NotFound"
PERFORMANCE: slog is allocation-efficient. The JSON handler pre-allocates buffers. Logging adds ~1-2 microseconds per RPC — negligible vs network.
func RecoveryStreamInterceptor ¶
func RecoveryStreamInterceptor() grpc.StreamServerInterceptor
RecoveryStreamInterceptor is the streaming counterpart of RecoveryUnaryInterceptor: it catches panics in streaming handlers and converts them to a gRPC Internal error instead of crashing the goroutine.
func RecoveryUnaryInterceptor ¶
func RecoveryUnaryInterceptor() grpc.UnaryServerInterceptor
RecoveryUnaryInterceptor catches panics in handlers and returns a gRPC Internal error instead of crashing the server.
WHY: Go panics are unrecoverable by default — they kill the goroutine. In a gRPC server, each RPC runs in its own goroutine. A panic crashes that goroutine, leaving the client hanging (no response, eventual timeout). Worse: if the panic happens in a shared resource (e.g., map write without mutex), it can crash the entire process.
HOW: Go's recover() function catches panics when called from a deferred function. We defer the recovery, catch any panic value, log the stack trace, and return a proper gRPC error. The server continues serving other requests.
FAILURE MODE: If the panic corrupted shared state (e.g., a sync.Map), subsequent requests may also fail. The health check should detect this (readiness fails → K8s stops routing traffic → pod restarts).
Types ¶
type AuthOption ¶
type AuthOption func(*authConfig)
AuthOption configures the auth interceptor.
func WithSkipMethods ¶
func WithSkipMethods(methods ...string) AuthOption
WithSkipMethods specifies gRPC methods that bypass authentication. Common skip targets: health checks, gRPC reflection, Login RPC.
WHY skip instead of "require" list:
Default-deny is more secure. New RPCs are automatically protected. A "require" list risks forgetting to add a new sensitive RPC. Skip list is explicit: you consciously exempt specific methods.
type Claims ¶
Claims represents the authenticated user's identity and permissions. Extracted from JWT tokens or API keys by the auth interceptor.
type Server ¶
type Server struct {
// GRPC is the underlying server. Register handlers on it:
// authpb.RegisterAuthServiceServer(srv.GRPC, handler)
GRPC *grpc.Server
// contains filtered or unexported fields
}
Server bundles the gRPC server with its health server and provides graceful shutdown. Register your service handlers on the embedded GRPC field.
WHY a wrapper instead of returning *grpc.Server directly:
Two things every service needs can't be done through a bare *grpc.Server: (1) flipping the gRPC health status to SERVING only once dependencies are ready (a bare server reports SERVING immediately, so probes pass before the service can actually serve), and (2) a single Serve(ctx) that drains in-flight RPCs on SIGTERM. The wrapper owns both.
func NewServer ¶
func NewServer(opts ...ServerOption) *Server
NewServer creates a Server with the standard interceptor chains, tracing, health service, and the given options.
The gRPC health status starts as NOT_SERVING. The service must call SetServing(true) once its dependencies (DB, NATS, etc.) are ready — this is what makes a gRPC readiness probe meaningful instead of always-green.
USAGE:
srv := grpcutil.NewServer(
grpcutil.WithLogger(logger),
grpcutil.WithAuthValidator(validator, "/forgepoint.auth.v1.AuthService/Login"),
grpcutil.WithReflection(),
)
authpb.RegisterAuthServiceServer(srv.GRPC, handler)
srv.SetServing(true) // dependencies are ready
srv.Serve(ctx, lis) // blocks; drains on ctx cancel (SIGTERM)
func (*Server) Serve ¶
Serve starts the server on lis and blocks until the server stops or ctx is cancelled. On cancellation (typically SIGTERM in K8s) it marks the server NOT_SERVING — so in-flight readiness probes fail fast and the pod is pulled from the Service endpoints — then GracefulStop drains in-flight RPCs before returning.
WHY GracefulStop over Stop: Stop hard-kills in-flight RPCs (clients see broken connections mid-call). GracefulStop stops accepting new RPCs and waits for active ones to finish — the correct behavior for a K8s rolling update, which gives a 30s termination grace period for exactly this.
func (*Server) SetServing ¶
SetServing flips the gRPC health status. Call SetServing(true) once dependencies are ready, and SetServing(false) to drain before shutdown.
type ServerOption ¶
type ServerOption func(*serverConfig)
ServerOption configures a gRPC server.
func WithAuthValidator ¶
func WithAuthValidator(validator TokenValidator, skipMethods ...string) ServerOption
WithAuthValidator enables the auth interceptors with the given validator. If not set, no auth interceptors are added (useful for tests).
WHY optional auth: Some services may have public RPCs (e.g., the auth service's Login RPC). The auth interceptor is still added but with skip methods for those public RPCs.
func WithLogger ¶
func WithLogger(logger *slog.Logger) ServerOption
WithLogger sets the logger for the logging interceptor. If not set, a default slog logger is used.
func WithReflection ¶
func WithReflection() ServerOption
WithReflection enables gRPC server reflection.
WHY: Reflection lets clients (grpcurl, grpcui, Postman) discover available services and methods without having .proto files locally. Essential for debugging in dev. In production, you may disable it to reduce attack surface (service discovery is an info leak).
grpcurl -plaintext localhost:9090 list grpcurl -plaintext localhost:9090 describe forgepoint.auth.v1.AuthService
func WithStreamInterceptors ¶
func WithStreamInterceptors(interceptors ...grpc.StreamServerInterceptor) ServerOption
WithStreamInterceptors adds custom stream interceptors that run AFTER the standard stream chain (recovery → logging → auth).
func WithUnaryInterceptors ¶
func WithUnaryInterceptors(interceptors ...grpc.UnaryServerInterceptor) ServerOption
WithUnaryInterceptors adds custom unary interceptors that run AFTER the standard chain (recovery → logging → auth). Use for service-specific interceptors like rate limiting or request validation — they run post-authentication, so they can read Claims via ClaimsFromContext.
type TokenValidator ¶
TokenValidator validates authentication tokens and returns claims. Implementations include:
- Local JWT validation (for services that have the JWT secret)
- Remote validation via auth service gRPC call
- API key validation via auth service gRPC call
Source Files