Kratos middleware seems to only apply to unary gRPC operations

[wscalf]

What happened:

We've started using Kratos middleware (recovery, logging, validate) in our project, and noticed that, on the gRPC side, they only seem to apply to unary operations. Our streaming operations don't get validated or logged, and a panic takes down the server (versus on a unary endpoint, they're validated and logged as expected, and panics are handled.)

What you expected to happen:

We expected to either have the middleware apply seamlessly to streaming requests or to have separate streaming middleware we can register (which, to be fair, may exist, but we weren't able to find it.)

How to reproduce it (as minimally and precisely as possible):

Repro project: https://github.com/wscalf/streaming-repro

• "SayHello" is the normal hello world operation + a 50/50 chance of panicking to test the recovery middleware.
• "KeepSayingHello" is an added operation that streams the response every second until the client terminates the connection, at which point it panics (intentionally, to test
  Commands:
• grpcurl -plaintext -d '{"name":""}' localhost:9000 helloworld.v1.Greeter.SayHello: gets correctly rejected with a validation error
• grpcurl -plaintext -d '{"name":""}' localhost:9000 helloworld.v1.Greeter.KeepSayingHello: should get rejected (it's using the same message) but does not. Also, is coded to panic when the client closes, which should trigger the recovery middleware, but it doesn't- it just exits
• grpcurl -plaintext -d '{"name":"Kratos"}' localhost:9000 helloworld.v1.Greeter.SayHello: is valid, and is coded to have a 50/50 chance of panicking, which does correctly trigger the recovery middleware, returning an error to the client and not exiting.

Anything else we need to know?:

Environment:

• Kratos version (use kratos -v): kratos version v2.7.3
• Go version (use go version): go version go1.22.2 linux/amd64
• OS (e.g: cat /etc/os-release):

NAME="Fedora Linux"
VERSION="40 (Workstation Edition)"
ID=fedora
VERSION_ID=40
VERSION_CODENAME=""
PLATFORM_ID="platform:f40"
PRETTY_NAME="Fedora Linux 40 (Workstation Edition)"
ANSI_COLOR="0;38;2;60;110;180"
LOGO=fedora-logo-icon
CPE_NAME="cpe:/o:fedoraproject:fedora:40"
DEFAULT_HOSTNAME="fedora"
HOME_URL="https://fedoraproject.org/"
DOCUMENTATION_URL="https://docs.fedoraproject.org/en-US/fedora/f40/system-administrators-guide/"
SUPPORT_URL="https://ask.fedoraproject.org/"
BUG_REPORT_URL="https://bugzilla.redhat.com/"
REDHAT_BUGZILLA_PRODUCT="Fedora"
REDHAT_BUGZILLA_PRODUCT_VERSION=40
REDHAT_SUPPORT_PRODUCT="Fedora"
REDHAT_SUPPORT_PRODUCT_VERSION=40
SUPPORT_END=2025-05-13
VARIANT="Workstation Edition"
VARIANT_ID=workstation

• Others:

[shenqidebaozi]

Yes, adaptation to stream middleware is rudimentary because the current interface model does not have a good abstraction

[wscalf]

Would there be interest in proposals around this? We haven't looked super closely at it, but it seems like there would be a few options:

• Could approach it the way gRPC does with separate interfaces and implementations for unary and streaming endpoints. This is what we've done for now- reimplementing the middleware we need as gRPC stream interceptors and injecting them: https://github.com/project-kessel/relations-api/pull/105/files#diff-85d6db9d4d33f7064a02bbe45907ffdaca1cd1382bbfdb5ab8c4a7e77841cea5R25
• Or, if there's a desire to unify unary and streaming middleware under one interface, that should be possible if the interface is streaming-first, since a unary request could be thought of as a streaming request with one incoming and one outgoing messages. I could work up something more concrete around this if there's interest.

[shenqidebaozi]

@wscalf OK

[shenqidebaozi]

But we need to think about how to do it better

[wscalf]

FWIW, I did some poking around at what it would take to have a unified middleware interface for unary and streaming endpoints. It does look possible, and there's some precedent for doing so: go-grpc-middleware defines a Reportable interface and interceptor wrappers that allows for the same implementations to be used with both kinds of endpoints. It's only used for logging, though it seems like something similar could apply generally.

I think it would be a breaking change to the middleware interface to be able to observe and react to multiple incoming or outgoing messages, but existing middleware could still be called from the UnaryServerInterceptor and accessed by the HTTP transport as they are today, alongside the unified middleware, so a piece-by-piece migration should be possible without breaking current functionality or end-user authored middleware.

Maybe something like: type Handler func(ctx context.Context, receiveMsg, sendMsg func(msg any) error, info CallInfo) error, which would allow implementations to keep more or less the natural flow they have now while being compatible with streaming.

This would allow for implementations like:

func Validator() middleware.Middleware {
	return func(handler middleware.Handler) middleware.Handler {
		return func(ctx context.Context, receiveMsg, sendMsg func(msg any) error, info CallInfo) error {
			recv := func (req any) error {
				if v, ok := req.(validator); ok {
					if err := v.Validate(); err != nil {
						return errors.BadRequest("VALIDATOR", err.Error()).WithCause(err)
					}
				}
				return nil
			}
			
			return handler(ctx, recv, func(any) error {return nil}, info)
		}
	}
}

Or if a middleware wanted the flexibility to treat streaming endpoints differently (ex: if logging middleware didn't want to log streamed inputs), that would still be possible by observing properties on CallInfo (which atm is intended to indicate whether the requests and/or responses are streaming, which would be a passthrough of grpc.StreamServerInfo for streaming gRPC requests and false/false for unary and HTTP requests.)

Thoughts/feedback? I've been mostly exploring so far but could put a draft PR together to provide something more concrete if there's interest, or pivot if this seems totally off the mark.

[akoserwal]

@go-kratos/contributor
@shenqidebaozi

[shenqidebaozi]

I've been quite busy lately, let me take a look now