Interceptors: middleware для gRPC
Interceptors: middleware для gRPC
Interceptors - аналог HTTP middleware для gRPC. Они оборачивают каждый RPC-вызов и решают cross-cutting concerns: логирование, аутентификация, метрики, recovery от паник, rate limiting, tracing. Без них код handler-ов превращается в копипасту boilerplate.
Unary interceptor - сигнатура
type UnaryServerInterceptor func(
ctx context.Context,
req any,
info *grpc.UnaryServerInfo,
handler grpc.UnaryHandler,
) (any, error)
info.FullMethod даёт имя метода (/user.UserService/CreateUser). handler - следующий шаг в цепочке (либо следующий interceptor, либо собственно метод сервиса).
<?php
declare(strict_types=1);
namespace App\Grpc\Interceptor;
use Spiral\Interceptors\Context\CallContextInterface;
use Spiral\Interceptors\HandlerInterface;
use Spiral\Interceptors\InterceptorInterface;
interface LoggingInterceptorContract extends InterceptorInterface
{
// Сигнатура из spiral/interceptors:
// public function intercept(CallContextInterface $context, HandlerInterface $handler): mixed;
}
В PHP/RoadRunner interceptors реализуются через InterceptorInterface из spiral/roadrunner-grpc. Это middleware-цепочка, идейно такая же, как в Symfony Messenger или PSR-15.
Аналогия с Go: CallContextInterface ~ (ctx, req, info), HandlerInterface::handle() ~ вызов handler(ctx, req). Цепочка соответствует ChainUnaryInterceptor.
Logging Interceptor
В проде обычно логируют не всё: на success - debug, на error - info/warn. Чувствительные поля req не пишут (пароли, токены).
func loggingInterceptor(
ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler,
) (any, error) {
start := time.Now()
resp, err := handler(ctx, req)
slog.Info("gRPC call",
"method", info.FullMethod,
"duration", time.Since(start),
"code", status.Code(err),
)
return resp, err
}
<?php
declare(strict_types=1);
namespace App\Grpc\Interceptor;
use Psr\Log\LoggerInterface;
use Spiral\Interceptors\Context\CallContextInterface;
use Spiral\Interceptors\HandlerInterface;
use Spiral\Interceptors\InterceptorInterface;
use Throwable;
final readonly class LoggingInterceptor implements InterceptorInterface
{
public function __construct(
private LoggerInterface $logger,
) {}
public function intercept(CallContextInterface $context, HandlerInterface $handler): mixed
{
$startedAt = hrtime(true);
$method = $context->getTarget()->getPath();
$code = 'OK';
try {
return $handler->handle($context);
} catch (Throwable $e) {
$code = 'ERROR';
throw $e;
} finally {
$durationMs = (int)((hrtime(true) - $startedAt) / 1_000_000);
$this->logger->info('gRPC call', [
'method' => implode('/', $method),
'duration' => $durationMs,
'code' => $code,
]);
}
}
}
Auth Interceptor
func authInterceptor(
ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler,
) (any, error) {
// Healthcheck без авторизации
if strings.HasPrefix(info.FullMethod, "/grpc.health.v1.") {
return handler(ctx, req)
}
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
return nil, status.Error(codes.Unauthenticated, "no metadata")
}
tokens := md.Get("authorization")
if len(tokens) == 0 {
return nil, status.Error(codes.Unauthenticated, "no token")
}
userID, err := validateToken(tokens[0])
if err != nil {
return nil, status.Error(codes.Unauthenticated, "invalid token")
}
// Прокидываем user_id в context для handler-ов
ctx = context.WithValue(ctx, userIDKey, userID)
return handler(ctx, req)
}
<?php
declare(strict_types=1);
namespace App\Grpc\Interceptor;
use App\Auth\TokenValidator;
use Spiral\Interceptors\Context\CallContextInterface;
use Spiral\Interceptors\HandlerInterface;
use Spiral\Interceptors\InterceptorInterface;
use Spiral\RoadRunner\GRPC\Exception\GRPCException;
use Spiral\RoadRunner\GRPC\StatusCode;
final readonly class AuthInterceptor implements InterceptorInterface
{
public function __construct(
private TokenValidator $validator,
) {}
public function intercept(CallContextInterface $context, HandlerInterface $handler): mixed
{
$path = implode('/', $context->getTarget()->getPath());
if (str_starts_with($path, 'grpc.health.v1.')) {
return $handler->handle($context);
}
$ctx = $context->getArguments()[0]; // ContextInterface gRPC
$tokens = $ctx->getValue('authorization') ?? [];
$token = $tokens[0] ?? null;
if ($token === null) {
throw new GRPCException('no token', StatusCode::UNAUTHENTICATED);
}
try {
$userId = $this->validator->validate($token);
} catch (\Throwable) {
throw new GRPCException('invalid token', StatusCode::UNAUTHENTICATED);
}
// Кладём user_id в attributes контекста для handler-ов
return $handler->handle($context->withAttribute('user_id', $userId));
}
}
Стандартная практика: проверка токена один раз в interceptor, user_id - в context. Все handler-ы читают user_id из context, не повторяя auth-логику.
Recovery Interceptor
func recoveryInterceptor(
ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler,
) (resp any, err error) {
defer func() {
if r := recover(); r != nil {
slog.Error("panic recovered",
"method", info.FullMethod,
"recovered", r,
"stack", string(debug.Stack()),
)
err = status.Error(codes.Internal, "internal error")
}
}()
return handler(ctx, req)
}
<?php
declare(strict_types=1);
namespace App\Grpc\Interceptor;
use Psr\Log\LoggerInterface;
use Spiral\Interceptors\Context\CallContextInterface;
use Spiral\Interceptors\HandlerInterface;
use Spiral\Interceptors\InterceptorInterface;
use Spiral\RoadRunner\GRPC\Exception\GRPCException;
use Spiral\RoadRunner\GRPC\StatusCode;
use Throwable;
final readonly class RecoveryInterceptor implements InterceptorInterface
{
public function __construct(
private LoggerInterface $logger,
) {}
public function intercept(CallContextInterface $context, HandlerInterface $handler): mixed
{
try {
return $handler->handle($context);
} catch (GRPCException $e) {
throw $e; // явные gRPC-ошибки пропускаем
} catch (Throwable $e) {
$this->logger->error('panic recovered', [
'method' => implode('/', $context->getTarget()->getPath()),
'message' => $e->getMessage(),
'trace' => $e->getTraceAsString(),
]);
throw new GRPCException('internal error', StatusCode::INTERNAL);
}
}
}
В PHP recover() не нужен - есть try/catch. Превращаем любое неотловленное исключение в Internal.
Без recovery panic в одном handler-е роняет весь сервер. Это первый interceptor в цепочке - должен обернуть всё.
Rate Limiting Interceptor
func rateLimitInterceptor(limiter *rate.Limiter) grpc.UnaryServerInterceptor {
return func(
ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler,
) (any, error) {
if !limiter.Allow() {
return nil, status.Error(codes.ResourceExhausted, "rate limit exceeded")
}
return handler(ctx, req)
}
}
golang.org/x/time/rate даёт token bucket из коробки. Для per-IP лимитирования держат map[ip]*rate.Limiter с TTL-cleanup.
<?php
declare(strict_types=1);
namespace App\Grpc\Interceptor;
use Spiral\Interceptors\Context\CallContextInterface;
use Spiral\Interceptors\HandlerInterface;
use Spiral\Interceptors\InterceptorInterface;
use Spiral\RoadRunner\GRPC\Exception\GRPCException;
use Spiral\RoadRunner\GRPC\StatusCode;
use Symfony\Component\RateLimiter\LimiterInterface;
final readonly class RateLimitInterceptor implements InterceptorInterface
{
public function __construct(
private LimiterInterface $limiter,
) {}
public function intercept(CallContextInterface $context, HandlerInterface $handler): mixed
{
if (!$this->limiter->consume(1)->isAccepted()) {
throw new GRPCException('rate limit exceeded', StatusCode::RESOURCE_EXHAUSTED);
}
return $handler->handle($context);
}
}
В PHP token bucket удобно сделать на symfony/rate-limiter.
Metrics Interceptor
var (
rpcDuration = prometheus.NewHistogramVec(
prometheus.HistogramOpts{Name: "grpc_server_duration_seconds"},
[]string{"method", "code"},
)
)
func metricsInterceptor(
ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler,
) (any, error) {
start := time.Now()
resp, err := handler(ctx, req)
rpcDuration.WithLabelValues(info.FullMethod, status.Code(err).String()).
Observe(time.Since(start).Seconds())
return resp, err
}
Готовая библиотека - github.com/grpc-ecosystem/go-grpc-prometheus - даёт стандартные метрики из коробки.
<?php
declare(strict_types=1);
namespace App\Grpc\Interceptor;
use Prometheus\Histogram;
use Spiral\Interceptors\Context\CallContextInterface;
use Spiral\Interceptors\HandlerInterface;
use Spiral\Interceptors\InterceptorInterface;
use Spiral\RoadRunner\GRPC\Exception\GRPCException;
use Throwable;
final readonly class MetricsInterceptor implements InterceptorInterface
{
public function __construct(
private Histogram $rpcDuration,
) {}
public function intercept(CallContextInterface $context, HandlerInterface $handler): mixed
{
$startedAt = hrtime(true);
$method = implode('/', $context->getTarget()->getPath());
$code = 'OK';
try {
return $handler->handle($context);
} catch (GRPCException $e) {
$code = (string)$e->getCode();
throw $e;
} catch (Throwable $e) {
$code = 'INTERNAL';
throw $e;
} finally {
$durationSec = (hrtime(true) - $startedAt) / 1_000_000_000;
$this->rpcDuration->observe($durationSec, [$method, $code]);
}
}
}
PHP-эквивалент - promphp/prometheus_client_php плюс interceptor.
Цепочка interceptors
grpcServer := grpc.NewServer(
grpc.ChainUnaryInterceptor(
recoveryInterceptor, // 1. защита от panic
otelgrpc.UnaryServerInterceptor(), // 2. tracing - раньше всего, чтобы видеть в трейсе
metricsInterceptor, // 3. метрики
loggingInterceptor, // 4. логирование
authInterceptor, // 5. auth
rateLimitInterceptor(limiter), // 6. rate limit (после auth - лимитируем по user)
),
)
<?php
declare(strict_types=1);
use App\Grpc\Interceptor\AuthInterceptor;
use App\Grpc\Interceptor\LoggingInterceptor;
use App\Grpc\Interceptor\MetricsInterceptor;
use App\Grpc\Interceptor\RateLimitInterceptor;
use App\Grpc\Interceptor\RecoveryInterceptor;
use App\Grpc\Interceptor\TracingInterceptor;
use Spiral\Interceptors\PipelineBuilder;
$pipeline = (new PipelineBuilder())
->withInterceptors(
$container->get(RecoveryInterceptor::class), // 1. ловим всё
$container->get(TracingInterceptor::class), // 2. tracing - перед остальным
$container->get(MetricsInterceptor::class), // 3. метрики
$container->get(LoggingInterceptor::class), // 4. логи
$container->get(AuthInterceptor::class), // 5. auth
$container->get(RateLimitInterceptor::class), // 6. rate limit
)
->build($finalHandler);
В RoadRunner цепочка собирается из списка InterceptorInterface через PipelineBuilder.
Порядок важен: recovery обёртывает всё (включая последующие interceptors), tracing идёт рано чтобы покрыть весь обработчик, auth/rate limit ближе к handler-у.
Stream interceptors
type StreamServerInterceptor func(
srv any,
ss grpc.ServerStream,
info *grpc.StreamServerInfo,
handler grpc.StreamHandler,
) error
Сложнее unary: нужно обернуть ServerStream в proxy-объект, чтобы перехватывать Send/Recv. Например, для логирования каждого сообщения в stream-е.
type wrappedStream struct {
grpc.ServerStream
logger *slog.Logger
}
func (w *wrappedStream) RecvMsg(m any) error {
err := w.ServerStream.RecvMsg(m)
w.logger.Debug("recv", "msg", m, "err", err)
return err
}
<?php
declare(strict_types=1);
namespace App\Grpc\Interceptor;
use Psr\Log\LoggerInterface;
use Spiral\RoadRunner\GRPC\ServerStream;
final readonly class LoggingServerStream
{
public function __construct(
private ServerStream $inner,
private LoggerInterface $logger,
private string $method,
) {}
public function recv(string $class): ?object
{
$msg = $this->inner->recv($class);
$this->logger->debug('recv', ['method' => $this->method, 'class' => $class]);
return $msg;
}
public function send(object $msg): void
{
$this->logger->debug('send', ['method' => $this->method]);
$this->inner->send($msg);
}
}
В PHP/RoadRunner отдельной сигнатуры под stream нет - тот же InterceptorInterface. Для перехвата Send/Recv заворачиваем ServerStream в декоратор.
Client-side interceptors
Симметричны server-side: тоже Unary и Stream варианты, оборачивают исходящие RPC. Применяются для retry, добавления auth-токена, метрик клиента.
conn, _ := grpc.Dial(addr,
grpc.WithUnaryInterceptor(otelgrpc.UnaryClientInterceptor()),
grpc.WithUnaryInterceptor(retryInterceptor(3)),
)
<?php
declare(strict_types=1);
namespace App\Grpc\Client;
use Grpc\UnaryCall;
use User\UserServiceClient;
final class RetryingUserServiceClient extends UserServiceClient
{
public function __construct(
string $hostname,
array $opts,
private readonly int $maxAttempts = 3,
) {
parent::__construct($hostname, $opts);
}
protected function _simpleRequest(
string $method,
$argument,
$deserialize,
array $metadata = [],
array $options = []
): UnaryCall {
$attempt = 0;
while (true) {
$call = parent::_simpleRequest($method, $argument, $deserialize, $metadata, $options);
[$response, $status] = $call->wait();
$retryable = in_array(
$status->code,
[\Grpc\STATUS_UNAVAILABLE, \Grpc\STATUS_DEADLINE_EXCEEDED],
true,
);
if (!$retryable || ++$attempt >= $this->maxAttempts) {
return $call;
}
usleep(100_000 * $attempt); // линейный backoff
}
}
}
В PHP client-side interceptors реализуются переопределением BaseStub::_simpleRequest в потомке или декорированием stub-а на уровне DI. Symfony Messenger middleware - точная аналогия паттерна.
Circuit Breaker (client-side)
Retry без circuit breaker = «retry storm»: упавший downstream-сервис добивается лавиной повторов. Circuit breaker «размыкает цепь» на N секунд после порога ошибок - даёт сервису восстановиться, а текущим запросам быстрый fail вместо таймаута.
import "github.com/sony/gobreaker"
func circuitBreakerInterceptor(cb *gobreaker.CircuitBreaker) grpc.UnaryClientInterceptor {
return func(
ctx context.Context, method string, req, reply any,
cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption,
) error {
_, err := cb.Execute(func() (any, error) {
return nil, invoker(ctx, method, req, reply, cc, opts...)
})
return err
}
}
cb := gobreaker.NewCircuitBreaker(gobreaker.Settings{
Name: "user-service",
MaxRequests: 3, // полуоткрытое состояние: 3 пробных запроса
Interval: 60 * time.Second,
Timeout: 30 * time.Second, // как долго быть «открытым»
ReadyToTrip: func(c gobreaker.Counts) bool {
return c.ConsecutiveFailures > 5 // открыть после 5 подряд ошибок
},
})
<?php
declare(strict_types=1);
namespace App\Grpc\Client;
use Ackintosh\Ganesha;
use Grpc\UnaryCall;
use User\UserServiceClient;
final class CircuitBreakerUserServiceClient extends UserServiceClient
{
private const SERVICE = 'user-service';
public function __construct(
string $hostname,
array $opts,
private readonly Ganesha $ganesha,
) {
parent::__construct($hostname, $opts);
}
protected function _simpleRequest(
string $method,
$argument,
$deserialize,
array $metadata = [],
array $options = []
): UnaryCall {
if (!$this->ganesha->isAvailable(self::SERVICE)) {
throw new \RuntimeException('circuit open: user-service');
}
$call = parent::_simpleRequest($method, $argument, $deserialize, $metadata, $options);
[$response, $status] = $call->wait();
$isFailure = in_array(
$status->code,
[\Grpc\STATUS_UNAVAILABLE, \Grpc\STATUS_DEADLINE_EXCEEDED, \Grpc\STATUS_INTERNAL],
true,
);
$isFailure
? $this->ganesha->failure(self::SERVICE)
: $this->ganesha->success(self::SERVICE);
return $call;
}
}
В PHP можно взять ackintosh/ganesha - готовый circuit breaker с in-memory или Redis-storage. Декорируем gRPC-stub.
Состояния: Поведение
──────────────── ─────────────────────────────────────
closed (норма) все запросы летят к сервису
open (сработал) запросы сразу fail (ErrOpenState),
сервис не нагружается
half-open (проба) пускаем N пробных; OK - закрываем,
fail - снова open
Какие коды считать «ошибкой»: Unavailable, DeadlineExceeded, Internal - да. NotFound, InvalidArgument - нет (это не «сервис упал», это правильный ответ на плохой запрос).
Error model: status.Error, codes, error wrapping
gRPC использует отдельный механизм ошибок - не возвращайте обычные Go errors из handler-ов:
// Плохо: клиент получит code=Unknown с непонятным сообщением
func (s *server) Get(ctx context.Context, req *pb.GetReq) (*pb.GetResp, error) {
if req.Id == "" {
return nil, errors.New("id is required")
}
user, err := s.repo.Get(ctx, req.Id)
if err != nil {
return nil, err // утечёт текст SQL-ошибки!
}
return &pb.GetResp{User: user}, nil
}
// Хорошо: статусы и понятные коды
func (s *server) Get(ctx context.Context, req *pb.GetReq) (*pb.GetResp, error) {
if req.Id == "" {
return nil, status.Error(codes.InvalidArgument, "id is required")
}
user, err := s.repo.Get(ctx, req.Id)
switch {
case errors.Is(err, repo.ErrNotFound):
return nil, status.Error(codes.NotFound, "user not found")
case err != nil:
slog.Error("repo.Get failed", "err", err) // детали - в лог
return nil, status.Error(codes.Internal, "internal error") // клиенту - обобщённо
}
return &pb.GetResp{User: user}, nil
}
Что приходит клиенту Это
────────────────────────── ────────────────────────────────────────
err из вызова RPC Go error; работает errors.Is/As
status.FromError(err) *status.Status - code, message, details
st.Code() codes.NotFound, codes.Internal, ...
status.Code(err) удобный шорткат → codes.Code
st.Details() структурированные детали (BadRequest и др.)
<?php
declare(strict_types=1);
namespace App\Grpc;
use App\Domain\Exception\UserNotFoundException;
use App\Domain\UserRepository;
use Psr\Log\LoggerInterface;
use Spiral\RoadRunner\GRPC\ContextInterface;
use Spiral\RoadRunner\GRPC\Exception\GRPCException;
use Spiral\RoadRunner\GRPC\StatusCode;
use Throwable;
use User\GetReq;
use User\GetResp;
final readonly class UserGetHandler
{
public function __construct(
private UserRepository $repo,
private LoggerInterface $logger,
) {}
public function Get(ContextInterface $ctx, GetReq $req): GetResp
{
if ($req->getId() === '') {
throw new GRPCException('id is required', StatusCode::INVALID_ARGUMENT);
}
try {
$user = $this->repo->get($req->getId());
} catch (UserNotFoundException) {
throw new GRPCException('user not found', StatusCode::NOT_FOUND);
} catch (Throwable $e) {
$this->logger->error('repo.get failed', ['err' => $e->getMessage()]);
throw new GRPCException('internal error', StatusCode::INTERNAL);
}
return (new GetResp())->setUser($user);
}
}
В PHP та же логика - доменные исключения транслируем в GRPCException с явным StatusCode.
Маппинг доменных ошибок в gRPC-коды - задача handler-а или error mapping interceptor-а:
func errorMapInterceptor(
ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler,
) (any, error) {
resp, err := handler(ctx, req)
if err == nil {
return resp, nil
}
if _, ok := status.FromError(err); ok {
return nil, err // уже status - пропускаем
}
switch {
case errors.Is(err, domain.ErrNotFound):
return nil, status.Error(codes.NotFound, err.Error())
case errors.Is(err, domain.ErrValidation):
return nil, status.Error(codes.InvalidArgument, err.Error())
case errors.Is(err, context.DeadlineExceeded):
return nil, status.Error(codes.DeadlineExceeded, "request timed out")
default:
slog.Error("unhandled", "method", info.FullMethod, "err", err)
return nil, status.Error(codes.Internal, "internal error")
}
}
<?php
declare(strict_types=1);
namespace App\Grpc\Interceptor;
use App\Domain\Exception\DomainException;
use App\Domain\Exception\NotFoundException;
use App\Domain\Exception\ValidationException;
use Psr\Log\LoggerInterface;
use Spiral\Interceptors\Context\CallContextInterface;
use Spiral\Interceptors\HandlerInterface;
use Spiral\Interceptors\InterceptorInterface;
use Spiral\RoadRunner\GRPC\Exception\GRPCException;
use Spiral\RoadRunner\GRPC\StatusCode;
use Throwable;
final readonly class ErrorMapInterceptor implements InterceptorInterface
{
public function __construct(
private LoggerInterface $logger,
) {}
public function intercept(CallContextInterface $context, HandlerInterface $handler): mixed
{
try {
return $handler->handle($context);
} catch (GRPCException $e) {
throw $e; // уже gRPC - пропускаем
} catch (DomainException $e) {
throw new GRPCException($e->getMessage(), match (true) {
$e instanceof NotFoundException => StatusCode::NOT_FOUND,
$e instanceof ValidationException => StatusCode::INVALID_ARGUMENT,
default => StatusCode::INTERNAL,
});
} catch (Throwable $e) {
$this->logger->error('unhandled', [
'method' => implode('/', $context->getTarget()->getPath()),
'err' => $e->getMessage(),
]);
throw new GRPCException('internal error', StatusCode::INTERNAL);
}
}
}
PHP-эквивалент - отдельный interceptor, который ловит доменные исключения и маппит их в gRPC-коды через match.
Так handler-ы возвращают доменные ошибки (domain.ErrNotFound), а перевод в gRPC-семантику централизован.
Готовые библиотеки
github.com/grpc-ecosystem/go-grpc-middleware - коллекция готовых interceptors: auth (jwt, oauth), logging (zap, logrus, zerolog), validator (proto-gen-validate), recovery, ratelimit, retry. В большинстве случаев свой interceptor писать не нужно - берите готовый.
Мини-практика
Напиши цепочку interceptors для production-ready gRPC-сервера: recovery от паники с логом стека, OpenTelemetry tracing, Prometheus метрики (RPS, latency histogram, error rate по коду), JWT-auth с прокидыванием user_id в context, rate limiting per-user (100 RPC/мин). Покрой тестами panic-recovery и unauthorized-кейсы.