GODRIVER-1593 Implement a proxy dialer for integration tests (#378)

Author: Divjot Arora

.evergreen/config.yml

@@ -838,6 +838,20 @@ tasks:
           AUTH: "auth"
           SSL: "ssl"
 
+  - name: test-standalone-auth-nossl
+    tags: ["test", "standalone", "authssl"]
+    commands:
+      - func: bootstrap-mongo-orchestration
+        vars:
+          TOPOLOGY: "server"
+          AUTH: "auth"
+          SSL: "nossl"
+      - func: run-tests
+        vars:
+          TOPOLOGY: "server"
+          AUTH: "auth"
+          SSL: "nossl"
+
   - name: test-standalone-auth-ssl-snappy-compression
     tags: ["test", "standalone", "authssl", "compression", "snappy"]
     commands:
@@ -1367,6 +1381,20 @@ tasks:
           SSL: "ssl"
           MONGO_GO_DRIVER_COMPRESSOR: "zstd"
 
+  - name: test-sharded-auth-nossl
+    tags: ["test", "sharded", "authssl"]
+    commands:
+      - func: bootstrap-mongo-orchestration
+        vars:
+          TOPOLOGY: "sharded_cluster"
+          AUTH: "auth"
+          SSL: "nossl"
+      - func: run-tests
+        vars:
+          TOPOLOGY: "sharded_cluster"
+          AUTH: "auth"
+          SSL: "nossl"
+
   - name: test-enterprise-auth-plain
     tags: ["test", "enterprise-auth"]
     commands:

mongo/integration/client_test.go

@@ -7,15 +7,10 @@
 package integration
 
 import (
-	"bytes"
-	"context"
 	"fmt"
-	"io"
-	"net"
 	"path"
 	"reflect"
 	"strings"
-	"sync"
 	"testing"
 	"time"
 
@@ -29,7 +24,6 @@ import (
 	"go.mongodb.org/mongo-driver/mongo/readpref"
 	"go.mongodb.org/mongo-driver/x/bsonx/bsoncore"
 	"go.mongodb.org/mongo-driver/x/mongo/driver"
-	"go.mongodb.org/mongo-driver/x/mongo/driver/drivertest"
 	"go.mongodb.org/mongo-driver/x/mongo/driver/wiremessage"
 )
 
@@ -381,37 +375,31 @@ func TestClient(t *testing.T) {
 	})
 
 	testAppName := "foo"
-	hosts := options.Client().ApplyURI(mt.ConnString()).Hosts
-	appNameProxyDialer := newProxyDialer()
-	appNameDialerOpts := options.Client().
-		SetDialer(appNameProxyDialer).
-		SetHosts(hosts[:1]).
-		SetDirect(true).
+	appNameClientOpts := options.Client().
 		SetAppName(testAppName)
 	appNameMtOpts := mtest.NewOptions().
-		ClientOptions(appNameDialerOpts).
-		Topologies(mtest.Single).
-		Auth(false) // Can't run with auth because the proxy dialer won't work with TLS enabled.
+		ClientType(mtest.Proxy).
+		ClientOptions(appNameClientOpts).
+		Topologies(mtest.Single)
 	mt.RunOpts("app name is always sent", appNameMtOpts, func(mt *mtest.T) {
 		err := mt.Client.Ping(mtest.Background, mtest.PrimaryRp)
 		assert.Nil(mt, err, "Ping error: %v", err)
 
-		msgPairs := appNameProxyDialer.messages
+		msgPairs := mt.GetProxiedMessages()
 		assert.True(mt, len(msgPairs) >= 2, "expected at least 2 events sent, got %v", len(msgPairs))
 
 		// First two messages should be connection handshakes: one for the heartbeat connection and the other for the
 		// application connection.
 		for idx, pair := range msgPairs[:2] {
-			cmd, err := drivertest.GetCommandFromQueryWireMessage(pair.sent)
-			assert.Nil(mt, err, "GetCommandFromQueryWireMessage error at index %d: %v", idx, err)
-			heartbeatCmdName := cmd.Index(0).Key()
-			assert.Equal(mt, "isMaster", heartbeatCmdName,
-				"expected command name isMaster at index %d, got %v", idx, heartbeatCmdName)
-
-			appNameVal, err := cmd.LookupErr("client", "application", "name")
-			assert.Nil(mt, err, "expected command %s at index %d to contain app name", cmd, idx)
+			assert.Equal(mt, pair.CommandName, "isMaster", "expected command name isMaster at index %d, got %s", idx,
+				pair.CommandName)
+
+			sent := pair.Sent
+			appNameVal, err := sent.Command.LookupErr("client", "application", "name")
+			assert.Nil(mt, err, "expected command %s at index %d to contain app name", sent.Command, idx)
 			appName := appNameVal.StringValue()
-			assert.Equal(mt, testAppName, appName, "expected app name %v at index %d, got %v", testAppName, idx, appName)
+			assert.Equal(mt, testAppName, appName, "expected app name %v at index %d, got %v", testAppName, idx,
+				appName)
 		}
 	})
 
@@ -446,101 +434,3 @@ type proxyMessage struct {
 	sent          wiremessage.WireMessage
 	received      wiremessage.WireMessage
 }
-
-// proxyDialer is a ContextDialer implementation that wraps a net.Dialer and records the messages sent and received
-// using connections created through it.
-type proxyDialer struct {
-	*net.Dialer
-	sync.Mutex
-	messages []proxyMessage
-	sentMap  sync.Map
-}
-
-var _ options.ContextDialer = (*proxyDialer)(nil)
-
-func newProxyDialer() *proxyDialer {
-	return &proxyDialer{
-		Dialer: &net.Dialer{Timeout: 30 * time.Second},
-	}
-}
-
-// DialContext creates a new proxyConnection.
-func (p *proxyDialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
-	netConn, err := p.Dialer.DialContext(ctx, network, address)
-	if err != nil {
-		return netConn, err
-	}
-
-	proxy := &proxyConn{
-		Conn:           netConn,
-		dialer:         p,
-		currentReading: bytes.NewBuffer(nil),
-	}
-	return proxy, nil
-}
-
-// storeSentMessage stores a copy of the wire message being sent to the server.
-func (p *proxyDialer) storeSentMessage(msg []byte) {
-	p.Lock()
-	defer p.Unlock()
-
-	msgCopy := make(wiremessage.WireMessage, len(msg))
-	copy(msgCopy, msg)
-
-	_, requestID, _, _, _, _ := wiremessage.ReadHeader(msgCopy)
-	p.sentMap.Store(requestID, msgCopy)
-}
-
-// storeReceivedMessage stores a copy of the wire message being received from the server.
-func (p *proxyDialer) storeReceivedMessage(msg []byte) {
-	p.Lock()
-	defer p.Unlock()
-
-	msgCopy := make(wiremessage.WireMessage, len(msg))
-	copy(msgCopy, msg)
-
-	_, _, responseTo, _, _, _ := wiremessage.ReadHeader(msgCopy)
-	sentMsg, _ := p.sentMap.Load(responseTo)
-	p.sentMap.Delete(responseTo)
-
-	proxyMsg := proxyMessage{
-		sent:     sentMsg.(wiremessage.WireMessage),
-		received: msgCopy,
-	}
-	p.messages = append(p.messages, proxyMsg)
-}
-
-// proxyConn is a net.Conn that wraps a network connection. All messages sent/received through a proxyConn are stored
-// in the associated proxyDialer and are forwarded over the wrapped connection.
-type proxyConn struct {
-	net.Conn
-	dialer         *proxyDialer
-	currentReading *bytes.Buffer // The current message being read.
-}
-
-// Write stores the given message in the proxyDialer associated with this connection and forwards the message to the
-// server.
-func (pc *proxyConn) Write(msg []byte) (n int, err error) {
-	pc.dialer.storeSentMessage(msg)
-	return pc.Conn.Write(msg)
-}
-
-// Read reads the message from the server into the given buffer and stores the read message in the proxyDialer
-// associated with this connection.
-func (pc *proxyConn) Read(buffer []byte) (int, error) {
-	n, err := pc.Conn.Read(buffer)
-	if err != nil {
-		return n, err
-	}
-
-	_, err = io.Copy(pc.currentReading, bytes.NewReader(buffer))
-	if err != nil {
-		return 0, fmt.Errorf("error copying to mock: %v", err)
-	}
-	if len(buffer) != 4 {
-		pc.dialer.storeReceivedMessage(pc.currentReading.Bytes())
-		pc.currentReading.Reset()
-	}
-
-	return n, err
-}

mongo/integration/mtest/mongotest.go

@@ -86,14 +86,15 @@ type T struct {
 	mockDeployment   *mockDeployment // nil if the test is not being run against a mock
 	mockResponses    []bson.D
 	createdColls     []*Collection // collections created in this test
+	proxyDialer      *proxyDialer
 	dbName, collName string
 	failPointNames   []string
 	minServerVersion string
 	maxServerVersion string
 	validTopologies  []TopologyKind
 	auth             *bool
-	ssl              *bool
 	enterprise       *bool
+	ssl              *bool
 	collCreateOpts   bson.D
 	connsCheckedOut  int // net number of connections checked out during test execution
 
@@ -342,6 +343,15 @@ func (t *T) FilterFailedEvents(filter func(*event.CommandFailedEvent) bool) {
 	t.failed = newEvents
 }
 
+// GetProxiedMessages returns the messages proxied to the server by the test. If the client type is not Proxy, this
+// returns nil.
+func (t *T) GetProxiedMessages() []*ProxyMessage {
+	if t.proxyDialer == nil {
+		return nil
+	}
+	return t.proxyDialer.messages
+}
+
 // ClearEvents clears the existing command monitoring events.
 func (t *T) ClearEvents() {
 	t.started = t.started[:0]
@@ -580,14 +590,6 @@ func (t *T) createTestClient() {
 
 	var err error
 	switch t.clientType {
-	case Default:
-		// only specify URI if the deployment is not set to avoid setting topology/server options along with the deployment
-		var uriOpts *options.ClientOptions
-		if clientOpts.Deployment == nil {
-			uriOpts = options.Client().ApplyURI(testContext.connString.Original)
-		}
-		// Specify the URI-based options first so the test can override them.
-		t.Client, err = mongo.NewClient(uriOpts, clientOpts)
 	case Pinned:
 		// pin to first mongos
 		pinnedHostList := []string{testContext.connString.Hosts[0]}
@@ -599,6 +601,24 @@ func (t *T) createTestClient() {
 		t.mockDeployment = newMockDeployment()
 		clientOpts.Deployment = t.mockDeployment
 		t.Client, err = mongo.NewClient(clientOpts)
+	case Proxy:
+		t.proxyDialer = newProxyDialer()
+		clientOpts.SetDialer(t.proxyDialer)
+
+		// After setting the Dialer, fall-through to the Default case to apply the correct URI
+		fallthrough
+	case Default:
+		// Use a different set of options to specify the URI because clientOpts may already have a URI or host seedlist
+		// specified.
+		var uriOpts *options.ClientOptions
+		if clientOpts.Deployment == nil {
+			// Only specify URI if the deployment is not set to avoid setting topology/server options along with the
+			// deployment.
+			uriOpts = options.Client().ApplyURI(testContext.connString.Original)
+		}
+
+		// Pass in uriOpts first so clientOpts wins if there are any conflicting settings.
+		t.Client, err = mongo.NewClient(uriOpts, clientOpts)
 	}
 	if err != nil {
 		t.Fatalf("error creating client: %v", err)

mongo/integration/mtest/options.go

@@ -33,6 +33,13 @@ const (
 	Pinned
 	// Mock specifies a client that communicates with a mock deployment.
 	Mock
+	// Proxy specifies a client that proxies messages to the server and also stores parsed copies. The proxied
+	// messages can be retrieved via T.GetProxiedMessages or T.GetRawProxiedMessages.
+	Proxy
+)
+
+var (
+	falseBool = false
 )
 
 // RunOnBlock describes a constraint for a test.
@@ -125,10 +132,15 @@ func (op *Options) DatabaseName(dbName string) *Options {
 }
 
 // ClientType specifies the type of client that should be created for a test. This option will be propagated to all
-// sub-tests.
+// sub-tests. If the provided ClientType is Proxy, the SSL(false) option will be also be added because the internal
+// proxy dialer and connection types do not support SSL.
 func (op *Options) ClientType(ct ClientType) *Options {
 	op.optFuncs = append(op.optFuncs, func(t *T) {
 		t.clientType = ct
+
+		if ct == Proxy {
+			t.ssl = &falseBool
+		}
 	})
 	return op
 }