Mirroristas/etcd
2
0
Fork 0
mirror of https://github.com/etcd-io/etcd.git synced 2024-09-27 06:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(grpcproxy): support grpc context propagation in grpc proxy;fix the problem that watch exits when recvLoop exits but sendLoop still running

Browse Source 
Signed-off-by: ximenzaoshi <wanglu_bx@163.com>
This commit is contained in:
saiwl 2024-05-15 10:42:16 +08:00 committed by ximenzaoshi
parent 7b9013da46
commit 1282b04219
No known key found for this signature in database
GPG Key ID: 8FB607D82F7A671A
2 changed files with 20 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
server/etcdmain/grpc_proxy.go
View File

@ -40,6 +40,7 @@ import (
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/grpclog" "google.golang.org/grpc/grpclog"
"google.golang.org/grpc/keepalive" "google.golang.org/grpc/keepalive"
"google.golang.org/grpc/metadata"
pb "go.etcd.io/etcd/api/v3/etcdserverpb" pb "go.etcd.io/etcd/api/v3/etcdserverpb"
"go.etcd.io/etcd/client/pkg/v3/logutil" "go.etcd.io/etcd/client/pkg/v3/logutil"
@ -422,6 +423,20 @@ func mustListenCMux(lg *zap.Logger, tlsinfo *transport.TLSInfo) cmux.CMux {
return cmux.New(l) return cmux.New(l)
} }
func contextPropagationUnaryServerInterceptor() grpc.UnaryServerInterceptor {
return func(
ctx context.Context,
req interface{},
info *grpc.UnaryServerInfo,
handler grpc.UnaryHandler,
) (interface{}, error) {
if md, ok := metadata.FromIncomingContext(ctx); ok {
ctx = metadata.NewOutgoingContext(ctx, md)
}
return handler(ctx, req)
}
}
func newGRPCProxyServer(lg *zap.Logger, client *clientv3.Client) *grpc.Server { func newGRPCProxyServer(lg *zap.Logger, client *clientv3.Client) *grpc.Server {
if grpcProxyEnableOrdering { if grpcProxyEnableOrdering {
vf := ordering.NewOrderViolationSwitchEndpointClosure(client) vf := ordering.NewOrderViolationSwitchEndpointClosure(client)
@ -467,6 +482,7 @@ func newGRPCProxyServer(lg *zap.Logger, client *clientv3.Client) *grpc.Server {
} }
grpcChainUnaryList := []grpc.UnaryServerInterceptor{ grpcChainUnaryList := []grpc.UnaryServerInterceptor{
grpc_prometheus.UnaryServerInterceptor, grpc_prometheus.UnaryServerInterceptor,
contextPropagationUnaryServerInterceptor(),
} }
if grpcProxyEnableLogging { if grpcProxyEnableLogging {
grpcChainStreamList = append(grpcChainStreamList, grpcChainStreamList = append(grpcChainStreamList,

7
server/proxy/grpcproxy/watch.go
View File

@ -123,7 +123,8 @@ func (wp *watchProxy) Watch(stream pb.Watch_WatchServer) (err error) {
// post to stopc => terminate server stream; can't use a waitgroup // post to stopc => terminate server stream; can't use a waitgroup
// since all goroutines will only terminate after Watch() exits. // since all goroutines will only terminate after Watch() exits.
stopc := make(chan struct{}, 3) stopc := make(chan struct{}, 2)
leaderc := make(chan struct{}, 1)
go func() { go func() {
defer func() { stopc <- struct{}{} }() defer func() { stopc <- struct{}{} }()
wps.recvLoop() wps.recvLoop()
@ -134,7 +135,7 @@ func (wp *watchProxy) Watch(stream pb.Watch_WatchServer) (err error) {
}() }()
// tear down watch if leader goes down or entire watch proxy is terminated // tear down watch if leader goes down or entire watch proxy is terminated
go func() { go func() {
defer func() { stopc <- struct{}{} }() defer func() { leaderc <- struct{}{} }()
select { select {
case <-lostLeaderC: case <-lostLeaderC:
case <-ctx.Done(): case <-ctx.Done():
@ -142,7 +143,7 @@ func (wp *watchProxy) Watch(stream pb.Watch_WatchServer) (err error) {
} }
}() }()
<-stopc <-leaderc
cancel() cancel()
// recv/send may only shutdown after function exits; // recv/send may only shutdown after function exits;