From d71481e770066f86bf7ee4fb5baf0b1bae7d1a4d Mon Sep 17 00:00:00 2001 From: Qingyang Hu <103950869+qingyang-hu@users.noreply.github.com> Date: Tue, 29 Aug 2023 18:07:18 -0400 Subject: [PATCH 01/15] GODRIVER-2423 Include pinned connections in WaitQueueTimeoutError only for load balanced clusters. (#1353) --- x/mongo/driver/topology/errors.go | 38 ++++++++++++++++++++----------- x/mongo/driver/topology/pool.go | 25 +++++++++++++++----- x/mongo/driver/topology/server.go | 1 + 3 files changed, 45 insertions(+), 19 deletions(-) diff --git a/x/mongo/driver/topology/errors.go b/x/mongo/driver/topology/errors.go index 4f7b485405..7ce41864e6 100644 --- a/x/mongo/driver/topology/errors.go +++ b/x/mongo/driver/topology/errors.go @@ -9,6 +9,7 @@ package topology import ( "context" "fmt" + "time" "go.mongodb.org/mongo-driver/mongo/description" ) @@ -69,11 +70,17 @@ func (e ServerSelectionError) Unwrap() error { // WaitQueueTimeoutError represents a timeout when requesting a connection from the pool type WaitQueueTimeoutError struct { - Wrapped error - PinnedCursorConnections uint64 - PinnedTransactionConnections uint64 - maxPoolSize uint64 - totalConnectionCount int + Wrapped error + pinnedConnections *pinnedConnections + maxPoolSize uint64 + totalConnections int + availableConnections int + waitDuration time.Duration +} + +type pinnedConnections struct { + cursorConnections uint64 + transactionConnections uint64 } // Error implements the error interface. @@ -95,14 +102,19 @@ func (w WaitQueueTimeoutError) Error() string { ) } - return fmt.Sprintf( - "%s; maxPoolSize: %d, connections in use by cursors: %d"+ - ", connections in use by transactions: %d, connections in use by other operations: %d", - errorMsg, - w.maxPoolSize, - w.PinnedCursorConnections, - w.PinnedTransactionConnections, - uint64(w.totalConnectionCount)-w.PinnedCursorConnections-w.PinnedTransactionConnections) + msg := fmt.Sprintf("%s; total connections: %d, maxPoolSize: %d, ", errorMsg, w.totalConnections, w.maxPoolSize) + if pinnedConnections := w.pinnedConnections; pinnedConnections != nil { + openConnectionCount := uint64(w.totalConnections) - + pinnedConnections.cursorConnections - + pinnedConnections.transactionConnections + msg += fmt.Sprintf("connections in use by cursors: %d, connections in use by transactions: %d, connections in use by other operations: %d, ", + pinnedConnections.cursorConnections, + pinnedConnections.transactionConnections, + openConnectionCount, + ) + } + msg += fmt.Sprintf("idle connections: %d, wait duration: %s", w.availableConnections, w.waitDuration.String()) + return msg } // Unwrap returns the underlying error. diff --git a/x/mongo/driver/topology/pool.go b/x/mongo/driver/topology/pool.go index 5d2369352e..6e150344db 100644 --- a/x/mongo/driver/topology/pool.go +++ b/x/mongo/driver/topology/pool.go @@ -74,6 +74,7 @@ type poolConfig struct { MaxConnecting uint64 MaxIdleTime time.Duration MaintainInterval time.Duration + LoadBalanced bool PoolMonitor *event.PoolMonitor Logger *logger.Logger handshakeErrFn func(error, uint64, *primitive.ObjectID) @@ -93,6 +94,7 @@ type pool struct { minSize uint64 maxSize uint64 maxConnecting uint64 + loadBalanced bool monitor *event.PoolMonitor logger *logger.Logger @@ -206,6 +208,7 @@ func newPool(config poolConfig, connOpts ...ConnectionOption) *pool { minSize: config.MinPoolSize, maxSize: config.MaxPoolSize, maxConnecting: maxConnecting, + loadBalanced: config.LoadBalanced, monitor: config.PoolMonitor, logger: config.Logger, handshakeErrFn: config.handshakeErrFn, @@ -574,6 +577,7 @@ func (p *pool) checkOut(ctx context.Context) (conn *connection, err error) { p.stateMu.RUnlock() // Wait for either the wantConn to be ready or for the Context to time out. + start := time.Now() select { case <-w.ready: if w.err != nil { @@ -615,6 +619,8 @@ func (p *pool) checkOut(ctx context.Context) (conn *connection, err error) { } return w.conn, nil case <-ctx.Done(): + duration := time.Since(start) + if mustLogPoolMessage(p) { keysAndValues := logger.KeyValues{ logger.KeyReason, logger.ReasonConnCheckoutFailedTimout, @@ -632,13 +638,20 @@ func (p *pool) checkOut(ctx context.Context) (conn *connection, err error) { }) } - return nil, WaitQueueTimeoutError{ - Wrapped: ctx.Err(), - PinnedCursorConnections: atomic.LoadUint64(&p.pinnedCursorConnections), - PinnedTransactionConnections: atomic.LoadUint64(&p.pinnedTransactionConnections), - maxPoolSize: p.maxSize, - totalConnectionCount: p.totalConnectionCount(), + err := WaitQueueTimeoutError{ + Wrapped: ctx.Err(), + maxPoolSize: p.maxSize, + totalConnections: p.totalConnectionCount(), + availableConnections: p.availableConnectionCount(), + waitDuration: duration, + } + if p.loadBalanced { + err.pinnedConnections = &pinnedConnections{ + cursorConnections: atomic.LoadUint64(&p.pinnedCursorConnections), + transactionConnections: atomic.LoadUint64(&p.pinnedTransactionConnections), + } } + return nil, err } } diff --git a/x/mongo/driver/topology/server.go b/x/mongo/driver/topology/server.go index 600797df40..88b93b15e6 100644 --- a/x/mongo/driver/topology/server.go +++ b/x/mongo/driver/topology/server.go @@ -177,6 +177,7 @@ func NewServer(addr address.Address, topologyID primitive.ObjectID, opts ...Serv MaxConnecting: cfg.maxConnecting, MaxIdleTime: cfg.poolMaxIdleTime, MaintainInterval: cfg.poolMaintainInterval, + LoadBalanced: cfg.loadBalanced, PoolMonitor: cfg.poolMonitor, Logger: cfg.logger, handshakeErrFn: s.ProcessHandshakeError, From 9c094db39fc530672d8b1ab52372852e46270651 Mon Sep 17 00:00:00 2001 From: Kevin Albertson Date: Wed, 30 Aug 2023 15:05:35 -0400 Subject: [PATCH 02/15] GODRIVER-2963 Use more environment variables in Azure KMS test (#1367) * use environment variables for key_name and key_vault_endpoint This may prevent future code changes if Azure test data changes. --- .evergreen/config.yml | 4 ++-- cmd/testkms/main.go | 20 ++++++++++++++++++-- 2 files changed, 20 insertions(+), 4 deletions(-) diff --git a/.evergreen/config.yml b/.evergreen/config.yml index db6b51f680..81738eb513 100644 --- a/.evergreen/config.yml +++ b/.evergreen/config.yml @@ -2228,7 +2228,7 @@ tasks: export AZUREKMS_VMNAME=${AZUREKMS_VMNAME} echo '${testazurekms_privatekey}' > /tmp/testazurekms.prikey export AZUREKMS_PRIVATEKEYPATH=/tmp/testazurekms.prikey - AZUREKMS_CMD="LD_LIBRARY_PATH=./install/libmongocrypt/lib MONGODB_URI='mongodb://localhost:27017' PROVIDER='azure' ./testkms" $DRIVERS_TOOLS/.evergreen/csfle/azurekms/run-command.sh + AZUREKMS_CMD="LD_LIBRARY_PATH=./install/libmongocrypt/lib MONGODB_URI='mongodb://localhost:27017' PROVIDER='azure' AZUREKMS_KEY_NAME='${AZUREKMS_KEY_NAME}' AZUREKMS_KEY_VAULT_ENDPOINT='${AZUREKMS_KEY_VAULT_ENDPOINT}' ./testkms" $DRIVERS_TOOLS/.evergreen/csfle/azurekms/run-command.sh - name: "testazurekms-fail-task" # testazurekms-fail-task runs without environment variables. @@ -2250,7 +2250,7 @@ tasks: LD_LIBRARY_PATH=./install/libmongocrypt/lib \ MONGODB_URI='mongodb://localhost:27017' \ EXPECT_ERROR='unable to retrieve azure credentials' \ - PROVIDER='azure' \ + PROVIDER='azure' AZUREKMS_KEY_NAME='${AZUREKMS_KEY_NAME}' AZUREKMS_KEY_VAULT_ENDPOINT='${AZUREKMS_KEY_VAULT_ENDPOINT}' \ ./testkms - name: "test-fuzz" diff --git a/cmd/testkms/main.go b/cmd/testkms/main.go index 5d7532c4b2..af86eca523 100644 --- a/cmd/testkms/main.go +++ b/cmd/testkms/main.go @@ -24,8 +24,8 @@ var datakeyopts = map[string]primitive.M{ "key": "arn:aws:kms:us-east-1:579766882180:key/89fcc2c4-08b0-4bd9-9f25-e30687b580d0", }, "azure": bson.M{ - "keyVaultEndpoint": "https://keyvault-drivers-2411.vault.azure.net/keys/", - "keyName": "KEY-NAME", + "keyVaultEndpoint": "", + "keyName": "", }, "gcp": bson.M{ "projectId": "devprod-drivers", @@ -53,6 +53,20 @@ func main() { default: ok = true } + if provider == "azure" { + azureKmsKeyName := os.Getenv("AZUREKMS_KEY_NAME") + azureKmsKeyVaultEndpoint := os.Getenv("AZUREKMS_KEY_VAULT_ENDPOINT") + if azureKmsKeyName == "" { + fmt.Println("ERROR: Please set required AZUREKMS_KEY_NAME environment variable.") + ok = false + } + if azureKmsKeyVaultEndpoint == "" { + fmt.Println("ERROR: Please set required AZUREKMS_KEY_VAULT_ENDPOINT environment variable.") + ok = false + } + datakeyopts["azure"]["keyName"] = azureKmsKeyName + datakeyopts["azure"]["keyVaultEndpoint"] = azureKmsKeyVaultEndpoint + } if !ok { providers := make([]string, 0, len(datakeyopts)) for p := range datakeyopts { @@ -63,6 +77,8 @@ func main() { fmt.Println("- MONGODB_URI as a MongoDB URI. Example: 'mongodb://localhost:27017'") fmt.Println("- EXPECT_ERROR as an optional expected error substring.") fmt.Println("- PROVIDER as a KMS provider, which supports:", strings.Join(providers, ", ")) + fmt.Println("- AZUREKMS_KEY_NAME as the Azure key name. Required if PROVIDER=azure.") + fmt.Println("- AZUREKMS_KEY_VAULT_ENDPOINT as the Azure key name. Required if PROVIDER=azure.") os.Exit(1) } From 2f372fdae551b566fa65e6400e6bb52d745f9f2a Mon Sep 17 00:00:00 2001 From: Qingyang Hu <103950869+qingyang-hu@users.noreply.github.com> Date: Wed, 30 Aug 2023 18:15:18 -0400 Subject: [PATCH 03/15] GODRIVER-2872 Duplicate slice passed to mongocrypt.newBinaryFromBytes(). (#1359) --- .../client_side_encryption_prose_test.go | 3 --- x/mongo/driver/mongocrypt/binary.go | 17 ++++++++++++----- 2 files changed, 12 insertions(+), 8 deletions(-) diff --git a/mongo/integration/client_side_encryption_prose_test.go b/mongo/integration/client_side_encryption_prose_test.go index f650aa6b47..4eac5364ca 100644 --- a/mongo/integration/client_side_encryption_prose_test.go +++ b/mongo/integration/client_side_encryption_prose_test.go @@ -378,9 +378,6 @@ func TestClientSideEncryptionProse(t *testing.T) { } }) mt.Run("4. bson size limits", func(mt *mtest.T) { - // TODO(GODRIVER-2872): Fix and unskip this test case. - mt.Skip("Test fails frequently, skipping. See GODRIVER-2872") - kmsProviders := map[string]map[string]interface{}{ "local": { "key": localMasterKey, diff --git a/x/mongo/driver/mongocrypt/binary.go b/x/mongo/driver/mongocrypt/binary.go index 9e887375a9..4e4b51d74b 100644 --- a/x/mongo/driver/mongocrypt/binary.go +++ b/x/mongo/driver/mongocrypt/binary.go @@ -9,7 +9,10 @@ package mongocrypt -// #include +/* +#include +#include +*/ import "C" import ( "unsafe" @@ -17,6 +20,7 @@ import ( // binary is a wrapper type around a mongocrypt_binary_t* type binary struct { + p *C.uint8_t wrapped *C.mongocrypt_binary_t } @@ -33,11 +37,11 @@ func newBinaryFromBytes(data []byte) *binary { return newBinary() } - // We don't need C.CBytes here because data cannot go out of scope. Any mongocrypt function that takes a - // mongocrypt_binary_t will make a copy of the data so the data can be garbage collected after calling. - addr := (*C.uint8_t)(unsafe.Pointer(&data[0])) // uint8_t* - dataLen := C.uint32_t(len(data)) // uint32_t + // TODO: Consider using runtime.Pinner to replace the C.CBytes after using go1.21.0. + addr := (*C.uint8_t)(C.CBytes(data)) // uint8_t* + dataLen := C.uint32_t(len(data)) // uint32_t return &binary{ + p: addr, wrapped: C.mongocrypt_binary_new_from_data(addr, dataLen), } } @@ -52,5 +56,8 @@ func (b *binary) toBytes() []byte { // close cleans up any resources associated with the given binary instance. func (b *binary) close() { + if b.p != nil { + C.free(unsafe.Pointer(b.p)) + } C.mongocrypt_binary_destroy(b.wrapped) } From 02d5139081ebd4d150ec15b476d20543200e9d8e Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 1 Sep 2023 06:06:06 -0500 Subject: [PATCH 04/15] GODRIVER-2958 Use AWS Secrets for Atlas tests (#1365) * GODRIVER-2958 Use AWS Secrets for Atlas tests * fix handling of shell * fix cleanup * print finished * try wip branch * cleanup * cleanup * cleanup * cleanup * clean up secrets handling * GODRIVER-2958 Use AWS Secrets for Atlas tests * restore drivers-eg-tools checkout --- .evergreen/config.yml | 31 ++++++------------------------- .gitignore | 6 +++++- Makefile | 5 ----- cmd/testatlas/main.go | 6 ++++++ etc/get_aws_secrets.sh | 12 ++++++++++++ etc/run-atlas-test.sh | 11 +++++++++++ 6 files changed, 40 insertions(+), 31 deletions(-) create mode 100644 etc/get_aws_secrets.sh create mode 100644 etc/run-atlas-test.sh diff --git a/.evergreen/config.yml b/.evergreen/config.yml index 81738eb513..0603f385ac 100644 --- a/.evergreen/config.yml +++ b/.evergreen/config.yml @@ -437,37 +437,18 @@ functions: make -s evg-test-enterprise-auth run-atlas-test: + - command: ec2.assume_role + params: + role_arn: "${aws_test_secrets_role}" - command: shell.exec type: test params: shell: "bash" working_dir: src/go.mongodb.org/mongo-driver + include_expansions_in_env: ["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_SESSION_TOKEN"] script: | - # DO NOT ECHO WITH XTRACE - if [ "Windows_NT" = "$OS" ]; then - export GOPATH=$(cygpath -w $(dirname $(dirname $(dirname `pwd`)))) - export GOCACHE=$(cygpath -w "$(pwd)/.cache") - else - export GOPATH=$(dirname $(dirname $(dirname `pwd`))) - export GOCACHE="$(pwd)/.cache" - fi; - export GOPATH="$GOPATH" - export GOROOT="${GO_DIST}" - export GOCACHE="$GOCACHE" - export PATH="${GCC_PATH}:${GO_DIST}/bin:$PATH" - export ATLAS_FREE="${atlas_free_tier_uri}" - export ATLAS_REPLSET="${atlas_replica_set_uri}" - export ATLAS_SHARD="${atlas_sharded_uri}" - export ATLAS_TLS11="${atlas_tls_v11_uri}" - export ATLAS_TLS12="${atlas_tls_v12_uri}" - export ATLAS_FREE_SRV="${atlas_free_tier_uri_srv}" - export ATLAS_REPLSET_SRV="${atlas_replica_set_uri_srv}" - export ATLAS_SHARD_SRV="${atlas_sharded_uri_srv}" - export ATLAS_TLS11_SRV="${atlas_tls_v11_uri_srv}" - export ATLAS_TLS12_SRV="${atlas_tls_v12_uri_srv}" - export ATLAS_SERVERLESS="${atlas_serverless_uri}" - export ATLAS_SERVERLESS_SRV="${atlas_serverless_uri_srv}" - make -s evg-test-atlas + ${PREPARE_SHELL} + bash etc/run-atlas-test.sh run-ocsp-test: - command: shell.exec diff --git a/.gitignore b/.gitignore index e9609ce1e9..16b52325e4 100644 --- a/.gitignore +++ b/.gitignore @@ -24,4 +24,8 @@ internal/test/compilecheck/compilecheck.so # Ignore api report files api-report.md -api-report.txt \ No newline at end of file +api-report.txt + +# Ignore secrets files +secrets-expansion.yml +secrets-export.sh diff --git a/Makefile b/Makefile index 5531b61a29..66f5b137e5 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,3 @@ -ATLAS_URIS = "$(ATLAS_FREE)" "$(ATLAS_REPLSET)" "$(ATLAS_SHARD)" "$(ATLAS_TLS11)" "$(ATLAS_TLS12)" "$(ATLAS_FREE_SRV)" "$(ATLAS_REPLSET_SRV)" "$(ATLAS_SHARD_SRV)" "$(ATLAS_TLS11_SRV)" "$(ATLAS_TLS12_SRV)" "$(ATLAS_SERVERLESS)" "$(ATLAS_SERVERLESS_SRV)" TEST_TIMEOUT = 1800 ### Utility targets. ### @@ -128,10 +127,6 @@ build-aws-ecs-test: evg-test: go test -exec "env PKG_CONFIG_PATH=$(PKG_CONFIG_PATH) LD_LIBRARY_PATH=$(LD_LIBRARY_PATH)" $(BUILD_TAGS) -v -timeout $(TEST_TIMEOUT)s -p 1 ./... >> test.suite -.PHONY: evg-test-atlas -evg-test-atlas: - go run ./cmd/testatlas/main.go $(ATLAS_URIS) - .PHONY: evg-test-atlas-data-lake evg-test-atlas-data-lake: ATLAS_DATA_LAKE_INTEGRATION_TEST=true go test -v ./mongo/integration -run TestUnifiedSpecs/atlas-data-lake-testing >> spec_test.suite diff --git a/cmd/testatlas/main.go b/cmd/testatlas/main.go index d98631d468..fa50d7cde7 100644 --- a/cmd/testatlas/main.go +++ b/cmd/testatlas/main.go @@ -23,7 +23,11 @@ func main() { uris := flag.Args() ctx := context.Background() + fmt.Printf("Running atlas tests for %d uris\n", len(uris)) + for idx, uri := range uris { + fmt.Printf("Running test %d\n", idx) + // Set a low server selection timeout so we fail fast if there are errors. clientOpts := options.Client(). ApplyURI(uri). @@ -41,6 +45,8 @@ func main() { panic(fmt.Sprintf("error running test with tlsInsecure at index %d: %v", idx, err)) } } + + fmt.Println("Finished!") } func runTest(ctx context.Context, clientOpts *options.ClientOptions) error { diff --git a/etc/get_aws_secrets.sh b/etc/get_aws_secrets.sh new file mode 100644 index 0000000000..894016553b --- /dev/null +++ b/etc/get_aws_secrets.sh @@ -0,0 +1,12 @@ +#!/usr/bin/env bash +# get-aws-secrets +# Gets AWS secrets from the vault +set -eu + +if [ -z "$DRIVERS_TOOLS" ]; then + echo "Please define DRIVERS_TOOLS variable" + exit 1 +fi + +bash $DRIVERS_TOOLS/.evergreen/auth_aws/setup_secrets.sh $@ +. ./secrets-export.sh diff --git a/etc/run-atlas-test.sh b/etc/run-atlas-test.sh new file mode 100644 index 0000000000..aa89b2dd4b --- /dev/null +++ b/etc/run-atlas-test.sh @@ -0,0 +1,11 @@ +#!/usr/bin/env bash +# run-atlas-test +# Run atlas connectivity tests. +set -eu +set +x + +# Get the atlas secrets. +. etc/get_aws_secrets.sh drivers/atlas_connect + +echo "Running cmd/testatlas/main.go" +go run ./cmd/testatlas/main.go "$ATLAS_REPL" "$ATLAS_SHRD" "$ATLAS_FREE" "$ATLAS_TLS11" "$ATLAS_TLS12" "$ATLAS_SERVERLESS" "$ATLAS_SRV_REPL" "$ATLAS_SRV_SHRD" "$ATLAS_SRV_FREE" "$ATLAS_SRV_TLS11" "$ATLAS_SRV_TLS12" "$ATLAS_SRV_SERVERLESS" From a3fcd7604ce98c28aa7952c007c5aff9cba7f2b2 Mon Sep 17 00:00:00 2001 From: Qingyang Hu <103950869+qingyang-hu@users.noreply.github.com> Date: Tue, 5 Sep 2023 09:34:46 -0400 Subject: [PATCH 05/15] GODRIVER-2964 Fix failing "TestClient/tls_connection". (#1374) --- mongo/integration/client_test.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/mongo/integration/client_test.go b/mongo/integration/client_test.go index 914ca863b7..d564fab2f8 100644 --- a/mongo/integration/client_test.go +++ b/mongo/integration/client_test.go @@ -126,8 +126,6 @@ func TestClient(t *testing.T) { "expected security field to be type %v, got %v", bson.TypeMaxKey, security.Type) _, found := security.Document().LookupErr("SSLServerSubjectName") assert.Nil(mt, found, "SSLServerSubjectName not found in result") - _, found = security.Document().LookupErr("SSLServerHasCertificateAuthority") - assert.Nil(mt, found, "SSLServerHasCertificateAuthority not found in result") }) mt.RunOpts("x509", mtest.NewOptions().Auth(true).SSL(true), func(mt *mtest.T) { testCases := []struct { From 84a43854bbd7c243ce5a68d7b56b3ba4ba10ed05 Mon Sep 17 00:00:00 2001 From: Charlie Vieth Date: Tue, 5 Sep 2023 13:40:00 -0400 Subject: [PATCH 06/15] GODRIVER-2914 x/mongo/driver: enable parallel zlib compression and improve zstd decompression (#1320) Co-authored-by: Matt Dale <9760375+matthewdale@users.noreply.github.com> --- x/mongo/driver/compression.go | 114 ++++++++++++------- x/mongo/driver/compression_test.go | 137 ++++++++++++++++++++-- x/mongo/driver/testdata/compression.go | 151 +++++++++++++++++++++++++ 3 files changed, 349 insertions(+), 53 deletions(-) create mode 100644 x/mongo/driver/testdata/compression.go diff --git a/x/mongo/driver/compression.go b/x/mongo/driver/compression.go index 7f355f61a4..d79b024b74 100644 --- a/x/mongo/driver/compression.go +++ b/x/mongo/driver/compression.go @@ -26,48 +26,72 @@ type CompressionOpts struct { UncompressedSize int32 } -var zstdEncoders sync.Map // map[zstd.EncoderLevel]*zstd.Encoder +// mustZstdNewWriter creates a zstd.Encoder with the given level and a nil +// destination writer. It panics on any errors and should only be used at +// package initialization time. +func mustZstdNewWriter(lvl zstd.EncoderLevel) *zstd.Encoder { + enc, err := zstd.NewWriter(nil, zstd.WithEncoderLevel(lvl)) + if err != nil { + panic(err) + } + return enc +} + +var zstdEncoders = [zstd.SpeedBestCompression + 1]*zstd.Encoder{ + 0: nil, // zstd.speedNotSet + zstd.SpeedFastest: mustZstdNewWriter(zstd.SpeedFastest), + zstd.SpeedDefault: mustZstdNewWriter(zstd.SpeedDefault), + zstd.SpeedBetterCompression: mustZstdNewWriter(zstd.SpeedBetterCompression), + zstd.SpeedBestCompression: mustZstdNewWriter(zstd.SpeedBestCompression), +} func getZstdEncoder(level zstd.EncoderLevel) (*zstd.Encoder, error) { - if v, ok := zstdEncoders.Load(level); ok { - return v.(*zstd.Encoder), nil - } - encoder, err := zstd.NewWriter(nil, zstd.WithEncoderLevel(level)) - if err != nil { - return nil, err + if zstd.SpeedFastest <= level && level <= zstd.SpeedBestCompression { + return zstdEncoders[level], nil } - zstdEncoders.Store(level, encoder) - return encoder, nil + // The level is outside the expected range, return an error. + return nil, fmt.Errorf("invalid zstd compression level: %d", level) } -var zlibEncoders sync.Map // map[int /*level*/]*zlibEncoder +// zlibEncodersOffset is the offset into the zlibEncoders array for a given +// compression level. +const zlibEncodersOffset = -zlib.HuffmanOnly // HuffmanOnly == -2 + +var zlibEncoders [zlib.BestCompression + zlibEncodersOffset + 1]sync.Pool func getZlibEncoder(level int) (*zlibEncoder, error) { - if v, ok := zlibEncoders.Load(level); ok { - return v.(*zlibEncoder), nil - } - writer, err := zlib.NewWriterLevel(nil, level) - if err != nil { - return nil, err + if zlib.HuffmanOnly <= level && level <= zlib.BestCompression { + if enc, _ := zlibEncoders[level+zlibEncodersOffset].Get().(*zlibEncoder); enc != nil { + return enc, nil + } + writer, err := zlib.NewWriterLevel(nil, level) + if err != nil { + return nil, err + } + enc := &zlibEncoder{writer: writer, level: level} + return enc, nil } - encoder := &zlibEncoder{writer: writer, buf: new(bytes.Buffer)} - zlibEncoders.Store(level, encoder) + // The level is outside the expected range, return an error. + return nil, fmt.Errorf("invalid zlib compression level: %d", level) +} - return encoder, nil +func putZlibEncoder(enc *zlibEncoder) { + if enc != nil { + zlibEncoders[enc.level+zlibEncodersOffset].Put(enc) + } } type zlibEncoder struct { - mu sync.Mutex writer *zlib.Writer - buf *bytes.Buffer + buf bytes.Buffer + level int } func (e *zlibEncoder) Encode(dst, src []byte) ([]byte, error) { - e.mu.Lock() - defer e.mu.Unlock() + defer putZlibEncoder(e) e.buf.Reset() - e.writer.Reset(e.buf) + e.writer.Reset(&e.buf) _, err := e.writer.Write(src) if err != nil { @@ -105,8 +129,15 @@ func CompressPayload(in []byte, opts CompressionOpts) ([]byte, error) { } } +var zstdReaderPool = sync.Pool{ + New: func() interface{} { + r, _ := zstd.NewReader(nil) + return r + }, +} + // DecompressPayload takes a byte slice that has been compressed and undoes it according to the options passed -func DecompressPayload(in []byte, opts CompressionOpts) (uncompressed []byte, err error) { +func DecompressPayload(in []byte, opts CompressionOpts) ([]byte, error) { switch opts.Compressor { case wiremessage.CompressorNoOp: return in, nil @@ -117,34 +148,29 @@ func DecompressPayload(in []byte, opts CompressionOpts) (uncompressed []byte, er } else if int32(l) != opts.UncompressedSize { return nil, fmt.Errorf("unexpected decompression size, expected %v but got %v", opts.UncompressedSize, l) } - uncompressed = make([]byte, opts.UncompressedSize) - return snappy.Decode(uncompressed, in) + out := make([]byte, opts.UncompressedSize) + return snappy.Decode(out, in) case wiremessage.CompressorZLib: r, err := zlib.NewReader(bytes.NewReader(in)) if err != nil { return nil, err } - defer func() { - err = r.Close() - }() - uncompressed = make([]byte, opts.UncompressedSize) - _, err = io.ReadFull(r, uncompressed) - if err != nil { + out := make([]byte, opts.UncompressedSize) + if _, err := io.ReadFull(r, out); err != nil { return nil, err } - return uncompressed, nil - case wiremessage.CompressorZstd: - r, err := zstd.NewReader(bytes.NewBuffer(in)) - if err != nil { - return nil, err - } - defer r.Close() - uncompressed = make([]byte, opts.UncompressedSize) - _, err = io.ReadFull(r, uncompressed) - if err != nil { + if err := r.Close(); err != nil { return nil, err } - return uncompressed, nil + return out, nil + case wiremessage.CompressorZstd: + buf := make([]byte, 0, opts.UncompressedSize) + // Using a pool here is about ~20% faster + // than using a single global zstd.Reader + r := zstdReaderPool.Get().(*zstd.Decoder) + out, err := r.DecodeAll(in, buf) + zstdReaderPool.Put(r) + return out, err default: return nil, fmt.Errorf("unknown compressor ID %v", opts.Compressor) } diff --git a/x/mongo/driver/compression_test.go b/x/mongo/driver/compression_test.go index b477cb32c1..75a7ff072b 100644 --- a/x/mongo/driver/compression_test.go +++ b/x/mongo/driver/compression_test.go @@ -7,9 +7,14 @@ package driver import ( + "bytes" + "compress/zlib" "os" "testing" + "github.com/golang/snappy" + "github.com/klauspost/compress/zstd" + "go.mongodb.org/mongo-driver/internal/assert" "go.mongodb.org/mongo-driver/x/mongo/driver/wiremessage" ) @@ -41,6 +46,43 @@ func TestCompression(t *testing.T) { } } +func TestCompressionLevels(t *testing.T) { + in := []byte("abc") + wr := new(bytes.Buffer) + + t.Run("ZLib", func(t *testing.T) { + opts := CompressionOpts{ + Compressor: wiremessage.CompressorZLib, + } + for lvl := zlib.HuffmanOnly - 2; lvl < zlib.BestCompression+2; lvl++ { + opts.ZlibLevel = lvl + _, err1 := CompressPayload(in, opts) + _, err2 := zlib.NewWriterLevel(wr, lvl) + if err2 != nil { + assert.Error(t, err1, "expected an error for ZLib level %d", lvl) + } else { + assert.NoError(t, err1, "unexpected error for ZLib level %d", lvl) + } + } + }) + + t.Run("Zstd", func(t *testing.T) { + opts := CompressionOpts{ + Compressor: wiremessage.CompressorZstd, + } + for lvl := zstd.SpeedFastest - 2; lvl < zstd.SpeedBestCompression+2; lvl++ { + opts.ZstdLevel = int(lvl) + _, err1 := CompressPayload(in, opts) + _, err2 := zstd.NewWriter(wr, zstd.WithEncoderLevel(zstd.EncoderLevelFromZstd(opts.ZstdLevel))) + if err2 != nil { + assert.Error(t, err1, "expected an error for Zstd level %d", lvl) + } else { + assert.NoError(t, err1, "unexpected error for Zstd level %d", lvl) + } + } + }) +} + func TestDecompressFailures(t *testing.T) { t.Parallel() @@ -62,18 +104,57 @@ func TestDecompressFailures(t *testing.T) { }) } -func BenchmarkCompressPayload(b *testing.B) { - payload := func() []byte { - buf, err := os.ReadFile("compression.go") +var ( + compressionPayload []byte + compressedSnappyPayload []byte + compressedZLibPayload []byte + compressedZstdPayload []byte +) + +func initCompressionPayload(b *testing.B) { + if compressionPayload != nil { + return + } + data, err := os.ReadFile("testdata/compression.go") + if err != nil { + b.Fatal(err) + } + for i := 1; i < 10; i++ { + data = append(data, data...) + } + compressionPayload = data + + compressedSnappyPayload = snappy.Encode(compressedSnappyPayload[:0], data) + + { + var buf bytes.Buffer + enc, err := zstd.NewWriter(&buf, zstd.WithEncoderLevel(zstd.SpeedDefault)) if err != nil { - b.Log(err) - b.FailNow() + b.Fatal(err) } - for i := 1; i < 10; i++ { - buf = append(buf, buf...) + compressedZstdPayload = enc.EncodeAll(data, nil) + } + + { + var buf bytes.Buffer + enc := zlib.NewWriter(&buf) + if _, err := enc.Write(data); err != nil { + b.Fatal(err) } - return buf - }() + if err := enc.Close(); err != nil { + b.Fatal(err) + } + if err := enc.Close(); err != nil { + b.Fatal(err) + } + compressedZLibPayload = append(compressedZLibPayload[:0], buf.Bytes()...) + } + + b.ResetTimer() +} + +func BenchmarkCompressPayload(b *testing.B) { + initCompressionPayload(b) compressors := []wiremessage.CompressorID{ wiremessage.CompressorSnappy, @@ -88,6 +169,9 @@ func BenchmarkCompressPayload(b *testing.B) { ZlibLevel: wiremessage.DefaultZlibLevel, ZstdLevel: wiremessage.DefaultZstdLevel, } + payload := compressionPayload + b.SetBytes(int64(len(payload))) + b.ReportAllocs() b.RunParallel(func(pb *testing.PB) { for pb.Next() { _, err := CompressPayload(payload, opts) @@ -99,3 +183,38 @@ func BenchmarkCompressPayload(b *testing.B) { }) } } + +func BenchmarkDecompressPayload(b *testing.B) { + initCompressionPayload(b) + + benchmarks := []struct { + compressor wiremessage.CompressorID + payload []byte + }{ + {wiremessage.CompressorSnappy, compressedSnappyPayload}, + {wiremessage.CompressorZLib, compressedZLibPayload}, + {wiremessage.CompressorZstd, compressedZstdPayload}, + } + + for _, bench := range benchmarks { + b.Run(bench.compressor.String(), func(b *testing.B) { + opts := CompressionOpts{ + Compressor: bench.compressor, + ZlibLevel: wiremessage.DefaultZlibLevel, + ZstdLevel: wiremessage.DefaultZstdLevel, + UncompressedSize: int32(len(compressionPayload)), + } + payload := bench.payload + b.SetBytes(int64(len(compressionPayload))) + b.ReportAllocs() + b.RunParallel(func(pb *testing.PB) { + for pb.Next() { + _, err := DecompressPayload(payload, opts) + if err != nil { + b.Fatal(err) + } + } + }) + }) + } +} diff --git a/x/mongo/driver/testdata/compression.go b/x/mongo/driver/testdata/compression.go new file mode 100644 index 0000000000..7f355f61a4 --- /dev/null +++ b/x/mongo/driver/testdata/compression.go @@ -0,0 +1,151 @@ +// Copyright (C) MongoDB, Inc. 2017-present. +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. You may obtain +// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 + +package driver + +import ( + "bytes" + "compress/zlib" + "fmt" + "io" + "sync" + + "github.com/golang/snappy" + "github.com/klauspost/compress/zstd" + "go.mongodb.org/mongo-driver/x/mongo/driver/wiremessage" +) + +// CompressionOpts holds settings for how to compress a payload +type CompressionOpts struct { + Compressor wiremessage.CompressorID + ZlibLevel int + ZstdLevel int + UncompressedSize int32 +} + +var zstdEncoders sync.Map // map[zstd.EncoderLevel]*zstd.Encoder + +func getZstdEncoder(level zstd.EncoderLevel) (*zstd.Encoder, error) { + if v, ok := zstdEncoders.Load(level); ok { + return v.(*zstd.Encoder), nil + } + encoder, err := zstd.NewWriter(nil, zstd.WithEncoderLevel(level)) + if err != nil { + return nil, err + } + zstdEncoders.Store(level, encoder) + return encoder, nil +} + +var zlibEncoders sync.Map // map[int /*level*/]*zlibEncoder + +func getZlibEncoder(level int) (*zlibEncoder, error) { + if v, ok := zlibEncoders.Load(level); ok { + return v.(*zlibEncoder), nil + } + writer, err := zlib.NewWriterLevel(nil, level) + if err != nil { + return nil, err + } + encoder := &zlibEncoder{writer: writer, buf: new(bytes.Buffer)} + zlibEncoders.Store(level, encoder) + + return encoder, nil +} + +type zlibEncoder struct { + mu sync.Mutex + writer *zlib.Writer + buf *bytes.Buffer +} + +func (e *zlibEncoder) Encode(dst, src []byte) ([]byte, error) { + e.mu.Lock() + defer e.mu.Unlock() + + e.buf.Reset() + e.writer.Reset(e.buf) + + _, err := e.writer.Write(src) + if err != nil { + return nil, err + } + err = e.writer.Close() + if err != nil { + return nil, err + } + dst = append(dst[:0], e.buf.Bytes()...) + return dst, nil +} + +// CompressPayload takes a byte slice and compresses it according to the options passed +func CompressPayload(in []byte, opts CompressionOpts) ([]byte, error) { + switch opts.Compressor { + case wiremessage.CompressorNoOp: + return in, nil + case wiremessage.CompressorSnappy: + return snappy.Encode(nil, in), nil + case wiremessage.CompressorZLib: + encoder, err := getZlibEncoder(opts.ZlibLevel) + if err != nil { + return nil, err + } + return encoder.Encode(nil, in) + case wiremessage.CompressorZstd: + encoder, err := getZstdEncoder(zstd.EncoderLevelFromZstd(opts.ZstdLevel)) + if err != nil { + return nil, err + } + return encoder.EncodeAll(in, nil), nil + default: + return nil, fmt.Errorf("unknown compressor ID %v", opts.Compressor) + } +} + +// DecompressPayload takes a byte slice that has been compressed and undoes it according to the options passed +func DecompressPayload(in []byte, opts CompressionOpts) (uncompressed []byte, err error) { + switch opts.Compressor { + case wiremessage.CompressorNoOp: + return in, nil + case wiremessage.CompressorSnappy: + l, err := snappy.DecodedLen(in) + if err != nil { + return nil, fmt.Errorf("decoding compressed length %w", err) + } else if int32(l) != opts.UncompressedSize { + return nil, fmt.Errorf("unexpected decompression size, expected %v but got %v", opts.UncompressedSize, l) + } + uncompressed = make([]byte, opts.UncompressedSize) + return snappy.Decode(uncompressed, in) + case wiremessage.CompressorZLib: + r, err := zlib.NewReader(bytes.NewReader(in)) + if err != nil { + return nil, err + } + defer func() { + err = r.Close() + }() + uncompressed = make([]byte, opts.UncompressedSize) + _, err = io.ReadFull(r, uncompressed) + if err != nil { + return nil, err + } + return uncompressed, nil + case wiremessage.CompressorZstd: + r, err := zstd.NewReader(bytes.NewBuffer(in)) + if err != nil { + return nil, err + } + defer r.Close() + uncompressed = make([]byte, opts.UncompressedSize) + _, err = io.ReadFull(r, uncompressed) + if err != nil { + return nil, err + } + return uncompressed, nil + default: + return nil, fmt.Errorf("unknown compressor ID %v", opts.Compressor) + } +} From 68bf155d63446251f63603eebd74a12707bcdf51 Mon Sep 17 00:00:00 2001 From: Preston Vasquez Date: Thu, 7 Sep 2023 10:41:47 -0600 Subject: [PATCH 07/15] GODRIVER-2972 Fix wiremessage RequestID race in operation.Execute (#1375) --- x/mongo/driver/operation.go | 23 ++++++++++++++--------- x/mongo/driver/operation_test.go | 9 +++++---- x/mongo/driver/wiremessage/wiremessage.go | 3 --- 3 files changed, 19 insertions(+), 16 deletions(-) diff --git a/x/mongo/driver/operation.go b/x/mongo/driver/operation.go index 90573daa53..8e52773503 100644 --- a/x/mongo/driver/operation.go +++ b/x/mongo/driver/operation.go @@ -322,7 +322,7 @@ func (op Operation) shouldEncrypt() bool { } // selectServer handles performing server selection for an operation. -func (op Operation) selectServer(ctx context.Context) (Server, error) { +func (op Operation) selectServer(ctx context.Context, requestID int32) (Server, error) { if err := op.Validate(); err != nil { return nil, err } @@ -340,14 +340,14 @@ func (op Operation) selectServer(ctx context.Context) (Server, error) { } ctx = logger.WithOperationName(ctx, op.Name) - ctx = logger.WithOperationID(ctx, wiremessage.CurrentRequestID()) + ctx = logger.WithOperationID(ctx, requestID) return op.Deployment.SelectServer(ctx, selector) } // getServerAndConnection should be used to retrieve a Server and Connection to execute an operation. -func (op Operation) getServerAndConnection(ctx context.Context) (Server, Connection, error) { - server, err := op.selectServer(ctx) +func (op Operation) getServerAndConnection(ctx context.Context, requestID int32) (Server, Connection, error) { + server, err := op.selectServer(ctx, requestID) if err != nil { if op.Client != nil && !(op.Client.Committing || op.Client.Aborting) && op.Client.TransactionRunning() { @@ -530,11 +530,11 @@ func (op Operation) Execute(ctx context.Context) error { } }() for { - wiremessage.NextRequestID() + requestID := wiremessage.NextRequestID() // If the server or connection are nil, try to select a new server and get a new connection. if srvr == nil || conn == nil { - srvr, conn, err = op.getServerAndConnection(ctx) + srvr, conn, err = op.getServerAndConnection(ctx, requestID) if err != nil { // If the returned error is retryable and there are retries remaining (negative // retries means retry indefinitely), then retry the operation. Set the server @@ -629,7 +629,7 @@ func (op Operation) Execute(ctx context.Context) error { } var startedInfo startedInformation - *wm, startedInfo, err = op.createMsgWireMessage(ctx, maxTimeMS, (*wm)[:0], desc, conn) + *wm, startedInfo, err = op.createMsgWireMessage(ctx, maxTimeMS, (*wm)[:0], desc, conn, requestID) if err != nil { return err @@ -1103,8 +1103,13 @@ func (op Operation) addBatchArray(dst []byte) []byte { return dst } -func (op Operation) createMsgWireMessage(ctx context.Context, maxTimeMS uint64, dst []byte, desc description.SelectedServer, +func (op Operation) createMsgWireMessage( + ctx context.Context, + maxTimeMS uint64, + dst []byte, + desc description.SelectedServer, conn Connection, + requestID int32, ) ([]byte, startedInformation, error) { var info startedInformation var flags wiremessage.MsgFlag @@ -1120,7 +1125,7 @@ func (op Operation) createMsgWireMessage(ctx context.Context, maxTimeMS uint64, flags |= wiremessage.ExhaustAllowed } - info.requestID = wiremessage.CurrentRequestID() + info.requestID = requestID wmindex, dst = wiremessage.AppendHeaderStart(dst, info.requestID, 0, wiremessage.OpMsg) dst = wiremessage.AppendMsgFlags(dst, flags) // Body diff --git a/x/mongo/driver/operation_test.go b/x/mongo/driver/operation_test.go index d4c5a1b6a0..8509b5da9b 100644 --- a/x/mongo/driver/operation_test.go +++ b/x/mongo/driver/operation_test.go @@ -62,7 +62,7 @@ func TestOperation(t *testing.T) { t.Run("selectServer", func(t *testing.T) { t.Run("returns validation error", func(t *testing.T) { op := &Operation{} - _, err := op.selectServer(context.Background()) + _, err := op.selectServer(context.Background(), 1) if err == nil { t.Error("Expected a validation error from selectServer, but got ") } @@ -76,7 +76,7 @@ func TestOperation(t *testing.T) { Database: "testing", Selector: want, } - _, err := op.selectServer(context.Background()) + _, err := op.selectServer(context.Background(), 1) noerr(t, err) got := d.params.selector if !cmp.Equal(got, want) { @@ -90,7 +90,7 @@ func TestOperation(t *testing.T) { Deployment: d, Database: "testing", } - _, err := op.selectServer(context.Background()) + _, err := op.selectServer(context.Background(), 1) noerr(t, err) if d.params.selector == nil { t.Error("The selectServer method should use a default selector when not specified on Operation, but it passed .") @@ -652,7 +652,8 @@ func TestOperation(t *testing.T) { } func createExhaustServerResponse(response bsoncore.Document, moreToCome bool) []byte { - idx, wm := wiremessage.AppendHeaderStart(nil, 0, wiremessage.CurrentRequestID()+1, wiremessage.OpMsg) + const psuedoRequestID = 1 + idx, wm := wiremessage.AppendHeaderStart(nil, 0, psuedoRequestID, wiremessage.OpMsg) var flags wiremessage.MsgFlag if moreToCome { flags = wiremessage.MoreToCome diff --git a/x/mongo/driver/wiremessage/wiremessage.go b/x/mongo/driver/wiremessage/wiremessage.go index c4d2567bf0..abf09c15bd 100644 --- a/x/mongo/driver/wiremessage/wiremessage.go +++ b/x/mongo/driver/wiremessage/wiremessage.go @@ -19,9 +19,6 @@ type WireMessage []byte var globalRequestID int32 -// CurrentRequestID returns the current request ID. -func CurrentRequestID() int32 { return atomic.LoadInt32(&globalRequestID) } - // NextRequestID returns the next request ID. func NextRequestID() int32 { return atomic.AddInt32(&globalRequestID, 1) } From 3bef0e4b998bd276a80c7be51de801a258d52c4b Mon Sep 17 00:00:00 2001 From: Preston Vasquez Date: Thu, 7 Sep 2023 15:06:09 -0600 Subject: [PATCH 08/15] GODRIVER-2935 Use OP_QUERY in connection handshakes (#1377) --- mongo/integration/client_test.go | 4 +- mongo/integration/mtest/sent_message.go | 65 +++++++++++ x/mongo/driver/auth/speculative_scram_test.go | 4 +- x/mongo/driver/auth/speculative_x509_test.go | 4 +- x/mongo/driver/drivertest/channel_conn.go | 32 ++++++ x/mongo/driver/legacy.go | 1 + x/mongo/driver/operation.go | 105 +++++++++++++++++- x/mongo/driver/operation/hello.go | 34 +++++- x/mongo/driver/topology/server_test.go | 39 ++++++- 9 files changed, 272 insertions(+), 16 deletions(-) diff --git a/mongo/integration/client_test.go b/mongo/integration/client_test.go index d564fab2f8..0dfe740358 100644 --- a/mongo/integration/client_test.go +++ b/mongo/integration/client_test.go @@ -731,8 +731,8 @@ func TestClient(t *testing.T) { pair := msgPairs[0] assert.Equal(mt, handshake.LegacyHello, pair.CommandName, "expected command name %s at index 0, got %s", handshake.LegacyHello, pair.CommandName) - assert.Equal(mt, wiremessage.OpMsg, pair.Sent.OpCode, - "expected 'OP_MSG' OpCode in wire message, got %q", pair.Sent.OpCode.String()) + assert.Equal(mt, wiremessage.OpQuery, pair.Sent.OpCode, + "expected 'OP_QUERY' OpCode in wire message, got %q", pair.Sent.OpCode.String()) // Look for a saslContinue in the remaining proxied messages and assert that it uses the OP_MSG OpCode, as wire // version is now known to be >= 6. diff --git a/mongo/integration/mtest/sent_message.go b/mongo/integration/mtest/sent_message.go index d36075bf81..6b96e061bc 100644 --- a/mongo/integration/mtest/sent_message.go +++ b/mongo/integration/mtest/sent_message.go @@ -37,6 +37,8 @@ type sentMsgParseFn func([]byte) (*SentMessage, error) func getSentMessageParser(opcode wiremessage.OpCode) (sentMsgParseFn, bool) { switch opcode { + case wiremessage.OpQuery: + return parseOpQuery, true case wiremessage.OpMsg: return parseSentOpMsg, true case wiremessage.OpCompressed: @@ -46,6 +48,69 @@ func getSentMessageParser(opcode wiremessage.OpCode) (sentMsgParseFn, bool) { } } +func parseOpQuery(wm []byte) (*SentMessage, error) { + var ok bool + + if _, wm, ok = wiremessage.ReadQueryFlags(wm); !ok { + return nil, errors.New("failed to read query flags") + } + if _, wm, ok = wiremessage.ReadQueryFullCollectionName(wm); !ok { + return nil, errors.New("failed to read full collection name") + } + if _, wm, ok = wiremessage.ReadQueryNumberToSkip(wm); !ok { + return nil, errors.New("failed to read number to skip") + } + if _, wm, ok = wiremessage.ReadQueryNumberToReturn(wm); !ok { + return nil, errors.New("failed to read number to return") + } + + query, wm, ok := wiremessage.ReadQueryQuery(wm) + if !ok { + return nil, errors.New("failed to read query") + } + + // If there is no read preference document, the command document is query. + // Otherwise, query is in the format {$query: , $readPreference: }. + commandDoc := query + var rpDoc bsoncore.Document + + dollarQueryVal, err := query.LookupErr("$query") + if err == nil { + commandDoc = dollarQueryVal.Document() + + rpVal, err := query.LookupErr("$readPreference") + if err != nil { + return nil, fmt.Errorf("query %s contains $query but not $readPreference fields", query) + } + rpDoc = rpVal.Document() + } + + // For OP_QUERY, inserts, updates, and deletes are sent as a BSON array of documents inside the main command + // document. Pull these sequences out into an ArrayStyle DocumentSequence. + var docSequence *bsoncore.DocumentSequence + cmdElems, _ := commandDoc.Elements() + for _, elem := range cmdElems { + switch elem.Key() { + case "documents", "updates", "deletes": + docSequence = &bsoncore.DocumentSequence{ + Style: bsoncore.ArrayStyle, + Data: elem.Value().Array(), + } + } + if docSequence != nil { + // There can only be one of these arrays in a well-formed command, so we exit the loop once one is found. + break + } + } + + sm := &SentMessage{ + Command: commandDoc, + ReadPreference: rpDoc, + DocumentSequence: docSequence, + } + return sm, nil +} + func parseSentMessage(wm []byte) (*SentMessage, error) { // Re-assign the wire message to "remaining" so "wm" continues to point to the entire message after parsing. _, requestID, _, opcode, remaining, ok := wiremessage.ReadHeader(wm) diff --git a/x/mongo/driver/auth/speculative_scram_test.go b/x/mongo/driver/auth/speculative_scram_test.go index f2234e227c..a159891adc 100644 --- a/x/mongo/driver/auth/speculative_scram_test.go +++ b/x/mongo/driver/auth/speculative_scram_test.go @@ -93,7 +93,7 @@ func TestSpeculativeSCRAM(t *testing.T) { // Assert that the driver sent hello with the speculative authentication message. assert.Equal(t, len(tc.payloads), len(conn.Written), "expected %d wire messages to be sent, got %d", len(tc.payloads), (conn.Written)) - helloCmd, err := drivertest.GetCommandFromMsgWireMessage(<-conn.Written) + helloCmd, err := drivertest.GetCommandFromQueryWireMessage(<-conn.Written) assert.Nil(t, err, "error parsing hello command: %v", err) assertCommandName(t, helloCmd, handshake.LegacyHello) @@ -177,7 +177,7 @@ func TestSpeculativeSCRAM(t *testing.T) { assert.Equal(t, numResponses, len(conn.Written), "expected %d wire messages to be sent, got %d", numResponses, len(conn.Written)) - hello, err := drivertest.GetCommandFromMsgWireMessage(<-conn.Written) + hello, err := drivertest.GetCommandFromQueryWireMessage(<-conn.Written) assert.Nil(t, err, "error parsing hello command: %v", err) assertCommandName(t, hello, handshake.LegacyHello) _, err = hello.LookupErr("speculativeAuthenticate") diff --git a/x/mongo/driver/auth/speculative_x509_test.go b/x/mongo/driver/auth/speculative_x509_test.go index 13fdf2b185..cf46de6ffd 100644 --- a/x/mongo/driver/auth/speculative_x509_test.go +++ b/x/mongo/driver/auth/speculative_x509_test.go @@ -58,7 +58,7 @@ func TestSpeculativeX509(t *testing.T) { assert.Equal(t, numResponses, len(conn.Written), "expected %d wire messages to be sent, got %d", numResponses, len(conn.Written)) - hello, err := drivertest.GetCommandFromMsgWireMessage(<-conn.Written) + hello, err := drivertest.GetCommandFromQueryWireMessage(<-conn.Written) assert.Nil(t, err, "error parsing hello command: %v", err) assertCommandName(t, hello, handshake.LegacyHello) @@ -103,7 +103,7 @@ func TestSpeculativeX509(t *testing.T) { assert.Equal(t, numResponses, len(conn.Written), "expected %d wire messages to be sent, got %d", numResponses, len(conn.Written)) - hello, err := drivertest.GetCommandFromMsgWireMessage(<-conn.Written) + hello, err := drivertest.GetCommandFromQueryWireMessage(<-conn.Written) assert.Nil(t, err, "error parsing hello command: %v", err) assertCommandName(t, hello, handshake.LegacyHello) _, err = hello.LookupErr("speculativeAuthenticate") diff --git a/x/mongo/driver/drivertest/channel_conn.go b/x/mongo/driver/drivertest/channel_conn.go index d2ae8df248..27be4c264d 100644 --- a/x/mongo/driver/drivertest/channel_conn.go +++ b/x/mongo/driver/drivertest/channel_conn.go @@ -99,6 +99,38 @@ func MakeReply(doc bsoncore.Document) []byte { return bsoncore.UpdateLength(dst, idx, int32(len(dst[idx:]))) } +// GetCommandFromQueryWireMessage returns the command sent in an OP_QUERY wire message. +func GetCommandFromQueryWireMessage(wm []byte) (bsoncore.Document, error) { + var ok bool + _, _, _, _, wm, ok = wiremessage.ReadHeader(wm) + if !ok { + return nil, errors.New("could not read header") + } + _, wm, ok = wiremessage.ReadQueryFlags(wm) + if !ok { + return nil, errors.New("could not read flags") + } + _, wm, ok = wiremessage.ReadQueryFullCollectionName(wm) + if !ok { + return nil, errors.New("could not read fullCollectionName") + } + _, wm, ok = wiremessage.ReadQueryNumberToSkip(wm) + if !ok { + return nil, errors.New("could not read numberToSkip") + } + _, wm, ok = wiremessage.ReadQueryNumberToReturn(wm) + if !ok { + return nil, errors.New("could not read numberToReturn") + } + + var query bsoncore.Document + query, wm, ok = wiremessage.ReadQueryQuery(wm) + if !ok { + return nil, errors.New("could not read query") + } + return query, nil +} + // GetCommandFromMsgWireMessage returns the command document sent in an OP_MSG wire message. func GetCommandFromMsgWireMessage(wm []byte) (bsoncore.Document, error) { var ok bool diff --git a/x/mongo/driver/legacy.go b/x/mongo/driver/legacy.go index 9f3b8a39ac..c40f1f8091 100644 --- a/x/mongo/driver/legacy.go +++ b/x/mongo/driver/legacy.go @@ -19,4 +19,5 @@ const ( LegacyKillCursors LegacyListCollections LegacyListIndexes + LegacyHandshake ) diff --git a/x/mongo/driver/operation.go b/x/mongo/driver/operation.go index 8e52773503..229988e133 100644 --- a/x/mongo/driver/operation.go +++ b/x/mongo/driver/operation.go @@ -19,6 +19,7 @@ import ( "time" "go.mongodb.org/mongo-driver/bson" + "go.mongodb.org/mongo-driver/bson/bsontype" "go.mongodb.org/mongo-driver/bson/primitive" "go.mongodb.org/mongo-driver/event" "go.mongodb.org/mongo-driver/internal/csot" @@ -629,7 +630,7 @@ func (op Operation) Execute(ctx context.Context) error { } var startedInfo startedInformation - *wm, startedInfo, err = op.createMsgWireMessage(ctx, maxTimeMS, (*wm)[:0], desc, conn, requestID) + *wm, startedInfo, err = op.createWireMessage(ctx, maxTimeMS, (*wm)[:0], desc, conn, requestID) if err != nil { return err @@ -1103,6 +1104,85 @@ func (op Operation) addBatchArray(dst []byte) []byte { return dst } +func (op Operation) createLegacyHandshakeWireMessage( + maxTimeMS uint64, + dst []byte, + desc description.SelectedServer, +) ([]byte, startedInformation, error) { + var info startedInformation + flags := op.secondaryOK(desc) + var wmindex int32 + info.requestID = wiremessage.NextRequestID() + wmindex, dst = wiremessage.AppendHeaderStart(dst, info.requestID, 0, wiremessage.OpQuery) + dst = wiremessage.AppendQueryFlags(dst, flags) + + dollarCmd := [...]byte{'.', '$', 'c', 'm', 'd'} + + // FullCollectionName + dst = append(dst, op.Database...) + dst = append(dst, dollarCmd[:]...) + dst = append(dst, 0x00) + dst = wiremessage.AppendQueryNumberToSkip(dst, 0) + dst = wiremessage.AppendQueryNumberToReturn(dst, -1) + + wrapper := int32(-1) + rp, err := op.createReadPref(desc, true) + if err != nil { + return dst, info, err + } + if len(rp) > 0 { + wrapper, dst = bsoncore.AppendDocumentStart(dst) + dst = bsoncore.AppendHeader(dst, bsontype.EmbeddedDocument, "$query") + } + idx, dst := bsoncore.AppendDocumentStart(dst) + dst, err = op.CommandFn(dst, desc) + if err != nil { + return dst, info, err + } + + if op.Batches != nil && len(op.Batches.Current) > 0 { + dst = op.addBatchArray(dst) + } + + dst, err = op.addReadConcern(dst, desc) + if err != nil { + return dst, info, err + } + + dst, err = op.addWriteConcern(dst, desc) + if err != nil { + return dst, info, err + } + + dst, err = op.addSession(dst, desc) + if err != nil { + return dst, info, err + } + + dst = op.addClusterTime(dst, desc) + dst = op.addServerAPI(dst) + // If maxTimeMS is greater than 0 append it to wire message. A maxTimeMS value of 0 only explicitly + // specifies the default behavior of no timeout server-side. + if maxTimeMS > 0 { + dst = bsoncore.AppendInt64Element(dst, "maxTimeMS", int64(maxTimeMS)) + } + + dst, _ = bsoncore.AppendDocumentEnd(dst, idx) + // Command monitoring only reports the document inside $query + info.cmd = dst[idx:] + + if len(rp) > 0 { + var err error + dst = bsoncore.AppendDocumentElement(dst, "$readPreference", rp) + dst, err = bsoncore.AppendDocumentEnd(dst, wrapper) + if err != nil { + return dst, info, err + } + } + + return bsoncore.UpdateLength(dst, wmindex, int32(len(dst[wmindex:]))), info, nil +} + func (op Operation) createMsgWireMessage( ctx context.Context, maxTimeMS uint64, @@ -1191,6 +1271,29 @@ func (op Operation) createMsgWireMessage( return bsoncore.UpdateLength(dst, wmindex, int32(len(dst[wmindex:]))), info, nil } +// isLegacyHandshake returns True if the operation is the first message of +// the initial handshake and should use a legacy hello. +func isLegacyHandshake(op Operation, desc description.SelectedServer) bool { + isInitialHandshake := desc.WireVersion == nil || desc.WireVersion.Max == 0 + + return op.Legacy == LegacyHandshake && isInitialHandshake +} + +func (op Operation) createWireMessage( + ctx context.Context, + maxTimeMS uint64, + dst []byte, + desc description.SelectedServer, + conn Connection, + requestID int32, +) ([]byte, startedInformation, error) { + if isLegacyHandshake(op, desc) { + return op.createLegacyHandshakeWireMessage(maxTimeMS, dst, desc) + } + + return op.createMsgWireMessage(ctx, maxTimeMS, dst, desc, conn, requestID) +} + // addCommandFields adds the fields for a command to the wire message in dst. This assumes that the start of the document // has already been added and does not add the final 0 byte. func (op Operation) addCommandFields(ctx context.Context, dst []byte, desc description.SelectedServer) ([]byte, error) { diff --git a/x/mongo/driver/operation/hello.go b/x/mongo/driver/operation/hello.go index 3cfa2d450a..16d5809130 100644 --- a/x/mongo/driver/operation/hello.go +++ b/x/mongo/driver/operation/hello.go @@ -537,8 +537,16 @@ func (h *Hello) StreamResponse(ctx context.Context, conn driver.StreamerConnecti return h.createOperation().ExecuteExhaust(ctx, conn) } +// isLegacyHandshake returns True if server API version is not requested and +// loadBalanced is False. If this is the case, then the drivers MUST use legacy +// hello for the first message of the initial handshake with the OP_QUERY +// protocol +func isLegacyHandshake(srvAPI *driver.ServerAPIOptions, deployment driver.Deployment) bool { + return srvAPI == nil && deployment.Kind() != description.LoadBalanced +} + func (h *Hello) createOperation() driver.Operation { - return driver.Operation{ + op := driver.Operation{ Clock: h.clock, CommandFn: h.command, Database: "admin", @@ -549,23 +557,36 @@ func (h *Hello) createOperation() driver.Operation { }, ServerAPI: h.serverAPI, } + + if isLegacyHandshake(h.serverAPI, h.d) { + op.Legacy = driver.LegacyHandshake + } + + return op } // GetHandshakeInformation performs the MongoDB handshake for the provided connection and returns the relevant // information about the server. This function implements the driver.Handshaker interface. func (h *Hello) GetHandshakeInformation(ctx context.Context, _ address.Address, c driver.Connection) (driver.HandshakeInformation, error) { - err := driver.Operation{ + deployment := driver.SingleConnectionDeployment{C: c} + + op := driver.Operation{ Clock: h.clock, CommandFn: h.handshakeCommand, - Deployment: driver.SingleConnectionDeployment{C: c}, + Deployment: deployment, Database: "admin", ProcessResponseFn: func(info driver.ResponseInfo) error { h.res = info.ServerResponse return nil }, ServerAPI: h.serverAPI, - }.Execute(ctx) - if err != nil { + } + + if isLegacyHandshake(h.serverAPI, deployment) { + op.Legacy = driver.LegacyHandshake + } + + if err := op.Execute(ctx); err != nil { return driver.HandshakeInformation{}, err } @@ -578,6 +599,9 @@ func (h *Hello) GetHandshakeInformation(ctx context.Context, _ address.Address, if serverConnectionID, ok := h.res.Lookup("connectionId").AsInt64OK(); ok { info.ServerConnectionID = &serverConnectionID } + + var err error + // Cast to bson.Raw to lookup saslSupportedMechs to avoid converting from bsoncore.Value to bson.RawValue for the // StringSliceFromRawValue call. if saslSupportedMechs, lookupErr := bson.Raw(h.res).LookupErr("saslSupportedMechs"); lookupErr == nil { diff --git a/x/mongo/driver/topology/server_test.go b/x/mongo/driver/topology/server_test.go index a2abd1fb1f..ba92b6dd94 100644 --- a/x/mongo/driver/topology/server_test.go +++ b/x/mongo/driver/topology/server_test.go @@ -31,9 +31,11 @@ import ( "go.mongodb.org/mongo-driver/internal/require" "go.mongodb.org/mongo-driver/mongo/address" "go.mongodb.org/mongo-driver/mongo/description" + "go.mongodb.org/mongo-driver/x/bsonx/bsoncore" "go.mongodb.org/mongo-driver/x/mongo/driver" "go.mongodb.org/mongo-driver/x/mongo/driver/auth" "go.mongodb.org/mongo-driver/x/mongo/driver/drivertest" + "go.mongodb.org/mongo-driver/x/mongo/driver/wiremessage" ) type channelNetConnDialer struct{} @@ -1207,12 +1209,41 @@ func TestServer_ProcessError(t *testing.T) { func includesClientMetadata(t *testing.T, wm []byte) bool { t.Helper() - doc, err := drivertest.GetCommandFromMsgWireMessage(wm) - assert.NoError(t, err) + var ok bool + _, _, _, _, wm, ok = wiremessage.ReadHeader(wm) + if !ok { + t.Fatal("could not read header") + } + _, wm, ok = wiremessage.ReadQueryFlags(wm) + if !ok { + t.Fatal("could not read flags") + } + _, wm, ok = wiremessage.ReadQueryFullCollectionName(wm) + if !ok { + t.Fatal("could not read fullCollectionName") + } + _, wm, ok = wiremessage.ReadQueryNumberToSkip(wm) + if !ok { + t.Fatal("could not read numberToSkip") + } + _, wm, ok = wiremessage.ReadQueryNumberToReturn(wm) + if !ok { + t.Fatal("could not read numberToReturn") + } + var query bsoncore.Document + query, wm, ok = wiremessage.ReadQueryQuery(wm) + if !ok { + t.Fatal("could not read query") + } - _, err = doc.LookupErr("client") + if _, err := query.LookupErr("client"); err == nil { + return true + } + if _, err := query.LookupErr("$query", "client"); err == nil { + return true + } - return err == nil + return false } // processErrorTestConn is a driver.Connection implementation used by tests From b191b72002985fe8c4f7a8bbcbc9c3a56472b175 Mon Sep 17 00:00:00 2001 From: Matt Dale <9760375+matthewdale@users.noreply.github.com> Date: Fri, 8 Sep 2023 11:32:11 -0700 Subject: [PATCH 09/15] GODRIVER-2603 Use errors.Is/As for IsTimeout and IsDuplicateKeyError. (#1371) --- mongo/errors.go | 74 +++++++++++++++++--------------- mongo/integration/client_test.go | 2 +- x/mongo/driver/errors.go | 9 +++- 3 files changed, 48 insertions(+), 37 deletions(-) diff --git a/mongo/errors.go b/mongo/errors.go index 5f2b1b819b..72c3bcc243 100644 --- a/mongo/errors.go +++ b/mongo/errors.go @@ -102,50 +102,56 @@ func replaceErrors(err error) error { return err } -// IsDuplicateKeyError returns true if err is a duplicate key error +// IsDuplicateKeyError returns true if err is a duplicate key error. func IsDuplicateKeyError(err error) bool { - // handles SERVER-7164 and SERVER-11493 - for ; err != nil; err = unwrap(err) { - if e, ok := err.(ServerError); ok { - return e.HasErrorCode(11000) || e.HasErrorCode(11001) || e.HasErrorCode(12582) || - e.HasErrorCodeWithMessage(16460, " E11000 ") - } + if se := ServerError(nil); errors.As(err, &se) { + return se.HasErrorCode(11000) || // Duplicate key error. + se.HasErrorCode(11001) || // Duplicate key error on update. + // Duplicate key error in a capped collection. See SERVER-7164. + se.HasErrorCode(12582) || + // Mongos insert error caused by a duplicate key error. See + // SERVER-11493. + se.HasErrorCodeWithMessage(16460, " E11000 ") } return false } -// IsTimeout returns true if err is from a timeout +// timeoutErrs is a list of error values that indicate a timeout happened. +var timeoutErrs = [...]error{ + context.DeadlineExceeded, + driver.ErrDeadlineWouldBeExceeded, + topology.ErrServerSelectionTimeout, +} + +// IsTimeout returns true if err was caused by a timeout. For error chains, +// IsTimeout returns true if any error in the chain was caused by a timeout. func IsTimeout(err error) bool { - for ; err != nil; err = unwrap(err) { - // check unwrappable errors together - if err == context.DeadlineExceeded { - return true - } - if err == driver.ErrDeadlineWouldBeExceeded { - return true - } - if err == topology.ErrServerSelectionTimeout { - return true - } - if _, ok := err.(topology.WaitQueueTimeoutError); ok { - return true - } - if ce, ok := err.(CommandError); ok && ce.IsMaxTimeMSExpiredError() { + // Check if the error chain contains any of the timeout error values. + for _, target := range timeoutErrs { + if errors.Is(err, target) { return true } - if we, ok := err.(WriteException); ok && we.WriteConcernError != nil && - we.WriteConcernError.IsMaxTimeMSExpiredError() { + } + + // Check if the error chain contains any error types that can indicate + // timeout. + if errors.As(err, &topology.WaitQueueTimeoutError{}) { + return true + } + if ce := (CommandError{}); errors.As(err, &ce) && ce.IsMaxTimeMSExpiredError() { + return true + } + if we := (WriteException{}); errors.As(err, &we) && we.WriteConcernError != nil && we.WriteConcernError.IsMaxTimeMSExpiredError() { + return true + } + if ne := net.Error(nil); errors.As(err, &ne) { + return ne.Timeout() + } + // Check timeout error labels. + if le := LabeledError(nil); errors.As(err, &le) { + if le.HasErrorLabel("NetworkTimeoutError") || le.HasErrorLabel("ExceededTimeLimitError") { return true } - if ne, ok := err.(net.Error); ok { - return ne.Timeout() - } - //timeout error labels - if le, ok := err.(LabeledError); ok { - if le.HasErrorLabel("NetworkTimeoutError") || le.HasErrorLabel("ExceededTimeLimitError") { - return true - } - } } return false diff --git a/mongo/integration/client_test.go b/mongo/integration/client_test.go index 0dfe740358..038ed25d72 100644 --- a/mongo/integration/client_test.go +++ b/mongo/integration/client_test.go @@ -709,7 +709,7 @@ func TestClient(t *testing.T) { err := mt.Client.Ping(ctx, nil) cancel() assert.NotNil(mt, err, "expected Ping to return an error") - assert.True(mt, mongo.IsTimeout(err), "expected a timeout error: got %v", err) + assert.True(mt, mongo.IsTimeout(err), "expected a timeout error, got: %v", err) } // Assert that the Ping timeouts result in no connections being closed. diff --git a/x/mongo/driver/errors.go b/x/mongo/driver/errors.go index 55f2fb37eb..177aa1234b 100644 --- a/x/mongo/driver/errors.go +++ b/x/mongo/driver/errors.go @@ -264,10 +264,15 @@ func (e Error) UnsupportedStorageEngine() bool { // Error implements the error interface. func (e Error) Error() string { + var msg string if e.Name != "" { - return fmt.Sprintf("(%v) %v", e.Name, e.Message) + msg = fmt.Sprintf("(%v)", e.Name) } - return e.Message + msg += " " + e.Message + if e.Wrapped != nil { + msg += ": " + e.Wrapped.Error() + } + return msg } // Unwrap returns the underlying error. From d92f20d7e599800a0045cb48da0f0759ced9d972 Mon Sep 17 00:00:00 2001 From: Preston Vasquez Date: Tue, 12 Sep 2023 06:57:26 -0600 Subject: [PATCH 10/15] GODRIVER-2101 Direct read/write retries to another mongos if possible (#1358) * GODRIVER-2101 Expand test to use pigeonhole principle * GODRIVER-2101 Direct read/write retries to another mongos if possible * GODRIVER-2101 Revert unecessary changes * GODRIVER-2101 revert changes to collection and cursor * GODRIVER-2101 Apply opServerSelector * GODRIVER-2101 Fix static analysis errors * GODRIVER-2101 Remove empty line * GODRIVER-2101 Use map 'ok' value --- .../integration/retryable_reads_prose_test.go | 92 ++++++++++++ .../retryable_writes_prose_test.go | 92 ++++++++++++ x/mongo/driver/operation.go | 100 ++++++++++++- x/mongo/driver/operation_test.go | 137 +++++++++++++++++- 4 files changed, 408 insertions(+), 13 deletions(-) diff --git a/mongo/integration/retryable_reads_prose_test.go b/mongo/integration/retryable_reads_prose_test.go index 80d7937e8c..80f4d3329a 100644 --- a/mongo/integration/retryable_reads_prose_test.go +++ b/mongo/integration/retryable_reads_prose_test.go @@ -16,6 +16,7 @@ import ( "go.mongodb.org/mongo-driver/event" "go.mongodb.org/mongo-driver/internal/assert" "go.mongodb.org/mongo-driver/internal/eventtest" + "go.mongodb.org/mongo-driver/internal/require" "go.mongodb.org/mongo-driver/mongo/integration/mtest" "go.mongodb.org/mongo-driver/mongo/options" ) @@ -102,4 +103,95 @@ func TestRetryableReadsProse(t *testing.T) { "expected a find event, got a(n) %v event", cmdEvt.CommandName) } }) + + mtOpts = mtest.NewOptions().Topologies(mtest.Sharded).MinServerVersion("4.2") + mt.RunOpts("retrying in sharded cluster", mtOpts, func(mt *mtest.T) { + tests := []struct { + name string + + // Note that setting this value greater than 2 will result in false + // negatives. The current specification does not account for CSOT, which + // might allow for an "inifinite" number of retries over a period of time. + // Because of this, we only track the "previous server". + hostCount int + failpointErrorCode int32 + expectedFailCount int + expectedSuccessCount int + }{ + { + name: "retry on different mongos", + hostCount: 2, + failpointErrorCode: 6, // HostUnreachable + expectedFailCount: 2, + expectedSuccessCount: 0, + }, + { + name: "retry on same mongos", + hostCount: 1, + failpointErrorCode: 6, // HostUnreachable + expectedFailCount: 1, + expectedSuccessCount: 1, + }, + } + + for _, tc := range tests { + mt.Run(tc.name, func(mt *mtest.T) { + hosts := options.Client().ApplyURI(mtest.ClusterURI()).Hosts + require.GreaterOrEqualf(mt, len(hosts), tc.hostCount, + "test cluster must have at least %v mongos hosts", tc.hostCount) + + // Configure the failpoint options for each mongos. + failPoint := mtest.FailPoint{ + ConfigureFailPoint: "failCommand", + Mode: mtest.FailPointMode{ + Times: 1, + }, + Data: mtest.FailPointData{ + FailCommands: []string{"find"}, + ErrorCode: tc.failpointErrorCode, + CloseConnection: false, + }, + } + + // In order to ensure that each mongos in the hostCount-many mongos + // hosts are tried at least once (i.e. failures are deprioritized), we + // set a failpoint on all mongos hosts. The idea is that if we get + // hostCount-many failures, then by the pigeonhole principal all mongos + // hosts must have been tried. + for i := 0; i < tc.hostCount; i++ { + mt.ResetClient(options.Client().SetHosts([]string{hosts[i]})) + mt.SetFailPoint(failPoint) + + // The automatic failpoint clearing may not clear failpoints set on + // specific hosts, so manually clear the failpoint we set on the + // specific mongos when the test is done. + defer mt.ResetClient(options.Client().SetHosts([]string{hosts[i]})) + defer mt.ClearFailPoints() + } + + failCount := 0 + successCount := 0 + + commandMonitor := &event.CommandMonitor{ + Failed: func(context.Context, *event.CommandFailedEvent) { + failCount++ + }, + Succeeded: func(context.Context, *event.CommandSucceededEvent) { + successCount++ + }, + } + + // Reset the client with exactly hostCount-many mongos hosts. + mt.ResetClient(options.Client(). + SetHosts(hosts[:tc.hostCount]). + SetRetryReads(true). + SetMonitor(commandMonitor)) + + mt.Coll.FindOne(context.Background(), bson.D{}) + + assert.Equal(mt, tc.expectedFailCount, failCount) + assert.Equal(mt, tc.expectedSuccessCount, successCount) + }) + } + }) } diff --git a/mongo/integration/retryable_writes_prose_test.go b/mongo/integration/retryable_writes_prose_test.go index b378cdcbb5..1c8d353f14 100644 --- a/mongo/integration/retryable_writes_prose_test.go +++ b/mongo/integration/retryable_writes_prose_test.go @@ -284,4 +284,96 @@ func TestRetryableWritesProse(t *testing.T) { // Assert that the "ShutdownInProgress" error is returned. require.True(mt, err.(mongo.WriteException).HasErrorCode(int(shutdownInProgressErrorCode))) }) + + mtOpts = mtest.NewOptions().Topologies(mtest.Sharded).MinServerVersion("4.2") + mt.RunOpts("retrying in sharded cluster", mtOpts, func(mt *mtest.T) { + tests := []struct { + name string + + // Note that setting this value greater than 2 will result in false + // negatives. The current specification does not account for CSOT, which + // might allow for an "inifinite" number of retries over a period of time. + // Because of this, we only track the "previous server". + hostCount int + failpointErrorCode int32 + expectedFailCount int + expectedSuccessCount int + }{ + { + name: "retry on different mongos", + hostCount: 2, + failpointErrorCode: 6, // HostUnreachable + expectedFailCount: 2, + expectedSuccessCount: 0, + }, + { + name: "retry on same mongos", + hostCount: 1, + failpointErrorCode: 6, // HostUnreachable + expectedFailCount: 1, + expectedSuccessCount: 1, + }, + } + + for _, tc := range tests { + mt.Run(tc.name, func(mt *mtest.T) { + hosts := options.Client().ApplyURI(mtest.ClusterURI()).Hosts + require.GreaterOrEqualf(mt, len(hosts), tc.hostCount, + "test cluster must have at least %v mongos hosts", tc.hostCount) + + // Configure the failpoint options for each mongos. + failPoint := mtest.FailPoint{ + ConfigureFailPoint: "failCommand", + Mode: mtest.FailPointMode{ + Times: 1, + }, + Data: mtest.FailPointData{ + FailCommands: []string{"insert"}, + ErrorLabels: &[]string{"RetryableWriteError"}, + ErrorCode: tc.failpointErrorCode, + CloseConnection: false, + }, + } + + // In order to ensure that each mongos in the hostCount-many mongos + // hosts are tried at least once (i.e. failures are deprioritized), we + // set a failpoint on all mongos hosts. The idea is that if we get + // hostCount-many failures, then by the pigeonhole principal all mongos + // hosts must have been tried. + for i := 0; i < tc.hostCount; i++ { + mt.ResetClient(options.Client().SetHosts([]string{hosts[i]})) + mt.SetFailPoint(failPoint) + + // The automatic failpoint clearing may not clear failpoints set on + // specific hosts, so manually clear the failpoint we set on the + // specific mongos when the test is done. + defer mt.ResetClient(options.Client().SetHosts([]string{hosts[i]})) + defer mt.ClearFailPoints() + } + + failCount := 0 + successCount := 0 + + commandMonitor := &event.CommandMonitor{ + Failed: func(context.Context, *event.CommandFailedEvent) { + failCount++ + }, + Succeeded: func(context.Context, *event.CommandSucceededEvent) { + successCount++ + }, + } + + // Reset the client with exactly hostCount-many mongos hosts. + mt.ResetClient(options.Client(). + SetHosts(hosts[:tc.hostCount]). + SetRetryWrites(true). + SetMonitor(commandMonitor)) + + _, _ = mt.Coll.InsertOne(context.Background(), bson.D{}) + + assert.Equal(mt, tc.expectedFailCount, failCount) + assert.Equal(mt, tc.expectedSuccessCount, successCount) + }) + } + }) } diff --git a/x/mongo/driver/operation.go b/x/mongo/driver/operation.go index 229988e133..6b56191a01 100644 --- a/x/mongo/driver/operation.go +++ b/x/mongo/driver/operation.go @@ -322,8 +322,73 @@ func (op Operation) shouldEncrypt() bool { return op.Crypt != nil && !op.Crypt.BypassAutoEncryption() } +// filterDeprioritizedServers will filter out the server candidates that have +// been deprioritized by the operation due to failure. +// +// The server selector should try to select a server that is not in the +// deprioritization list. However, if this is not possible (e.g. there are no +// other healthy servers in the cluster), the selector may return a +// deprioritized server. +func filterDeprioritizedServers(candidates, deprioritized []description.Server) []description.Server { + if len(deprioritized) == 0 { + return candidates + } + + dpaSet := make(map[address.Address]*description.Server) + for i, srv := range deprioritized { + dpaSet[srv.Addr] = &deprioritized[i] + } + + allowed := []description.Server{} + + // Iterate over the candidates and append them to the allowdIndexes slice if + // they are not in the deprioritizedServers list. + for _, candidate := range candidates { + if srv, ok := dpaSet[candidate.Addr]; !ok || !srv.Equal(candidate) { + allowed = append(allowed, candidate) + } + } + + // If nothing is allowed, then all available servers must have been + // deprioritized. In this case, return the candidates list as-is so that the + // selector can find a suitable server + if len(allowed) == 0 { + return candidates + } + + return allowed +} + +// opServerSelector is a wrapper for the server selector that is assigned to the +// operation. The purpose of this wrapper is to filter candidates with +// operation-specific logic, such as deprioritizing failing servers. +type opServerSelector struct { + selector description.ServerSelector + deprioritizedServers []description.Server +} + +// SelectServer will filter candidates with operation-specific logic before +// passing them onto the user-defined or default selector. +func (oss *opServerSelector) SelectServer( + topo description.Topology, + candidates []description.Server, +) ([]description.Server, error) { + selectedServers, err := oss.selector.SelectServer(topo, candidates) + if err != nil { + return nil, err + } + + filteredServers := filterDeprioritizedServers(selectedServers, oss.deprioritizedServers) + + return filteredServers, nil +} + // selectServer handles performing server selection for an operation. -func (op Operation) selectServer(ctx context.Context, requestID int32) (Server, error) { +func (op Operation) selectServer( + ctx context.Context, + requestID int32, + deprioritized []description.Server, +) (Server, error) { if err := op.Validate(); err != nil { return nil, err } @@ -340,15 +405,24 @@ func (op Operation) selectServer(ctx context.Context, requestID int32) (Server, }) } + oss := &opServerSelector{ + selector: selector, + deprioritizedServers: deprioritized, + } + ctx = logger.WithOperationName(ctx, op.Name) ctx = logger.WithOperationID(ctx, requestID) - return op.Deployment.SelectServer(ctx, selector) + return op.Deployment.SelectServer(ctx, oss) } // getServerAndConnection should be used to retrieve a Server and Connection to execute an operation. -func (op Operation) getServerAndConnection(ctx context.Context, requestID int32) (Server, Connection, error) { - server, err := op.selectServer(ctx, requestID) +func (op Operation) getServerAndConnection( + ctx context.Context, + requestID int32, + deprioritized []description.Server, +) (Server, Connection, error) { + server, err := op.selectServer(ctx, requestID, deprioritized) if err != nil { if op.Client != nil && !(op.Client.Committing || op.Client.Aborting) && op.Client.TransactionRunning() { @@ -481,6 +555,11 @@ func (op Operation) Execute(ctx context.Context) error { first := true currIndex := 0 + // deprioritizedServers are a running list of servers that should be + // deprioritized during server selection. Per the specifications, we should + // only ever deprioritize the "previous server". + var deprioritizedServers []description.Server + // resetForRetry records the error that caused the retry, decrements retries, and resets the // retry loop variables to request a new server and a new connection for the next attempt. resetForRetry := func(err error) { @@ -506,11 +585,18 @@ func (op Operation) Execute(ctx context.Context) error { } } - // If we got a connection, close it immediately to release pool resources for - // subsequent retries. + // If we got a connection, close it immediately to release pool resources + // for subsequent retries. if conn != nil { + // If we are dealing with a sharded cluster, then mark the failed server + // as "deprioritized". + if desc := conn.Description; desc != nil && op.Deployment.Kind() == description.Sharded { + deprioritizedServers = []description.Server{conn.Description()} + } + conn.Close() } + // Set the server and connection to nil to request a new server and connection. srvr = nil conn = nil @@ -535,7 +621,7 @@ func (op Operation) Execute(ctx context.Context) error { // If the server or connection are nil, try to select a new server and get a new connection. if srvr == nil || conn == nil { - srvr, conn, err = op.getServerAndConnection(ctx, requestID) + srvr, conn, err = op.getServerAndConnection(ctx, requestID, deprioritizedServers) if err != nil { // If the returned error is retryable and there are retries remaining (negative // retries means retry indefinitely), then retry the operation. Set the server diff --git a/x/mongo/driver/operation_test.go b/x/mongo/driver/operation_test.go index 8509b5da9b..e6c9d4cf95 100644 --- a/x/mongo/driver/operation_test.go +++ b/x/mongo/driver/operation_test.go @@ -20,6 +20,7 @@ import ( "go.mongodb.org/mongo-driver/internal/assert" "go.mongodb.org/mongo-driver/internal/csot" "go.mongodb.org/mongo-driver/internal/handshake" + "go.mongodb.org/mongo-driver/internal/require" "go.mongodb.org/mongo-driver/internal/uuid" "go.mongodb.org/mongo-driver/mongo/address" "go.mongodb.org/mongo-driver/mongo/description" @@ -62,7 +63,7 @@ func TestOperation(t *testing.T) { t.Run("selectServer", func(t *testing.T) { t.Run("returns validation error", func(t *testing.T) { op := &Operation{} - _, err := op.selectServer(context.Background(), 1) + _, err := op.selectServer(context.Background(), 1, nil) if err == nil { t.Error("Expected a validation error from selectServer, but got ") } @@ -76,11 +77,15 @@ func TestOperation(t *testing.T) { Database: "testing", Selector: want, } - _, err := op.selectServer(context.Background(), 1) + _, err := op.selectServer(context.Background(), 1, nil) noerr(t, err) - got := d.params.selector - if !cmp.Equal(got, want) { - t.Errorf("Did not get expected server selector. got %v; want %v", got, want) + + // Assert the the selector is an operation selector wrapper. + oss, ok := d.params.selector.(*opServerSelector) + require.True(t, ok) + + if !cmp.Equal(oss.selector, want) { + t.Errorf("Did not get expected server selector. got %v; want %v", oss.selector, want) } }) t.Run("uses a default server selector", func(t *testing.T) { @@ -90,7 +95,7 @@ func TestOperation(t *testing.T) { Deployment: d, Database: "testing", } - _, err := op.selectServer(context.Background(), 1) + _, err := op.selectServer(context.Background(), 1, nil) noerr(t, err) if d.params.selector == nil { t.Error("The selectServer method should use a default selector when not specified on Operation, but it passed .") @@ -881,3 +886,123 @@ func TestDecodeOpReply(t *testing.T) { assert.Equal(t, []bsoncore.Document(nil), reply.documents) }) } + +func TestFilterDeprioritizedServers(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + deprioritized []description.Server + candidates []description.Server + want []description.Server + }{ + { + name: "empty", + candidates: []description.Server{}, + want: []description.Server{}, + }, + { + name: "nil candidates", + candidates: nil, + want: []description.Server{}, + }, + { + name: "nil deprioritized server list", + candidates: []description.Server{ + { + Addr: address.Address("mongodb://localhost:27017"), + }, + }, + want: []description.Server{ + { + Addr: address.Address("mongodb://localhost:27017"), + }, + }, + }, + { + name: "deprioritize single server candidate list", + candidates: []description.Server{ + { + Addr: address.Address("mongodb://localhost:27017"), + }, + }, + deprioritized: []description.Server{ + { + Addr: address.Address("mongodb://localhost:27017"), + }, + }, + want: []description.Server{ + // Since all available servers were deprioritized, then the selector + // should return all candidates. + { + Addr: address.Address("mongodb://localhost:27017"), + }, + }, + }, + { + name: "depriotirize one server in multi server candidate list", + candidates: []description.Server{ + { + Addr: address.Address("mongodb://localhost:27017"), + }, + { + Addr: address.Address("mongodb://localhost:27018"), + }, + { + Addr: address.Address("mongodb://localhost:27019"), + }, + }, + deprioritized: []description.Server{ + { + Addr: address.Address("mongodb://localhost:27017"), + }, + }, + want: []description.Server{ + { + Addr: address.Address("mongodb://localhost:27018"), + }, + { + Addr: address.Address("mongodb://localhost:27019"), + }, + }, + }, + { + name: "depriotirize multiple servers in multi server candidate list", + deprioritized: []description.Server{ + { + Addr: address.Address("mongodb://localhost:27017"), + }, + { + Addr: address.Address("mongodb://localhost:27018"), + }, + }, + candidates: []description.Server{ + { + Addr: address.Address("mongodb://localhost:27017"), + }, + { + Addr: address.Address("mongodb://localhost:27018"), + }, + { + Addr: address.Address("mongodb://localhost:27019"), + }, + }, + want: []description.Server{ + { + Addr: address.Address("mongodb://localhost:27019"), + }, + }, + }, + } + + for _, tc := range tests { + tc := tc // Capture the range variable. + + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + got := filterDeprioritizedServers(tc.candidates, tc.deprioritized) + assert.ElementsMatch(t, got, tc.want) + }) + } +} From 101d1e70ccbe184efcbc2679b8f3bd76c062f9b2 Mon Sep 17 00:00:00 2001 From: Qingyang Hu <103950869+qingyang-hu@users.noreply.github.com> Date: Tue, 12 Sep 2023 09:48:12 -0400 Subject: [PATCH 11/15] GODRIVER-2904 Add environment log. (#1373) * GODRIVER-2904 Add environment log. * minor updates * fix typos * fix typos * log before SRV lookup * improve logic --- .../topology/polling_srv_records_test.go | 1 + x/mongo/driver/topology/topology.go | 78 ++++-- x/mongo/driver/topology/topology_test.go | 234 ++++++++++++++++++ 3 files changed, 291 insertions(+), 22 deletions(-) diff --git a/x/mongo/driver/topology/polling_srv_records_test.go b/x/mongo/driver/topology/polling_srv_records_test.go index 0ca5c7cbce..7484109d4e 100644 --- a/x/mongo/driver/topology/polling_srv_records_test.go +++ b/x/mongo/driver/topology/polling_srv_records_test.go @@ -105,6 +105,7 @@ func (ss serverSorter) Less(i, j int) bool { } func compareHosts(t *testing.T, received []description.Server, expected []string) { + t.Helper() if len(received) != len(expected) { t.Fatalf("Number of hosts in topology does not match expected value. Got %v; want %v.", len(received), len(expected)) } diff --git a/x/mongo/driver/topology/topology.go b/x/mongo/driver/topology/topology.go index b0683021ee..b79efed4ed 100644 --- a/x/mongo/driver/topology/topology.go +++ b/x/mongo/driver/topology/topology.go @@ -161,13 +161,13 @@ func New(cfg *Config) (*Topology, error) { return t, nil } -func mustLogTopologyMessage(topo *Topology) bool { +func mustLogTopologyMessage(topo *Topology, level logger.Level) bool { return topo.cfg.logger != nil && topo.cfg.logger.LevelComponentEnabled( - logger.LevelDebug, logger.ComponentTopology) + level, logger.ComponentTopology) } -func logTopologyMessage(topo *Topology, msg string, keysAndValues ...interface{}) { - topo.cfg.logger.Print(logger.LevelDebug, +func logTopologyMessage(topo *Topology, level logger.Level, msg string, keysAndValues ...interface{}) { + topo.cfg.logger.Print(level, logger.ComponentTopology, msg, logger.SerializeTopology(logger.Topology{ @@ -176,6 +176,36 @@ func logTopologyMessage(topo *Topology, msg string, keysAndValues ...interface{} }, keysAndValues...)...) } +func logTopologyThirdPartyUsage(topo *Topology, parsedHosts []string) { + thirdPartyMessages := [2]string{ + `You appear to be connected to a CosmosDB cluster. For more information regarding feature compatibility and support please visit https://www.mongodb.com/supportability/cosmosdb`, + `You appear to be connected to a DocumentDB cluster. For more information regarding feature compatibility and support please visit https://www.mongodb.com/supportability/documentdb`, + } + + thirdPartySuffixes := map[string]int{ + ".cosmos.azure.com": 0, + ".docdb.amazonaws.com": 1, + ".docdb-elastic.amazonaws.com": 1, + } + + hostSet := make([]bool, len(thirdPartyMessages)) + for _, host := range parsedHosts { + if h, _, err := net.SplitHostPort(host); err == nil { + host = h + } + for suffix, env := range thirdPartySuffixes { + if !strings.HasSuffix(host, suffix) { + continue + } + if hostSet[env] { + break + } + hostSet[env] = true + logTopologyMessage(topo, logger.LevelInfo, thirdPartyMessages[env]) + } + } +} + func mustLogServerSelection(topo *Topology, level logger.Level) bool { return topo.cfg.logger != nil && topo.cfg.logger.LevelComponentEnabled( level, logger.ComponentServerSelection) @@ -183,8 +213,8 @@ func mustLogServerSelection(topo *Topology, level logger.Level) bool { func logServerSelection( ctx context.Context, - level logger.Level, topo *Topology, + level logger.Level, msg string, srvSelector description.ServerSelector, keysAndValues ...interface{}, @@ -224,7 +254,7 @@ func logServerSelectionSucceeded( portInt64, _ := strconv.ParseInt(port, 10, 32) - logServerSelection(ctx, logger.LevelDebug, topo, logger.ServerSelectionSucceeded, srvSelector, + logServerSelection(ctx, topo, logger.LevelDebug, logger.ServerSelectionSucceeded, srvSelector, logger.KeyServerHost, host, logger.KeyServerPort, portInt64) } @@ -235,7 +265,7 @@ func logServerSelectionFailed( srvSelector description.ServerSelector, err error, ) { - logServerSelection(ctx, logger.LevelDebug, topo, logger.ServerSelectionFailed, srvSelector, + logServerSelection(ctx, topo, logger.LevelDebug, logger.ServerSelectionFailed, srvSelector, logger.KeyFailure, err.Error()) } @@ -321,13 +351,17 @@ func (t *Topology) Connect() error { } t.serversLock.Unlock() + uri, err := url.Parse(t.cfg.URI) + if err != nil { + return err + } + parsedHosts := strings.Split(uri.Host, ",") + if mustLogTopologyMessage(t, logger.LevelInfo) { + logTopologyThirdPartyUsage(t, parsedHosts) + } if t.pollingRequired { - uri, err := url.Parse(t.cfg.URI) - if err != nil { - return err - } // sanity check before passing the hostname to resolver - if parsedHosts := strings.Split(uri.Host, ","); len(parsedHosts) != 1 { + if len(parsedHosts) != 1 { return fmt.Errorf("URI with SRV must include one and only one hostname") } _, _, err = net.SplitHostPort(uri.Host) @@ -492,7 +526,7 @@ func (t *Topology) SelectServer(ctx context.Context, ss description.ServerSelect if !doneOnce { if mustLogServerSelection(t, logger.LevelDebug) { - logServerSelection(ctx, logger.LevelDebug, t, logger.ServerSelectionStarted, ss) + logServerSelection(ctx, t, logger.LevelDebug, logger.ServerSelectionStarted, ss) } // for the first pass, select a server from the current description. @@ -531,7 +565,7 @@ func (t *Topology) SelectServer(ctx context.Context, ss description.ServerSelect elapsed := time.Since(startTime) remainingTimeMS := t.cfg.ServerSelectionTimeout - elapsed - logServerSelection(ctx, logger.LevelInfo, t, logger.ServerSelectionWaiting, ss, + logServerSelection(ctx, t, logger.LevelInfo, logger.ServerSelectionWaiting, ss, logger.KeyRemainingTimeMS, remainingTimeMS.Milliseconds()) } @@ -970,7 +1004,7 @@ func (t *Topology) publishServerClosedEvent(addr address.Address) { t.cfg.ServerMonitor.ServerClosed(serverClosed) } - if mustLogTopologyMessage(t) { + if mustLogTopologyMessage(t, logger.LevelDebug) { serverHost, serverPort, err := net.SplitHostPort(addr.String()) if err != nil { serverHost = addr.String() @@ -979,7 +1013,7 @@ func (t *Topology) publishServerClosedEvent(addr address.Address) { portInt64, _ := strconv.ParseInt(serverPort, 10, 32) - logTopologyMessage(t, logger.TopologyServerClosed, + logTopologyMessage(t, logger.LevelDebug, logger.TopologyServerClosed, logger.KeyServerHost, serverHost, logger.KeyServerPort, portInt64) } @@ -997,8 +1031,8 @@ func (t *Topology) publishTopologyDescriptionChangedEvent(prev description.Topol t.cfg.ServerMonitor.TopologyDescriptionChanged(topologyDescriptionChanged) } - if mustLogTopologyMessage(t) { - logTopologyMessage(t, logger.TopologyDescriptionChanged, + if mustLogTopologyMessage(t, logger.LevelDebug) { + logTopologyMessage(t, logger.LevelDebug, logger.TopologyDescriptionChanged, logger.KeyPreviousDescription, prev.String(), logger.KeyNewDescription, current.String()) } @@ -1014,8 +1048,8 @@ func (t *Topology) publishTopologyOpeningEvent() { t.cfg.ServerMonitor.TopologyOpening(topologyOpening) } - if mustLogTopologyMessage(t) { - logTopologyMessage(t, logger.TopologyOpening) + if mustLogTopologyMessage(t, logger.LevelDebug) { + logTopologyMessage(t, logger.LevelDebug, logger.TopologyOpening) } } @@ -1029,7 +1063,7 @@ func (t *Topology) publishTopologyClosedEvent() { t.cfg.ServerMonitor.TopologyClosed(topologyClosed) } - if mustLogTopologyMessage(t) { - logTopologyMessage(t, logger.TopologyClosed) + if mustLogTopologyMessage(t, logger.LevelDebug) { + logTopologyMessage(t, logger.LevelDebug, logger.TopologyClosed) } } diff --git a/x/mongo/driver/topology/topology_test.go b/x/mongo/driver/topology/topology_test.go index 773a8b6475..6cf540a95e 100644 --- a/x/mongo/driver/topology/topology_test.go +++ b/x/mongo/driver/topology/topology_test.go @@ -693,6 +693,240 @@ func TestTopologyConstruction(t *testing.T) { }) } +type mockLogSink struct { + msgs []string +} + +func (s *mockLogSink) Info(_ int, msg string, _ ...interface{}) { + s.msgs = append(s.msgs, msg) +} +func (*mockLogSink) Error(error, string, ...interface{}) { + // Do nothing. +} + +// Note: SRV connection strings are intentionally untested, since initial +// lookup responses cannot be easily mocked. +func TestTopologyConstructionLogging(t *testing.T) { + const ( + cosmosDBMsg = `You appear to be connected to a CosmosDB cluster. For more information regarding feature compatibility and support please visit https://www.mongodb.com/supportability/cosmosdb` + documentDBMsg = `You appear to be connected to a DocumentDB cluster. For more information regarding feature compatibility and support please visit https://www.mongodb.com/supportability/documentdb` + ) + + newLoggerOptions := func(sink options.LogSink) *options.LoggerOptions { + return options. + Logger(). + SetSink(sink). + SetComponentLevel(options.LogComponentTopology, options.LogLevelInfo) + } + + t.Run("CosmosDB URIs", func(t *testing.T) { + t.Parallel() + + testCases := []struct { + name string + uri string + msgs []string + }{ + { + name: "normal", + uri: "mongodb://a.mongo.cosmos.azure.com:19555/", + msgs: []string{cosmosDBMsg}, + }, + { + name: "multiple hosts", + uri: "mongodb://a.mongo.cosmos.azure.com:1955,b.mongo.cosmos.azure.com:19555/", + msgs: []string{cosmosDBMsg}, + }, + { + name: "case-insensitive matching", + uri: "mongodb://a.MONGO.COSMOS.AZURE.COM:19555/", + msgs: []string{}, + }, + { + name: "Mixing genuine and nongenuine hosts (unlikely in practice)", + uri: "mongodb://a.example.com:27017,b.mongo.cosmos.azure.com:19555/", + msgs: []string{cosmosDBMsg}, + }, + } + for _, tc := range testCases { + tc := tc + + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + sink := &mockLogSink{} + cfg, err := NewConfig(options.Client().ApplyURI(tc.uri).SetLoggerOptions(newLoggerOptions(sink)), nil) + require.Nil(t, err, "error constructing topology config: %v", err) + + topo, err := New(cfg) + require.Nil(t, err, "topology.New error: %v", err) + + err = topo.Connect() + assert.Nil(t, err, "Connect error: %v", err) + + assert.ElementsMatch(t, tc.msgs, sink.msgs, "expected messages to be %v, got %v", tc.msgs, sink.msgs) + }) + } + }) + t.Run("DocumentDB URIs", func(t *testing.T) { + t.Parallel() + + testCases := []struct { + name string + uri string + msgs []string + }{ + { + name: "normal", + uri: "mongodb://a.docdb.amazonaws.com:27017/", + msgs: []string{documentDBMsg}, + }, + { + name: "normal", + uri: "mongodb://a.docdb-elastic.amazonaws.com:27017/", + msgs: []string{documentDBMsg}, + }, + { + name: "multiple hosts", + uri: "mongodb://a.docdb.amazonaws.com:27017,a.docdb-elastic.amazonaws.com:27017/", + msgs: []string{documentDBMsg}, + }, + { + name: "case-insensitive matching", + uri: "mongodb://a.DOCDB.AMAZONAWS.COM:27017/", + msgs: []string{}, + }, + { + name: "case-insensitive matching", + uri: "mongodb://a.DOCDB-ELASTIC.AMAZONAWS.COM:27017/", + msgs: []string{}, + }, + { + name: "Mixing genuine and nongenuine hosts (unlikely in practice)", + uri: "mongodb://a.example.com:27017,b.docdb.amazonaws.com:27017/", + msgs: []string{documentDBMsg}, + }, + { + name: "Mixing genuine and nongenuine hosts (unlikely in practice)", + uri: "mongodb://a.example.com:27017,b.docdb-elastic.amazonaws.com:27017/", + msgs: []string{documentDBMsg}, + }, + } + for _, tc := range testCases { + tc := tc + + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + sink := &mockLogSink{} + cfg, err := NewConfig(options.Client().ApplyURI(tc.uri).SetLoggerOptions(newLoggerOptions(sink)), nil) + require.Nil(t, err, "error constructing topology config: %v", err) + + topo, err := New(cfg) + require.Nil(t, err, "topology.New error: %v", err) + + err = topo.Connect() + assert.Nil(t, err, "Connect error: %v", err) + + assert.ElementsMatch(t, tc.msgs, sink.msgs, "expected messages to be %v, got %v", tc.msgs, sink.msgs) + }) + } + }) + t.Run("Mixing CosmosDB and DocumentDB URIs", func(t *testing.T) { + t.Parallel() + + testCases := []struct { + name string + uri string + msgs []string + }{ + { + name: "Mixing hosts", + uri: "mongodb://a.mongo.cosmos.azure.com:19555,a.docdb.amazonaws.com:27017/", + msgs: []string{cosmosDBMsg, documentDBMsg}, + }, + } + for _, tc := range testCases { + tc := tc + + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + sink := &mockLogSink{} + cfg, err := NewConfig(options.Client().ApplyURI(tc.uri).SetLoggerOptions(newLoggerOptions(sink)), nil) + require.Nil(t, err, "error constructing topology config: %v", err) + + topo, err := New(cfg) + require.Nil(t, err, "topology.New error: %v", err) + + err = topo.Connect() + assert.Nil(t, err, "Connect error: %v", err) + + assert.ElementsMatch(t, tc.msgs, sink.msgs, "expected messages to be %v, got %v", tc.msgs, sink.msgs) + }) + } + }) + t.Run("genuine URIs", func(t *testing.T) { + t.Parallel() + + testCases := []struct { + name string + uri string + msgs []string + }{ + { + name: "normal", + uri: "mongodb://a.example.com:27017/", + msgs: []string{}, + }, + { + name: "srv", + uri: "mongodb+srv://test22.test.build.10gen.cc/?srvServiceName=customname", + msgs: []string{}, + }, + { + name: "multiple hosts", + uri: "mongodb://a.example.com:27017,b.example.com:27017/", + msgs: []string{}, + }, + { + name: "unexpected suffix", + uri: "mongodb://a.mongo.cosmos.azure.com.tld:19555/", + msgs: []string{}, + }, + { + name: "unexpected suffix", + uri: "mongodb://a.docdb.amazonaws.com.tld:27017/", + msgs: []string{}, + }, + { + name: "unexpected suffix", + uri: "mongodb://a.docdb-elastic.amazonaws.com.tld:27017/", + msgs: []string{}, + }, + } + for _, tc := range testCases { + tc := tc + + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + sink := &mockLogSink{} + cfg, err := NewConfig(options.Client().ApplyURI(tc.uri).SetLoggerOptions(newLoggerOptions(sink)), nil) + require.Nil(t, err, "error constructing topology config: %v", err) + + topo, err := New(cfg) + require.Nil(t, err, "topology.New error: %v", err) + + err = topo.Connect() + assert.Nil(t, err, "Connect error: %v", err) + + assert.ElementsMatch(t, tc.msgs, sink.msgs, "expected messages to be %v, got %v", tc.msgs, sink.msgs) + }) + } + }) +} + type inWindowServer struct { Address string `json:"address"` Type string `json:"type"` From 5a5a23150c386afc1e7341da13b30bf58036f414 Mon Sep 17 00:00:00 2001 From: Charlie Vieth Date: Tue, 12 Sep 2023 15:12:36 -0400 Subject: [PATCH 12/15] GODRIVER-2914 bsoncodec/bsonrw: eliminate encoding allocations (#1323) Co-authored-by: Matt Dale <9760375+matthewdale@users.noreply.github.com> Co-authored-by: Qingyang Hu <103950869+qingyang-hu@users.noreply.github.com> --- bson/bsoncodec/slice_codec.go | 2 +- bson/bsoncodec/struct_codec.go | 73 ++++++++++------------------ bson/bsoncodec/struct_codec_test.go | 3 +- bson/bsoncodec/types.go | 1 + bson/bsonrw/copier.go | 6 +-- bson/bsonrw/value_reader.go | 12 ++++- bson/bsonrw/value_writer.go | 74 +++++++++++++++++++---------- bson/marshal.go | 32 +++++++++++-- 8 files changed, 119 insertions(+), 84 deletions(-) diff --git a/bson/bsoncodec/slice_codec.go b/bson/bsoncodec/slice_codec.go index 20c3e7549c..a43daf005f 100644 --- a/bson/bsoncodec/slice_codec.go +++ b/bson/bsoncodec/slice_codec.go @@ -62,7 +62,7 @@ func (sc SliceCodec) EncodeValue(ec EncodeContext, vw bsonrw.ValueWriter, val re } // If we have a []primitive.E we want to treat it as a document instead of as an array. - if val.Type().ConvertibleTo(tD) { + if val.Type() == tD || val.Type().ConvertibleTo(tD) { d := val.Convert(tD).Interface().(primitive.D) dw, err := vw.WriteDocument() diff --git a/bson/bsoncodec/struct_codec.go b/bson/bsoncodec/struct_codec.go index 29ea76d19c..4cde0a4d6b 100644 --- a/bson/bsoncodec/struct_codec.go +++ b/bson/bsoncodec/struct_codec.go @@ -190,15 +190,14 @@ func (sc *StructCodec) EncodeValue(ec EncodeContext, vw bsonrw.ValueWriter, val encoder := desc.encoder var zero bool - rvInterface := rv.Interface() if cz, ok := encoder.(CodecZeroer); ok { - zero = cz.IsTypeZero(rvInterface) + zero = cz.IsTypeZero(rv.Interface()) } else if rv.Kind() == reflect.Interface { // isZero will not treat an interface rv as an interface, so we need to check for the // zero interface separately. zero = rv.IsNil() } else { - zero = isZero(rvInterface, sc.EncodeOmitDefaultStruct || ec.omitZeroStruct) + zero = isZero(rv, sc.EncodeOmitDefaultStruct || ec.omitZeroStruct) } if desc.omitEmpty && zero { continue @@ -392,56 +391,32 @@ func (sc *StructCodec) DecodeValue(dc DecodeContext, vr bsonrw.ValueReader, val return nil } -func isZero(i interface{}, omitZeroStruct bool) bool { - v := reflect.ValueOf(i) - - // check the value validity - if !v.IsValid() { - return true +func isZero(v reflect.Value, omitZeroStruct bool) bool { + kind := v.Kind() + if (kind != reflect.Ptr || !v.IsNil()) && v.Type().Implements(tZeroer) { + return v.Interface().(Zeroer).IsZero() } - - if z, ok := v.Interface().(Zeroer); ok && (v.Kind() != reflect.Ptr || !v.IsNil()) { - return z.IsZero() - } - - switch v.Kind() { - case reflect.Array, reflect.Map, reflect.Slice, reflect.String: - return v.Len() == 0 - case reflect.Bool: - return !v.Bool() - case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64: - return v.Int() == 0 - case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64, reflect.Uintptr: - return v.Uint() == 0 - case reflect.Float32, reflect.Float64: - return v.Float() == 0 - case reflect.Interface, reflect.Ptr: - return v.IsNil() - case reflect.Struct: + if kind == reflect.Struct { if !omitZeroStruct { return false } - - // TODO(GODRIVER-2820): Update the logic to be able to handle private struct fields. - // TODO Use condition "reflect.Zero(v.Type()).Equal(v)" instead. - vt := v.Type() if vt == tTime { return v.Interface().(time.Time).IsZero() } - for i := 0; i < v.NumField(); i++ { - if vt.Field(i).PkgPath != "" && !vt.Field(i).Anonymous { + numField := vt.NumField() + for i := 0; i < numField; i++ { + ff := vt.Field(i) + if ff.PkgPath != "" && !ff.Anonymous { continue // Private field } - fld := v.Field(i) - if !isZero(fld.Interface(), omitZeroStruct) { + if !isZero(v.Field(i), omitZeroStruct) { return false } } return true } - - return false + return !v.IsValid() || v.IsZero() } type structDescription struct { @@ -708,21 +683,21 @@ func getInlineField(val reflect.Value, index []int) (reflect.Value, error) { // DeepZero returns recursive zero object func deepZero(st reflect.Type) (result reflect.Value) { - result = reflect.Indirect(reflect.New(st)) - - if result.Kind() == reflect.Struct { - for i := 0; i < result.NumField(); i++ { - if f := result.Field(i); f.Kind() == reflect.Ptr { - if f.CanInterface() { - if ft := reflect.TypeOf(f.Interface()); ft.Elem().Kind() == reflect.Struct { - result.Field(i).Set(recursivePointerTo(deepZero(ft.Elem()))) - } + if st.Kind() == reflect.Struct { + numField := st.NumField() + for i := 0; i < numField; i++ { + if result == emptyValue { + result = reflect.Indirect(reflect.New(st)) + } + f := result.Field(i) + if f.CanInterface() { + if f.Type().Kind() == reflect.Struct { + result.Field(i).Set(recursivePointerTo(deepZero(f.Type().Elem()))) } } } } - - return + return result } // recursivePointerTo calls reflect.New(v.Type) but recursively for its fields inside diff --git a/bson/bsoncodec/struct_codec_test.go b/bson/bsoncodec/struct_codec_test.go index 008fc11528..573b374b14 100644 --- a/bson/bsoncodec/struct_codec_test.go +++ b/bson/bsoncodec/struct_codec_test.go @@ -7,6 +7,7 @@ package bsoncodec import ( + "reflect" "testing" "time" @@ -147,7 +148,7 @@ func TestIsZero(t *testing.T) { t.Run(tc.description, func(t *testing.T) { t.Parallel() - got := isZero(tc.value, tc.omitZeroStruct) + got := isZero(reflect.ValueOf(tc.value), tc.omitZeroStruct) assert.Equal(t, tc.want, got, "expected and actual isZero return are different") }) } diff --git a/bson/bsoncodec/types.go b/bson/bsoncodec/types.go index 07f4b70e6d..6ade17b7d3 100644 --- a/bson/bsoncodec/types.go +++ b/bson/bsoncodec/types.go @@ -34,6 +34,7 @@ var tValueUnmarshaler = reflect.TypeOf((*ValueUnmarshaler)(nil)).Elem() var tMarshaler = reflect.TypeOf((*Marshaler)(nil)).Elem() var tUnmarshaler = reflect.TypeOf((*Unmarshaler)(nil)).Elem() var tProxy = reflect.TypeOf((*Proxy)(nil)).Elem() +var tZeroer = reflect.TypeOf((*Zeroer)(nil)).Elem() var tBinary = reflect.TypeOf(primitive.Binary{}) var tUndefined = reflect.TypeOf(primitive.Undefined{}) diff --git a/bson/bsonrw/copier.go b/bson/bsonrw/copier.go index 33d59bd258..c146d02e58 100644 --- a/bson/bsonrw/copier.go +++ b/bson/bsonrw/copier.go @@ -193,7 +193,7 @@ func (c Copier) AppendDocumentBytes(dst []byte, src ValueReader) ([]byte, error) } vw := vwPool.Get().(*valueWriter) - defer vwPool.Put(vw) + defer putValueWriter(vw) vw.reset(dst) @@ -213,7 +213,7 @@ func (c Copier) AppendArrayBytes(dst []byte, src ValueReader) ([]byte, error) { } vw := vwPool.Get().(*valueWriter) - defer vwPool.Put(vw) + defer putValueWriter(vw) vw.reset(dst) @@ -258,7 +258,7 @@ func (c Copier) AppendValueBytes(dst []byte, src ValueReader) (bsontype.Type, [] } vw := vwPool.Get().(*valueWriter) - defer vwPool.Put(vw) + defer putValueWriter(vw) start := len(dst) diff --git a/bson/bsonrw/value_reader.go b/bson/bsonrw/value_reader.go index 9bf24fae0b..a242bb57cf 100644 --- a/bson/bsonrw/value_reader.go +++ b/bson/bsonrw/value_reader.go @@ -739,8 +739,7 @@ func (vr *valueReader) ReadValue() (ValueReader, error) { return nil, ErrEOA } - _, err = vr.readCString() - if err != nil { + if err := vr.skipCString(); err != nil { return nil, err } @@ -794,6 +793,15 @@ func (vr *valueReader) readByte() (byte, error) { return vr.d[vr.offset-1], nil } +func (vr *valueReader) skipCString() error { + idx := bytes.IndexByte(vr.d[vr.offset:], 0x00) + if idx < 0 { + return io.EOF + } + vr.offset += int64(idx) + 1 + return nil +} + func (vr *valueReader) readCString() (string, error) { idx := bytes.IndexByte(vr.d[vr.offset:], 0x00) if idx < 0 { diff --git a/bson/bsonrw/value_writer.go b/bson/bsonrw/value_writer.go index a6dd8d34f5..311518a80d 100644 --- a/bson/bsonrw/value_writer.go +++ b/bson/bsonrw/value_writer.go @@ -28,6 +28,13 @@ var vwPool = sync.Pool{ }, } +func putValueWriter(vw *valueWriter) { + if vw != nil { + vw.w = nil // don't leak the writer + vwPool.Put(vw) + } +} + // BSONValueWriterPool is a pool for BSON ValueWriters. // // Deprecated: BSONValueWriterPool will not be supported in Go Driver 2.0. @@ -149,32 +156,21 @@ type valueWriter struct { } func (vw *valueWriter) advanceFrame() { - if vw.frame+1 >= int64(len(vw.stack)) { // We need to grow the stack - length := len(vw.stack) - if length+1 >= cap(vw.stack) { - // double it - buf := make([]vwState, 2*cap(vw.stack)+1) - copy(buf, vw.stack) - vw.stack = buf - } - vw.stack = vw.stack[:length+1] - } vw.frame++ + if vw.frame >= int64(len(vw.stack)) { + vw.stack = append(vw.stack, vwState{}) + } } func (vw *valueWriter) push(m mode) { vw.advanceFrame() // Clean the stack - vw.stack[vw.frame].mode = m - vw.stack[vw.frame].key = "" - vw.stack[vw.frame].arrkey = 0 - vw.stack[vw.frame].start = 0 + vw.stack[vw.frame] = vwState{mode: m} - vw.stack[vw.frame].mode = m switch m { case mDocument, mArray, mCodeWithScope: - vw.reserveLength() + vw.reserveLength() // WARN: this is not needed } } @@ -213,6 +209,7 @@ func newValueWriter(w io.Writer) *valueWriter { return vw } +// TODO: only used in tests func newValueWriterFromSlice(buf []byte) *valueWriter { vw := new(valueWriter) stack := make([]vwState, 1, 5) @@ -249,17 +246,16 @@ func (vw *valueWriter) invalidTransitionError(destination mode, name string, mod } func (vw *valueWriter) writeElementHeader(t bsontype.Type, destination mode, callerName string, addmodes ...mode) error { - switch vw.stack[vw.frame].mode { + frame := &vw.stack[vw.frame] + switch frame.mode { case mElement: - key := vw.stack[vw.frame].key + key := frame.key if !isValidCString(key) { return errors.New("BSON element key cannot contain null bytes") } - - vw.buf = bsoncore.AppendHeader(vw.buf, t, key) + vw.appendHeader(t, key) case mValue: - // TODO: Do this with a cache of the first 1000 or so array keys. - vw.buf = bsoncore.AppendHeader(vw.buf, t, strconv.Itoa(vw.stack[vw.frame].arrkey)) + vw.appendIntHeader(t, frame.arrkey) default: modes := []mode{mElement, mValue} if addmodes != nil { @@ -601,9 +597,11 @@ func (vw *valueWriter) writeLength() error { if length > maxSize { return errMaxDocumentSizeExceeded{size: int64(len(vw.buf))} } - length = length - int(vw.stack[vw.frame].start) - start := vw.stack[vw.frame].start + frame := &vw.stack[vw.frame] + length = length - int(frame.start) + start := frame.start + _ = vw.buf[start+3] // BCE vw.buf[start+0] = byte(length) vw.buf[start+1] = byte(length >> 8) vw.buf[start+2] = byte(length >> 16) @@ -612,5 +610,31 @@ func (vw *valueWriter) writeLength() error { } func isValidCString(cs string) bool { - return !strings.ContainsRune(cs, '\x00') + // Disallow the zero byte in a cstring because the zero byte is used as the + // terminating character. + // + // It's safe to check bytes instead of runes because all multibyte UTF-8 + // code points start with (binary) 11xxxxxx or 10xxxxxx, so 00000000 (i.e. + // 0) will never be part of a multibyte UTF-8 code point. This logic is the + // same as the "r < utf8.RuneSelf" case in strings.IndexRune but can be + // inlined. + // + // https://cs.opensource.google/go/go/+/refs/tags/go1.21.1:src/strings/strings.go;l=127 + return strings.IndexByte(cs, 0) == -1 +} + +// appendHeader is the same as bsoncore.AppendHeader but does not check if the +// key is a valid C string since the caller has already checked for that. +// +// The caller of this function must check if key is a valid C string. +func (vw *valueWriter) appendHeader(t bsontype.Type, key string) { + vw.buf = bsoncore.AppendType(vw.buf, t) + vw.buf = append(vw.buf, key...) + vw.buf = append(vw.buf, 0x00) +} + +func (vw *valueWriter) appendIntHeader(t bsontype.Type, key int) { + vw.buf = bsoncore.AppendType(vw.buf, t) + vw.buf = strconv.AppendInt(vw.buf, int64(key), 10) + vw.buf = append(vw.buf, 0x00) } diff --git a/bson/marshal.go b/bson/marshal.go index f2c48d049e..17ce6697e0 100644 --- a/bson/marshal.go +++ b/bson/marshal.go @@ -9,6 +9,7 @@ package bson import ( "bytes" "encoding/json" + "sync" "go.mongodb.org/mongo-driver/bson/bsoncodec" "go.mongodb.org/mongo-driver/bson/bsonrw" @@ -141,6 +142,13 @@ func MarshalAppendWithRegistry(r *bsoncodec.Registry, dst []byte, val interface{ return MarshalAppendWithContext(bsoncodec.EncodeContext{Registry: r}, dst, val) } +// Pool of buffers for marshalling BSON. +var bufPool = sync.Pool{ + New: func() interface{} { + return new(bytes.Buffer) + }, +} + // MarshalAppendWithContext will encode val as a BSON document using Registry r and EncodeContext ec and append the // bytes to dst. If dst is not large enough to hold the bytes, it will be grown. If val is not a type that can be // transformed into a document, MarshalValueAppendWithContext should be used instead. @@ -162,8 +170,26 @@ func MarshalAppendWithRegistry(r *bsoncodec.Registry, dst []byte, val interface{ // // See [Encoder] for more examples. func MarshalAppendWithContext(ec bsoncodec.EncodeContext, dst []byte, val interface{}) ([]byte, error) { - sw := new(bsonrw.SliceWriter) - *sw = dst + sw := bufPool.Get().(*bytes.Buffer) + defer func() { + // Proper usage of a sync.Pool requires each entry to have approximately + // the same memory cost. To obtain this property when the stored type + // contains a variably-sized buffer, we add a hard limit on the maximum + // buffer to place back in the pool. We limit the size to 16MiB because + // that's the maximum wire message size supported by any current MongoDB + // server. + // + // Comment based on + // https://cs.opensource.google/go/go/+/refs/tags/go1.19:src/fmt/print.go;l=147 + // + // Recycle byte slices that are smaller than 16MiB and at least half + // occupied. + if sw.Cap() < 16*1024*1024 && sw.Cap()/2 < sw.Len() { + bufPool.Put(sw) + } + }() + + sw.Reset() vw := bvwPool.Get(sw) defer bvwPool.Put(vw) @@ -184,7 +210,7 @@ func MarshalAppendWithContext(ec bsoncodec.EncodeContext, dst []byte, val interf return nil, err } - return *sw, nil + return append(dst, sw.Bytes()...), nil } // MarshalValue returns the BSON encoding of val. From 60ded84e486dda6eea4f446d642571489a3ed3d0 Mon Sep 17 00:00:00 2001 From: Preston Vasquez Date: Tue, 12 Sep 2023 14:06:22 -0600 Subject: [PATCH 13/15] GODRIVER-2792 Migrate CI to RHEL 8.7 (#1362) --- .evergreen/config.yml | 93 +++++------ .evergreen/run-tests.sh | 4 +- .../client_side_encryption_prose_test.go | 152 +++++++++++++++--- 3 files changed, 175 insertions(+), 74 deletions(-) diff --git a/.evergreen/config.yml b/.evergreen/config.yml index 0603f385ac..6e71fd660f 100644 --- a/.evergreen/config.yml +++ b/.evergreen/config.yml @@ -141,8 +141,8 @@ functions: export UPLOAD_BUCKET="$UPLOAD_BUCKET" export PROJECT="$PROJECT" export TMPDIR="$MONGO_ORCHESTRATION_HOME/db" - export PKG_CONFIG_PATH=$(pwd)/install/libmongocrypt/lib/pkgconfig:$(pwd)/install/mongo-c-driver/lib/pkgconfig - export LD_LIBRARY_PATH=$(pwd)/install/libmongocrypt/lib + export PKG_CONFIG_PATH=$(pwd)/install/libmongocrypt/lib64/pkgconfig:$(pwd)/install/mongo-c-driver/lib/pkgconfig + export LD_LIBRARY_PATH=$(pwd)/install/libmongocrypt/lib64 export PATH="$PATH" EOT # See what we variables we've set. @@ -245,6 +245,7 @@ functions: params: shell: "bash" script: | + set -x ${PREPARE_SHELL} MONGODB_VERSION=${VERSION} \ @@ -2082,7 +2083,7 @@ tasks: export GCPKMS_PROJECT=${GCPKMS_PROJECT} export GCPKMS_ZONE=${GCPKMS_ZONE} export GCPKMS_INSTANCENAME=${GCPKMS_INSTANCENAME} - tar czf testgcpkms.tgz ./testkms ./install/libmongocrypt/lib/libmongocrypt.* + tar czf testgcpkms.tgz ./testkms ./install/libmongocrypt/lib64/libmongocrypt.* GCPKMS_SRC=testgcpkms.tgz GCPKMS_DST=$GCPKMS_INSTANCENAME: $DRIVERS_TOOLS/.evergreen/csfle/gcpkms/copy-file.sh echo "Copying files ... end" @@ -2101,7 +2102,7 @@ tasks: export GCPKMS_PROJECT=${GCPKMS_PROJECT} export GCPKMS_ZONE=${GCPKMS_ZONE} export GCPKMS_INSTANCENAME=${GCPKMS_INSTANCENAME} - GCPKMS_CMD="LD_LIBRARY_PATH=./install/libmongocrypt/lib MONGODB_URI='mongodb://localhost:27017' PROVIDER='gcp' ./testkms" $DRIVERS_TOOLS/.evergreen/csfle/gcpkms/run-command.sh + GCPKMS_CMD="LD_LIBRARY_PATH=./install/libmongocrypt/lib64 MONGODB_URI='mongodb://localhost:27017' PROVIDER='gcp' ./testkms" $DRIVERS_TOOLS/.evergreen/csfle/gcpkms/run-command.sh - name: "testgcpkms-fail-task" # testgcpkms-fail-task runs in a non-GCE environment. @@ -2119,7 +2120,7 @@ tasks: PKG_CONFIG_PATH=$PKG_CONFIG_PATH \ make build-kms-test echo "Building build-kms-test ... end" - LD_LIBRARY_PATH=./install/libmongocrypt/lib \ + LD_LIBRARY_PATH=./install/libmongocrypt/lib64 \ MONGODB_URI='mongodb://localhost:27017/' \ EXPECT_ERROR='unable to retrieve GCP credentials' \ PROVIDER='gcp' \ @@ -2143,7 +2144,7 @@ tasks: export AWS_ACCESS_KEY_ID="${cse_aws_access_key_id}" export AWS_SECRET_ACCESS_KEY="${cse_aws_secret_access_key}" - LD_LIBRARY_PATH=./install/libmongocrypt/lib \ + LD_LIBRARY_PATH=./install/libmongocrypt/lib64 \ MONGODB_URI='${atlas_free_tier_uri}' \ PROVIDER='aws' \ ./testkms @@ -2165,9 +2166,9 @@ tasks: make build-kms-test echo "Building build-kms-test ... end" - LD_LIBRARY_PATH=./install/libmongocrypt/lib \ + LD_LIBRARY_PATH=./install/libmongocrypt/lib64 \ MONGODB_URI='${atlas_free_tier_uri}' \ - EXPECT_ERROR='unable to retrieve aws credentials' \ + EXPECT_ERROR='status=400' \ PROVIDER='aws' \ ./testkms @@ -2191,7 +2192,7 @@ tasks: export AZUREKMS_VMNAME=${AZUREKMS_VMNAME} echo '${testazurekms_privatekey}' > /tmp/testazurekms.prikey export AZUREKMS_PRIVATEKEYPATH=/tmp/testazurekms.prikey - tar czf testazurekms.tgz ./testkms ./install/libmongocrypt/lib/libmongocrypt.* + tar czf testazurekms.tgz ./testkms ./install/libmongocrypt/lib64/libmongocrypt.* AZUREKMS_SRC=testazurekms.tgz AZUREKMS_DST=/tmp $DRIVERS_TOOLS/.evergreen/csfle/azurekms/copy-file.sh echo "Copying files ... end" echo "Untarring file ... begin" @@ -2209,7 +2210,7 @@ tasks: export AZUREKMS_VMNAME=${AZUREKMS_VMNAME} echo '${testazurekms_privatekey}' > /tmp/testazurekms.prikey export AZUREKMS_PRIVATEKEYPATH=/tmp/testazurekms.prikey - AZUREKMS_CMD="LD_LIBRARY_PATH=./install/libmongocrypt/lib MONGODB_URI='mongodb://localhost:27017' PROVIDER='azure' AZUREKMS_KEY_NAME='${AZUREKMS_KEY_NAME}' AZUREKMS_KEY_VAULT_ENDPOINT='${AZUREKMS_KEY_VAULT_ENDPOINT}' ./testkms" $DRIVERS_TOOLS/.evergreen/csfle/azurekms/run-command.sh + AZUREKMS_CMD="LD_LIBRARY_PATH=./install/libmongocrypt/lib64 MONGODB_URI='mongodb://localhost:27017' PROVIDER='azure' AZUREKMS_KEY_NAME='${AZUREKMS_KEY_NAME}' AZUREKMS_KEY_VAULT_ENDPOINT='${AZUREKMS_KEY_VAULT_ENDPOINT}' ./testkms" $DRIVERS_TOOLS/.evergreen/csfle/azurekms/run-command.sh - name: "testazurekms-fail-task" # testazurekms-fail-task runs without environment variables. @@ -2228,7 +2229,7 @@ tasks: make build-kms-test echo "Building build-kms-test ... end" - LD_LIBRARY_PATH=./install/libmongocrypt/lib \ + LD_LIBRARY_PATH=./install/libmongocrypt/lib64 \ MONGODB_URI='mongodb://localhost:27017' \ EXPECT_ERROR='unable to retrieve azure credentials' \ PROVIDER='azure' AZUREKMS_KEY_NAME='${AZUREKMS_KEY_NAME}' AZUREKMS_KEY_VAULT_ENDPOINT='${AZUREKMS_KEY_VAULT_ENDPOINT}' \ @@ -2311,9 +2312,9 @@ axes: GCC_PATH: "/cygdrive/c/ProgramData/chocolatey/lib/mingw/tools/install/mingw64/bin" GO_DIST: "C:\\golang\\go1.20" VENV_BIN_DIR: "Scripts" - - id: "ubuntu1604-64-go-1-20" - display_name: "Ubuntu 16.04" - run_on: ubuntu1604-build + - id: "rhel87-64-go-1-20" + display_name: "RHEL 8.7" + run_on: rhel8.7-large variables: GO_DIST: "/opt/golang/go1.20" - id: "macos11-go-1-20" @@ -2335,9 +2336,9 @@ axes: GCC_PATH: "/cygdrive/c/ProgramData/chocolatey/lib/mingw/tools/install/mingw64/bin" GO_DIST: "C:\\golang\\go1.20" VENV_BIN_DIR: "Scripts" - - id: "ubuntu1804-64-go-1-20" - display_name: "Ubuntu 18.04" - run_on: ubuntu1804-build + - id: "rhel87-64-go-1-20" + display_name: "RHEL 8.7" + run_on: rhel8.7-large variables: GO_DIST: "/opt/golang/go1.20" - id: "macos11-go-1-20" @@ -2347,13 +2348,12 @@ axes: variables: GO_DIST: "/opt/golang/go1.20" - # OCSP linux tasks need to run against this OS since stapling is disabled on Ubuntu 18.04 (SERVER-51364) - - id: ocsp-rhel-70 + - id: ocsp-rhel-87 display_name: OS values: - - id: "rhel70-go-1-20" - display_name: "RHEL 7.0" - run_on: rhel70-build + - id: "rhel87-go-1-20" + display_name: "RHEL 8.7" + run_on: rhel8.7-large variables: GO_DIST: "/opt/golang/go1.20" @@ -2368,6 +2368,8 @@ axes: GCC_PATH: "/cygdrive/c/ProgramData/chocolatey/lib/mingw/tools/install/mingw64/bin" GO_DIST: "C:\\golang\\go1.20" SKIP_ECS_AUTH_TEST: true + # TODO(BUILD-17329): Update this to Ubuntu 22 after we add a new ECS task + # definition. - id: "ubuntu1804-64-go-1-20" display_name: "Ubuntu 18.04" run_on: ubuntu1804-test @@ -2386,18 +2388,18 @@ axes: - id: os-faas-80 display_name: OS values: - - id: "rhel80-large-go-1-20" - display_name: "RHEL 8.0" - run_on: rhel80-large + - id: "rhel87-large-go-1-20" + display_name: "RHEL 8.7" + run_on: rhel8.7-large variables: GO_DIST: "/opt/golang/go1.20" - id: os-serverless display_name: OS values: - - id: "ubuntu2204-go-1-20" - display_name: "Ubuntu 22.04" - run_on: ubuntu2204-small + - id: "rhel87-go-1-20" + display_name: "RHEL 8.7" + run_on: rhel8.7-small variables: GO_DIST: "/opt/golang/go1.20" @@ -2620,7 +2622,7 @@ buildvariants: - name: static-analysis display_name: "Static Analysis" run_on: - - ubuntu1804-build + - rhel8.7-large expansions: GO_DIST: "/opt/golang/go1.20" tasks: @@ -2629,7 +2631,7 @@ buildvariants: - name: perf display_name: "Performance" run_on: - - ubuntu1804-build + - rhel8.7-large expansions: GO_DIST: "/opt/golang/go1.20" tasks: @@ -2638,16 +2640,16 @@ buildvariants: - name: build-check display_name: "Compile Only Checks" run_on: - - ubuntu1804-test + - rhel8.7-large expansions: GO_DIST: "/opt/golang/go1.20" tasks: - name: ".compile-check" - + - name: atlas-test display_name: "Atlas test" run_on: - - ubuntu1804-build + - rhel8.7-large expansions: GO_DIST: "/opt/golang/go1.20" tasks: @@ -2656,7 +2658,7 @@ buildvariants: - name: atlas-data-lake-test display_name: "Atlas Data Lake Test" run_on: - - ubuntu1804-build + - rhel8.7-large expansions: GO_DIST: "/opt/golang/go1.20" tasks: @@ -2693,8 +2695,8 @@ buildvariants: - name: "aws-auth-test" - matrix_name: "ocsp-test" - matrix_spec: { version: ["4.4", "5.0", "6.0", "7.0", "latest"], ocsp-rhel-70: ["rhel70-go-1-20"] } - display_name: "OCSP ${version} ${ocsp-rhel-70}" + matrix_spec: { version: ["4.4", "5.0", "6.0", "7.0", "latest"], ocsp-rhel-87: ["rhel87-go-1-20"] } + display_name: "OCSP ${version} ${ocsp-rhel-87}" batchtime: 20160 # Use a batchtime of 14 days as suggested by the OCSP test README tasks: - name: ".ocsp" @@ -2716,7 +2718,7 @@ buildvariants: - name: ".ocsp-rsa !.ocsp-staple" - matrix_name: "race-test" - matrix_spec: { version: ["latest"], os-ssl-40: ["ubuntu1804-64-go-1-20"] } + matrix_spec: { version: ["latest"], os-ssl-40: ["rhel87-64-go-1-20"] } display_name: "Race Detector Test" tasks: - name: ".race" @@ -2728,14 +2730,13 @@ buildvariants: - name: ".versioned-api" - matrix_name: "kms-tls-test" - matrix_spec: { version: ["latest"], os-ssl-40: ["ubuntu1804-64-go-1-20"] } + matrix_spec: { version: ["latest"], os-ssl-40: ["rhel87-64-go-1-20"] } display_name: "KMS TLS ${os-ssl-40}" tasks: - name: ".kms-tls" - matrix_name: "load-balancer-test" - # The LB software is only available on Ubuntu 18.04, so we don't test on all OSes. - matrix_spec: { version: ["5.0", "6.0", "7.0", "latest", "rapid"], os-ssl-40: ["ubuntu1804-64-go-1-20"] } + matrix_spec: { version: ["5.0", "6.0", "7.0", "latest", "rapid"], os-ssl-40: ["rhel87-64-go-1-20"] } display_name: "Load Balancer Support ${version} ${os-ssl-40}" tasks: - name: ".load-balancer" @@ -2747,20 +2748,20 @@ buildvariants: - "serverless_task_group" - matrix_name: "kms-kmip-test" - matrix_spec: { version: ["latest"], os-ssl-40: ["ubuntu1804-64-go-1-20"] } + matrix_spec: { version: ["latest"], os-ssl-40: ["rhel87-64-go-1-20"] } display_name: "KMS KMIP ${os-ssl-40}" tasks: - name: ".kms-kmip" - matrix_name: "fuzz-test" - matrix_spec: { version: ["5.0"], os-ssl-40: ["ubuntu1804-64-go-1-20"] } + matrix_spec: { version: ["5.0"], os-ssl-40: ["rhel87-64-go-1-20"] } display_name: "Fuzz ${version} ${os-ssl-40}" tasks: - name: "test-fuzz" batchtime: 1440 # Run at most once per 24 hours. - matrix_name: "faas-test" - matrix_spec: { version: ["latest"], os-faas-80: ["rhel80-large-go-1-20"] } + matrix_spec: { version: ["latest"], os-faas-80: ["rhel87-large-go-1-20"] } display_name: "FaaS ${version} ${os-faas-80}" tasks: - test-aws-lambda-task-group @@ -2768,7 +2769,7 @@ buildvariants: - name: testgcpkms-variant display_name: "GCP KMS" run_on: - - debian11-small + - rhel8.7-small expansions: GO_DIST: "/opt/golang/go1.20" tasks: @@ -2779,7 +2780,7 @@ buildvariants: - name: testawskms-variant display_name: "AWS KMS" run_on: - - debian11-small + - rhel8.7-small expansions: GO_DIST: "/opt/golang/go1.20" tasks: @@ -2789,7 +2790,7 @@ buildvariants: - name: testazurekms-variant display_name: "AZURE KMS" run_on: - - debian11-small + - rhel8.7-small expansions: GO_DIST: "/opt/golang/go1.20" tasks: diff --git a/.evergreen/run-tests.sh b/.evergreen/run-tests.sh index 392531613a..a223f309f8 100644 --- a/.evergreen/run-tests.sh +++ b/.evergreen/run-tests.sh @@ -15,8 +15,8 @@ fi export GOROOT="${GOROOT}" export PATH="${GOROOT}/bin:${GCC_PATH}:$GOPATH/bin:$PATH" export PROJECT="${project}" -export PKG_CONFIG_PATH=$(pwd)/install/libmongocrypt/lib/pkgconfig:$(pwd)/install/mongo-c-driver/lib/pkgconfig -export LD_LIBRARY_PATH=$(pwd)/install/libmongocrypt/lib +export PKG_CONFIG_PATH=$(pwd)/install/libmongocrypt/lib64/pkgconfig:$(pwd)/install/mongo-c-driver/lib/pkgconfig +export LD_LIBRARY_PATH=$(pwd)/install/libmongocrypt/lib64 export GOFLAGS=-mod=vendor SSL=${SSL:-nossl} diff --git a/mongo/integration/client_side_encryption_prose_test.go b/mongo/integration/client_side_encryption_prose_test.go index 4eac5364ca..45a2ef01c8 100644 --- a/mongo/integration/client_side_encryption_prose_test.go +++ b/mongo/integration/client_side_encryption_prose_test.go @@ -19,7 +19,6 @@ import ( "net/http" "os" "path/filepath" - "runtime" "strings" "testing" "time" @@ -55,6 +54,16 @@ const ( maxBsonObjSize = 16777216 // max bytes in BSON object ) +func containsSubstring(possibleSubstrings []string, str string) bool { + for _, possibleSubstring := range possibleSubstrings { + if strings.Contains(str, possibleSubstring) { + return true + } + } + + return false +} + func TestClientSideEncryptionProse(t *testing.T) { t.Parallel() @@ -865,26 +874,119 @@ func TestClientSideEncryptionProse(t *testing.T) { "endpoint": "doesnotexist.local:5698", } + const ( + errConnectionRefused = "connection refused" + errInvalidKMSResponse = "Invalid KMS response" + errMongocryptError = "mongocrypt error" + errNoSuchHost = "no such host" + errServerMisbehaving = "server misbehaving" + errWindowsTLSConnectionRefused = "No connection could be made because the target machine actively refused it" + ) + testCases := []struct { name string provider string masterKey interface{} - errorSubstring string + errorSubstring []string testInvalidClientEncryption bool - invalidClientEncryptionErrorSubstring string + invalidClientEncryptionErrorSubstring []string }{ - {"Case 1: aws success without endpoint", "aws", awsSuccessWithoutEndpoint, "", false, ""}, - {"Case 2: aws success with endpoint", "aws", awsSuccessWithEndpoint, "", false, ""}, - {"Case 3: aws success with https endpoint", "aws", awsSuccessWithHTTPSEndpoint, "", false, ""}, - {"Case 4: aws failure with connection error", "aws", awsFailureConnectionError, "connection refused", false, ""}, - {"Case 5: aws failure with wrong endpoint", "aws", awsFailureInvalidEndpoint, "mongocrypt error", false, ""}, - {"Case 6: aws failure with parse error", "aws", awsFailureParseError, "no such host", false, ""}, - {"Case 7: azure success", "azure", azure, "", true, "no such host"}, - {"Case 8: gcp success", "gcp", gcpSuccess, "", true, "no such host"}, - {"Case 9: gcp failure", "gcp", gcpFailure, "Invalid KMS response", false, ""}, - {"Case 10: kmip success without endpoint", "kmip", kmipSuccessWithoutEndpoint, "", true, "no such host"}, - {"Case 11: kmip success with endpoint", "kmip", kmipSuccessWithEndpoint, "", false, ""}, - {"Case 12: kmip failure with invalid endpoint", "kmip", kmipFailureInvalidEndpoint, "no such host", false, ""}, + { + name: "Case 1: aws success without endpoint", + provider: "aws", + masterKey: awsSuccessWithoutEndpoint, + errorSubstring: []string{}, + testInvalidClientEncryption: false, + invalidClientEncryptionErrorSubstring: []string{}, + }, + { + name: "Case 2: aws success with endpoint", + provider: "aws", + masterKey: awsSuccessWithEndpoint, + errorSubstring: []string{}, + testInvalidClientEncryption: false, + invalidClientEncryptionErrorSubstring: []string{}, + }, + { + name: "Case 3: aws success with https endpoint", + provider: "aws", + masterKey: awsSuccessWithHTTPSEndpoint, + errorSubstring: []string{}, + testInvalidClientEncryption: false, + invalidClientEncryptionErrorSubstring: []string{}, + }, + { + name: "Case 4: aws failure with connection error", + provider: "aws", + masterKey: awsFailureConnectionError, + errorSubstring: []string{errConnectionRefused, errWindowsTLSConnectionRefused}, + testInvalidClientEncryption: false, + invalidClientEncryptionErrorSubstring: []string{}, + }, + { + name: "Case 5: aws failure with wrong endpoint", + provider: "aws", + masterKey: awsFailureInvalidEndpoint, + errorSubstring: []string{errMongocryptError}, + testInvalidClientEncryption: false, + invalidClientEncryptionErrorSubstring: []string{}, + }, + { + name: "Case 6: aws failure with parse error", + provider: "aws", + masterKey: awsFailureParseError, + errorSubstring: []string{errNoSuchHost, errServerMisbehaving}, + testInvalidClientEncryption: false, + invalidClientEncryptionErrorSubstring: []string{}, + }, + { + name: "Case 7: azure success", + provider: "azure", + masterKey: azure, + errorSubstring: []string{}, + testInvalidClientEncryption: true, + invalidClientEncryptionErrorSubstring: []string{errNoSuchHost, errServerMisbehaving}, + }, + { + name: "Case 8: gcp success", + provider: "gcp", + masterKey: gcpSuccess, + errorSubstring: []string{}, + testInvalidClientEncryption: true, + invalidClientEncryptionErrorSubstring: []string{errNoSuchHost, errServerMisbehaving}, + }, + { + name: "Case 9: gcp failure", + provider: "gcp", + masterKey: gcpFailure, + errorSubstring: []string{errInvalidKMSResponse}, + testInvalidClientEncryption: false, + invalidClientEncryptionErrorSubstring: []string{}, + }, + { + name: "Case 10: kmip success without endpoint", + provider: "kmip", + masterKey: kmipSuccessWithoutEndpoint, + errorSubstring: []string{}, + testInvalidClientEncryption: true, + invalidClientEncryptionErrorSubstring: []string{errNoSuchHost, errServerMisbehaving}, + }, + { + name: "Case 11: kmip success with endpoint", + provider: "kmip", + masterKey: kmipSuccessWithEndpoint, + errorSubstring: []string{}, + testInvalidClientEncryption: false, + invalidClientEncryptionErrorSubstring: []string{}, + }, + { + name: "Case 12: kmip failure with invalid endpoint", + provider: "kmip", + masterKey: kmipFailureInvalidEndpoint, + errorSubstring: []string{errNoSuchHost, errServerMisbehaving}, + testInvalidClientEncryption: false, + invalidClientEncryptionErrorSubstring: []string{}, + }, } for _, tc := range testCases { mt.Run(tc.name, func(mt *mtest.T) { @@ -896,16 +998,12 @@ func TestClientSideEncryptionProse(t *testing.T) { dkOpts := options.DataKey().SetMasterKey(tc.masterKey) createdKey, err := cpt.clientEnc.CreateDataKey(context.Background(), tc.provider, dkOpts) - if tc.errorSubstring != "" { + if len(tc.errorSubstring) > 0 { assert.NotNil(mt, err, "expected error, got nil") - errSubstr := tc.errorSubstring - if runtime.GOOS == "windows" && errSubstr == "connection refused" { - // tls.Dial returns an error that does not contain the substring "connection refused" - // on Windows machines - errSubstr = "No connection could be made because the target machine actively refused it" - } - assert.True(mt, strings.Contains(err.Error(), errSubstr), - "expected error '%s' to contain '%s'", err.Error(), errSubstr) + + assert.True(t, containsSubstring(tc.errorSubstring, err.Error()), + "expected tc.errorSubstring=%v to contain %v, but it didn't", tc.errorSubstring, err.Error()) + return } assert.Nil(mt, err, "CreateDataKey error: %v", err) @@ -932,8 +1030,10 @@ func TestClientSideEncryptionProse(t *testing.T) { invalidKeyOpts := options.DataKey().SetMasterKey(tc.masterKey) _, err = invalidClientEncryption.CreateDataKey(context.Background(), tc.provider, invalidKeyOpts) assert.NotNil(mt, err, "expected CreateDataKey error, got nil") - assert.True(mt, strings.Contains(err.Error(), tc.invalidClientEncryptionErrorSubstring), - "expected error %v to contain substring '%v'", err, tc.invalidClientEncryptionErrorSubstring) + + assert.True(t, containsSubstring(tc.invalidClientEncryptionErrorSubstring, err.Error()), + "expected tc.invalidClientEncryptionErrorSubstring=%v to contain %v, but it didn't", + tc.invalidClientEncryptionErrorSubstring, err.Error()) }) } }) From 27ba3724c94e572fc817597b2319e0fc94589c3e Mon Sep 17 00:00:00 2001 From: Qingyang Hu <103950869+qingyang-hu@users.noreply.github.com> Date: Fri, 15 Sep 2023 10:19:10 -0400 Subject: [PATCH 14/15] Quick update for api changes report. (#1384) --- etc/api_report.sh | 1 + 1 file changed, 1 insertion(+) diff --git a/etc/api_report.sh b/etc/api_report.sh index 1cfbac5025..ffdac7f975 100755 --- a/etc/api_report.sh +++ b/etc/api_report.sh @@ -10,6 +10,7 @@ if [ -z $cmd ]; then fi branch=${GITHUB_BASE_REF:-master} +git fetch origin $branch:$branch sha=$(git merge-base $branch HEAD) gorelease -base=$sha > api-report.txt || true From 59f7519e8b7c4e549a39ce769b8536f2bdfd0446 Mon Sep 17 00:00:00 2001 From: Preston Vasquez Date: Fri, 15 Sep 2023 16:39:11 -0600 Subject: [PATCH 15/15] GODRIVER-2986 Resolve failures in Race Detector Test (#1380) --- mongo/integration/change_stream_test.go | 2 +- mongo/integration/unified/client_entity.go | 16 ++++++++++++ mongo/integration/unified/logger.go | 25 +++++++++++++++---- .../unified/unified_spec_runner.go | 2 +- 4 files changed, 38 insertions(+), 7 deletions(-) diff --git a/mongo/integration/change_stream_test.go b/mongo/integration/change_stream_test.go index 868706ad1e..b3d0469c36 100644 --- a/mongo/integration/change_stream_test.go +++ b/mongo/integration/change_stream_test.go @@ -770,7 +770,7 @@ func TestChangeStream_ReplicaSet(t *testing.T) { require.NoError(mt, err, "failed to update idValue") }() - nextCtx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + nextCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) t.Cleanup(cancel) type splitEvent struct { diff --git a/mongo/integration/unified/client_entity.go b/mongo/integration/unified/client_entity.go index e63c891039..ff7d9d5fc3 100644 --- a/mongo/integration/unified/client_entity.go +++ b/mongo/integration/unified/client_entity.go @@ -66,6 +66,7 @@ type clientEntity struct { eventsCountLock sync.RWMutex serverDescriptionChangedEventsCountLock sync.RWMutex + eventProcessMu sync.RWMutex entityMap *EntityMap @@ -471,6 +472,9 @@ func (c *clientEntity) processPoolEvent(evt *event.PoolEvent) { } func (c *clientEntity) processServerDescriptionChangedEvent(evt *event.ServerDescriptionChangedEvent) { + c.eventProcessMu.Lock() + defer c.eventProcessMu.Unlock() + if !c.getRecordEvents() { return } @@ -487,6 +491,9 @@ func (c *clientEntity) processServerDescriptionChangedEvent(evt *event.ServerDes } func (c *clientEntity) processServerHeartbeatFailedEvent(evt *event.ServerHeartbeatFailedEvent) { + c.eventProcessMu.Lock() + defer c.eventProcessMu.Unlock() + if !c.getRecordEvents() { return } @@ -499,6 +506,9 @@ func (c *clientEntity) processServerHeartbeatFailedEvent(evt *event.ServerHeartb } func (c *clientEntity) processServerHeartbeatStartedEvent(evt *event.ServerHeartbeatStartedEvent) { + c.eventProcessMu.Lock() + defer c.eventProcessMu.Unlock() + if !c.getRecordEvents() { return } @@ -511,6 +521,9 @@ func (c *clientEntity) processServerHeartbeatStartedEvent(evt *event.ServerHeart } func (c *clientEntity) processServerHeartbeatSucceededEvent(evt *event.ServerHeartbeatSucceededEvent) { + c.eventProcessMu.Lock() + defer c.eventProcessMu.Unlock() + if !c.getRecordEvents() { return } @@ -523,6 +536,9 @@ func (c *clientEntity) processServerHeartbeatSucceededEvent(evt *event.ServerHea } func (c *clientEntity) processTopologyDescriptionChangedEvent(evt *event.TopologyDescriptionChangedEvent) { + c.eventProcessMu.Lock() + defer c.eventProcessMu.Unlock() + if !c.getRecordEvents() { return } diff --git a/mongo/integration/unified/logger.go b/mongo/integration/unified/logger.go index 6dcadacf4a..1d9a612092 100644 --- a/mongo/integration/unified/logger.go +++ b/mongo/integration/unified/logger.go @@ -7,6 +7,8 @@ package unified import ( + "sync" + "go.mongodb.org/mongo-driver/internal/logger" ) @@ -20,9 +22,19 @@ type orderedLogMessage struct { // Logger is the Sink used to captured log messages for logger verification in // the unified spec tests. type Logger struct { + // bufSize is the number of logs expected to be sent to the logger for a + // unified spec test. + bufSize int + + // lastOrder increments each time the "Info" method is called, and is used to + // determine when to close the logQueue. lastOrder int - logQueue chan orderedLogMessage - bufSize int + + // orderMu guards the order value, which increments each time the "Info" + // method is called. This is necessary since "Info" could be called from + // multiple go routines, e.g. SDAM logs. + orderMu sync.RWMutex + logQueue chan orderedLogMessage } func newLogger(olm *observeLogMessages, bufSize int) *Logger { @@ -44,14 +56,17 @@ func (log *Logger) Info(level int, msg string, args ...interface{}) { return } - defer func() { log.lastOrder++ }() - // If the order is greater than the buffer size, we must return. This // would indicate that the logQueue channel has been closed. if log.lastOrder > log.bufSize { return } + log.orderMu.Lock() + defer log.orderMu.Unlock() + + defer func() { log.lastOrder++ }() + // Add the Diff back to the level, as there is no need to create a // logging offset. level = level + logger.DiffToInfo @@ -68,7 +83,7 @@ func (log *Logger) Info(level int, msg string, args ...interface{}) { logMessage: logMessage, } - // If the order has reached the buffer size, then close the channe. + // If the order has reached the buffer size, then close the channel. if log.lastOrder == log.bufSize { close(log.logQueue) } diff --git a/mongo/integration/unified/unified_spec_runner.go b/mongo/integration/unified/unified_spec_runner.go index 1b1cbeb533..7b92d07204 100644 --- a/mongo/integration/unified/unified_spec_runner.go +++ b/mongo/integration/unified/unified_spec_runner.go @@ -224,7 +224,7 @@ func (tc *TestCase) Run(ls LoggerSkipper) error { } // Count the number of expected log messages over all clients. - expectedLogCount := 0 + var expectedLogCount int for _, clientLog := range tc.ExpectLogMessages { expectedLogCount += len(clientLog.LogMessages) }