Gurl Proxy

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.
+
+

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 (
@@ -38,10 +37,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")
@@ -68,33 +69,44 @@ 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
+		req := k8.PortForwardRequest{
+			Context: parsedURI.Context,
+			// TODO: Make this namespace configurable via URI
+			Namespace: "default",
+			Service:   parsedURI.Host,
+			Port:      parsedURI.Port,
 		}
-		defer pf.Close()
 
-		address = fmt.Sprintf("localhost:%s", pf.LocalPort())
+		connector = jsonpb.NewK8Connector(k8.DefaultConfig(), req)
 	}
 
+	// Set up the JSONPB client
 	cfg := &jsonpb.Config{
-		Address:      address,
 		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
-	response, err := client.Call(callOptions.ContextWithOptions(context.Background()), parsedURI.Service, parsedURI.RPC, []byte(data))
+	address := fmt.Sprintf("%s:%s", parsedURI.Host, parsedURI.Port)
+	req := &jsonpb.Request{
+		Connector:   connector,
+		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)
 	}
@@ -108,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,
-	}
-}

cmd/proxy/proxy.go

@@ -0,0 +1,45 @@
+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"
+)
+
+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.DefaultConfig()
+	//	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)
+	return proxySrv.Run()
+}

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() {

pkg/jsonpb/client.go

@@ -7,46 +7,46 @@ 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
 type Client struct {
-	stub grpcdynamic.Stub
-	// TODO: Might want to turn this into an interface?
 	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 {
 		return nil, err
 	}
 
 	return &Client{
-		stub:      grpcdynamic.NewStub(conn),
 		collector: protobuf.NewCollector(descriptors),
 	}, nil
 }
 
-// Call takes in a context, service, RPC, and message as JSON string to convert to protobuf and
-// send across the wire.
-func (c *Client) Call(ctx context.Context, service, rpc string, rawMsg []byte) ([]byte, error) {
-	serviceDescriptor, err := c.collector.GetService(service)
+// Invoke makes a unary call across the wire.
+func (c *Client) Invoke(ctx context.Context, req *Request) ([]byte, error) {
+	// Run any connection logic
+	conn, err := req.Connect()
+	if err != nil {
+		return nil, err
+	}
+	defer req.Close()
+
+	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(rpc)
+	methodDescriptor := serviceDescriptor.FindMethodByName(req.RPC)
 	if methodDescriptor == nil {
-		err := fmt.Errorf("No method %s found", service)
+		err := fmt.Errorf("No method %s found", req.Service)
 		return nil, err
 	}
 
@@ -58,7 +58,7 @@ func (c *Client) Call(ctx context.Context, service, rpc string, rawMsg []byte) (
 		return nil, err
 	}
 
-	message, err := protobuf.Construct(messageDescriptor, rawMsg)
+	message, err := protobuf.Construct(messageDescriptor, req.Message)
 	if err != nil {
 		return nil, err
 	}
@@ -69,7 +69,7 @@ func (c *Client) Call(ctx context.Context, service, rpc string, rawMsg []byte) (
 	methodProto.ClientStreaming = &disableStreaming
 	methodProto.ServerStreaming = &disableStreaming
 
-	response, err := c.stub.InvokeRpc(ctx, methodDescriptor, message)
+	response, err := stub.InvokeRpc(ctx, methodDescriptor, message)
 	if err != nil {
 		return nil, err
 	}

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

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
+}