change: use gRPC proxying instead of third_party package to route gRP…

…C requests (lavanet#440)

* change: use grpc proxying instead of third_party pkg

* adding reflection api to cosmossdk spec for the proxy to work

* finally fix all

* remove unneeded things

* fixing lint error

* fixing conflict

* resolving conflict

* fix lint

---------

Co-authored-by: testinginprod <[email protected]>
Co-authored-by: Ran Mishael <[email protected]>
Co-authored-by: unknown unknown <unknown@unknown>

config.yml

@@ -1,66 +1,66 @@
-build:
-  binary: "lavad"
-  main: "cmd/lavad"
-accounts:
-  - name: alice
-    coins: ["50000000000000ulava"]
-  - name: bob
-    coins: ["50000000000000ulava"]
-  - name: user1
-    coins: ["50000000000000ulava"]
-  - name: user2
-    coins: ["50000000000000ulava"]
-  - name: user3
-    coins: ["50000000000000ulava"]
-  - name: user4
-    coins: ["50000000000000ulava"]
-  - name: servicer1
-    coins: ["50000000000000ulava"]
-  - name: servicer2
-    coins: ["50000000000000ulava"]
-  - name: servicer3
-    coins: ["50000000000000ulava"]
-  - name: servicer4
-    coins: ["50000000000000ulava"]
-  - name: servicer5
-    coins: ["50000000000000ulava"]
-  - name: servicer6
-    coins: ["50000000000000ulava"]
-  - name: servicer7
-    coins: ["50000000000000ulava"]
-  - name: servicer8
-    coins: ["50000000000000ulava"]
-  - name: servicer9
-    coins: ["50000000000000ulava"]
-  - name: servicer10
-    coins: ["50000000000000ulava"]
-validator:
-  name: alice
-  staked: "100000000000ulava"
-client:
-  openapi:
-    path: "docs/static/openapi.yml"
-  vuex:
-    path: "vue/src/store"
-faucet:
-  name: bob
-  coins: ["100000ulava"]
-init:
-  app:
-    minimum-gas-prices: "0.000000001ulava"
-genesis:
-  chain_id: "lava"
-  app_state:
-    gov:
-      deposit_params:
-        min_deposit:
-        - amount: "10000000"
-          denom: "ulava"
-      voting_params:
-        voting_period: "3s"
-    staking:
-      params:
-        bond_denom: "ulava"
-    mint:
-      params:
+build:
+  binary: "lavad"
+  main: "cmd/lavad"
+accounts:
+  - name: alice
+    coins: ["50000000000000ulava"]
+  - name: bob
+    coins: ["50000000000000ulava"]
+  - name: user1
+    coins: ["50000000000000ulava"]
+  - name: user2
+    coins: ["50000000000000ulava"]
+  - name: user3
+    coins: ["50000000000000ulava"]
+  - name: user4
+    coins: ["50000000000000ulava"]
+  - name: servicer1
+    coins: ["50000000000000ulava"]
+  - name: servicer2
+    coins: ["50000000000000ulava"]
+  - name: servicer3
+    coins: ["50000000000000ulava"]
+  - name: servicer4
+    coins: ["50000000000000ulava"]
+  - name: servicer5
+    coins: ["50000000000000ulava"]
+  - name: servicer6
+    coins: ["50000000000000ulava"]
+  - name: servicer7
+    coins: ["50000000000000ulava"]
+  - name: servicer8
+    coins: ["50000000000000ulava"]
+  - name: servicer9
+    coins: ["50000000000000ulava"]
+  - name: servicer10
+    coins: ["50000000000000ulava"]
+validator:
+  name: alice
+  staked: "100000000000ulava"
+client:
+  openapi:
+    path: "docs/static/openapi.yml"
+  vuex:
+    path: "vue/src/store"
+faucet:
+  name: bob
+  coins: ["100000ulava"]
+init:
+  app:
+    minimum-gas-prices: "0.000000001ulava"
+genesis:
+  chain_id: "lava"
+  app_state:
+    gov:
+      deposit_params:
+        min_deposit:
+          - amount: "10000000"
+            denom: "ulava"
+      voting_params:
+        voting_period: "3s"
+    staking:
+      params:
+        bond_denom: "ulava"
+    mint:
+      params:
         mint_denom: "ulava"

cookbook/specs/spec_add_cosmossdk.json

@@ -2848,6 +2848,30 @@
                             }
                         ]
                     },
