add forwad implementation + tests

Author: eddyzags

go.mod

@@ -111,6 +111,7 @@ require (
 require (
 	github.com/docker/go-units v0.5.0
 	github.com/go-chi/chi/v5 v5.2.1
+	github.com/go-playground/validator/v10 v10.11.1
 	github.com/hashicorp/go-version v1.7.0
 )
 
@@ -257,6 +258,8 @@ require (
 	github.com/go-logfmt/logfmt v0.6.0 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/go-playground/locales v0.14.0 // indirect
+	github.com/go-playground/universal-translator v0.18.0 // indirect
 	github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
 	github.com/go-toolsmith/astcast v1.1.0 // indirect
 	github.com/go-toolsmith/astcopy v1.1.0 // indirect
@@ -343,6 +346,7 @@ require (
 	github.com/ldez/grignotin v0.9.0 // indirect
 	github.com/ldez/tagliatelle v0.7.1 // indirect
 	github.com/ldez/usetesting v0.4.2 // indirect
+	github.com/leodido/go-urn v1.2.1 // indirect
 	github.com/leonklingele/grouper v1.1.2 // indirect
 	github.com/lib/pq v1.10.9 // indirect
 	github.com/linxGnu/grocksdb v1.8.14 // indirect

go.sum

@@ -673,6 +673,7 @@ github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
 github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
+github.com/go-playground/assert/v2 v2.0.1 h1:MsBgLAaY856+nPRTKrp3/OZK38U/wa0CcBYNjji3q3A=
 github.com/go-playground/assert/v2 v2.0.1/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4=
 github.com/go-playground/locales v0.13.0/go.mod h1:taPMhCMXrRLJO55olJkUXHZBHCxTMfnGwq/HNwmWNS8=
 github.com/go-playground/locales v0.14.0 h1:u50s323jtVGugKlcYeyzC0etD1HifMjqmJqb8WugfUU=
@@ -1056,6 +1057,8 @@ github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxv
 github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
+github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
@@ -1360,6 +1363,8 @@ github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUc
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
+github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
+github.com/rogpeppe/go-internal v1.8.0/go.mod h1:WmiCO8CzOY8rg0OYDC4/i/2WRWAB6poM+XZ2dLUbcbE=
 github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
 github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
 github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
@@ -1631,6 +1636,7 @@ golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200728195943-123391ffb6de/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
+golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/crypto v0.13.0/go.mod h1:y6Z2r+Rw4iayiXXAIxJIDAJ1zMW4yaTpebo8fPOliYc=
 golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4=
 golang.org/x/crypto v0.15.0/go.mod h1:4ChreQoLWfG3xLDer1WdlH5NdlQ3+mwnQq1YTKY+72g=
@@ -1736,6 +1742,7 @@ golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96b
 golang.org/x/net v0.0.0-20210503060351-7fd8e65b6420/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
+golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
 golang.org/x/net v0.0.0-20220225172249-27dd8689420f/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
 golang.org/x/net v0.0.0-20220325170049-de3da57026de/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=

makefiles/relay.mk

@@ -65,4 +65,5 @@ relayminer_forward_http_rest: ## Forward request to the rest service.
 .PHONY: relayminer_forward_request_websocket_anvilws
 relayminer_forward_request_websocket_anvilws: ## Forward websocket request to the anvilws service.
 	@websocat ws://localhost:10001/services/anvilws/forward \
-		-H "token: 8cc09793290cd64d8a9bc80eaae4fbeef5f7cf797b0c70e078d2a5b81d74f12c"
+		-H "token: 8cc09793290cd64d8a9bc80eaae4fbeef5f7cf797b0c70e078d2a5b81d74f12c" \
+		-H "Rpc-Type: 2" # RPCType_WEBSOCKET

pkg/relayer/forward_server.go

@@ -0,0 +1,59 @@
+package relayer
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"net/http"
+
+	"github.com/go-chi/chi/v5"
+)
+
+// ServeForward exposes a forward HTTP server for administrators to send request to
+// specific service.
+func (rel *relayMiner) ServeForward(ctx context.Context, network, addr, token string) error {
+	ln, err := net.Listen(network, addr)
+	if err != nil {
+		return fmt.Errorf("net listen: %w", err)
+	}
+
+	muxRouter := chi.NewRouter()
+	muxRouter.HandleFunc("/services/{service_id}/forward", rel.newForwardHandlerFn(ctx, token))
+
+	go func() {
+		if err := http.Serve(ln, muxRouter); err != nil {
+			rel.logger.Error().Err(err).
+				Msg("unexpected error occurred while serving forward server")
+			return
+		}
+	}()
+
+	return nil
+}
+
+func (rel *relayMiner) newForwardHandlerFn(ctx context.Context, token string) http.HandlerFunc {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		reqToken := r.Header.Get("token")
+		if reqToken != token {
+			w.WriteHeader(http.StatusUnauthorized)
+			return
+		}
+
+		serviceID := chi.URLParam(r, "service_id")
+		if serviceID == "" {
+			rel.logger.Error().Msg("service id not found in URL while forwarding request")
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+
+		rel.logger.Debug().Str("service_id", serviceID).
+			Msg("forwarding request to supplier...")
+
+		if err := rel.relayerProxy.Forward(ctx, serviceID, w, r); err != nil {
+			rel.logger.Error().Err(err).
+				Msg("unable to forward request")
+			w.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+	})
+}

pkg/relayer/interface.go

@@ -7,6 +7,7 @@ package relayer
 
 import (
 	"context"
+	"net/http"
 
 	"github.com/pokt-network/smt"
 
@@ -59,6 +60,9 @@ type RelayerProxy interface {
 
 	// PingAll tests the connectivity between all the managed relay servers and their respective backend URLs.
 	PingAll(ctx context.Context) error
+
+	// Forward sends a request to the relay server for the given service ID.
+	Forward(ctx context.Context, serviceID string, w http.ResponseWriter, r *http.Request) error
 }
 
 type RelayerProxyOption func(RelayerProxy)
@@ -101,6 +105,9 @@ type RelayServer interface {
 
 	// Ping tests the connection between the relay server and its backend URL.
 	Ping(ctx context.Context) error
+
+	// Forward sends a request to the supplier service for the given service ID.
+	Forward(ctx context.Context, serviceID string, w http.ResponseWriter, r *http.Request) error
 }
 
 // RelayServers aggregates a slice of RelayServer interface.

pkg/relayer/proxy/async.go

@@ -2,11 +2,13 @@ package proxy
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 
 	"github.com/gorilla/websocket"
 
 	"github.com/pokt-network/poktroll/pkg/polylog"
+	"github.com/pokt-network/poktroll/pkg/relayer/config"
 	proxyws "github.com/pokt-network/poktroll/pkg/relayer/proxy/websockets"
 	sharedtypes "github.com/pokt-network/poktroll/x/shared/types"
 )
@@ -121,3 +123,62 @@ func (server *relayMinerHTTPServer) handleAsyncConnection(
 
 	return nil
 }
+
+// forwardAsyncConnection instantiates two websocket connections that:
+// - receive and forward message from the client to the supplier (backend URL).
+// - receive and forward message from the supplier (backend URL) to the client.
+func (server *relayMinerHTTPServer) forwardAsyncConnection(ctx context.Context, supplierConfig *config.RelayMinerSupplierConfig, w http.ResponseWriter, req *http.Request) error {
+	upgrader := websocket.Upgrader{
+		CheckOrigin: func(r *http.Request) bool { return true },
+	}
+
+	clientConn, err := upgrader.Upgrade(w, req, nil)
+	if err != nil {
+		return fmt.Errorf("client connection upgrade client to ws: %w", err)
+	}
+
+	serviceConn, err := proxyws.ConnectServiceBackend(supplierConfig.ServiceConfig.BackendUrl, supplierConfig.ServiceConfig.GetHeadersHTTP())
+	if err != nil {
+		return fmt.Errorf("service connection upgrade to ws: %w", err)
+	}
+
+	forwardFn := func(from, to *websocket.Conn) {
+		defer from.Close()
+		defer to.Close()
+
+		isNormalCloseConnection := func(err error) bool {
+			return websocket.IsCloseError(err,
+				websocket.CloseNormalClosure,
+				websocket.CloseGoingAway,
+				websocket.CloseAbnormalClosure)
+		}
+
+		for {
+			msgType, msg, err := from.ReadMessage()
+			if err != nil {
+				if isNormalCloseConnection(err) {
+					return
+				}
+
+				server.logger.Error().
+					Msgf("from read message: %w", err)
+				return
+			}
+
+			if err := to.WriteMessage(msgType, msg); err != nil {
+				if isNormalCloseConnection(err) {
+					return
+				}
+
+				server.logger.Error().
+					Msgf("to write message: %w", err)
+				return
+			}
+		}
+	}
+
+	go forwardFn(clientConn, serviceConn)
+	forwardFn(serviceConn, clientConn)
+
+	return nil
+}

pkg/relayer/proxy/errors.go

@@ -20,4 +20,5 @@ var (
 	ErrRelayerProxyResponseLimitExceeded     = sdkerrors.Register(codespace, 12, "response limit exceed")
 	ErrRelayerProxyRequestLimitExceeded      = sdkerrors.Register(codespace, 13, "request limit exceed")
 	ErrRelayerProxyUnmarshalingRelayRequest  = sdkerrors.Register(codespace, 14, "failed to unmarshal relay request")
+	ErrRelayerProxyServiceIDNotFound         = sdkerrors.Register(codespace, 15, "service id not found")
 )

pkg/relayer/proxy/http_server.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"net"
 	"net/http"
-	"strconv"
 	"time"
 
 	codectypes "github.com/cosmos/cosmos-sdk/codec/types"
@@ -14,7 +13,6 @@ import (
 	"github.com/pokt-network/poktroll/pkg/relayer"
 	"github.com/pokt-network/poktroll/pkg/relayer/config"
 	"github.com/pokt-network/poktroll/x/service/types"
-	sharedtypes "github.com/pokt-network/poktroll/x/shared/types"
 )
 
 // rpcTypeHeader is the header key for the RPC type, provided by the client.
@@ -137,11 +135,27 @@ func (server *relayMinerHTTPServer) Start(ctx context.Context) error {
 	return server.server.Serve(listener)
 }
 
+// Forward sends request to the appropriate service.
+// - It checks if the service id is managed by the relayminer.
 // Stop terminates the service server and returns an error if it fails.
 func (server *relayMinerHTTPServer) Stop(ctx context.Context) error {
 	return server.server.Shutdown(ctx)
 }
 
+// - It checks wether it needs to forward a websocket connection or send a http request.
+func (server *relayMinerHTTPServer) Forward(ctx context.Context, serviceID string, w http.ResponseWriter, req *http.Request) error {
+	supplierConfig, ok := server.serverConfig.SupplierConfigsMap[serviceID]
+	if !ok {
+		return ErrRelayerProxyServiceIDNotFound.Wrapf("service ID: %s", serviceID)
+	}
+
+	if isWebSocketRequest(req) {
+		return server.forwardAsyncConnection(ctx, supplierConfig, w, req)
+	} else {
+		return server.forwardHTTP(ctx, supplierConfig, w, req)
+	}
+}
+
 // ServeHTTP listens for incoming relay requests. It implements the respective
 // method of the http.Handler interface. It is called by http.ListenAndServe()
 // when relayMinerHTTPServer is used as an http.Handler with an http.Server.
@@ -156,13 +170,6 @@ func (server *relayMinerHTTPServer) ServeHTTP(writer http.ResponseWriter, reques
 		"remote_addr", request.RemoteAddr,
 	)
 
-	// isWebSocketRequest checks if the request is trying to upgrade to WebSocket.
-	isWebSocketRequest := func(r *http.Request) bool {
-		// The request must have the "Rpc-Type" header set to "websocket".
-		// This will be handled in the client, likely a PATH gateway.
-		return r.Header.Get(RPCTypeHeader) == strconv.Itoa(int(sharedtypes.RPCType_WEBSOCKET))
-	}
-
 	// Determine whether the request is upgrading to websocket.
 	if isWebSocketRequest(request) {
 		logger.ProbabilisticDebugInfo(relayProbabilisticDebugProb).Msg("🔍 detected asynchronous relay request")