From 7b29398e23d6e086a6ba904f15789105c52a02e7 Mon Sep 17 00:00:00 2001 From: Cat Cai Date: Wed, 7 Nov 2018 09:43:29 -0800 Subject: [PATCH 01/14] WIP --- pkg/proxy/handler.go | 17 +++++++++++++++++ pkg/proxy/proxy.go | 44 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 61 insertions(+) create mode 100644 pkg/proxy/handler.go create mode 100644 pkg/proxy/proxy.go diff --git a/pkg/proxy/handler.go b/pkg/proxy/handler.go new file mode 100644 index 00000000..89e37fb2 --- /dev/null +++ b/pkg/proxy/handler.go @@ -0,0 +1,17 @@ +package proxy + +import ( + "net/http" + + "github.com/gorilla/mux" +) + +// RPCRouteHandler takes in a route resembling the following +// :// +// localhost:50051/fairapis.pubsub.v1.Publish/ListTopics +// It pulls the service and the RPC name off of the route, generates +// the proto message, and then routes it to the proper destination. +func RPCRouteHandler(rw http.ResponseWriter, req *http.Request) { + // Pull the variables off of the request + vars := mux.Vars(req) +} diff --git a/pkg/proxy/proxy.go b/pkg/proxy/proxy.go new file mode 100644 index 00000000..43c0d594 --- /dev/null +++ b/pkg/proxy/proxy.go @@ -0,0 +1,44 @@ +package proxy + +import ( + "net/http" + + "github.com/gorilla/mux" +) + +// Config wraps all configs for the proxy +type Config struct { + Addr string + RouteHandler func(http.ResponseWriter, *http.Request) +} + +// Proxy encapsulates all gURL proxy server logic +type Proxy struct { + router *mux.Router + server *http.Server + handler func(http.ResponseWriter, *http.Request) +} + +// NewProxy returns an instance of Proxy +func NewProxy(cfg *Config) *Proxy { + r := mux.NewRouter() + s := &http.Server{ + Addr: cfg.Addr, + Handler: r, + } + return &Proxy{ + router: r, + server: s, + } +} + +// Configures routes +func (p *Proxy) configure() error { + p.router.HandleFunc("/{service}/{method}", p.handler) + return nil +} + +// Run the proxy server +func (p *Proxy) Run() error { + return p.server.ListenAndServe() +} From 71c5612bdc804d355ce2fdede4e7d93a78db83c7 Mon Sep 17 00:00:00 2001 From: Cat Cai Date: Wed, 22 May 2019 20:37:40 -0700 Subject: [PATCH 02/14] Proxy cmd --- Gopkg.lock | 18 ++++++++++++++++++ cmd/call/call.go | 6 ++++-- cmd/proxy/proxy.go | 33 +++++++++++++++++++++++++++++++++ cmd/root.go | 2 ++ pkg/proxy/handler.go | 17 ++++++++++++++++- pkg/proxy/header.go | 15 +++++++++++++++ pkg/proxy/proxy.go | 5 +++-- 7 files changed, 91 insertions(+), 5 deletions(-) create mode 100644 cmd/proxy/proxy.go create mode 100644 pkg/proxy/header.go diff --git a/Gopkg.lock b/Gopkg.lock index dec93774..0bf3c294 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -1,6 +1,14 @@ # This file is autogenerated, do not edit; changes may be undone by the next 'dep ensure'. +[[projects]] + digest = "1:ffe9824d294da03b391f44e1ae8281281b4afc1bdaa9588c9097785e3af10cec" + name = "github.com/davecgh/go-spew" + packages = ["spew"] + pruneopts = "UT" + revision = "8991bc29aa16c548c550c7ff78260e27b9ab7c73" + version = "v1.1.1" + [[projects]] branch = "master" digest = "1:dbb3d1675f5beeb37de6e9b95cc460158ff212902a916e67688b01e0660f41bd" @@ -87,6 +95,14 @@ revision = "ee43cbb60db7bd22502942cccbc39059117352ab" version = "v0.1.0" +[[projects]] + digest = "1:3af6be4fee7c08f81f13d36f04ffb63ad4b6b5aaba12cce96095c7c2863d4912" + name = "github.com/gorilla/mux" + packages = ["."] + pruneopts = "UT" + revision = "ed099d42384823742bba0bf9a72b53b55c9e2e38" + version = "v1.7.2" + [[projects]] branch = "master" digest = "1:cfca58687d969bf8184ddbe5058cbabc2ecf26b371ef56f9e57791ab151b16fb" @@ -471,8 +487,10 @@ analyzer-name = "dep" analyzer-version = 1 input-imports = [ + "github.com/davecgh/go-spew/spew", "github.com/golang/glog", "github.com/golang/protobuf/proto", + "github.com/gorilla/mux", "github.com/grpc-ecosystem/grpc-gateway/runtime", "github.com/jhump/protoreflect/desc", "github.com/jhump/protoreflect/desc/protoparse", diff --git a/cmd/call/call.go b/cmd/call/call.go index 6036d05f..1d90ae3e 100644 --- a/cmd/call/call.go +++ b/cmd/call/call.go @@ -38,10 +38,12 @@ var CallCmd = &cobra.Command{ } func init() { - flags := CallCmd.Flags() // Add any flags that were registered on the built-in flag package. - flags.AddGoFlagSet(flag.CommandLine) + // This is specifically for configuring glog. We want this to be a persistent + // flag since we want to be able to handle log configuration for subcommands as well. + CallCmd.PersistentFlags().AddGoFlagSet(flag.CommandLine) + flags := CallCmd.Flags() flags.StringVarP(&uri, "uri", "u", "", "gRPC URI in the form of host:port/service_name/method_name") flags.StringVarP(&data, "data", "d", "", "Data, as JSON string, to send to the gRPC service") CallCmd.MarkFlagRequired("uri") diff --git a/cmd/proxy/proxy.go b/cmd/proxy/proxy.go new file mode 100644 index 00000000..032f32dc --- /dev/null +++ b/cmd/proxy/proxy.go @@ -0,0 +1,33 @@ +package proxy + +import ( + "fmt" + + "github.com/spf13/cobra" + "github.com/wearefair/gurl/pkg/log" + "github.com/wearefair/gurl/pkg/proxy" +) + +var ( + port int +) + +var ProxyCmd = &cobra.Command{ + Use: "proxy", + Short: "Run the gURL proxy", + RunE: runProxy, +} + +func init() { + ProxyCmd.Flags().IntVarP(&port, "port", "p", 3030, "Port for the proxy to run on") +} + +func runProxy(cmd *cobra.Command, args []string) error { + cfg := &proxy.Config{ + Addr: fmt.Sprintf(":%d", port), + } + + proxySrv := proxy.New(cfg) + log.Infof("Starting server at %d\n", port) + return proxySrv.Run() +} diff --git a/cmd/root.go b/cmd/root.go index 27bcf4f1..3cc7c858 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -6,6 +6,7 @@ import ( "github.com/wearefair/gurl/cmd/call" configcmd "github.com/wearefair/gurl/cmd/config" "github.com/wearefair/gurl/cmd/list" + proxycmd "github.com/wearefair/gurl/cmd/proxy" "github.com/wearefair/gurl/pkg/config" ) @@ -19,6 +20,7 @@ func init() { cobra.OnInitialize(initConfig) call.CallCmd.AddCommand(list.ListServicesCmd) call.CallCmd.AddCommand(configcmd.ConfigCmd) + call.CallCmd.AddCommand(proxycmd.ProxyCmd) } func initConfig() { diff --git a/pkg/proxy/handler.go b/pkg/proxy/handler.go index 89e37fb2..24cf5fe7 100644 --- a/pkg/proxy/handler.go +++ b/pkg/proxy/handler.go @@ -3,15 +3,30 @@ package proxy import ( "net/http" + "github.com/davecgh/go-spew/spew" "github.com/gorilla/mux" ) // RPCRouteHandler takes in a route resembling the following -// :// +// // // localhost:50051/fairapis.pubsub.v1.Publish/ListTopics // It pulls the service and the RPC name off of the route, generates // the proto message, and then routes it to the proper destination. +// The destination must be set in the headers under the key +// 'x-proxy-host'. func RPCRouteHandler(rw http.ResponseWriter, req *http.Request) { + // Pull the host off the headers + headers := req.Header.Get(ProxyTargetHeader) + + // If the header is not set... return a 422, because we really + // can't process this request. + if headers == "" { + rw.WriteHeader(http.StatusUnprocessableEntity) + return + } + // Pull the variables off of the request vars := mux.Vars(req) + + spew.Dump(vars) } diff --git a/pkg/proxy/header.go b/pkg/proxy/header.go new file mode 100644 index 00000000..eb29f710 --- /dev/null +++ b/pkg/proxy/header.go @@ -0,0 +1,15 @@ +package proxy + +const ( + // This is the default proxy header for forwarding the request + DefaultProxyTargetHeader = "x-proxy-target" +) + +var ( + ProxyTargetHeader = DefaultProxyTargetHeader +) + +// SetProxyTargetHeader sets the header key +func SetProxyTargetHeader(headerKey string) { + ProxyTargetHeader = headerKey +} diff --git a/pkg/proxy/proxy.go b/pkg/proxy/proxy.go index 43c0d594..ba7e14f7 100644 --- a/pkg/proxy/proxy.go +++ b/pkg/proxy/proxy.go @@ -19,8 +19,9 @@ type Proxy struct { handler func(http.ResponseWriter, *http.Request) } -// NewProxy returns an instance of Proxy -func NewProxy(cfg *Config) *Proxy { +// New returns an instance of Proxy +func New(cfg *Config) *Proxy { + // TODO: Allow for overriding of the handler r := mux.NewRouter() s := &http.Server{ Addr: cfg.Addr, From 1bd9cffd3821336ce637e862f4c6100f29d00078 Mon Sep 17 00:00:00 2001 From: Cat Cai Date: Mon, 10 Jun 2019 21:38:37 -0700 Subject: [PATCH 03/14] WIP logging middleware --- Gopkg.lock | 8 ++++++++ cmd/proxy/proxy.go | 3 ++- pkg/middleware/log/log.go | 35 +++++++++++++++++++++++++++++++++++ pkg/proxy/handler.go | 2 ++ pkg/proxy/proxy.go | 13 ++++++++----- 5 files changed, 55 insertions(+), 6 deletions(-) create mode 100644 pkg/middleware/log/log.go diff --git a/Gopkg.lock b/Gopkg.lock index 0bf3c294..4c6af603 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -1,6 +1,14 @@ # This file is autogenerated, do not edit; changes may be undone by the next 'dep ensure'. +[[projects]] + branch = "master" + digest = "1:8b8357ca201567d857f6644124f597a6b0fc20396576da4951bc70ae1ff86bc5" + name = "github.com/blocktop/go-glog-cobra" + packages = ["."] + pruneopts = "UT" + revision = "1a2e8060f5d24a33c21382d5fcc75346d3e260a5" + [[projects]] digest = "1:ffe9824d294da03b391f44e1ae8281281b4afc1bdaa9588c9097785e3af10cec" name = "github.com/davecgh/go-spew" diff --git a/cmd/proxy/proxy.go b/cmd/proxy/proxy.go index 032f32dc..ca56bf0f 100644 --- a/cmd/proxy/proxy.go +++ b/cmd/proxy/proxy.go @@ -24,7 +24,8 @@ func init() { func runProxy(cmd *cobra.Command, args []string) error { cfg := &proxy.Config{ - Addr: fmt.Sprintf(":%d", port), + Addr: fmt.Sprintf(":%d", port), + RouteHandler: proxy.RPCRouteHandler, } proxySrv := proxy.New(cfg) diff --git a/pkg/middleware/log/log.go b/pkg/middleware/log/log.go new file mode 100644 index 00000000..d92edd61 --- /dev/null +++ b/pkg/middleware/log/log.go @@ -0,0 +1,35 @@ +package log + +import ( + "net/http" + + "github.com/wearefair/gurl/pkg/log" +) + +// Middleware is logging middleware that logs the request/response info +func Middleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + rw := &responseWrapper{ResponseWriter: w} + next.ServeHTTP(rw, r) + log.Infof("%s %s -%d", r.Method, r.URL, rw.status) + }) +} + +// responseWrapper allows us to grab the info about a response and log it +type responseWrapper struct { + http.ResponseWriter + status int +} + +func (o *responseWrapper) Header() http.Header { + return o.ResponseWriter.Header() +} + +func (o *responseWrapper) Write(b []byte) (int, error) { + return o.ResponseWriter.Write(b) +} + +func (o *responseWrapper) WriteHeader(code int) { + o.status = code + o.ResponseWriter.WriteHeader(code) +} diff --git a/pkg/proxy/handler.go b/pkg/proxy/handler.go index 24cf5fe7..dd7e5407 100644 --- a/pkg/proxy/handler.go +++ b/pkg/proxy/handler.go @@ -29,4 +29,6 @@ func RPCRouteHandler(rw http.ResponseWriter, req *http.Request) { vars := mux.Vars(req) spew.Dump(vars) + + rw.WriteHeader(http.StatusOK) } diff --git a/pkg/proxy/proxy.go b/pkg/proxy/proxy.go index ba7e14f7..c43ca83b 100644 --- a/pkg/proxy/proxy.go +++ b/pkg/proxy/proxy.go @@ -4,6 +4,7 @@ import ( "net/http" "github.com/gorilla/mux" + logmw "github.com/wearefair/gurl/pkg/middleware/log" ) // Config wraps all configs for the proxy @@ -28,18 +29,20 @@ func New(cfg *Config) *Proxy { Handler: r, } return &Proxy{ - router: r, - server: s, + router: r, + server: s, + handler: cfg.RouteHandler, } } // Configures routes -func (p *Proxy) configure() error { - p.router.HandleFunc("/{service}/{method}", p.handler) - return nil +func (p *Proxy) configure() { + p.router.HandleFunc("/{service}/{method}", p.handler).Methods("POST") + p.router.Use(logmw.Middleware) } // Run the proxy server func (p *Proxy) Run() error { + p.configure() return p.server.ListenAndServe() } From 4c9f59e29f50e7ad8e4fe7a8b65f337c4ea6be31 Mon Sep 17 00:00:00 2001 From: Cat Cai Date: Mon, 10 Jun 2019 21:47:40 -0700 Subject: [PATCH 04/14] Handler --- pkg/middleware/log/log.go | 2 +- pkg/proxy/handler.go | 11 +++++++++-- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/pkg/middleware/log/log.go b/pkg/middleware/log/log.go index d92edd61..7379360f 100644 --- a/pkg/middleware/log/log.go +++ b/pkg/middleware/log/log.go @@ -11,7 +11,7 @@ func Middleware(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { rw := &responseWrapper{ResponseWriter: w} next.ServeHTTP(rw, r) - log.Infof("%s %s -%d", r.Method, r.URL, rw.status) + log.Infof("%s - [%d] - %s", r.Method, rw.status, r.URL) }) } diff --git a/pkg/proxy/handler.go b/pkg/proxy/handler.go index dd7e5407..759b7e95 100644 --- a/pkg/proxy/handler.go +++ b/pkg/proxy/handler.go @@ -7,6 +7,11 @@ import ( "github.com/gorilla/mux" ) +const ( + ServiceKey = "service" + MethodKey = "method" +) + // RPCRouteHandler takes in a route resembling the following // // // localhost:50051/fairapis.pubsub.v1.Publish/ListTopics @@ -19,7 +24,7 @@ func RPCRouteHandler(rw http.ResponseWriter, req *http.Request) { headers := req.Header.Get(ProxyTargetHeader) // If the header is not set... return a 422, because we really - // can't process this request. + // can't process this request. Where are we supposed to forward this to? if headers == "" { rw.WriteHeader(http.StatusUnprocessableEntity) return @@ -27,8 +32,10 @@ func RPCRouteHandler(rw http.ResponseWriter, req *http.Request) { // Pull the variables off of the request vars := mux.Vars(req) - spew.Dump(vars) + // service := vars[ServiceKey] + // method := vars[MethodKey] + rw.WriteHeader(http.StatusOK) } From 50ce69b1f60d32764d185025ab70a1cc4e2311d1 Mon Sep 17 00:00:00 2001 From: Cat Cai Date: Fri, 14 Jun 2019 12:05:50 -0700 Subject: [PATCH 05/14] JSON pb client --- cmd/proxy/proxy.go | 3 +-- pkg/jsonpb/client.go | 2 +- pkg/proxy/config.go | 33 ++++++++++++++++++++++++++++++++ pkg/proxy/handler.go | 38 +++++++++++++++++++++++++++---------- pkg/proxy/header.go | 7 +------ pkg/proxy/proxy.go | 45 ++++++++++++++++++++++---------------------- 6 files changed, 87 insertions(+), 41 deletions(-) create mode 100644 pkg/proxy/config.go diff --git a/cmd/proxy/proxy.go b/cmd/proxy/proxy.go index ca56bf0f..032f32dc 100644 --- a/cmd/proxy/proxy.go +++ b/cmd/proxy/proxy.go @@ -24,8 +24,7 @@ func init() { func runProxy(cmd *cobra.Command, args []string) error { cfg := &proxy.Config{ - Addr: fmt.Sprintf(":%d", port), - RouteHandler: proxy.RPCRouteHandler, + Addr: fmt.Sprintf(":%d", port), } proxySrv := proxy.New(cfg) diff --git a/pkg/jsonpb/client.go b/pkg/jsonpb/client.go index cd40a5c2..d7da5e20 100644 --- a/pkg/jsonpb/client.go +++ b/pkg/jsonpb/client.go @@ -13,7 +13,7 @@ import ( // Client handles constructing and dialing a gRPC service type Client struct { stub grpcdynamic.Stub - // TODO: Might want to turn this into an interface? + // TODO: Might want to turn this into an interface and make this reusable/overrideable? collector *protobuf.Collector } diff --git a/pkg/proxy/config.go b/pkg/proxy/config.go new file mode 100644 index 00000000..16010965 --- /dev/null +++ b/pkg/proxy/config.go @@ -0,0 +1,33 @@ +package proxy + +import ( + "net/http" + + "github.com/gorilla/mux" + logmw "github.com/wearefair/gurl/pkg/middleware/log" +) + +// Config wraps all configs for the proxy +type Config struct { + // Addr is the address that the proxy will run at + Addr string + // Middlewares you want to register with the router + Middlewares []func(http.Handler) http.Handler + // You can overwrite the ProxyTargetHeader to target calls at, if desired. + ProxyTargetHeader string + // If you want to override the default mux.Router. All of the middleware + // passed in will be registered on the router on creation of the Proxy struct. + Router *mux.Router +} + +// DefaultConfig returns a sane base template for Proxy configs +func DefaultConfig() *Config { + return &Config{ + Addr: ":3030", + Middlewares: []func(http.Handler) http.Handler{ + logmw.Middleware, + }, + Router: mux.NewRouter(), + ProxyTargetHeader: DefaultProxyTargetHeader, + } +} diff --git a/pkg/proxy/handler.go b/pkg/proxy/handler.go index 759b7e95..83c1c990 100644 --- a/pkg/proxy/handler.go +++ b/pkg/proxy/handler.go @@ -1,6 +1,8 @@ package proxy import ( + "context" + "io/ioutil" "net/http" "github.com/davecgh/go-spew/spew" @@ -9,19 +11,19 @@ import ( const ( ServiceKey = "service" - MethodKey = "method" + RpcKey = "rpc" ) -// RPCRouteHandler takes in a route resembling the following -// // -// localhost:50051/fairapis.pubsub.v1.Publish/ListTopics +// Handler takes in a route resembling the following +// format: ://. +// Ex: localhost:50051/fairapis.pubsub.v1.Publish/ListTopics // It pulls the service and the RPC name off of the route, generates // the proto message, and then routes it to the proper destination. -// The destination must be set in the headers under the key -// 'x-proxy-host'. -func RPCRouteHandler(rw http.ResponseWriter, req *http.Request) { +// The destination must be set in the headers under the proxy target header +// which defaults to x-gurl-proxy-target +func (p *Proxy) Handler(rw http.ResponseWriter, req *http.Request) { // Pull the host off the headers - headers := req.Header.Get(ProxyTargetHeader) + headers := req.Header.Get(p.proxyTargetHeader) // If the header is not set... return a 422, because we really // can't process this request. Where are we supposed to forward this to? @@ -30,12 +32,28 @@ func RPCRouteHandler(rw http.ResponseWriter, req *http.Request) { return } + defer req.Body.Close() + msg, err := ioutil.ReadAll(req.Body) + if err != nil { + rw.WriteHeader(http.StatusBadRequest) + return + } + // Pull the variables off of the request vars := mux.Vars(req) + // TODO: Remove spew.Dump(vars) - // service := vars[ServiceKey] - // method := vars[MethodKey] + // TODO: Validation errors on either one of these if they're nil + service := vars[ServiceKey] + rpc := vars[RpcKey] + + response, err := p.client.Call(context.Background(), service, rpc, msg) + if err != nil { + rw.WriteHeader(http.StatusBadRequest) + return + } rw.WriteHeader(http.StatusOK) + rw.Write(response) } diff --git a/pkg/proxy/header.go b/pkg/proxy/header.go index eb29f710..a9bf9500 100644 --- a/pkg/proxy/header.go +++ b/pkg/proxy/header.go @@ -2,14 +2,9 @@ package proxy const ( // This is the default proxy header for forwarding the request - DefaultProxyTargetHeader = "x-proxy-target" + DefaultProxyTargetHeader = "x-gurl-proxy-target" ) var ( ProxyTargetHeader = DefaultProxyTargetHeader ) - -// SetProxyTargetHeader sets the header key -func SetProxyTargetHeader(headerKey string) { - ProxyTargetHeader = headerKey -} diff --git a/pkg/proxy/proxy.go b/pkg/proxy/proxy.go index c43ca83b..2906511c 100644 --- a/pkg/proxy/proxy.go +++ b/pkg/proxy/proxy.go @@ -4,45 +4,46 @@ import ( "net/http" "github.com/gorilla/mux" - logmw "github.com/wearefair/gurl/pkg/middleware/log" + "github.com/wearefair/gurl/pkg/jsonpb" ) -// Config wraps all configs for the proxy -type Config struct { - Addr string - RouteHandler func(http.ResponseWriter, *http.Request) -} - // Proxy encapsulates all gURL proxy server logic type Proxy struct { - router *mux.Router - server *http.Server - handler func(http.ResponseWriter, *http.Request) + client *jsonpb.Client + router *mux.Router + server *http.Server + proxyTargetHeader string } // New returns an instance of Proxy func New(cfg *Config) *Proxy { - // TODO: Allow for overriding of the handler - r := mux.NewRouter() + configureMiddleware(cfg.Router, cfg.Middlewares) + s := &http.Server{ Addr: cfg.Addr, - Handler: r, + Handler: cfg.Router, } + return &Proxy{ - router: r, - server: s, - handler: cfg.RouteHandler, + client: cfg.Client, + proxyTargetHeader: cfg.ProxyTargetHeader, + server: s, } } -// Configures routes -func (p *Proxy) configure() { - p.router.HandleFunc("/{service}/{method}", p.handler).Methods("POST") - p.router.Use(logmw.Middleware) -} - // Run the proxy server func (p *Proxy) Run() error { p.configure() return p.server.ListenAndServe() } + +// Proxy configure configures the handler and routes +func (p *Proxy) configure() { + p.router.HandleFunc("/{service}/{rpc}", p.Handler) +} + +func configureMiddleware(r *mux.Router, middlewares []func(http.Handler) http.Handler) { + for _, middleware := range middlewares { + r.Use(middleware) + } +} From 6f99a9e726d6545dfba6febde709a4ae8afcf4ca Mon Sep 17 00:00:00 2001 From: Cat Cai Date: Tue, 18 Jun 2019 11:00:52 -0700 Subject: [PATCH 06/14] Test push --- README.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/README.md b/README.md index b89e5ff3..0da177f4 100644 --- a/README.md +++ b/README.md @@ -93,3 +93,5 @@ If you don't see anything listed, feel free to open an issue or pull request to ### Special Thanks Huge props to [jhump](https://github.com/jhump) for his [protoreflect](https://github.com/jhump/protoreflect) package, which gURL makes heavy use of. + + From bce014ce6abde1be37c5626524137a49b789fdcc Mon Sep 17 00:00:00 2001 From: Cat Cai Date: Sun, 23 Jun 2019 19:15:00 -0700 Subject: [PATCH 07/14] WIP --- cmd/proxy/proxy.go | 6 +---- pkg/jsonpb/client.go | 57 ++++++++++++++++++++++++++++++++++++++++++ pkg/jsonpb/request.go | 18 +++++++++++++ pkg/protobuf/cacher.go | 16 ++++++++++++ pkg/proxy/config.go | 7 ++++++ pkg/proxy/handler.go | 24 +++++++++++++++--- pkg/proxy/proxy.go | 7 +++--- 7 files changed, 124 insertions(+), 11 deletions(-) create mode 100644 pkg/jsonpb/request.go create mode 100644 pkg/protobuf/cacher.go diff --git a/cmd/proxy/proxy.go b/cmd/proxy/proxy.go index 032f32dc..873aeae2 100644 --- a/cmd/proxy/proxy.go +++ b/cmd/proxy/proxy.go @@ -1,8 +1,6 @@ package proxy import ( - "fmt" - "github.com/spf13/cobra" "github.com/wearefair/gurl/pkg/log" "github.com/wearefair/gurl/pkg/proxy" @@ -23,9 +21,7 @@ func init() { } func runProxy(cmd *cobra.Command, args []string) error { - cfg := &proxy.Config{ - Addr: fmt.Sprintf(":%d", port), - } + cfg := proxy.DefaultConfig() proxySrv := proxy.New(cfg) log.Infof("Starting server at %d\n", port) diff --git a/pkg/jsonpb/client.go b/pkg/jsonpb/client.go index d7da5e20..c2111573 100644 --- a/pkg/jsonpb/client.go +++ b/pkg/jsonpb/client.go @@ -12,6 +12,7 @@ import ( // Client handles constructing and dialing a gRPC service type Client struct { + // TODO: We need this to be overrideable stub grpcdynamic.Stub // TODO: Might want to turn this into an interface and make this reusable/overrideable? collector *protobuf.Collector @@ -35,8 +36,64 @@ func NewClient(cfg *Config) (*Client, error) { }, nil } +// Invoke makes a unary call across the wire. +func (c *Client) Invoke(ctx context.Context, req *Request) ([]byte, error) { + conn, err := grpc.Dial(req.Address, req.DialOptions...) + if err != nil { + return nil, err + } + + stub := grpcdynamic.NewStub(conn) + + serviceDescriptor, err := c.collector.GetService(req.Service) + if err != nil { + return nil, err + } + + // Find the RPC attached to the service via the URI + methodDescriptor := serviceDescriptor.FindMethodByName(req.RPC) + if methodDescriptor == nil { + err := fmt.Errorf("No method %s found", req.Service) + return nil, err + } + + methodProto := methodDescriptor.AsMethodDescriptorProto() + messageDescriptor, err := c.collector.GetMessage( + protobuf.NormalizeMessageName(*methodProto.InputType), + ) + if err != nil { + return nil, err + } + + message, err := protobuf.Construct(messageDescriptor, req.Message) + if err != nil { + return nil, err + } + + // TODO: Allow for streaming calls. This locks us to unary calls + // Disabled server and client streaming calls + disableStreaming := false + methodProto.ClientStreaming = &disableStreaming + methodProto.ServerStreaming = &disableStreaming + + response, err := stub.InvokeRpc(ctx, methodDescriptor, message) + if err != nil { + return nil, err + } + + marshaler := &runtime.JSONPb{} + // Marshals PB response into JSON + responseJSON, err := marshaler.Marshal(response) + if err != nil { + return nil, err + } + + return responseJSON, nil +} + // Call takes in a context, service, RPC, and message as JSON string to convert to protobuf and // send across the wire. +// TODO: Deprecate this in favor of the Invoke call func (c *Client) Call(ctx context.Context, service, rpc string, rawMsg []byte) ([]byte, error) { serviceDescriptor, err := c.collector.GetService(service) if err != nil { diff --git a/pkg/jsonpb/request.go b/pkg/jsonpb/request.go new file mode 100644 index 00000000..69fb0e75 --- /dev/null +++ b/pkg/jsonpb/request.go @@ -0,0 +1,18 @@ +package jsonpb + +import "google.golang.org/grpc" + +// Request encompasses all of the args required to make a request +type Request struct { + // The address the gRPC client will be making a call to + Address string + // DialOptions to pass down to the gRPC client + DialOptions []grpc.DialOption + + // The FQDN of the gRPC service + Service string + // The method name of the RPC to target + RPC string + // The JSON message in bytes to send to the service + Message []byte +} diff --git a/pkg/protobuf/cacher.go b/pkg/protobuf/cacher.go new file mode 100644 index 00000000..24cf7bb7 --- /dev/null +++ b/pkg/protobuf/cacher.go @@ -0,0 +1,16 @@ +package protobuf + +import "github.com/jhump/protoreflect/desc" + +// Cacher is the public interface for registering FileDescriptors and getting back +// Message and ServiceDescriptors. +type Cacher interface { + // AddDescriptors takes in a slice of file descriptors, walks them, and collects + // all message/service descriptors so that they can be returned in the GetMessage + // and GetService methods. + AddDescriptors(fileDescriptors []*desc.FileDescriptor) + // GetMessage takes a message descriptor's FQDN and returns the descriptor + GetMessage(fqdn string) (*desc.MessageDescriptor, error) + // GetService takes a service descriptor's FQDN and returns the descriptor + GetService(fqdn string) (*desc.ServiceDescriptor, error) +} diff --git a/pkg/proxy/config.go b/pkg/proxy/config.go index 16010965..d77b3b53 100644 --- a/pkg/proxy/config.go +++ b/pkg/proxy/config.go @@ -5,12 +5,19 @@ import ( "github.com/gorilla/mux" logmw "github.com/wearefair/gurl/pkg/middleware/log" + "google.golang.org/grpc" ) // Config wraps all configs for the proxy type Config struct { // Addr is the address that the proxy will run at Addr string + // DialOptions to pass down to the jsonpb client + DialOptions []grpc.DialOption + // ImportPaths are the import paths to pass down to the jsonpb client + ImportPaths []string + // ServicePaths are the service paths to pass down to the jsonpb client + ServicePaths []string // Middlewares you want to register with the router Middlewares []func(http.Handler) http.Handler // You can overwrite the ProxyTargetHeader to target calls at, if desired. diff --git a/pkg/proxy/handler.go b/pkg/proxy/handler.go index 83c1c990..ab8aef1c 100644 --- a/pkg/proxy/handler.go +++ b/pkg/proxy/handler.go @@ -7,6 +7,8 @@ import ( "github.com/davecgh/go-spew/spew" "github.com/gorilla/mux" + "github.com/wearefair/gurl/pkg/jsonpb" + "github.com/wearefair/gurl/pkg/log" ) const ( @@ -23,11 +25,12 @@ const ( // which defaults to x-gurl-proxy-target func (p *Proxy) Handler(rw http.ResponseWriter, req *http.Request) { // Pull the host off the headers - headers := req.Header.Get(p.proxyTargetHeader) + target := req.Header.Get(p.proxyTargetHeader) // If the header is not set... return a 422, because we really // can't process this request. Where are we supposed to forward this to? - if headers == "" { + if target == "" { + log.Error("Proxy target header is empty!") rw.WriteHeader(http.StatusUnprocessableEntity) return } @@ -35,6 +38,7 @@ func (p *Proxy) Handler(rw http.ResponseWriter, req *http.Request) { defer req.Body.Close() msg, err := ioutil.ReadAll(req.Body) if err != nil { + log.Error(err.Error()) rw.WriteHeader(http.StatusBadRequest) return } @@ -48,8 +52,22 @@ func (p *Proxy) Handler(rw http.ResponseWriter, req *http.Request) { service := vars[ServiceKey] rpc := vars[RpcKey] - response, err := p.client.Call(context.Background(), service, rpc, msg) + cfg := &jsonpb.Config{ + Address: target, + ImportPaths: p.importPaths, + ServicePaths: p.servicePaths, + } + + client, err := jsonpb.NewClient(cfg) + if err != nil { + log.Error(err) + rw.WriteHeader(http.StatusInternalServerError) + return + } + + response, err := client.Call(context.Background(), service, rpc, msg) if err != nil { + log.Error(err) rw.WriteHeader(http.StatusBadRequest) return } diff --git a/pkg/proxy/proxy.go b/pkg/proxy/proxy.go index 2906511c..385d28f3 100644 --- a/pkg/proxy/proxy.go +++ b/pkg/proxy/proxy.go @@ -4,15 +4,16 @@ import ( "net/http" "github.com/gorilla/mux" - "github.com/wearefair/gurl/pkg/jsonpb" ) // Proxy encapsulates all gURL proxy server logic type Proxy struct { - client *jsonpb.Client router *mux.Router server *http.Server proxyTargetHeader string + + importPaths []string + servicePaths []string } // New returns an instance of Proxy @@ -25,8 +26,8 @@ func New(cfg *Config) *Proxy { } return &Proxy{ - client: cfg.Client, proxyTargetHeader: cfg.ProxyTargetHeader, + router: cfg.Router, server: s, } } From 6cb24ef66efc17336a37ead7b07175a045833d49 Mon Sep 17 00:00:00 2001 From: Cat Cai Date: Thu, 27 Jun 2019 14:57:16 -0700 Subject: [PATCH 08/14] Update JSONPB client logic --- cmd/call/call.go | 11 +++++++-- cmd/proxy/proxy.go | 3 +++ pkg/jsonpb/client.go | 58 -------------------------------------------- pkg/proxy/handler.go | 23 ++++++++++++++---- pkg/proxy/proxy.go | 3 +++ 5 files changed, 33 insertions(+), 65 deletions(-) diff --git a/cmd/call/call.go b/cmd/call/call.go index 1d90ae3e..5e51d07e 100644 --- a/cmd/call/call.go +++ b/cmd/call/call.go @@ -84,7 +84,6 @@ func runCall(cmd *cobra.Command, args []string) error { } cfg := &jsonpb.Config{ - Address: address, DialOptions: callOptions.DialOptions(), ImportPaths: config.Instance().Local.ImportPaths, ServicePaths: config.Instance().Local.ServicePaths, @@ -96,7 +95,15 @@ func runCall(cmd *cobra.Command, args []string) error { } // Send request and get response - response, err := client.Call(callOptions.ContextWithOptions(context.Background()), parsedURI.Service, parsedURI.RPC, []byte(data)) + req := &jsonpb.Request{ + Address: address, + DialOptions: callOptions.DialOptions(), + Service: parsedURI.Service, + RPC: parsedURI.RPC, + Message: []byte(data), + } + + response, err := client.Invoke(callOptions.ContextWithOptions(context.Background()), req) if err != nil { return log.LogAndReturn(err) } diff --git a/cmd/proxy/proxy.go b/cmd/proxy/proxy.go index 873aeae2..5764f24e 100644 --- a/cmd/proxy/proxy.go +++ b/cmd/proxy/proxy.go @@ -2,6 +2,7 @@ package proxy import ( "github.com/spf13/cobra" + "github.com/wearefair/gurl/pkg/config" "github.com/wearefair/gurl/pkg/log" "github.com/wearefair/gurl/pkg/proxy" ) @@ -22,6 +23,8 @@ func init() { func runProxy(cmd *cobra.Command, args []string) error { cfg := proxy.DefaultConfig() + cfg.ImportPaths = config.Instance().Local.ImportPaths + cfg.ServicePaths = config.Instance().Local.ServicePaths proxySrv := proxy.New(cfg) log.Infof("Starting server at %d\n", port) diff --git a/pkg/jsonpb/client.go b/pkg/jsonpb/client.go index c2111573..772ec0a6 100644 --- a/pkg/jsonpb/client.go +++ b/pkg/jsonpb/client.go @@ -12,18 +12,11 @@ import ( // Client handles constructing and dialing a gRPC service type Client struct { - // TODO: We need this to be overrideable - stub grpcdynamic.Stub - // TODO: Might want to turn this into an interface and make this reusable/overrideable? collector *protobuf.Collector } // NewClient creates a client with a Stub func NewClient(cfg *Config) (*Client, error) { - conn, err := grpc.Dial(cfg.Address, cfg.DialOptions...) - if err != nil { - return nil, err - } // Walks the proto import and service paths defined in the config and returns all descriptors descriptors, err := protobuf.Collect(cfg.ImportPaths, cfg.ServicePaths) if err != nil { @@ -31,7 +24,6 @@ func NewClient(cfg *Config) (*Client, error) { } return &Client{ - stub: grpcdynamic.NewStub(conn), collector: protobuf.NewCollector(descriptors), }, nil } @@ -90,53 +82,3 @@ func (c *Client) Invoke(ctx context.Context, req *Request) ([]byte, error) { return responseJSON, nil } - -// Call takes in a context, service, RPC, and message as JSON string to convert to protobuf and -// send across the wire. -// TODO: Deprecate this in favor of the Invoke call -func (c *Client) Call(ctx context.Context, service, rpc string, rawMsg []byte) ([]byte, error) { - serviceDescriptor, err := c.collector.GetService(service) - if err != nil { - return nil, err - } - - // Find the RPC attached to the service via the URI - methodDescriptor := serviceDescriptor.FindMethodByName(rpc) - if methodDescriptor == nil { - err := fmt.Errorf("No method %s found", service) - return nil, err - } - - methodProto := methodDescriptor.AsMethodDescriptorProto() - messageDescriptor, err := c.collector.GetMessage( - protobuf.NormalizeMessageName(*methodProto.InputType), - ) - if err != nil { - return nil, err - } - - message, err := protobuf.Construct(messageDescriptor, rawMsg) - if err != nil { - return nil, err - } - - // TODO: Allow for streaming calls. This locks us to unary calls - // Disabled server and client streaming calls - disableStreaming := false - methodProto.ClientStreaming = &disableStreaming - methodProto.ServerStreaming = &disableStreaming - - response, err := c.stub.InvokeRpc(ctx, methodDescriptor, message) - if err != nil { - return nil, err - } - - marshaler := &runtime.JSONPb{} - // Marshals PB response into JSON - responseJSON, err := marshaler.Marshal(response) - if err != nil { - return nil, err - } - - return responseJSON, nil -} diff --git a/pkg/proxy/handler.go b/pkg/proxy/handler.go index ab8aef1c..20590220 100644 --- a/pkg/proxy/handler.go +++ b/pkg/proxy/handler.go @@ -1,14 +1,13 @@ package proxy import ( - "context" "io/ioutil" "net/http" - "github.com/davecgh/go-spew/spew" "github.com/gorilla/mux" "github.com/wearefair/gurl/pkg/jsonpb" "github.com/wearefair/gurl/pkg/log" + "google.golang.org/grpc" ) const ( @@ -24,6 +23,8 @@ const ( // The destination must be set in the headers under the proxy target header // which defaults to x-gurl-proxy-target func (p *Proxy) Handler(rw http.ResponseWriter, req *http.Request) { + ctx := req.Context() + // Pull the host off the headers target := req.Header.Get(p.proxyTargetHeader) @@ -45,8 +46,6 @@ func (p *Proxy) Handler(rw http.ResponseWriter, req *http.Request) { // Pull the variables off of the request vars := mux.Vars(req) - // TODO: Remove - spew.Dump(vars) // TODO: Validation errors on either one of these if they're nil service := vars[ServiceKey] @@ -56,6 +55,20 @@ func (p *Proxy) Handler(rw http.ResponseWriter, req *http.Request) { Address: target, ImportPaths: p.importPaths, ServicePaths: p.servicePaths, + // TODO + DialOptions: []grpc.DialOption{ + grpc.WithInsecure(), + }, + } + + jsonpbReq := &jsonpb.Request{ + Address: target, + DialOptions: []grpc.DialOption{ + grpc.WithInsecure(), + }, + Service: service, + RPC: rpc, + Message: msg, } client, err := jsonpb.NewClient(cfg) @@ -65,7 +78,7 @@ func (p *Proxy) Handler(rw http.ResponseWriter, req *http.Request) { return } - response, err := client.Call(context.Background(), service, rpc, msg) + response, err := client.Invoke(ctx, jsonpbReq) if err != nil { log.Error(err) rw.WriteHeader(http.StatusBadRequest) diff --git a/pkg/proxy/proxy.go b/pkg/proxy/proxy.go index 385d28f3..8a191aa7 100644 --- a/pkg/proxy/proxy.go +++ b/pkg/proxy/proxy.go @@ -29,6 +29,9 @@ func New(cfg *Config) *Proxy { proxyTargetHeader: cfg.ProxyTargetHeader, router: cfg.Router, server: s, + + importPaths: cfg.ImportPaths, + servicePaths: cfg.ServicePaths, } } From 68c3299ec6e3294130b66c8f98d153ad30b2c01b Mon Sep 17 00:00:00 2001 From: Cat Cai Date: Thu, 27 Jun 2019 21:04:24 -0700 Subject: [PATCH 09/14] Add tests --- Gopkg.lock | 33 +++++++++++++++------------------ pkg/proxy/config.go | 8 +++++--- pkg/proxy/handler.go | 33 +++++++++++++++++++-------------- pkg/proxy/handler_test.go | 38 ++++++++++++++++++++++++++++++++++++++ pkg/proxy/proxy.go | 3 +++ 5 files changed, 80 insertions(+), 35 deletions(-) create mode 100644 pkg/proxy/handler_test.go diff --git a/Gopkg.lock b/Gopkg.lock index 4c6af603..68fafcec 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -1,22 +1,6 @@ # This file is autogenerated, do not edit; changes may be undone by the next 'dep ensure'. -[[projects]] - branch = "master" - digest = "1:8b8357ca201567d857f6644124f597a6b0fc20396576da4951bc70ae1ff86bc5" - name = "github.com/blocktop/go-glog-cobra" - packages = ["."] - pruneopts = "UT" - revision = "1a2e8060f5d24a33c21382d5fcc75346d3e260a5" - -[[projects]] - digest = "1:ffe9824d294da03b391f44e1ae8281281b4afc1bdaa9588c9097785e3af10cec" - name = "github.com/davecgh/go-spew" - packages = ["spew"] - pruneopts = "UT" - revision = "8991bc29aa16c548c550c7ff78260e27b9ab7c73" - version = "v1.1.1" - [[projects]] branch = "master" digest = "1:dbb3d1675f5beeb37de6e9b95cc460158ff212902a916e67688b01e0660f41bd" @@ -83,6 +67,20 @@ pruneopts = "UT" revision = "316fb6d3f031ae8f4d457c6c5186b9e3ded70435" +[[projects]] + digest = "1:bf40199583e5143d1472fc34d10d6f4b69d97572142acf343b3e43136da40823" + name = "github.com/google/go-cmp" + packages = [ + "cmp", + "cmp/internal/diff", + "cmp/internal/flags", + "cmp/internal/function", + "cmp/internal/value", + ] + pruneopts = "UT" + revision = "6f77996f0c42f7b84e5a2b252227263f93432e9b" + version = "v0.3.0" + [[projects]] branch = "master" digest = "1:3ee90c0d94da31b442dde97c99635aaafec68d0b8a3c12ee2075c6bdabeec6bb" @@ -495,9 +493,8 @@ analyzer-name = "dep" analyzer-version = 1 input-imports = [ - "github.com/davecgh/go-spew/spew", "github.com/golang/glog", - "github.com/golang/protobuf/proto", + "github.com/google/go-cmp/cmp", "github.com/gorilla/mux", "github.com/grpc-ecosystem/grpc-gateway/runtime", "github.com/jhump/protoreflect/desc", diff --git a/pkg/proxy/config.go b/pkg/proxy/config.go index d77b3b53..20ea0ed5 100644 --- a/pkg/proxy/config.go +++ b/pkg/proxy/config.go @@ -5,15 +5,16 @@ import ( "github.com/gorilla/mux" logmw "github.com/wearefair/gurl/pkg/middleware/log" - "google.golang.org/grpc" + "github.com/wearefair/gurl/pkg/options" + "google.golang.org/grpc/metadata" ) // Config wraps all configs for the proxy type Config struct { // Addr is the address that the proxy will run at Addr string - // DialOptions to pass down to the jsonpb client - DialOptions []grpc.DialOption + // Options to pass down to the jsonpb client + Options *options.Options // ImportPaths are the import paths to pass down to the jsonpb client ImportPaths []string // ServicePaths are the service paths to pass down to the jsonpb client @@ -34,6 +35,7 @@ func DefaultConfig() *Config { Middlewares: []func(http.Handler) http.Handler{ logmw.Middleware, }, + Options: &options.Options{Metadata: metadata.MD{}}, Router: mux.NewRouter(), ProxyTargetHeader: DefaultProxyTargetHeader, } diff --git a/pkg/proxy/handler.go b/pkg/proxy/handler.go index 20590220..76030f96 100644 --- a/pkg/proxy/handler.go +++ b/pkg/proxy/handler.go @@ -1,13 +1,14 @@ package proxy import ( + "context" "io/ioutil" "net/http" "github.com/gorilla/mux" "github.com/wearefair/gurl/pkg/jsonpb" "github.com/wearefair/gurl/pkg/log" - "google.golang.org/grpc" + "google.golang.org/grpc/metadata" ) const ( @@ -23,8 +24,6 @@ const ( // The destination must be set in the headers under the proxy target header // which defaults to x-gurl-proxy-target func (p *Proxy) Handler(rw http.ResponseWriter, req *http.Request) { - ctx := req.Context() - // Pull the host off the headers target := req.Header.Get(p.proxyTargetHeader) @@ -55,20 +54,14 @@ func (p *Proxy) Handler(rw http.ResponseWriter, req *http.Request) { Address: target, ImportPaths: p.importPaths, ServicePaths: p.servicePaths, - // TODO - DialOptions: []grpc.DialOption{ - grpc.WithInsecure(), - }, } jsonpbReq := &jsonpb.Request{ - Address: target, - DialOptions: []grpc.DialOption{ - grpc.WithInsecure(), - }, - Service: service, - RPC: rpc, - Message: msg, + Address: target, + DialOptions: p.opts.DialOptions(), + Service: service, + RPC: rpc, + Message: msg, } client, err := jsonpb.NewClient(cfg) @@ -78,6 +71,9 @@ func (p *Proxy) Handler(rw http.ResponseWriter, req *http.Request) { return } + outgoingMd := mergeHttpHeadersToMetadata(metadata.MD{}, req.Header) + ctx := metadata.NewOutgoingContext(context.Background(), outgoingMd) + response, err := client.Invoke(ctx, jsonpbReq) if err != nil { log.Error(err) @@ -88,3 +84,12 @@ func (p *Proxy) Handler(rw http.ResponseWriter, req *http.Request) { rw.WriteHeader(http.StatusOK) rw.Write(response) } + +func mergeHttpHeadersToMetadata(md metadata.MD, headers http.Header) metadata.MD { + mdCopy := md.Copy() + for key, vals := range headers { + mdCopy.Append(key, vals...) + } + + return mdCopy +} diff --git a/pkg/proxy/handler_test.go b/pkg/proxy/handler_test.go new file mode 100644 index 00000000..cf214309 --- /dev/null +++ b/pkg/proxy/handler_test.go @@ -0,0 +1,38 @@ +package proxy + +import ( + "net/http" + "testing" + + "github.com/google/go-cmp/cmp" + "google.golang.org/grpc/metadata" +) + +func TestMergeHttpHeadersToMetadata(t *testing.T) { + testCases := []struct { + Md metadata.MD + Headers http.Header + Expected metadata.MD + }{ + // Empty metadata merges with HTTP headers + { + Md: metadata.MD{}, + Headers: map[string][]string{"foo": []string{"bar"}}, + Expected: metadata.Pairs("foo", "bar"), + }, + // Headers with multiple values merge properly + { + Md: metadata.MD{}, + Headers: map[string][]string{"hello": []string{"world", "boo"}}, + Expected: metadata.Pairs("hello", "world", "hello", "boo"), + }, + } + + for i, testCase := range testCases { + res := mergeHttpHeadersToMetadata(testCase.Md, testCase.Headers) + + if !cmp.Equal(res, testCase.Expected) { + t.Errorf("[%d] - Expected: %+v\nActual: %+v\n", i, testCase.Expected, res) + } + } +} diff --git a/pkg/proxy/proxy.go b/pkg/proxy/proxy.go index 8a191aa7..b74a2852 100644 --- a/pkg/proxy/proxy.go +++ b/pkg/proxy/proxy.go @@ -4,12 +4,14 @@ import ( "net/http" "github.com/gorilla/mux" + "github.com/wearefair/gurl/pkg/options" ) // Proxy encapsulates all gURL proxy server logic type Proxy struct { router *mux.Router server *http.Server + opts *options.Options proxyTargetHeader string importPaths []string @@ -26,6 +28,7 @@ func New(cfg *Config) *Proxy { } return &Proxy{ + opts: cfg.Options, proxyTargetHeader: cfg.ProxyTargetHeader, router: cfg.Router, server: s, From 8129f4ca0250453072887611ab62fcc1fb18f3a0 Mon Sep 17 00:00:00 2001 From: Cat Cai Date: Fri, 28 Jun 2019 09:45:35 -0700 Subject: [PATCH 10/14] Handler logic --- pkg/proxy/handler.go | 2 +- pkg/proxy/handler_test.go | 5 +++++ 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/pkg/proxy/handler.go b/pkg/proxy/handler.go index 76030f96..d7bc3227 100644 --- a/pkg/proxy/handler.go +++ b/pkg/proxy/handler.go @@ -71,7 +71,7 @@ func (p *Proxy) Handler(rw http.ResponseWriter, req *http.Request) { return } - outgoingMd := mergeHttpHeadersToMetadata(metadata.MD{}, req.Header) + outgoingMd := mergeHttpHeadersToMetadata(p.opts.Metadata, req.Header) ctx := metadata.NewOutgoingContext(context.Background(), outgoingMd) response, err := client.Invoke(ctx, jsonpbReq) diff --git a/pkg/proxy/handler_test.go b/pkg/proxy/handler_test.go index cf214309..47068961 100644 --- a/pkg/proxy/handler_test.go +++ b/pkg/proxy/handler_test.go @@ -26,6 +26,11 @@ func TestMergeHttpHeadersToMetadata(t *testing.T) { Headers: map[string][]string{"hello": []string{"world", "boo"}}, Expected: metadata.Pairs("hello", "world", "hello", "boo"), }, + // Original metadata is not touched if headers are empty + { + Md: metadata.Pairs("foo", "bar"), + Expected: metadata.Pairs("foo", "bar"), + }, } for i, testCase := range testCases { From 87655da8eb89101c86b48d8bf48ef4bbcb1a07eb Mon Sep 17 00:00:00 2001 From: Cat Cai Date: Sat, 6 Jul 2019 09:58:41 -0700 Subject: [PATCH 11/14] Yay this works --- cmd/call/call.go | 14 +++++------ pkg/jsonpb/client.go | 5 ++-- pkg/jsonpb/connector.go | 53 +++++++++++++++++++++++++++++++++++++++++ pkg/jsonpb/request.go | 32 +++++++++++++++++++++++-- 4 files changed, 92 insertions(+), 12 deletions(-) create mode 100644 pkg/jsonpb/connector.go diff --git a/cmd/call/call.go b/cmd/call/call.go index 5e51d07e..f1b98a74 100644 --- a/cmd/call/call.go +++ b/cmd/call/call.go @@ -70,32 +70,30 @@ func runCall(cmd *cobra.Command, args []string) error { } log.Infof("Parsed URI: %#v", parsedURI) - address := fmt.Sprintf("%s:%s", parsedURI.Host, parsedURI.Port) + // Set up connector + var connector jsonpb.Connector if parsedURI.Protocol == util.K8Protocol { // Set up port forward, then send request req := uriToPortForwardRequest(parsedURI) - pf, err := k8.StartPortForward(k8Config(), req) - if err != nil { - return err - } - defer pf.Close() - address = fmt.Sprintf("localhost:%s", pf.LocalPort()) + connector = jsonpb.NewK8Connector(k8Config(), req) } + // Set up the JSONPB client cfg := &jsonpb.Config{ DialOptions: callOptions.DialOptions(), ImportPaths: config.Instance().Local.ImportPaths, ServicePaths: config.Instance().Local.ServicePaths, } - client, err := jsonpb.NewClient(cfg) if err != nil { return log.LogAndReturn(err) } // Send request and get response + address := fmt.Sprintf("%s:%s", parsedURI.Host, parsedURI.Port) req := &jsonpb.Request{ + Connector: connector, Address: address, DialOptions: callOptions.DialOptions(), Service: parsedURI.Service, diff --git a/pkg/jsonpb/client.go b/pkg/jsonpb/client.go index 772ec0a6..4462ef5b 100644 --- a/pkg/jsonpb/client.go +++ b/pkg/jsonpb/client.go @@ -7,7 +7,6 @@ import ( "github.com/grpc-ecosystem/grpc-gateway/runtime" "github.com/jhump/protoreflect/dynamic/grpcdynamic" "github.com/wearefair/gurl/pkg/protobuf" - "google.golang.org/grpc" ) // Client handles constructing and dialing a gRPC service @@ -30,10 +29,12 @@ func NewClient(cfg *Config) (*Client, error) { // Invoke makes a unary call across the wire. func (c *Client) Invoke(ctx context.Context, req *Request) ([]byte, error) { - conn, err := grpc.Dial(req.Address, req.DialOptions...) + // Run any connection logic + conn, err := req.Connect() if err != nil { return nil, err } + defer req.Close() stub := grpcdynamic.NewStub(conn) diff --git a/pkg/jsonpb/connector.go b/pkg/jsonpb/connector.go new file mode 100644 index 00000000..73e8e13a --- /dev/null +++ b/pkg/jsonpb/connector.go @@ -0,0 +1,53 @@ +package jsonpb + +import ( + "fmt" + + "github.com/wearefair/gurl/pkg/k8" + "k8s.io/client-go/tools/clientcmd" +) + +// Connector wraps logic for connecting to a service, and then closing +// that connection. +type Connector interface { + // Connect connects sand then returns the address that the connector is expected + // to connect at, if any + Connect() (address string, err error) + Close() error +} + +// K8Connector implements the +type K8Connector struct { + k8Cfg clientcmd.ClientConfig + req k8.PortForwardRequest + pf *k8.PortForward +} + +// NewK8Connector opens up a new portforward request to K8 +func NewK8Connector(cfg clientcmd.ClientConfig, req k8.PortForwardRequest) *K8Connector { + return &K8Connector{ + k8Cfg: cfg, + req: req, + } +} + +// Connect on the K8Connector opens up a portforward, and then returns the portforward +// address to dial and connect to +func (c *K8Connector) Connect() (string, error) { + var address string + pf, err := k8.StartPortForward(c.k8Cfg, c.req) + if err != nil { + return address, err + } + c.pf = pf + + address = fmt.Sprintf("localhost:%s", pf.LocalPort()) + return address, nil +} + +// Close closest the portforward connection +func (c *K8Connector) Close() error { + c.pf.Close() + + return nil +} diff --git a/pkg/jsonpb/request.go b/pkg/jsonpb/request.go index 69fb0e75..6427605e 100644 --- a/pkg/jsonpb/request.go +++ b/pkg/jsonpb/request.go @@ -1,11 +1,17 @@ package jsonpb -import "google.golang.org/grpc" +import ( + "google.golang.org/grpc" +) // Request encompasses all of the args required to make a request type Request struct { - // The address the gRPC client will be making a call to + // Connector handles things like K8 portforward logic + Connector Connector + + // Address is where something will connect to Address string + // DialOptions to pass down to the gRPC client DialOptions []grpc.DialOption @@ -16,3 +22,25 @@ type Request struct { // The JSON message in bytes to send to the service Message []byte } + +func (r *Request) Connect() (*grpc.ClientConn, error) { + var err error + address := r.Address + + if r.Connector != nil { + address, err = r.Connector.Connect() + if err != nil { + return nil, err + } + } + + return grpc.Dial(address, r.DialOptions...) +} + +func (r *Request) Close() error { + if r.Connector == nil { + return nil + } + + return r.Connector.Close() +} From fddc4e0bc744675f417211e56fee31a60d823718 Mon Sep 17 00:00:00 2001 From: Cat Cai Date: Sat, 6 Jul 2019 10:27:17 -0700 Subject: [PATCH 12/14] Trying to make this decomposable and useful --- cmd/proxy/proxy.go | 17 +++++++++++++++-- pkg/jsonpb/config.go | 1 - pkg/proxy/caller.go | 12 ++++++++++++ pkg/proxy/config.go | 16 +++++++++++----- pkg/proxy/handler.go | 16 ++-------------- pkg/proxy/proxy.go | 5 +++-- 6 files changed, 43 insertions(+), 24 deletions(-) create mode 100644 pkg/proxy/caller.go diff --git a/cmd/proxy/proxy.go b/cmd/proxy/proxy.go index 5764f24e..32d118f1 100644 --- a/cmd/proxy/proxy.go +++ b/cmd/proxy/proxy.go @@ -3,6 +3,7 @@ package proxy import ( "github.com/spf13/cobra" "github.com/wearefair/gurl/pkg/config" + "github.com/wearefair/gurl/pkg/jsonpb" "github.com/wearefair/gurl/pkg/log" "github.com/wearefair/gurl/pkg/proxy" ) @@ -23,8 +24,20 @@ func init() { func runProxy(cmd *cobra.Command, args []string) error { cfg := proxy.DefaultConfig() - cfg.ImportPaths = config.Instance().Local.ImportPaths - cfg.ServicePaths = config.Instance().Local.ServicePaths + // cfg.ImportPaths = config.Instance().Local.ImportPaths + // cfg.ServicePaths = config.Instance().Local.ServicePaths + + jsonpbCfg := &jsonpb.Config{ + ImportPaths: config.Instance().Local.ImportPaths, + ServicePaths: config.Instance().Local.ServicePaths, + } + + client, err := jsonpb.NewClient(jsonpbCfg) + if err != nil { + return err + } + + cfg.Caller = client proxySrv := proxy.New(cfg) log.Infof("Starting server at %d\n", port) diff --git a/pkg/jsonpb/config.go b/pkg/jsonpb/config.go index e55e961b..ce4cd345 100644 --- a/pkg/jsonpb/config.go +++ b/pkg/jsonpb/config.go @@ -6,7 +6,6 @@ import ( // Config handles everything necessary to construct a Client type Config struct { - Address string DialOptions []grpc.DialOption ImportPaths []string ServicePaths []string diff --git a/pkg/proxy/caller.go b/pkg/proxy/caller.go new file mode 100644 index 00000000..4f21ff9f --- /dev/null +++ b/pkg/proxy/caller.go @@ -0,0 +1,12 @@ +package proxy + +import ( + "context" + + "github.com/wearefair/gurl/pkg/jsonpb" +) + +// Caller is the interface that wraps calling across the wire +type Caller interface { + Invoke(context.Context, *jsonpb.Request) ([]byte, error) +} diff --git a/pkg/proxy/config.go b/pkg/proxy/config.go index 20ea0ed5..6f6bab75 100644 --- a/pkg/proxy/config.go +++ b/pkg/proxy/config.go @@ -15,14 +15,20 @@ type Config struct { Addr string // Options to pass down to the jsonpb client Options *options.Options - // ImportPaths are the import paths to pass down to the jsonpb client - ImportPaths []string - // ServicePaths are the service paths to pass down to the jsonpb client - ServicePaths []string + // Caller is what's used to pull in the import and service paths for handling + // the JSON to protobuf behavior + Caller Caller + // // ImportPaths are the import paths to pass down to the jsonpb client + // ImportPaths []string + // // ServicePaths are the service paths to pass down to the jsonpb client + // ServicePaths []string // Middlewares you want to register with the router Middlewares []func(http.Handler) http.Handler - // You can overwrite the ProxyTargetHeader to target calls at, if desired. + // You can override the ProxyTargetHeader to target calls at, if desired. ProxyTargetHeader string + // You can override the default handler. This should only be passed in if + // you don't want to override the Router. + ProxyHandler http.HandlerFunc // If you want to override the default mux.Router. All of the middleware // passed in will be registered on the router on creation of the Proxy struct. Router *mux.Router diff --git a/pkg/proxy/handler.go b/pkg/proxy/handler.go index d7bc3227..7bd0e503 100644 --- a/pkg/proxy/handler.go +++ b/pkg/proxy/handler.go @@ -50,12 +50,6 @@ func (p *Proxy) Handler(rw http.ResponseWriter, req *http.Request) { service := vars[ServiceKey] rpc := vars[RpcKey] - cfg := &jsonpb.Config{ - Address: target, - ImportPaths: p.importPaths, - ServicePaths: p.servicePaths, - } - jsonpbReq := &jsonpb.Request{ Address: target, DialOptions: p.opts.DialOptions(), @@ -64,17 +58,11 @@ func (p *Proxy) Handler(rw http.ResponseWriter, req *http.Request) { Message: msg, } - client, err := jsonpb.NewClient(cfg) - if err != nil { - log.Error(err) - rw.WriteHeader(http.StatusInternalServerError) - return - } - + // TODO: This ctx isn't used properly outgoingMd := mergeHttpHeadersToMetadata(p.opts.Metadata, req.Header) ctx := metadata.NewOutgoingContext(context.Background(), outgoingMd) - response, err := client.Invoke(ctx, jsonpbReq) + response, err := p.caller.Invoke(ctx, jsonpbReq) if err != nil { log.Error(err) rw.WriteHeader(http.StatusBadRequest) diff --git a/pkg/proxy/proxy.go b/pkg/proxy/proxy.go index b74a2852..ed6420ba 100644 --- a/pkg/proxy/proxy.go +++ b/pkg/proxy/proxy.go @@ -9,6 +9,7 @@ import ( // Proxy encapsulates all gURL proxy server logic type Proxy struct { + caller Caller router *mux.Router server *http.Server opts *options.Options @@ -33,8 +34,8 @@ func New(cfg *Config) *Proxy { router: cfg.Router, server: s, - importPaths: cfg.ImportPaths, - servicePaths: cfg.ServicePaths, + // importPaths: cfg.ImportPaths, + // servicePaths: cfg.ServicePaths, } } From e71484657e429bf7f845a0a1e04dd3eb237d31fe Mon Sep 17 00:00:00 2001 From: Cat Cai Date: Sun, 7 Jul 2019 09:46:37 -0700 Subject: [PATCH 13/14] Finally got portforwarding to work. Now to cleanup --- pkg/proxy/handler.go | 49 ++++++++++++++++++++++++++++++++++++++++++++ pkg/proxy/proxy.go | 4 +--- pkg/util/uri.go | 2 +- pkg/util/uri_test.go | 11 ++++++++++ 4 files changed, 62 insertions(+), 4 deletions(-) diff --git a/pkg/proxy/handler.go b/pkg/proxy/handler.go index 7bd0e503..50b9a7ad 100644 --- a/pkg/proxy/handler.go +++ b/pkg/proxy/handler.go @@ -2,13 +2,17 @@ package proxy import ( "context" + "fmt" "io/ioutil" "net/http" "github.com/gorilla/mux" "github.com/wearefair/gurl/pkg/jsonpb" + "github.com/wearefair/gurl/pkg/k8" "github.com/wearefair/gurl/pkg/log" + "github.com/wearefair/gurl/pkg/util" "google.golang.org/grpc/metadata" + "k8s.io/client-go/tools/clientcmd" ) const ( @@ -50,7 +54,30 @@ func (p *Proxy) Handler(rw http.ResponseWriter, req *http.Request) { service := vars[ServiceKey] rpc := vars[RpcKey] + if service == "" || rpc == "" { + log.Error(fmt.Errorf("Service or RPC cannot be blank, service: %s, rpc: %s\n", service, rpc)) + rw.WriteHeader(http.StatusBadRequest) + return + } + + parsedURI, err := util.ParseURI(target) + if err != nil { + log.Error(err) + rw.WriteHeader(http.StatusBadRequest) + return + } + + // TODO: This is duplicated + var connector jsonpb.Connector + if parsedURI.Protocol == util.K8Protocol { + // Set up port forward, then send request + req := uriToPortForwardRequest(parsedURI) + + connector = jsonpb.NewK8Connector(k8Config(), req) + } + jsonpbReq := &jsonpb.Request{ + Connector: connector, Address: target, DialOptions: p.opts.DialOptions(), Service: service, @@ -81,3 +108,25 @@ func mergeHttpHeadersToMetadata(md metadata.MD, headers http.Header) metadata.MD return mdCopy } + +// Reads K8 config from default location, which is $HOME/.kube/config +func k8Config() clientcmd.ClientConfig { + // if you want to change the loading rules (which files in which order), you can do so here + loadingRules := clientcmd.NewDefaultClientConfigLoadingRules() + + // if you want to change override values or bind them to flags, there are methods to help you + configOverrides := &clientcmd.ConfigOverrides{} + + return clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, configOverrides) +} + +// TODO: This is duplicated logic +func uriToPortForwardRequest(uri *util.URI) k8.PortForwardRequest { + return k8.PortForwardRequest{ + Context: uri.Context, + // TODO: Make this namespace configurable via URI + Namespace: "default", + Service: uri.Host, + Port: uri.Port, + } +} diff --git a/pkg/proxy/proxy.go b/pkg/proxy/proxy.go index ed6420ba..9f312449 100644 --- a/pkg/proxy/proxy.go +++ b/pkg/proxy/proxy.go @@ -29,13 +29,11 @@ func New(cfg *Config) *Proxy { } return &Proxy{ + caller: cfg.Caller, opts: cfg.Options, proxyTargetHeader: cfg.ProxyTargetHeader, router: cfg.Router, server: s, - - // importPaths: cfg.ImportPaths, - // servicePaths: cfg.ServicePaths, } } diff --git a/pkg/util/uri.go b/pkg/util/uri.go index b39eac44..e1abdb04 100644 --- a/pkg/util/uri.go +++ b/pkg/util/uri.go @@ -41,7 +41,7 @@ const ( // Examples of valid URI for gurl: // http://localhost:50051/hello.world.package.Foo/Bar // k8://fake-context/foo-service:50051/hello.world.package.Foo/Bar - uriRegex = `((?P[a-z0-9]{2,5})(?:\:\/\/)((?P[0-9a-z._-]+)(?:\/))?)?(?P[0-9a-z-_.]+)(?:\:)(?P[0-9]{2,5})(?:\/)(?P[0-9a-zA-Z._-]+)(?:\/)(?P[0-9a-zA-Z._-]+)` + uriRegex = `((?P[a-z0-9]{2,5})(?:\:\/\/)((?P[0-9a-z._-]+)(?:\/))?)?(?P[0-9a-z-_.]+)(?:\:)(?P[0-9]{2,5})((?:\/)(?P[0-9a-zA-Z._-]+)(?:\/)(?P[0-9a-zA-Z._-]+))?` ) var ( diff --git a/pkg/util/uri_test.go b/pkg/util/uri_test.go index eae19a7d..305db3af 100644 --- a/pkg/util/uri_test.go +++ b/pkg/util/uri_test.go @@ -59,6 +59,17 @@ func TestParseURI(t *testing.T) { }, Err: nil, }, + // Parse input without service or RPC + { + Input: "k8://fake-context/k8-service:3000/", + Expected: &URI{ + Protocol: "k8", + Host: "k8-service", + Port: "3000", + Context: "fake-context", + }, + Err: nil, + }, // Input that's a completely hot garbage returns error { Input: "fakeNews", From a4884ba70bdc57de21d9c3053c63fda69438624d Mon Sep 17 00:00:00 2001 From: Cat Cai Date: Sun, 7 Jul 2019 09:56:20 -0700 Subject: [PATCH 14/14] Clean up duplication logic from K8 --- cmd/call/call.go | 34 +++++++++------------------------- pkg/k8/config.go | 15 +++++++++++++++ pkg/proxy/handler.go | 35 +++++++++-------------------------- 3 files changed, 33 insertions(+), 51 deletions(-) create mode 100644 pkg/k8/config.go diff --git a/cmd/call/call.go b/cmd/call/call.go index f1b98a74..a0f6d786 100644 --- a/cmd/call/call.go +++ b/cmd/call/call.go @@ -15,7 +15,6 @@ import ( "github.com/wearefair/gurl/pkg/options" "github.com/wearefair/gurl/pkg/util" "google.golang.org/grpc/metadata" - "k8s.io/client-go/tools/clientcmd" ) var ( @@ -74,9 +73,15 @@ func runCall(cmd *cobra.Command, args []string) error { var connector jsonpb.Connector if parsedURI.Protocol == util.K8Protocol { // Set up port forward, then send request - req := uriToPortForwardRequest(parsedURI) - - connector = jsonpb.NewK8Connector(k8Config(), req) + req := k8.PortForwardRequest{ + Context: parsedURI.Context, + // TODO: Make this namespace configurable via URI + Namespace: "default", + Service: parsedURI.Host, + Port: parsedURI.Port, + } + + connector = jsonpb.NewK8Connector(k8.DefaultConfig(), req) } // Set up the JSONPB client @@ -115,24 +120,3 @@ func runCall(cmd *cobra.Command, args []string) error { fmt.Printf("Response:\n%s\n", prettyResponse.String()) return nil } - -// Reads K8 config from default location, which is $HOME/.kube/config -func k8Config() clientcmd.ClientConfig { - // if you want to change the loading rules (which files in which order), you can do so here - loadingRules := clientcmd.NewDefaultClientConfigLoadingRules() - - // if you want to change override values or bind them to flags, there are methods to help you - configOverrides := &clientcmd.ConfigOverrides{} - - return clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, configOverrides) -} - -func uriToPortForwardRequest(uri *util.URI) k8.PortForwardRequest { - return k8.PortForwardRequest{ - Context: uri.Context, - // TODO: Make this namespace configurable via URI - Namespace: "default", - Service: uri.Host, - Port: uri.Port, - } -} diff --git a/pkg/k8/config.go b/pkg/k8/config.go new file mode 100644 index 00000000..38b849b4 --- /dev/null +++ b/pkg/k8/config.go @@ -0,0 +1,15 @@ +package k8 + +import "k8s.io/client-go/tools/clientcmd" + +// K8Config is just a helper function for getting the default K8 configs from +// $HOME/.kube/config +func DefaultConfig() clientcmd.ClientConfig { + // if you want to change the loading rules (which files in which order), you can do so here + loadingRules := clientcmd.NewDefaultClientConfigLoadingRules() + + // if you want to change override values or bind them to flags, there are methods to help you + configOverrides := &clientcmd.ConfigOverrides{} + + return clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, configOverrides) +} diff --git a/pkg/proxy/handler.go b/pkg/proxy/handler.go index 50b9a7ad..6b22ef34 100644 --- a/pkg/proxy/handler.go +++ b/pkg/proxy/handler.go @@ -12,7 +12,6 @@ import ( "github.com/wearefair/gurl/pkg/log" "github.com/wearefair/gurl/pkg/util" "google.golang.org/grpc/metadata" - "k8s.io/client-go/tools/clientcmd" ) const ( @@ -71,9 +70,15 @@ func (p *Proxy) Handler(rw http.ResponseWriter, req *http.Request) { var connector jsonpb.Connector if parsedURI.Protocol == util.K8Protocol { // Set up port forward, then send request - req := uriToPortForwardRequest(parsedURI) - - connector = jsonpb.NewK8Connector(k8Config(), req) + req := k8.PortForwardRequest{ + Context: parsedURI.Context, + // TODO: Make this namespace configurable via URI + Namespace: "default", + Service: parsedURI.Host, + Port: parsedURI.Port, + } + + connector = jsonpb.NewK8Connector(k8.DefaultConfig(), req) } jsonpbReq := &jsonpb.Request{ @@ -108,25 +113,3 @@ func mergeHttpHeadersToMetadata(md metadata.MD, headers http.Header) metadata.MD return mdCopy } - -// Reads K8 config from default location, which is $HOME/.kube/config -func k8Config() clientcmd.ClientConfig { - // if you want to change the loading rules (which files in which order), you can do so here - loadingRules := clientcmd.NewDefaultClientConfigLoadingRules() - - // if you want to change override values or bind them to flags, there are methods to help you - configOverrides := &clientcmd.ConfigOverrides{} - - return clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, configOverrides) -} - -// TODO: This is duplicated logic -func uriToPortForwardRequest(uri *util.URI) k8.PortForwardRequest { - return k8.PortForwardRequest{ - Context: uri.Context, - // TODO: Make this namespace configurable via URI - Namespace: "default", - Service: uri.Host, - Port: uri.Port, - } -}