fix: connection errors (keploy#1727)

Signed-off-by: charankamarapu <kamarapucharan@gmail.com>
gouravkrosx · Mar 27, 2024 · 2ed1b79 · 2ed1b79
1 parent 0ce495f
commit 2ed1b79
Show file tree

Hide file tree

Showing 11 changed files with 33 additions and 38 deletions.
diff --git a/.github/workflows/sample-run.yml b/.github/workflows/sample-run.yml
@@ -15,7 +15,7 @@ jobs:
       - name: Setup Go
         uses: actions/setup-go@v2
         with:
-          go-version: "1.20"
+          go-version: "1.21"
 
       - name: Build Keploy
         run: |

diff --git a/pkg/core/proxy/integrations/generic/decode.go b/pkg/core/proxy/integrations/generic/decode.go
@@ -3,6 +3,7 @@ package generic
 
 import (
 	"context"
+	"io"
 	"net"
 	"time"
 
@@ -111,6 +112,9 @@ func decodeGeneric(ctx context.Context, logger *zap.Logger, reqBuf []byte, clien
 	case <-ctx.Done():
 		return ctx.Err()
 	case err := <-errCh:
+		if err == io.EOF {
+			return nil
+		}
 		return err
 	}
 }
diff --git a/pkg/core/proxy/integrations/generic/encode.go b/pkg/core/proxy/integrations/generic/encode.go
@@ -4,11 +4,13 @@ import (
 	"context"
 	"encoding/base64"
 	"errors"
-	"golang.org/x/sync/errgroup"
+	"io"
 	"net"
 	"strconv"
 	"time"
 
+	"golang.org/x/sync/errgroup"
+
 	"go.keploy.io/server/v2/pkg/core/proxy/integrations/util"
 	pUtil "go.keploy.io/server/v2/pkg/core/proxy/util"
 	"go.keploy.io/server/v2/pkg/models"
@@ -17,13 +19,6 @@ import (
 )
 
 func encodeGeneric(ctx context.Context, logger *zap.Logger, reqBuf []byte, clientConn, destConn net.Conn, mocks chan<- *models.Mock, _ models.OutgoingOptions) error {
-	//closing the destination conn
-	defer func(destConn net.Conn) {
-		err := destConn.Close()
-		if err != nil {
-			utils.LogError(logger, err, "failed to close the destination connection")
-		}
-	}(destConn)
 
 	var genericRequests []models.GenericPayload
 
@@ -203,6 +198,9 @@ func encodeGeneric(ctx context.Context, logger *zap.Logger, reqBuf []byte, clien
 			logger.Debug("the iteration for the generic response ends with no of genericReqs:" + strconv.Itoa(len(genericRequests)) + " and genericResps: " + strconv.Itoa(len(genericResponses)))
 			prevChunkWasReq = false
 		case err := <-errChan:
+			if err == io.EOF {
+				return nil
+			}
 			return err
 		}
 	}

diff --git a/pkg/core/proxy/integrations/grpc/encode.go b/pkg/core/proxy/integrations/grpc/encode.go
@@ -11,13 +11,6 @@ import (
 )
 
 func encodeGrpc(ctx context.Context, logger *zap.Logger, reqBuf []byte, clientConn, destConn net.Conn, mocks chan<- *models.Mock, _ models.OutgoingOptions) error {
-	//closing the destination conn
-	defer func(destConn net.Conn) {
-		err := destConn.Close()
-		if err != nil {
-			utils.LogError(logger, err, "failed to close the destination connection")
-		}
-	}(destConn)
 
 	// Send the client preface to the server. This should be the first thing sent from the client.
 	_, err := destConn.Write(reqBuf)
@@ -78,6 +71,9 @@ func encodeGrpc(ctx context.Context, logger *zap.Logger, reqBuf []byte, clientCo
 	case <-ctx.Done():
 		return ctx.Err()
 	case err := <-errCh:
+		if err == io.EOF {
+			return nil
+		}
 		return err
 	}
 	// This would practically be an infinite loop, unless the client closes the grpc conn

diff --git a/pkg/core/proxy/integrations/http/decode.go b/pkg/core/proxy/integrations/http/decode.go
@@ -182,6 +182,9 @@ func decodeHTTP(ctx context.Context, logger *zap.Logger, reqBuf []byte, clientCo
 	case <-ctx.Done():
 		return ctx.Err()
 	case err := <-errCh:
+		if err == io.EOF {
+			return nil
+		}
 		return err
 	}
 }
diff --git a/pkg/core/proxy/integrations/http/encode.go b/pkg/core/proxy/integrations/http/encode.go
@@ -18,13 +18,6 @@ import (
 
 // encodeHTTP function parses the HTTP request and response text messages to capture outgoing network calls as mocks.
 func encodeHTTP(ctx context.Context, logger *zap.Logger, reqBuf []byte, clientConn, destConn net.Conn, mocks chan<- *models.Mock, opts models.OutgoingOptions) error {
-	//closing the destination conn
-	defer func(destConn net.Conn) {
-		err := destConn.Close()
-		if err != nil {
-			utils.LogError(logger, err, "failed to close the destination connection")
-		}
-	}(destConn)
 
 	remoteAddr := destConn.RemoteAddr().(*net.TCPAddr)
 	destPort := uint(remoteAddr.Port)
@@ -255,6 +248,9 @@ func encodeHTTP(ctx context.Context, logger *zap.Logger, reqBuf []byte, clientCo
 	case <-ctx.Done():
 		return ctx.Err()
 	case err := <-errCh:
+		if err == io.EOF {
+			return nil
+		}
 		return err
 	}
 }
diff --git a/pkg/core/proxy/integrations/mongo/encode.go b/pkg/core/proxy/integrations/mongo/encode.go
@@ -17,13 +17,6 @@ import (
 )
 
 func encodeMongo(ctx context.Context, logger *zap.Logger, reqBuf []byte, clientConn, destConn net.Conn, mocks chan<- *models.Mock, _ models.OutgoingOptions) error {
-	//closing the destination conn
-	defer func(destConn net.Conn) {
-		err := destConn.Close()
-		if err != nil {
-			utils.LogError(logger, err, "failed to close the destination connection")
-		}
-	}(destConn)
 
 	errCh := make(chan error, 1)
 
@@ -272,6 +265,9 @@ func encodeMongo(ctx context.Context, logger *zap.Logger, reqBuf []byte, clientC
 	case <-ctx.Done():
 		return ctx.Err()
 	case err := <-errCh:
+		if err == io.EOF {
+			return nil
+		}
 		return err
 	}
 }

diff --git a/pkg/core/proxy/integrations/mysql/encode.go b/pkg/core/proxy/integrations/mysql/encode.go
@@ -14,13 +14,6 @@ import (
 )
 
 func encodeMySQL(ctx context.Context, logger *zap.Logger, clientConn, destConn net.Conn, mocks chan<- *models.Mock, _ models.OutgoingOptions) error {
-	//closing the destination conn
-	defer func(destConn net.Conn) {
-		err := destConn.Close()
-		if err != nil {
-			utils.LogError(logger, err, "failed to close the destination connection")
-		}
-	}(destConn)
 
 	var (
 		mysqlRequests  []models.MySQLRequest

diff --git a/pkg/core/proxy/integrations/postgres/v1/decode.go b/pkg/core/proxy/integrations/postgres/v1/decode.go
@@ -95,6 +95,9 @@ func decodePostgres(ctx context.Context, logger *zap.Logger, reqBuf []byte, clie
 	case <-ctx.Done():
 		return ctx.Err()
 	case err := <-errCh:
+		if err == io.EOF {
+			return nil
+		}
 		return err
 	}
 }

diff --git a/pkg/core/proxy/integrations/postgres/v1/encode.go b/pkg/core/proxy/integrations/postgres/v1/encode.go
@@ -3,6 +3,7 @@ package v1
 import (
 	"context"
 	"encoding/binary"
+	"io"
 	"net"
 	"strconv"
 	"time"
@@ -387,6 +388,9 @@ func encodePostgres(ctx context.Context, logger *zap.Logger, reqBuf []byte, clie
 			logger.Debug("the iteration for the postgres response ends with no of postgresReqs:" + strconv.Itoa(len(pgRequests)) + " and pgResps: " + strconv.Itoa(len(pgResponses)))
 			prevChunkWasReq = false
 		case err := <-errChan:
+			if err == io.EOF {
+				return nil
+			}
 			return err
 		}
 	}

diff --git a/pkg/core/proxy/util/util.go b/pkg/core/proxy/util/util.go
@@ -50,7 +50,9 @@ func ReadBuffConn(ctx context.Context, logger *zap.Logger, conn net.Conn, buffer
 				if ctx.Err() != nil { // to avoid sending buffer to closed channel if the context is cancelled
 					return
 				}
-				utils.LogError(logger, err, "failed to read the packet message in proxy")
+				if err != io.EOF {
+					utils.LogError(logger, err, "failed to read the packet message in proxy")
+				}
 				errChannel <- err
 				return
 			}