From 81a67c41f6f2f1421830d32900326af99792822f Mon Sep 17 00:00:00 2001 From: "Edward.Yang" Date: Sat, 4 Aug 2018 16:12:01 +0800 Subject: [PATCH 1/3] git add proxy --- .DS_Store | Bin 0 -> 6148 bytes proxy/handler.go | 13 ++++++++----- 2 files changed, 8 insertions(+), 5 deletions(-) create mode 100644 .DS_Store diff --git a/.DS_Store b/.DS_Store new file mode 100644 index 0000000000000000000000000000000000000000..5b48abc28d6388de8be9c0ae45ffefa8a8f9db05 GIT binary patch literal 6148 zcmeHK%}T>S5ZE2QMD#Z?B>LoCL*)9{i~Z2{sV2B~7hT3c&~P0rcw6 zd#DfLi+FctcPmQusHn`q>^GU6nazG1cC(BzUTp<)j9H8^0~E1fLi34WKWd*8v?+WrCft_v`oKH{z(7mCMh(Fi;#E8aB+4l35xpnTKv& zwB5|hTFt7L)~qQv3PRsY_o5)H?}W)(9G-^BX1d{JVY-jOJ3_<|plr|Ijly=#8eb7f z=0!=E%6h8C!CsUEqFoiOBxuTbU0m8*URH-X#1tsd3Akrdw2ivs8)f)=PG4PV-KFdct^*nncgTL(1w`-uHTL=>>`E`exkbTsA) zp$CMkR6v!=^%aAwbnx3c&e51FROyV%k)e(rnd=)0m!pHr5V#Ntsr0V)Oj YHX4AA##|wIKh($ literal 0 HcmV?d00001 diff --git a/proxy/handler.go b/proxy/handler.go index 83920f1..f145028 100644 --- a/proxy/handler.go +++ b/proxy/handler.go @@ -4,12 +4,14 @@ package proxy import ( + "fmt" "io" "golang.org/x/net/context" "google.golang.org/grpc" "google.golang.org/grpc/codes" - "google.golang.org/grpc/transport" + // "google.golang.org/grpc/transport" + // "google.golang.org/grpc/transport" ) var ( @@ -59,11 +61,12 @@ type handler struct { // It is invoked like any gRPC server stream and uses the gRPC server framing to get and receive bytes from the wire, // forwarding it to a ClientStream established against the relevant ClientConn. func (s *handler) handler(srv interface{}, serverStream grpc.ServerStream) error { + fmt.Println(serverStream.Context()) // little bit of gRPC internals never hurt anyone - lowLevelServerStream, ok := transport.StreamFromContext(serverStream.Context()) - if !ok { - return grpc.Errorf(codes.Internal, "lowLevelServerStream not exists in context") - } + lowLevelServerStream := grpc.ServerTransportStreamFromContext(serverStream.Context()) + // if !ok { + // return grpc.Errorf(codes.Internal, "lowLevelServerStream not exists in context") + // } fullMethodName := lowLevelServerStream.Method() // We require that the director's returned context inherits from the serverStream.Context(). outgoingCtx, backendConn, err := s.director(serverStream.Context(), fullMethodName) From 3577d6ab4b9b9a66e1e2e96a79076dfa39d32337 Mon Sep 17 00:00:00 2001 From: "Edward.Yang" Date: Wed, 15 Aug 2018 11:50:14 +0800 Subject: [PATCH 2/3] =?UTF-8?q?=E5=85=B3=E9=97=AD=E5=8F=A5=E6=9F=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- proxy/README.md | 84 +++++++++++++++++++++++++++++++++++++++++++++++- proxy/handler.go | 3 +- 2 files changed, 84 insertions(+), 3 deletions(-) mode change 120000 => 100644 proxy/README.md diff --git a/proxy/README.md b/proxy/README.md deleted file mode 120000 index 71bfc07..0000000 --- a/proxy/README.md +++ /dev/null @@ -1 +0,0 @@ -DOC.md \ No newline at end of file diff --git a/proxy/README.md b/proxy/README.md new file mode 100644 index 0000000..85c411a --- /dev/null +++ b/proxy/README.md @@ -0,0 +1,83 @@ +# proxy +-- + import "github.com/mwitkow/grpc-proxy/proxy" + +Package proxy provides a reverse proxy handler for gRPC. + +The implementation allows a `grpc.Server` to pass a received ServerStream to a +ClientStream without understanding the semantics of the messages exchanged. It +basically provides a transparent reverse-proxy. + +This package is intentionally generic, exposing a `StreamDirector` function that +allows users of this package to implement whatever logic of backend-picking, +dialing and service verification to perform. + +See examples on documented functions. + +## Usage + +#### func Codec + +```go +func Codec() grpc.Codec +``` +Codec returns a proxying grpc.Codec with the default protobuf codec as parent. + +See CodecWithParent. + +#### func CodecWithParent + +```go +func CodecWithParent(fallback grpc.Codec) grpc.Codec +``` +CodecWithParent returns a proxying grpc.Codec with a user provided codec as +parent. + +This codec is *crucial* to the functioning of the proxy. It allows the proxy +server to be oblivious to the schema of the forwarded messages. It basically +treats a gRPC message frame as raw bytes. However, if the server handler, or the +client caller are not proxy-internal functions it will fall back to trying to +decode the message using a fallback codec. + +#### func RegisterService + +```go +func RegisterService(server *grpc.Server, director StreamDirector, serviceName string, methodNames ...string) +``` +RegisterService sets up a proxy handler for a particular gRPC service and +method. The behaviour is the same as if you were registering a handler method, +e.g. from a codegenerated pb.go file. + +This can *only* be used if the `server` also uses grpcproxy.CodecForServer() +ServerOption. + +#### func TransparentHandler + +```go +func TransparentHandler(director StreamDirector) grpc.StreamHandler +``` +TransparentHandler returns a handler that attempts to proxy all requests that +are not registered in the server. The indented use here is as a transparent +proxy, where the server doesn't know about the services implemented by the +backends. It should be used as a `grpc.UnknownServiceHandler`. + +This can *only* be used if the `server` also uses grpcproxy.CodecForServer() +ServerOption. + +#### type StreamDirector + +```go +type StreamDirector func(ctx context.Context, fullMethodName string) (*grpc.ClientConn, error) +``` + +StreamDirector returns a gRPC ClientConn to be used to forward the call to. + +The presence of the `Context` allows for rich filtering, e.g. based on Metadata +(headers). If no handling is meant to be done, a `codes.NotImplemented` gRPC +error should be returned. + +It is worth noting that the StreamDirector will be fired *after* all server-side +stream interceptors are invoked. So decisions around authorization, monitoring +etc. are better to be handled there. + +See the rather rich example. diff --git a/proxy/handler.go b/proxy/handler.go index f145028..dad64ce 100644 --- a/proxy/handler.go +++ b/proxy/handler.go @@ -4,7 +4,6 @@ package proxy import ( - "fmt" "io" "golang.org/x/net/context" @@ -61,7 +60,6 @@ type handler struct { // It is invoked like any gRPC server stream and uses the gRPC server framing to get and receive bytes from the wire, // forwarding it to a ClientStream established against the relevant ClientConn. func (s *handler) handler(srv interface{}, serverStream grpc.ServerStream) error { - fmt.Println(serverStream.Context()) // little bit of gRPC internals never hurt anyone lowLevelServerStream := grpc.ServerTransportStreamFromContext(serverStream.Context()) // if !ok { @@ -74,6 +72,7 @@ func (s *handler) handler(srv interface{}, serverStream grpc.ServerStream) error if err != nil { return err } + defer backendConn.Close() // TODO(mwitkow): Add a `forwarded` header to metadata, https://en.wikipedia.org/wiki/X-Forwarded-For. clientStream, err := grpc.NewClientStream(clientCtx, clientStreamDescForProxying, backendConn, fullMethodName) if err != nil { From 084bf1bc72b84a519c2972429f412043976343ba Mon Sep 17 00:00:00 2001 From: "Edward.Yang" Date: Fri, 16 Nov 2018 01:19:14 +0800 Subject: [PATCH 3/3] update --- proxy/handler.go | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/proxy/handler.go b/proxy/handler.go index dad64ce..f43b05a 100644 --- a/proxy/handler.go +++ b/proxy/handler.go @@ -18,6 +18,7 @@ var ( ServerStreams: true, ClientStreams: true, } + HandleEndCallback func(context.Context, *grpc.ClientConn) ) // RegisterService sets up a proxy handler for a particular gRPC service and method. @@ -72,7 +73,11 @@ func (s *handler) handler(srv interface{}, serverStream grpc.ServerStream) error if err != nil { return err } - defer backendConn.Close() + if HandleEndCallback == nil { + defer backendConn.Close() + } else { + HandleEndCallback(clientCtx, backendConn) + } // TODO(mwitkow): Add a `forwarded` header to metadata, https://en.wikipedia.org/wiki/X-Forwarded-For. clientStream, err := grpc.NewClientStream(clientCtx, clientStreamDescForProxying, backendConn, fullMethodName) if err != nil {