README
¶
grpcclient: gRPC client middleware
gRPC client middleware (interceptors, not generated stubs): retry with backoff on retryable status codes, optional circuit-breaker integration, latency observation, and a metrics surface. Compose the interceptors onto your own dial.
Configuration is a ClientOptions struct (json + mapstructure tags).
Usage
NewMiddleware(opts ClientOptions) *Middleware.(*Middleware).UnaryClientInterceptor() grpc.UnaryClientInterceptor.(*Middleware).StreamClientInterceptor() grpc.StreamClientInterceptor.DialConn(opts ClientOptions) (*grpc.ClientConn, error)ready-to-use conn with the interceptors attached.
ClientOptions
Targetthe dial target (required forDialConn).ConnectTimeout,RequestTimeout.RetryMax,RetryCodes []codes.Code(which codes are retryable),RetryWaitMin,RetryWaitMaxexponential backoff.Breaker CircuitBreaker— non-nil wraps calls; nil disables.Latency LatencyObserver— non-nil receives durations; nil disables.
The Client wrapper exposes SetOnEvent(func(ClientEvent)) and
Metrics() ClientMetrics for observation.
Example
One-shot dial (interceptors built internally from the same options):
import "github.com/v8fg/kit4go/grpcclient"
conn, err := grpcclient.DialConn(grpcclient.ClientOptions{
Target: target,
RetryMax: 3,
RetryCodes: []codes.Code{codes.Unavailable},
RetryWaitMin: 100 * time.Millisecond,
RetryWaitMax: 2 * time.Second,
})
Or attach the interceptors to your own dial:
mw := grpcclient.NewMiddleware(opts)
grpc.Dial(opts.Target,
grpc.WithUnaryInterceptor(mw.UnaryClientInterceptor()),
grpc.WithStreamInterceptor(mw.StreamClientInterceptor()),
)
Documentation
¶
Overview ¶
Package grpcclient provides gRPC interceptor middleware (not stubs) that adds timeout, retry, circuit-breaker, and metrics to any gRPC client.
Quick start ¶
mw := grpcclient.NewMiddleware(grpcclient.ClientOptions{
Target: "bid-engine:50051",
RequestTimeout: 20 * time.Millisecond, // RTB hard budget
RetryMax: 2,
RetryCodes: []codes.Code{codes.Unavailable, codes.DeadlineExceeded},
Breaker: myBreaker, // optional
})
conn, err := grpc.Dial(opts.Target,
grpc.WithTransportCredentials(insecure.NewCredentials()),
grpc.WithUnaryInterceptor(mw.UnaryClientInterceptor()),
grpc.WithStreamInterceptor(mw.StreamClientInterceptor()),
)
// Use conn with any generated stub...
client := pb.NewBidServiceClient(conn)
Performance ¶
BenchmarkUnary 33 us 164 allocs (bufconn, no real network) BenchmarkUnary_Parallel 12 us 149 allocs (RunParallel, amortized) BenchmarkMiddleware_Metrics 2.4 ns 0 allocs
The middleware adds negligible overhead — allocs are dominated by gRPC serialization and the protobuf wire format.
Retry (unary only) ¶
Retries gRPC codes in RetryCodes (default: Unavailable) with exponential backoff + jitter. DeadlineExceeded is excluded by default (a server may have committed before its deadline — retrying risks a duplicate side effect); opt in only for idempotent RPCs. Never retries on context cancellation or non-retryable codes (NotFound, PermissionDenied, etc.). Streams are NOT retried (semantically unsafe — caller must reconnect).
RequestTimeout bounds the ENTIRE retry sequence, not each attempt: a slow first attempt can leave little budget for retries (and a backoff sleep consumes part of it). Budget per-attempt by shortening RequestTimeout, or retry manually around a single attempt.
Stream RPCs: the per-RPC context (carrying RequestTimeout) is freed only when the stream is drained to EOF/error via RecvMsg. Abandoning a stream (dropping it without draining) holds the context until RequestTimeout elapses — drain streams to release resources promptly.
Monitoring ¶
m := mw.Metrics()
// m.Total, m.Success, m.Failed, m.Retried
// m.Active — in-flight RPCs (real-time atomic)
mw.SetOnEvent(func(evt grpcclient.ClientEvent) {
// evt.Name: "request"|"retry"|"success"|"failed"
// evt.Method, evt.Code, evt.Attempt
})
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func DialConn ¶
func DialConn(opts ClientOptions) (*grpc.ClientConn, error)
DialConn is a convenience that dials opts.Target with the middleware's interceptors already wired in, using insecure transport credentials and non-blocking dial with the configured ConnectTimeout. It returns a *grpc.ClientConn ready for stub construction; the caller must Close it.
This helper uses insecure credentials (suited to in-cluster mTLS-stripped or local dev traffic). For TLS or custom dial options, construct the middleware with NewMiddleware and dial manually.
Example ¶
ExampleDialConn uses the convenience constructor that dials with insecure credentials and the middleware's interceptors already attached. With an empty Target it surfaces a clear error rather than panicking.
package main
import (
"fmt"
grpcclient "github.com/v8fg/kit4go/grpcclient"
)
func main() {
// A realistic call would pass a real Target and then use the returned conn
// with generated stubs:
//
// conn, err := grpcclient.DialConn(grpcclient.ClientOptions{
// Target: "localhost:50051",
// })
// if err != nil { return err }
// defer conn.Close()
// cli := pb.NewEchoerClient(conn)
// resp, err := cli.Echo(ctx, &pb.StringValue{Value: "hi"})
//
// Here we demonstrate the empty-Target guard, which is cheap to exercise
// without a live server.
_, err := grpcclient.DialConn(grpcclient.ClientOptions{Target: ""})
fmt.Println("error:", err)
}
Output: error: grpcclient: empty Target in ClientOptions
Types ¶
type BreakerAdapter ¶ added in v0.7.0
type BreakerAdapter struct {
// ExecuteCtx is the underlying breaker's Execute function.
ExecuteCtx func(ctx context.Context, fn func(ctx context.Context) error) error
}
BreakerAdapter wraps a *breaker.Breaker[T] (which returns (T, error)) into the CircuitBreaker interface (which returns only error). The T value is discarded — grpcclient's invoker returns error only.
Usage:
b := breaker.New[any](breaker.WithThreshold(5))
mw := grpcclient.NewMiddleware(grpcclient.ClientOptions{
Breaker: grpcclient.BreakerAdapter{b},
})
type CircuitBreaker ¶
type CircuitBreaker interface {
// Execute runs fn under the breaker's protection. Implementations should
// short-circuit with their own ErrCircuitOpen when the breaker is open
// rather than invoking fn, and record the outcome of fn for their sliding
// window when it is invoked. fn must honour ctx.
Execute(ctx context.Context, fn func(ctx context.Context) error) error
}
CircuitBreaker is the interface used by Middleware to optionally wrap each call in a circuit breaker. grpcclient does NOT import breaker (that would create a hard dependency for every caller). Users adapt a *breaker.Breaker[T] to this interface — the breaker package's Execute returns (T, error), so a thin adapter is needed (e.g. breakerAdapter below). A nil breaker on ClientOptions disables the integration and calls are issued directly.
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client holds the shared counters and event hook for a Middleware. It is a distinct type from Middleware so the metrics/hook surface is named clearly and can be reused independently of the interceptor wiring. All fields are accessed atomically; the zero value is a ready-to-use metrics sink.
func (*Client) Metrics ¶
func (c *Client) Metrics() ClientMetrics
Metrics returns a point-in-time snapshot of the middleware's counters.
func (*Client) SetOnEvent ¶
func (c *Client) SetOnEvent(fn func(evt ClientEvent))
SetOnEvent installs a hook invoked for every notable RPC lifecycle event. fn receives a ClientEvent describing the attempt and its outcome. Pass nil to disable a previously-installed hook.
The hook is intended for metrics/tracing and must be cheap and non-blocking: it fires synchronously on the goroutine issuing the RPC. Install it once at construction time (before traffic) for the cleanest ordering; SetOnEvent is nevertheless safe to call concurrently with in-flight RPCs.
type ClientEvent ¶
ClientEvent is passed to the hook installed via Middleware.SetOnEvent for every notable outcome of an RPC lifecycle. It is the integration point for metrics push (Prometheus counters/histograms, tracing spans, log4go alerts), mirroring the hook pattern used by the httpclient and breaker packages.
Name is one of:
- "request": an attempt was sent (one per send, including retries).
- "retry": an attempt failed with a retryable code and will be retried (fires before backoff).
- "success": the call completed with a nil error (status OK).
- "failed": the call completed with a non-OK status, or could not be sent (breaker open, context cancelled, etc.).
Method is the full gRPC method name (e.g. "/pkg.Service/Method"). Code is the gRPC status code name of the relevant attempt (e.g. "Unavailable"); empty when no status was obtained. Attempt is the 0-indexed attempt number this event pertains to (0 = the initial send).
type ClientMetrics ¶
type ClientMetrics struct {
// Total is the number of RPCs observed, regardless of outcome. Incremented
// once per logical call (retries do not inflate this counter).
Total uint64
// Success is the number of calls whose final attempt returned a nil error
// (gRPC status OK).
Success uint64
// Failed is the number of calls that did not end in OK: this covers both
// non-OK final statuses and any error that prevented the RPC from running.
Failed uint64
// Retried is the total number of retry attempts made (not counting the
// initial attempt). A call that required two retries contributes 2 here.
// Only unary RPCs are retried, so stream calls never inflate this counter.
Retried uint64
// Active is the number of in-flight RPCs at snapshot time — calls that have
// entered the unary (or stream) interceptor but not yet returned. Read via
// an atomic load, so zero-contention and safe to scrape on the hot path.
Active int32
}
ClientMetrics is a point-in-time snapshot of the counters maintained by a Middleware. Values are gathered via atomic loads and may be slightly inconsistent with one another under concurrent load; that is acceptable for monitoring/observability use.
type ClientOptions ¶
type ClientOptions struct {
// Target is the gRPC server address dialled by [DialConn], e.g.
// "localhost:50051" or "dns:///srv.example.com:443". It is not consumed by
// the interceptors themselves (they only see per-call method names); it is
// stored on the options purely so DialConn has everything it needs.
Target string `json:"target" mapstructure:"target"`
// ConnectTimeout bounds the initial grpc.Dial connection establishment.
// Applied by [DialConn] via grpc.WithConnectParams. Default 5s. A value <= 0
// keeps the package default.
ConnectTimeout time.Duration `json:"connect_timeout" mapstructure:"connect_timeout"`
// RequestTimeout is the per-RPC timeout applied via context.WithTimeout on
// every unary and stream call. Default 10s. A caller-supplied context
// deadline tighter than RequestTimeout always wins.
RequestTimeout time.Duration `json:"request_timeout" mapstructure:"request_timeout"`
// RetryMax is the maximum number of retry attempts after the first call,
// i.e. the total number of sends for a unary RPC is RetryMax+1. Default 2.
// Retries only apply to unary RPCs whose status code is in RetryCodes.
RetryMax int `json:"retry_max" mapstructure:"retry_max"`
// RetryCodes is the set of gRPC status codes that trigger a unary-RPC
// retry. Default [codes.Unavailable] — Unavailable means the RPC was NOT
// processed, so retrying is replay-safe. DeadlineExceeded is excluded by
// default because a server may have already committed the work before its
// deadline elapsed (retrying would duplicate a side effect); opt in to it
// only for idempotent RPCs. A nil/empty slice is replaced with the default
// by withDefaults. Pass a single-element slice containing codes.OK to
// effectively disable retry (no real RPC ever returns OK as an error).
RetryCodes []codes.Code `json:"-"`
// RetryWaitMin is the lower bound of the exponential backoff applied
// between unary retries (before jitter). Default 100ms.
RetryWaitMin time.Duration `json:"retry_wait_min" mapstructure:"retry_wait_min"`
// RetryWaitMax is the upper bound of the backoff, and the cap above which
// the exponential growth stops. Default 1s.
RetryWaitMax time.Duration `json:"retry_wait_max" mapstructure:"retry_wait_max"`
// Breaker, when non-nil, wraps every unary and stream call via
// Breaker.Execute. nil (the default) disables circuit-breaker integration.
Breaker CircuitBreaker `json:"-"`
// Latency, when non-nil, receives the end-to-end duration of every RPC
// (unary calls include retries; streams measure open time). nil (the
// default) disables latency observation.
Latency LatencyObserver `json:"-"`
}
ClientOptions configures a Middleware. Zero values are replaced with sensible defaults by withDefaults at construction time, so the zero ClientOptions is usable (it yields a middleware with all defaults). Breaker and RetryCodes are the only fields that opt into extra behaviour; everything else is a tunable.
Field tags carry both json and mapstructure names so the struct can be loaded from either a JSON config or a Viper-style mapstructure source. Breaker and RetryCodes are tagged "-" because a live breaker object and a codes.Code slice cannot be cleanly (de)serialised.
type LatencyObserver ¶
type LatencyObserver interface {
// Observe records a single latency sample. Must be safe for concurrent use.
Observe(time.Duration)
}
LatencyObserver receives the end-to-end duration of an RPC. grpcclient does not import the latency package; pass a *latency.Histogram (which satisfies this interface) or any other implementation. A nil Latency on ClientOptions disables observation — the call site is a single nil check, with no time.Now and no defer, so the disabled path is free.
type Middleware ¶
type Middleware struct {
// contains filtered or unexported fields
}
Middleware bundles a Client (metrics + event hook) with the frozen options used to build its interceptors. Construct one with NewMiddleware and pass Middleware.UnaryClientInterceptor / Middleware.StreamClientInterceptor to grpc.Dial. The interceptors share the same underlying Client, so metrics reflect every call regardless of which interceptor handled it.
func NewMiddleware ¶
func NewMiddleware(opts ClientOptions) *Middleware
NewMiddleware constructs a Middleware from opts, filling zero fields with the package defaults. The returned middleware owns its metrics counters and a fresh Client; pass its interceptors to grpc.Dial:
mw := grpcclient.NewMiddleware(opts)
conn, err := grpc.Dial(opts.Target,
grpc.WithTransportCredentials(insecure.NewCredentials()),
grpc.WithUnaryInterceptor(mw.UnaryClientInterceptor()),
grpc.WithStreamInterceptor(mw.StreamClientInterceptor()),
)
Example ¶
ExampleNewMiddleware shows the basic construction of a middleware. A zero ClientOptions yields sensible production defaults (5s connect, 10s request timeout, up to 2 retries on Unavailable/DeadlineExceeded). Override only the fields you need, then pass the interceptors to grpc.Dial.
package main
import (
"fmt"
"time"
grpcclient "github.com/v8fg/kit4go/grpcclient"
)
func main() {
mw := grpcclient.NewMiddleware(grpcclient.ClientOptions{
Target: "localhost:50051",
RequestTimeout: 5 * time.Second,
RetryMax: 2,
RetryWaitMin: 50 * time.Millisecond,
RetryWaitMax: time.Second,
})
_ = mw
fmt.Println("middleware ready")
}
Output: middleware ready
Example (Dial) ¶
ExampleNewMiddleware_dial wires the middleware's interceptors onto a real (here: would-be) *grpc.ClientConn. In production you would then build generated stubs from conn and call them as usual; the interceptors apply retry, timeout, metrics and breaker transparently.
package main
import (
"fmt"
"time"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
grpcclient "github.com/v8fg/kit4go/grpcclient"
)
func main() {
mw := grpcclient.NewMiddleware(grpcclient.ClientOptions{
Target: "localhost:50051",
RequestTimeout: 5 * time.Second,
})
// We don't actually dial a live server in this example, so we construct the
// dial options to show the wiring. grpc.Dial is non-blocking by default, so
// this returns a conn whose connection comes up lazily on first RPC.
opts := []grpc.DialOption{
grpc.WithTransportCredentials(insecure.NewCredentials()),
grpc.WithUnaryInterceptor(mw.UnaryClientInterceptor()),
grpc.WithStreamInterceptor(mw.StreamClientInterceptor()),
}
_ = opts
fmt.Println("dial options wired")
}
Output: dial options wired
func (*Middleware) Metrics ¶
func (m *Middleware) Metrics() ClientMetrics
Metrics returns a point-in-time snapshot of the middleware's counters. See Client.Metrics.
Example ¶
ExampleMiddleware_Metrics shows how to read the middleware's atomic counters for monitoring. Because we have no live server here, we drive a single invocation through the interceptors via a minimal in-process round to show the shape of the returned snapshot. See grpcclient_test.go for full coverage with a bufconn server.
srv := newEchoServer()
dialer, shutdown := startTestServer(srv)
defer shutdown()
mw := grpcclient.NewMiddleware(grpcclient.ClientOptions{RetryMax: 0})
conn := dialBufconn(fakeT{}, dialer, mw)
defer conn.Close()
for range 3 {
if _, err := echoUnary(context.Background(), conn, wrapperspb.String("ping")); err != nil {
fmt.Println("error:", err)
return
}
}
m := mw.Metrics()
fmt.Printf("total=%d success=%d failed=%d retried=%d\n", m.Total, m.Success, m.Failed, m.Retried)
Output: total=3 success=3 failed=0 retried=0
func (*Middleware) SetOnEvent ¶
func (m *Middleware) SetOnEvent(fn func(evt ClientEvent))
SetOnEvent installs a hook invoked for every notable RPC lifecycle event on the middleware's underlying Client. See Client.SetOnEvent.
func (*Middleware) StreamClientInterceptor ¶
func (m *Middleware) StreamClientInterceptor() grpc.StreamClientInterceptor
StreamClientInterceptor returns a grpc.StreamClientInterceptor that applies the per-RPC timeout and (if configured) circuit-breaker integration. It does NOT retry: retrying a stream is semantically unsafe (the server may have already started emitting messages), so a caller whose stream errored must reconnect explicitly.
func (*Middleware) UnaryClientInterceptor ¶
func (m *Middleware) UnaryClientInterceptor() grpc.UnaryClientInterceptor
UnaryClientInterceptor returns a grpc.UnaryClientInterceptor that applies, in order:
- Per-RPC timeout: RequestTimeout is applied via context.WithTimeout when the caller's ctx has no deadline.
- Retry: if the RPC returns a status code in RetryCodes and the attempt is below RetryMax, sleep with exponential backoff + jitter and retry.
- Circuit breaker: if Breaker is configured, the whole retry loop is run inside Breaker.Execute so an open breaker short-circuits the call.
- Metrics + event hooks: total/success/failed/retried counters and the SetOnEvent callback fire at the relevant points.
Retries are NOT attempted when the caller's context is cancelled or hits its deadline: a DeadlineExceeded caused by the per-RPC timeout is retried only if the underlying transport returned it (via a gRPC Unavailable/DeadlineExceeded status), not when ctx.Err() fires.
Source Files