fix: ingress issues for grpc and http dual connection (#741)

* revert: revert changes for adding cmux

* fix: increase write timeout due to grpc errors

Author: sbchaos

go.mod

@@ -38,7 +38,6 @@ require (
 	github.com/schollz/progressbar/v3 v3.8.5
 	github.com/sirupsen/logrus v1.8.1
 	github.com/slack-go/slack v0.9.1
-	github.com/soheilhy/cmux v0.1.5
 	github.com/spf13/afero v1.8.2
 	github.com/spf13/cobra v1.2.1
 	github.com/spf13/viper v1.8.1

go.sum

@@ -1373,7 +1373,6 @@ github.com/smartystreets/goconvey v1.6.4 h1:fv0U8FUIMPNf1L9lnHLvLhgicrIVChEkdzIK
 github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
 github.com/snowflakedb/gosnowflake v1.6.3/go.mod h1:6hLajn6yxuJ4xUHZegMekpq9rnQbGJ7TMwXjgTmA6lg=
 github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM=
-github.com/soheilhy/cmux v0.1.5 h1:jjzc5WVemNEDTLwv9tlmemhC73tI08BNOIGwBOo10Js=
 github.com/soheilhy/cmux v0.1.5/go.mod h1:T7TcVDs9LWfQgPlPsdngu6I6QIoyIFZDDC6sNE1GqG0=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=

server/optimus.go

@@ -5,7 +5,6 @@ import (
 	"context"
 	"fmt"
 	"io"
-	"net"
 	"net/http"
 	"os"
 
@@ -16,7 +15,6 @@ import (
 	"github.com/odpf/salt/log"
 	"github.com/prometheus/client_golang/prometheus"
 	slackapi "github.com/slack-go/slack"
-	"github.com/soheilhy/cmux"
 	"google.golang.org/grpc"
 
 	"github.com/odpf/optimus/config"
@@ -60,7 +58,6 @@ type OptimusServer struct {
 	key    *[keyLength]byte
 
 	serverAddr string
-	mux        cmux.CMux
 	grpcServer *grpc.Server
 	httpServer *http.Server
 
@@ -85,7 +82,6 @@ func New(conf config.ServerConfig) (*OptimusServer, error) {
 		server.setupTelemetry,
 		server.setupAppKey,
 		server.setupDB,
-		server.setupMux,
 		server.setupGRPCServer,
 		server.setupHandlers,
 		server.setupMonitoring,
@@ -172,16 +168,6 @@ func (s *OptimusServer) setupDB() error {
 	return nil
 }
 
-func (s *OptimusServer) setupMux() error {
-	ln, err := net.Listen("tcp", s.serverAddr)
-	if err != nil {
-		return err
-	}
-
-	s.mux = cmux.New(ln)
-	return nil
-}
-
 func (s *OptimusServer) setupGRPCServer() error {
 	var err error
 	s.grpcServer, err = setupGRPCServer(s.logger)
@@ -195,40 +181,20 @@ func (s *OptimusServer) setupMonitoring() error {
 }
 
 func (s *OptimusServer) setupHTTPProxy() error {
-	srv, cleanup, err := prepareHTTPProxy(s.serverAddr)
+	srv, cleanup, err := prepareHTTPProxy(s.serverAddr, s.grpcServer)
 	s.httpServer = srv
 	s.cleanupFn = append(s.cleanupFn, cleanup)
 	return err
 }
 
 func (s *OptimusServer) startListening() {
 	// run our server in a goroutine so that it doesn't block to wait for termination requests
-	// We first match a request first for grpc
-	grpcLn := s.mux.MatchWithWriters(
-		cmux.HTTP2MatchHeaderFieldSendSettings("content-type", "application/grpc"))
-	// if it does not match grpc, we send the request to http
-	httpL := s.mux.Match(cmux.Any())
-
-	go func() {
-		// Start listening on grpc server
-		if err := s.grpcServer.Serve(grpcLn); err != nil {
-			s.logger.Info("shutting down grpc server")
-		}
-	}()
-
-	go func() {
-		// Start listening on http server
-		err := s.httpServer.Serve(httpL)
-		if err != nil {
-			s.logger.Info("Shutting http proxy")
-		}
-	}()
-
 	go func() {
-		// Start mux server
 		s.logger.Info("Listening at", "address", s.serverAddr)
-		if err := s.mux.Serve(); err != nil {
-			s.logger.Info("server error", "error", err)
+		if err := s.httpServer.ListenAndServe(); err != nil {
+			if err != http.ErrServerClosed {
+				s.logger.Fatal("server error", "error", err)
+			}
 		}
 	}()
 }

server/server.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"net/http"
 	"net/url"
+	"strings"
 	"time"
 
 	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
@@ -18,6 +19,8 @@ import (
 	"github.com/sirupsen/logrus"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
+	"golang.org/x/net/http2"
+	"golang.org/x/net/http2/h2c"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/health"
@@ -100,7 +103,7 @@ func setupGRPCServer(l log.Logger) (*grpc.Server, error) {
 	return grpcServer, nil
 }
 
-func prepareHTTPProxy(grpcAddr string) (*http.Server, func(), error) {
+func prepareHTTPProxy(grpcAddr string, grpcServer *grpc.Server) (*http.Server, func(), error) {
 	timeoutGrpcDialCtx, grpcDialCancel := context.WithTimeout(context.Background(), DialTimeout)
 	defer grpcDialCancel()
 
@@ -166,12 +169,29 @@ func prepareHTTPProxy(grpcAddr string) (*http.Server, func(), error) {
 
 	//nolint: gomnd
 	srv := &http.Server{
-		Handler:      baseMux,
+		Handler:      grpcHandlerFunc(grpcServer, baseMux),
 		Addr:         grpcAddr,
 		ReadTimeout:  5 * time.Second,
-		WriteTimeout: 60 * time.Second,
-		IdleTimeout:  120 * time.Second,
+		WriteTimeout: 30 * time.Minute, // FIXME: Creating issues for grpc connection
+		IdleTimeout:  5 * time.Minute,
 	}
 
 	return srv, cleanup, nil
 }
+
+// grpcHandlerFunc routes http1 calls to baseMux and http2 with grpc header to grpcServer.
+// Using a single port for proxying both http1 & 2 protocols will degrade http performance
+// but for our use-case the convenience per performance tradeoff is better suited
+// if in the future, this does become a bottleneck(which I highly doubt), we can break the service
+// into two ports, default port for grpc and default+1 for grpc-gateway proxy.
+// We can also use something like a connection multiplexer
+// https://github.com/soheilhy/cmux to achieve the same.
+func grpcHandlerFunc(grpcServer *grpc.Server, otherHandler http.Handler) http.Handler {
+	return h2c.NewHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.ProtoMajor == 2 && strings.Contains(r.Header.Get("Content-Type"), "application/grpc") {
+			grpcServer.ServeHTTP(w, r)
+		} else {
+			otherHandler.ServeHTTP(w, r)
+		}
+	}), &http2.Server{})
+}