+                    {
+                        "name": "grpc.reflection.v1alpha.ServerReflection/ServerReflectionInfo",
+                        "block_parsing": {
+                            "parser_arg": [
+                                ""
+                            ],
+                            "parser_func": "EMPTY"
+                        },
+                        "compute_units": "10",
+                        "enabled": true,
+                        "api_interfaces": [
+                            {
+                                "category": {
+                                    "deterministic": true,
+                                    "local": false,
+                                    "subscription": false,
+                                    "stateful": 0
+                                },
+                                "interface": "grpc",
+                                "type": "",
+                                "extra_compute_units": "0"
+                            }
+                        ]
+                    },
                     {
                         "name": "cosmos.auth.v1beta1.Query/Accounts",
                         "block_parsing": {

grpcproxy/grpcproxy.go

@@ -0,0 +1,76 @@
+package grpcproxy
+
+import (
+	"context"
+	"net/http"
+
+	"github.com/improbable-eng/grpc-web/go/grpcweb"
+	"github.com/lavanet/lava/utils"
+	"golang.org/x/net/http2"
+	"golang.org/x/net/http2/h2c"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+type ProxyCallBack = func(ctx context.Context, method string, reqBody []byte) ([]byte, error)
+
+func NewGRPCProxy(cb ProxyCallBack) (*grpc.Server, *http.Server, error) {
+	s := grpc.NewServer(grpc.UnknownServiceHandler(makeProxyFunc(cb)), grpc.ForceServerCodec(rawBytesCodec{}))
+	wrappedServer := grpcweb.WrapServer(s)
+	handler := func(resp http.ResponseWriter, req *http.Request) {
+		// Set CORS headers
+		resp.Header().Set("Access-Control-Allow-Origin", "*")
+		resp.Header().Set("Access-Control-Allow-Headers", "Content-Type,x-grpc-web")
+
+		wrappedServer.ServeHTTP(resp, req)
+	}
+
+	httpServer := &http.Server{
+		Handler: h2c.NewHandler(http.HandlerFunc(handler), &http2.Server{}),
+	}
+	return s, httpServer, nil
+}
+
+func makeProxyFunc(callBack ProxyCallBack) grpc.StreamHandler {
+	return func(srv interface{}, stream grpc.ServerStream) error {
+		// currently the callback function does not account for headers.
+		methodName, ok := grpc.MethodFromServerStream(stream)
+		if !ok {
+			return status.Error(codes.Unavailable, "unable to get method name")
+		}
+		var reqBytes []byte
+		err := stream.RecvMsg(&reqBytes)
+		if err != nil {
+			return err
+		}
+		respBytes, err := callBack(stream.Context(), methodName[1:], reqBytes) // strip first '/' of the method name
+		if err != nil {
+			return err
+		}
+		return stream.SendMsg(respBytes)
+	}
+}
+
+type rawBytesCodec struct{}
+
+func (rawBytesCodec) Marshal(v interface{}) ([]byte, error) {
+	bytes, ok := v.([]byte)
+	if !ok {
+		return nil, utils.LavaFormatError("cannot encode type", nil, utils.Attribute{Key: "v", Value: v})
+	}
+	return bytes, nil
+}
+
+func (rawBytesCodec) Unmarshal(data []byte, v interface{}) error {
+	bufferPtr, ok := v.(*[]byte)
+	if !ok {
+		return utils.LavaFormatError("cannot decode into type", nil, utils.Attribute{Key: "v", Value: v})
+	}
+	*bufferPtr = data
+	return nil
+}
+
+func (rawBytesCodec) Name() string {
+	return "lava/grpc-proxy-codec"
+}

grpcproxy/grpcproxy_integration_test.go

@@ -0,0 +1 @@
+package grpcproxy_test

grpcproxy/grpcproxy_test.go

@@ -0,0 +1,55 @@
+package grpcproxy
+
+import (
+	"context"
+	"net"
+	"testing"
+
+	"github.com/lavanet/lava/grpcproxy/testproto"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+func newTestGRPCServer(t *testing.T, grpcSrv *grpc.Server) *grpc.ClientConn {
+	lis, err := net.Listen("tcp", ":0")
+	require.NoError(t, err)
+
+	go func() {
+		defer lis.Close()
+		if err := grpcSrv.Serve(lis); err != nil {
+			panic(err)
+		}
+	}()
+
+	// returns the listening port
+	conn, err := grpc.Dial(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	return conn
+}
+
+func TestGRPCProxy(t *testing.T) {
+	proxyGRPCSrv, _, err := NewGRPCProxy(func(ctx context.Context, method string, reqBody []byte) ([]byte, error) {
+		// the callback function just does echo proxying
+		req := new(testproto.TestRequest)
+		err := req.Unmarshal(reqBody)
+		require.NoError(t, err)
+		respBytes, err := (&testproto.TestResponse{Response: req.Request + "-callback"}).Marshal()
+		require.NoError(t, err)
+		return respBytes, nil
+	})
+	require.NoError(t, err)
+
+	client := testproto.NewTestClient(newTestGRPCServer(t, proxyGRPCSrv))
+	ctx := context.Background()
+
+	do := func() {
+		req := &testproto.TestRequest{Request: "echo"}
+		resp, err := client.Test(ctx, req)
+		require.NoError(t, err)
+		require.Equal(t, req.Request+"-callback", resp.Response)
+	}
+
+	do()
+	do()
+}