Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:27
📄 description.go // 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 description contains types and functions for describing the state of MongoDB clusters. package description // import "go.mongodb.org/mongo-driver/mongo/description" // Unknown is an unknown server or topology kind. const Unknown = 0	515 B	2025-05-25 12:27
📄 max_staleness_spec_test.go // 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 description import ( "path" "testing" "go.mongodb.org/mongo-driver/internal/spectest" ) const maxStalenessTestsDir = "../../testdata/max-staleness" // Test case for all max staleness spec tests. func TestMaxStalenessSpec(t *testing.T) { for _, topology := range [...]string{ "ReplicaSetNoPrimary", "ReplicaSetWithPrimary", "Sharded", "Single", "Unknown", } { for _, file := range spectest.FindJSONFilesInDir(t, path.Join(maxStalenessTestsDir, topology)) { runTest(t, maxStalenessTestsDir, topology, file) } } }	820 B	2025-05-25 12:27
📄 selector_spec_test.go // 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 description import ( "path" "testing" "go.mongodb.org/mongo-driver/internal/spectest" ) const selectorTestsDir = "../../testdata/server-selection/server_selection" // Test case for all SDAM spec tests. func TestServerSelectionSpec(t *testing.T) { for _, topology := range [...]string{ "ReplicaSetNoPrimary", "ReplicaSetWithPrimary", "Sharded", "Single", "Unknown", "LoadBalanced", } { for _, subdir := range [...]string{"read", "write"} { subdirPath := path.Join(topology, subdir) for _, file := range spectest.FindJSONFilesInDir(t, path.Join(selectorTestsDir, subdirPath)) { runTest(t, selectorTestsDir, subdirPath, file) } } } }	954 B	2025-05-25 12:27
📄 selector_test.go // 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 description import ( "testing" "time" "github.com/google/go-cmp/cmp" "go.mongodb.org/mongo-driver/internal/assert" "go.mongodb.org/mongo-driver/internal/require" "go.mongodb.org/mongo-driver/mongo/address" "go.mongodb.org/mongo-driver/mongo/readpref" "go.mongodb.org/mongo-driver/tag" ) func TestServerSelection(t testing.T) { noerr := func(t testing.T, err error) { if err != nil { t.Errorf("Unepexted error: %v", err) t.FailNow() } } t.Run("WriteSelector", func(t testing.T) { testCases := []struct { name string desc Topology start int end int }{ { name: "ReplicaSetWithPrimary", desc: Topology{ Kind: ReplicaSetWithPrimary, Servers: []Server{ {Addr: address.Address("localhost:27017"), Kind: RSPrimary}, {Addr: address.Address("localhost:27018"), Kind: RSSecondary}, {Addr: address.Address("localhost:27019"), Kind: RSSecondary}, }, }, start: 0, end: 1, }, { name: "ReplicaSetNoPrimary", desc: Topology{ Kind: ReplicaSetNoPrimary, Servers: []Server{ {Addr: address.Address("localhost:27018"), Kind: RSSecondary}, {Addr: address.Address("localhost:27019"), Kind: RSSecondary}, }, }, start: 0, end: 0, }, { name: "Sharded", desc: Topology{ Kind: Sharded, Servers: []Server{ {Addr: address.Address("localhost:27018"), Kind: Mongos}, {Addr: address.Address("localhost:27019"), Kind: Mongos}, }, }, start: 0, end: 2, }, { name: "Single", desc: Topology{ Kind: Single, Servers: []Server{ {Addr: address.Address("localhost:27018"), Kind: Standalone}, }, }, start: 0, end: 1, }, } for _, tc := range testCases { t.Run(tc.name, func(t testing.T) { result, err := WriteSelector().SelectServer(tc.desc, tc.desc.Servers) noerr(t, err) if len(result) != tc.end-tc.start { t.Errorf("Incorrect number of servers selected. got %d; want %d", len(result), tc.end-tc.start) } if diff := cmp.Diff(result, tc.desc.Servers[tc.start:tc.end]); diff != "" { t.Errorf("Incorrect servers selected (-got +want):\n%s", diff) } }) } }) t.Run("LatencySelector", func(t testing.T) { testCases := []struct { name string desc Topology start int end int }{ { name: "NoRTTSet", desc: Topology{ Servers: []Server{ {Addr: address.Address("localhost:27017")}, {Addr: address.Address("localhost:27018")}, {Addr: address.Address("localhost:27019")}, }, }, start: 0, end: 3, }, { name: "MultipleServers PartialNoRTTSet", desc: Topology{ Servers: []Server{ {Addr: address.Address("localhost:27017"), AverageRTT: 5 time.Second, AverageRTTSet: true}, {Addr: address.Address("localhost:27018"), AverageRTT: 10 * time.Second, AverageRTTSet: true}, {Addr: address.Address("localhost:27019")}, }, }, start: 0, end: 2, }, { name: "MultipleServers", desc: Topology{ Servers: []Server{ {Addr: address.Address("localhost:27017"), AverageRTT: 5 * time.Second, AverageRTTSet: true}, {Addr: address.Address("localhost:27018"), AverageRTT: 10 * time.Second, AverageRTTSet: true}, {Addr: address.Address("localhost:27019"), AverageRTT: 26 * time.Second, AverageRTTSet: true}, }, }, start: 0, end: 2, }, { name: "No Servers", desc: Topology{Servers: []Server{}}, start: 0, end: 0, }, { name: "1 Server", desc: Topology{ Servers: []Server{ {Addr: address.Address("localhost:27017"), AverageRTT: 26 * time.Second, AverageRTTSet: true}, }, }, start: 0, end: 1, }, } for _, tc := range testCases { t.Run(tc.name, func(t testing.T) { result, err := LatencySelector(20time.Second).SelectServer(tc.desc, tc.desc.Servers) noerr(t, err) if len(result) != tc.end-tc.start { t.Errorf("Incorrect number of servers selected. got %d; want %d", len(result), tc.end-tc.start) } if diff := cmp.Diff(result, tc.desc.Servers[tc.start:tc.end]); diff != "" { t.Errorf("Incorrect servers selected (-got +want):\n%s", diff) } }) } }) } var readPrefTestPrimary = Server{ Addr: address.Address("localhost:27017"), HeartbeatInterval: time.Duration(10) * time.Second, LastWriteTime: time.Date(2017, 2, 11, 14, 0, 0, 0, time.UTC), LastUpdateTime: time.Date(2017, 2, 11, 14, 0, 2, 0, time.UTC), Kind: RSPrimary, Tags: tag.Set{tag.Tag{Name: "a", Value: "1"}}, WireVersion: &VersionRange{Min: 6, Max: 21}, } var readPrefTestSecondary1 = Server{ Addr: address.Address("localhost:27018"), HeartbeatInterval: time.Duration(10) * time.Second, LastWriteTime: time.Date(2017, 2, 11, 13, 58, 0, 0, time.UTC), LastUpdateTime: time.Date(2017, 2, 11, 14, 0, 2, 0, time.UTC), Kind: RSSecondary, Tags: tag.Set{tag.Tag{Name: "a", Value: "1"}}, WireVersion: &VersionRange{Min: 6, Max: 21}, } var readPrefTestSecondary2 = Server{ Addr: address.Address("localhost:27018"), HeartbeatInterval: time.Duration(10) * time.Second, LastWriteTime: time.Date(2017, 2, 11, 14, 0, 0, 0, time.UTC), LastUpdateTime: time.Date(2017, 2, 11, 14, 0, 2, 0, time.UTC), Kind: RSSecondary, Tags: tag.Set{tag.Tag{Name: "a", Value: "2"}}, WireVersion: &VersionRange{Min: 6, Max: 21}, } var readPrefTestTopology = Topology{ Kind: ReplicaSetWithPrimary, Servers: []Server{readPrefTestPrimary, readPrefTestSecondary1, readPrefTestSecondary2}, } func TestSelector_Sharded(t testing.T) { t.Parallel() subject := readpref.Primary() s := Server{ Addr: address.Address("localhost:27017"), HeartbeatInterval: time.Duration(10) time.Second, LastWriteTime: time.Date(2017, 2, 11, 14, 0, 0, 0, time.UTC), LastUpdateTime: time.Date(2017, 2, 11, 14, 0, 2, 0, time.UTC), Kind: Mongos, WireVersion: &VersionRange{Min: 6, Max: 21}, } c := Topology{ Kind: Sharded, Servers: []Server{s}, } result, err := ReadPrefSelector(subject).SelectServer(c, c.Servers) require.NoError(t, err) require.Len(t, result, 1) require.Equal(t, []Server{s}, result) } func BenchmarkLatencySelector(b testing.B) { for _, bcase := range []struct { name string serversHook func(servers []Server) }{ { name: "AllFit", serversHook: func(servers []Server) {}, }, { name: "AllButOneFit", serversHook: func(servers []Server) { servers[0].AverageRTT = 2 time.Second }, }, { name: "HalfFit", serversHook: func(servers []Server) { for i := 0; i < len(servers); i += 2 { servers[i].AverageRTT = 2 * time.Second } }, }, { name: "OneFit", serversHook: func(servers []Server) { for i := 1; i < len(servers); i++ { servers[i].AverageRTT = 2 * time.Second } }, }, } { bcase := bcase b.Run(bcase.name, func(b testing.B) { s := Server{ Addr: address.Address("localhost:27017"), HeartbeatInterval: time.Duration(10) time.Second, LastWriteTime: time.Date(2017, 2, 11, 14, 0, 0, 0, time.UTC), LastUpdateTime: time.Date(2017, 2, 11, 14, 0, 2, 0, time.UTC), Kind: Mongos, WireVersion: &VersionRange{Min: 6, Max: 21}, AverageRTTSet: true, AverageRTT: time.Second, } servers := make([]Server, 100) for i := 0; i < len(servers); i++ { servers[i] = s } bcase.serversHook(servers) //this will make base 1 sec latency < min (0.5) + conf (1) //and high latency 2 higher than the threshold servers[99].AverageRTT = 500 * time.Millisecond c := Topology{ Kind: Sharded, Servers: servers, } b.ResetTimer() b.RunParallel(func(p testing.PB) { b.ReportAllocs() for p.Next() { _, _ = LatencySelector(time.Second).SelectServer(c, c.Servers) } }) }) } } func BenchmarkSelector_Sharded(b testing.B) { for _, bcase := range []struct { name string serversHook func(servers []Server) }{ { name: "AllFit", serversHook: func(servers []Server) {}, }, { name: "AllButOneFit", serversHook: func(servers []Server) { servers[0].Kind = LoadBalancer }, }, { name: "HalfFit", serversHook: func(servers []Server) { for i := 0; i < len(servers); i += 2 { servers[i].Kind = LoadBalancer } }, }, { name: "OneFit", serversHook: func(servers []Server) { for i := 1; i < len(servers); i++ { servers[i].Kind = LoadBalancer } }, }, } { bcase := bcase b.Run(bcase.name, func(b testing.B) { subject := readpref.Primary() s := Server{ Addr: address.Address("localhost:27017"), HeartbeatInterval: time.Duration(10) time.Second, LastWriteTime: time.Date(2017, 2, 11, 14, 0, 0, 0, time.UTC), LastUpdateTime: time.Date(2017, 2, 11, 14, 0, 2, 0, time.UTC), Kind: Mongos, WireVersion: &VersionRange{Min: 6, Max: 21}, } servers := make([]Server, 100) for i := 0; i < len(servers); i++ { servers[i] = s } bcase.serversHook(servers) c := Topology{ Kind: Sharded, Servers: servers, } b.ResetTimer() b.RunParallel(func(p testing.PB) { b.ReportAllocs() for p.Next() { _, _ = ReadPrefSelector(subject).SelectServer(c, c.Servers) } }) }) } } func Benchmark_SelectServer_SelectServer(b testing.B) { topology := Topology{Kind: ReplicaSet} // You can change the topology as needed candidates := []Server{ {Kind: Mongos}, {Kind: RSPrimary}, {Kind: Standalone}, } selector := writeServerSelector{} // Assuming this is the receiver type b.ReportAllocs() b.ResetTimer() for i := 0; i < b.N; i++ { _, err := selector.SelectServer(topology, candidates) if err != nil { b.Fatalf("Error selecting server: %v", err) } } } func TestSelector_Single(t testing.T) { t.Parallel() subject := readpref.Primary() s := Server{ Addr: address.Address("localhost:27017"), HeartbeatInterval: time.Duration(10) time.Second, LastWriteTime: time.Date(2017, 2, 11, 14, 0, 0, 0, time.UTC), LastUpdateTime: time.Date(2017, 2, 11, 14, 0, 2, 0, time.UTC), Kind: Mongos, WireVersion: &VersionRange{Min: 6, Max: 21}, } c := Topology{ Kind: Single, Servers: []Server{s}, } result, err := ReadPrefSelector(subject).SelectServer(c, c.Servers) require.NoError(t, err) require.Len(t, result, 1) require.Equal(t, []Server{s}, result) } func TestSelector_Primary(t testing.T) { t.Parallel() subject := readpref.Primary() result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, readPrefTestTopology.Servers) require.NoError(t, err) require.Len(t, result, 1) require.Equal(t, []Server{readPrefTestPrimary}, result) } func TestSelector_Primary_with_no_primary(t testing.T) { t.Parallel() subject := readpref.Primary() result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, []Server{readPrefTestSecondary1, readPrefTestSecondary2}) require.NoError(t, err) require.Len(t, result, 0) } func TestSelector_PrimaryPreferred(t testing.T) { t.Parallel() subject := readpref.PrimaryPreferred() result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, readPrefTestTopology.Servers) require.NoError(t, err) require.Len(t, result, 1) require.Equal(t, []Server{readPrefTestPrimary}, result) } func TestSelector_PrimaryPreferred_ignores_tags(t testing.T) { t.Parallel() subject := readpref.PrimaryPreferred( readpref.WithTags("a", "2"), ) result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, readPrefTestTopology.Servers) require.NoError(t, err) require.Len(t, result, 1) require.Equal(t, []Server{readPrefTestPrimary}, result) } func TestSelector_PrimaryPreferred_with_no_primary(t testing.T) { t.Parallel() subject := readpref.PrimaryPreferred() result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, []Server{readPrefTestSecondary1, readPrefTestSecondary2}) require.NoError(t, err) require.Len(t, result, 2) require.Equal(t, []Server{readPrefTestSecondary1, readPrefTestSecondary2}, result) } func TestSelector_PrimaryPreferred_with_no_primary_and_tags(t testing.T) { t.Parallel() subject := readpref.PrimaryPreferred( readpref.WithTags("a", "2"), ) result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, []Server{readPrefTestSecondary1, readPrefTestSecondary2}) require.NoError(t, err) require.Len(t, result, 1) require.Equal(t, []Server{readPrefTestSecondary2}, result) } func TestSelector_PrimaryPreferred_with_maxStaleness(t testing.T) { t.Parallel() subject := readpref.PrimaryPreferred( readpref.WithMaxStaleness(time.Duration(90) time.Second), ) result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, readPrefTestTopology.Servers) require.NoError(t, err) require.Len(t, result, 1) require.Equal(t, []Server{readPrefTestPrimary}, result) } func TestSelector_PrimaryPreferred_with_maxStaleness_and_no_primary(t testing.T) { t.Parallel() subject := readpref.PrimaryPreferred( readpref.WithMaxStaleness(time.Duration(90) time.Second), ) result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, []Server{readPrefTestSecondary1, readPrefTestSecondary2}) require.NoError(t, err) require.Len(t, result, 1) require.Equal(t, []Server{readPrefTestSecondary2}, result) } func TestSelector_SecondaryPreferred(t testing.T) { t.Parallel() subject := readpref.SecondaryPreferred() result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, readPrefTestTopology.Servers) require.NoError(t, err) require.Len(t, result, 2) require.Equal(t, []Server{readPrefTestSecondary1, readPrefTestSecondary2}, result) } func TestSelector_SecondaryPreferred_with_tags(t testing.T) { t.Parallel() subject := readpref.SecondaryPreferred( readpref.WithTags("a", "2"), ) result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, readPrefTestTopology.Servers) require.NoError(t, err) require.Len(t, result, 1) require.Equal(t, []Server{readPrefTestSecondary2}, result) } func TestSelector_SecondaryPreferred_with_tags_that_do_not_match(t testing.T) { t.Parallel() subject := readpref.SecondaryPreferred( readpref.WithTags("a", "3"), ) result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, readPrefTestTopology.Servers) require.NoError(t, err) require.Len(t, result, 1) require.Equal(t, []Server{readPrefTestPrimary}, result) } func TestSelector_SecondaryPreferred_with_tags_that_do_not_match_and_no_primary(t testing.T) { t.Parallel() subject := readpref.SecondaryPreferred( readpref.WithTags("a", "3"), ) result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, []Server{readPrefTestSecondary1, readPrefTestSecondary2}) require.NoError(t, err) require.Len(t, result, 0) } func TestSelector_SecondaryPreferred_with_no_secondaries(t testing.T) { t.Parallel() subject := readpref.SecondaryPreferred() result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, []Server{readPrefTestPrimary}) require.NoError(t, err) require.Len(t, result, 1) require.Equal(t, []Server{readPrefTestPrimary}, result) } func TestSelector_SecondaryPreferred_with_no_secondaries_or_primary(t testing.T) { t.Parallel() subject := readpref.SecondaryPreferred() result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, []Server{}) require.NoError(t, err) require.Len(t, result, 0) } func TestSelector_SecondaryPreferred_with_maxStaleness(t testing.T) { t.Parallel() subject := readpref.SecondaryPreferred( readpref.WithMaxStaleness(time.Duration(90) time.Second), ) result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, readPrefTestTopology.Servers) require.NoError(t, err) require.Len(t, result, 1) require.Equal(t, []Server{readPrefTestSecondary2}, result) } func TestSelector_SecondaryPreferred_with_maxStaleness_and_no_primary(t testing.T) { t.Parallel() subject := readpref.SecondaryPreferred( readpref.WithMaxStaleness(time.Duration(90) time.Second), ) result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, []Server{readPrefTestSecondary1, readPrefTestSecondary2}) require.NoError(t, err) require.Len(t, result, 1) require.Equal(t, []Server{readPrefTestSecondary2}, result) } func TestSelector_Secondary(t testing.T) { t.Parallel() subject := readpref.Secondary() result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, readPrefTestTopology.Servers) require.NoError(t, err) require.Len(t, result, 2) require.Equal(t, []Server{readPrefTestSecondary1, readPrefTestSecondary2}, result) } func TestSelector_Secondary_with_tags(t testing.T) { t.Parallel() subject := readpref.Secondary( readpref.WithTags("a", "2"), ) result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, readPrefTestTopology.Servers) require.NoError(t, err) require.Len(t, result, 1) require.Equal(t, []Server{readPrefTestSecondary2}, result) } func TestSelector_Secondary_with_empty_tag_set(t testing.T) { t.Parallel() primaryNoTags := Server{ Addr: address.Address("localhost:27017"), Kind: RSPrimary, WireVersion: &VersionRange{Min: 6, Max: 21}, } firstSecondaryNoTags := Server{ Addr: address.Address("localhost:27018"), Kind: RSSecondary, WireVersion: &VersionRange{Min: 6, Max: 21}, } secondSecondaryNoTags := Server{ Addr: address.Address("localhost:27019"), Kind: RSSecondary, WireVersion: &VersionRange{Min: 6, Max: 21}, } topologyNoTags := Topology{ Kind: ReplicaSetWithPrimary, Servers: []Server{primaryNoTags, firstSecondaryNoTags, secondSecondaryNoTags}, } nonMatchingSet := tag.Set{ {Name: "foo", Value: "bar"}, } emptyTagSet := tag.Set{} rp := readpref.Secondary( readpref.WithTagSets(nonMatchingSet, emptyTagSet), ) result, err := ReadPrefSelector(rp).SelectServer(topologyNoTags, topologyNoTags.Servers) assert.Nil(t, err, "SelectServer error: %v", err) expectedResult := []Server{firstSecondaryNoTags, secondSecondaryNoTags} assert.Equal(t, expectedResult, result, "expected result %v, got %v", expectedResult, result) } func TestSelector_Secondary_with_tags_that_do_not_match(t testing.T) { t.Parallel() subject := readpref.Secondary( readpref.WithTags("a", "3"), ) result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, readPrefTestTopology.Servers) require.NoError(t, err) require.Len(t, result, 0) } func TestSelector_Secondary_with_no_secondaries(t testing.T) { t.Parallel() subject := readpref.Secondary() result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, []Server{readPrefTestPrimary}) require.NoError(t, err) require.Len(t, result, 0) } func TestSelector_Secondary_with_maxStaleness(t testing.T) { t.Parallel() subject := readpref.Secondary( readpref.WithMaxStaleness(time.Duration(90) * time.Second), ) result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, readPrefTestTopology.Servers) require.NoError(t, err) require.Len(t, result, 1) require.Equal(t, []Server{readPrefTestSecondary2}, result) } func TestSelector_Secondary_with_maxStaleness_and_no_primary(t testing.T) { t.Parallel() subject := readpref.Secondary( readpref.WithMaxStaleness(time.Duration(90) time.Second), ) result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, []Server{readPrefTestSecondary1, readPrefTestSecondary2}) require.NoError(t, err) require.Len(t, result, 1) require.Equal(t, []Server{readPrefTestSecondary2}, result) } func TestSelector_Nearest(t testing.T) { t.Parallel() subject := readpref.Nearest() result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, readPrefTestTopology.Servers) require.NoError(t, err) require.Len(t, result, 3) require.Equal(t, []Server{readPrefTestPrimary, readPrefTestSecondary1, readPrefTestSecondary2}, result) } func TestSelector_Nearest_with_tags(t testing.T) { t.Parallel() subject := readpref.Nearest( readpref.WithTags("a", "1"), ) result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, readPrefTestTopology.Servers) require.NoError(t, err) require.Len(t, result, 2) require.Equal(t, []Server{readPrefTestPrimary, readPrefTestSecondary1}, result) } func TestSelector_Nearest_with_tags_that_do_not_match(t testing.T) { t.Parallel() subject := readpref.Nearest( readpref.WithTags("a", "3"), ) result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, readPrefTestTopology.Servers) require.NoError(t, err) require.Len(t, result, 0) } func TestSelector_Nearest_with_no_primary(t testing.T) { t.Parallel() subject := readpref.Nearest() result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, []Server{readPrefTestSecondary1, readPrefTestSecondary2}) require.NoError(t, err) require.Len(t, result, 2) require.Equal(t, []Server{readPrefTestSecondary1, readPrefTestSecondary2}, result) } func TestSelector_Nearest_with_no_secondaries(t testing.T) { t.Parallel() subject := readpref.Nearest() result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, []Server{readPrefTestPrimary}) require.NoError(t, err) require.Len(t, result, 1) require.Equal(t, []Server{readPrefTestPrimary}, result) } func TestSelector_Nearest_with_maxStaleness(t testing.T) { t.Parallel() subject := readpref.Nearest( readpref.WithMaxStaleness(time.Duration(90) * time.Second), ) result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, readPrefTestTopology.Servers) require.NoError(t, err) require.Len(t, result, 2) require.Equal(t, []Server{readPrefTestPrimary, readPrefTestSecondary2}, result) } func TestSelector_Nearest_with_maxStaleness_and_no_primary(t testing.T) { t.Parallel() subject := readpref.Nearest( readpref.WithMaxStaleness(time.Duration(90) time.Second), ) result, err := ReadPrefSelector(subject).SelectServer(readPrefTestTopology, []Server{readPrefTestSecondary1, readPrefTestSecondary2}) require.NoError(t, err) require.Len(t, result, 1) require.Equal(t, []Server{readPrefTestSecondary2}, result) } func TestSelector_Max_staleness_is_less_than_90_seconds(t testing.T) { t.Parallel() subject := readpref.Nearest( readpref.WithMaxStaleness(time.Duration(50) time.Second), ) s := Server{ Addr: address.Address("localhost:27017"), HeartbeatInterval: time.Duration(10) * time.Second, LastWriteTime: time.Date(2017, 2, 11, 14, 0, 0, 0, time.UTC), LastUpdateTime: time.Date(2017, 2, 11, 14, 0, 2, 0, time.UTC), Kind: RSPrimary, WireVersion: &VersionRange{Min: 6, Max: 21}, } c := Topology{ Kind: ReplicaSetWithPrimary, Servers: []Server{s}, } _, err := ReadPrefSelector(subject).SelectServer(c, c.Servers) require.Error(t, err) } func TestSelector_Max_staleness_is_too_low(t testing.T) { t.Parallel() subject := readpref.Nearest( readpref.WithMaxStaleness(time.Duration(100) time.Second), ) s := Server{ Addr: address.Address("localhost:27017"), HeartbeatInterval: time.Duration(100) * time.Second, LastWriteTime: time.Date(2017, 2, 11, 14, 0, 0, 0, time.UTC), LastUpdateTime: time.Date(2017, 2, 11, 14, 0, 2, 0, time.UTC), Kind: RSPrimary, WireVersion: &VersionRange{Min: 6, Max: 21}, } c := Topology{ Kind: ReplicaSetWithPrimary, Servers: []Server{s}, } _, err := ReadPrefSelector(subject).SelectServer(c, c.Servers) require.Error(t, err) }	23.79 KB	2025-05-25 12:27
📄 server.go // 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 description import ( "errors" "fmt" "time" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/bson/primitive" "go.mongodb.org/mongo-driver/internal/bsonutil" "go.mongodb.org/mongo-driver/internal/handshake" "go.mongodb.org/mongo-driver/internal/ptrutil" "go.mongodb.org/mongo-driver/mongo/address" "go.mongodb.org/mongo-driver/tag" ) // SelectedServer augments the Server type by also including the TopologyKind of the topology that includes the server. // This type should be used to track the state of a server that was selected to perform an operation. type SelectedServer struct { Server Kind TopologyKind } // Server contains information about a node in a cluster. This is created from hello command responses. If the value // of the Kind field is LoadBalancer, only the Addr and Kind fields will be set. All other fields will be set to the // zero value of the field's type. type Server struct { Addr address.Address Arbiters []string AverageRTT time.Duration AverageRTTSet bool Compression []string // compression methods returned by server CanonicalAddr address.Address ElectionID primitive.ObjectID HeartbeatInterval time.Duration HelloOK bool Hosts []string IsCryptd bool LastError error LastUpdateTime time.Time LastWriteTime time.Time MaxBatchCount uint32 MaxDocumentSize uint32 MaxMessageSize uint32 Members []address.Address Passives []string Passive bool Primary address.Address ReadOnly bool ServiceID primitive.ObjectID // Only set for servers that are deployed behind a load balancer. // Deprecated: Use SessionTimeoutMinutesPtr instead. SessionTimeoutMinutes uint32 SessionTimeoutMinutesPtr int64 SetName string SetVersion uint32 Tags tag.Set TopologyVersion TopologyVersion Kind ServerKind WireVersion VersionRange } // NewServer creates a new server description from the given hello command response. func NewServer(addr address.Address, response bson.Raw) Server { desc := Server{Addr: addr, CanonicalAddr: addr, LastUpdateTime: time.Now().UTC()} elements, err := response.Elements() if err != nil { desc.LastError = err return desc } var ok bool var isReplicaSet, isWritablePrimary, hidden, secondary, arbiterOnly bool var msg string var versionRange VersionRange for _, element := range elements { switch element.Key() { case "arbiters": var err error desc.Arbiters, err = stringSliceFromRawElement(element) if err != nil { desc.LastError = err return desc } case "arbiterOnly": arbiterOnly, ok = element.Value().BooleanOK() if !ok { desc.LastError = fmt.Errorf("expected 'arbiterOnly' to be a boolean but it's a BSON %s", element.Value().Type) return desc } case "compression": var err error desc.Compression, err = stringSliceFromRawElement(element) if err != nil { desc.LastError = err return desc } case "electionId": desc.ElectionID, ok = element.Value().ObjectIDOK() if !ok { desc.LastError = fmt.Errorf("expected 'electionId' to be a objectID but it's a BSON %s", element.Value().Type) return desc } case "iscryptd": desc.IsCryptd, ok = element.Value().BooleanOK() if !ok { desc.LastError = fmt.Errorf("expected 'iscryptd' to be a boolean but it's a BSON %s", element.Value().Type) return desc } case "helloOk": desc.HelloOK, ok = element.Value().BooleanOK() if !ok { desc.LastError = fmt.Errorf("expected 'helloOk' to be a boolean but it's a BSON %s", element.Value().Type) return desc } case "hidden": hidden, ok = element.Value().BooleanOK() if !ok { desc.LastError = fmt.Errorf("expected 'hidden' to be a boolean but it's a BSON %s", element.Value().Type) return desc } case "hosts": var err error desc.Hosts, err = stringSliceFromRawElement(element) if err != nil { desc.LastError = err return desc } case "isWritablePrimary": isWritablePrimary, ok = element.Value().BooleanOK() if !ok { desc.LastError = fmt.Errorf("expected 'isWritablePrimary' to be a boolean but it's a BSON %s", element.Value().Type) return desc } case handshake.LegacyHelloLowercase: isWritablePrimary, ok = element.Value().BooleanOK() if !ok { desc.LastError = fmt.Errorf("expected legacy hello to be a boolean but it's a BSON %s", element.Value().Type) return desc } case "isreplicaset": isReplicaSet, ok = element.Value().BooleanOK() if !ok { desc.LastError = fmt.Errorf("expected 'isreplicaset' to be a boolean but it's a BSON %s", element.Value().Type) return desc } case "lastWrite": lastWrite, ok := element.Value().DocumentOK() if !ok { desc.LastError = fmt.Errorf("expected 'lastWrite' to be a document but it's a BSON %s", element.Value().Type) return desc } dateTime, err := lastWrite.LookupErr("lastWriteDate") if err == nil { dt, ok := dateTime.DateTimeOK() if !ok { desc.LastError = fmt.Errorf("expected 'lastWriteDate' to be a datetime but it's a BSON %s", dateTime.Type) return desc } desc.LastWriteTime = time.Unix(dt/1000, dt%10001000000).UTC() } case "logicalSessionTimeoutMinutes": i64, ok := element.Value().AsInt64OK() if !ok { desc.LastError = fmt.Errorf("expected 'logicalSessionTimeoutMinutes' to be an integer but it's a BSON %s", element.Value().Type) return desc } desc.SessionTimeoutMinutes = uint32(i64) desc.SessionTimeoutMinutesPtr = &i64 case "maxBsonObjectSize": i64, ok := element.Value().AsInt64OK() if !ok { desc.LastError = fmt.Errorf("expected 'maxBsonObjectSize' to be an integer but it's a BSON %s", element.Value().Type) return desc } desc.MaxDocumentSize = uint32(i64) case "maxMessageSizeBytes": i64, ok := element.Value().AsInt64OK() if !ok { desc.LastError = fmt.Errorf("expected 'maxMessageSizeBytes' to be an integer but it's a BSON %s", element.Value().Type) return desc } desc.MaxMessageSize = uint32(i64) case "maxWriteBatchSize": i64, ok := element.Value().AsInt64OK() if !ok { desc.LastError = fmt.Errorf("expected 'maxWriteBatchSize' to be an integer but it's a BSON %s", element.Value().Type) return desc } desc.MaxBatchCount = uint32(i64) case "me": me, ok := element.Value().StringValueOK() if !ok { desc.LastError = fmt.Errorf("expected 'me' to be a string but it's a BSON %s", element.Value().Type) return desc } desc.CanonicalAddr = address.Address(me).Canonicalize() case "maxWireVersion": versionRange.Max, ok = element.Value().AsInt32OK() if !ok { desc.LastError = fmt.Errorf("expected 'maxWireVersion' to be an integer but it's a BSON %s", element.Value().Type) return desc } case "minWireVersion": versionRange.Min, ok = element.Value().AsInt32OK() if !ok { desc.LastError = fmt.Errorf("expected 'minWireVersion' to be an integer but it's a BSON %s", element.Value().Type) return desc } case "msg": msg, ok = element.Value().StringValueOK() if !ok { desc.LastError = fmt.Errorf("expected 'msg' to be a string but it's a BSON %s", element.Value().Type) return desc } case "ok": okay, ok := element.Value().AsInt32OK() if !ok { desc.LastError = fmt.Errorf("expected 'ok' to be a boolean but it's a BSON %s", element.Value().Type) return desc } if okay != 1 { desc.LastError = errors.New("not ok") return desc } case "passives": var err error desc.Passives, err = stringSliceFromRawElement(element) if err != nil { desc.LastError = err return desc } case "passive": desc.Passive, ok = element.Value().BooleanOK() if !ok { desc.LastError = fmt.Errorf("expected 'passive' to be a boolean but it's a BSON %s", element.Value().Type) return desc } case "primary": primary, ok := element.Value().StringValueOK() if !ok { desc.LastError = fmt.Errorf("expected 'primary' to be a string but it's a BSON %s", element.Value().Type) return desc } desc.Primary = address.Address(primary) case "readOnly": desc.ReadOnly, ok = element.Value().BooleanOK() if !ok { desc.LastError = fmt.Errorf("expected 'readOnly' to be a boolean but it's a BSON %s", element.Value().Type) return desc } case "secondary": secondary, ok = element.Value().BooleanOK() if !ok { desc.LastError = fmt.Errorf("expected 'secondary' to be a boolean but it's a BSON %s", element.Value().Type) return desc } case "serviceId": oid, ok := element.Value().ObjectIDOK() if !ok { desc.LastError = fmt.Errorf("expected 'serviceId' to be an ObjectId but it's a BSON %s", element.Value().Type) } desc.ServiceID = &oid case "setName": desc.SetName, ok = element.Value().StringValueOK() if !ok { desc.LastError = fmt.Errorf("expected 'setName' to be a string but it's a BSON %s", element.Value().Type) return desc } case "setVersion": i64, ok := element.Value().AsInt64OK() if !ok { desc.LastError = fmt.Errorf("expected 'setVersion' to be an integer but it's a BSON %s", element.Value().Type) return desc } desc.SetVersion = uint32(i64) case "tags": m, err := decodeStringMap(element, "tags") if err != nil { desc.LastError = err return desc } desc.Tags = tag.NewTagSetFromMap(m) case "topologyVersion": doc, ok := element.Value().DocumentOK() if !ok { desc.LastError = fmt.Errorf("expected 'topologyVersion' to be a document but it's a BSON %s", element.Value().Type) return desc } desc.TopologyVersion, err = NewTopologyVersion(doc) if err != nil { desc.LastError = err return desc } } } for _, host := range desc.Hosts { desc.Members = append(desc.Members, address.Address(host).Canonicalize()) } for _, passive := range desc.Passives { desc.Members = append(desc.Members, address.Address(passive).Canonicalize()) } for _, arbiter := range desc.Arbiters { desc.Members = append(desc.Members, address.Address(arbiter).Canonicalize()) } desc.Kind = Standalone if isReplicaSet { desc.Kind = RSGhost } else if desc.SetName != "" { if isWritablePrimary { desc.Kind = RSPrimary } else if hidden { desc.Kind = RSMember } else if secondary { desc.Kind = RSSecondary } else if arbiterOnly { desc.Kind = RSArbiter } else { desc.Kind = RSMember } } else if msg == "isdbgrid" { desc.Kind = Mongos } desc.WireVersion = &versionRange return desc } // NewDefaultServer creates a new unknown server description with the given address. func NewDefaultServer(addr address.Address) Server { return NewServerFromError(addr, nil, nil) } // NewServerFromError creates a new unknown server description with the given parameters. func NewServerFromError(addr address.Address, err error, tv TopologyVersion) Server { return Server{ Addr: addr, LastError: err, Kind: Unknown, TopologyVersion: tv, } } // SetAverageRTT sets the average round trip time for this server description. func (s Server) SetAverageRTT(rtt time.Duration) Server { s.AverageRTT = rtt s.AverageRTTSet = true return s } // DataBearing returns true if the server is a data bearing server. func (s Server) DataBearing() bool { return s.Kind == RSPrimary \|\| s.Kind == RSSecondary \|\| s.Kind == Mongos \|\| s.Kind == Standalone } // LoadBalanced returns true if the server is a load balancer or is behind a load balancer. func (s Server) LoadBalanced() bool { return s.Kind == LoadBalancer \|\| s.ServiceID != nil } // String implements the Stringer interface func (s Server) String() string { str := fmt.Sprintf("Addr: %s, Type: %s", s.Addr, s.Kind) if len(s.Tags) != 0 { str += fmt.Sprintf(", Tag sets: %s", s.Tags) } if s.AverageRTTSet { str += fmt.Sprintf(", Average RTT: %d", s.AverageRTT) } if s.LastError != nil { str += fmt.Sprintf(", Last error: %s", s.LastError) } return str } func decodeStringMap(element bson.RawElement, name string) (map[string]string, error) { doc, ok := element.Value().DocumentOK() if !ok { return nil, fmt.Errorf("expected '%s' to be a document but it's a BSON %s", name, element.Value().Type) } elements, err := doc.Elements() if err != nil { return nil, err } m := make(map[string]string) for _, element := range elements { key := element.Key() value, ok := element.Value().StringValueOK() if !ok { return nil, fmt.Errorf("expected '%s' to be a document of strings, but found a BSON %s", name, element.Value().Type) } m[key] = value } return m, nil } // Equal compares two server descriptions and returns true if they are equal func (s Server) Equal(other Server) bool { if s.CanonicalAddr.String() != other.CanonicalAddr.String() { return false } if !sliceStringEqual(s.Arbiters, other.Arbiters) { return false } if !sliceStringEqual(s.Hosts, other.Hosts) { return false } if !sliceStringEqual(s.Passives, other.Passives) { return false } if s.Primary != other.Primary { return false } if s.SetName != other.SetName { return false } if s.Kind != other.Kind { return false } if s.LastError != nil \|\| other.LastError != nil { if s.LastError == nil \|\| other.LastError == nil { return false } if s.LastError.Error() != other.LastError.Error() { return false } } if !s.WireVersion.Equals(other.WireVersion) { return false } if len(s.Tags) != len(other.Tags) \|\| !s.Tags.ContainsAll(other.Tags) { return false } if s.SetVersion != other.SetVersion { return false } if s.ElectionID != other.ElectionID { return false } if ptrutil.CompareInt64(s.SessionTimeoutMinutesPtr, other.SessionTimeoutMinutesPtr) != 0 { return false } // If TopologyVersion is nil for both servers, CompareToIncoming will return -1 because it assumes that the // incoming response is newer. We want the descriptions to be considered equal in this case, though, so an // explicit check is required. if s.TopologyVersion == nil && other.TopologyVersion == nil { return true } return s.TopologyVersion.CompareToIncoming(other.TopologyVersion) == 0 } func sliceStringEqual(a []string, b []string) bool { if len(a) != len(b) { return false } for i, v := range a { if v != b[i] { return false } } return true } // stringSliceFromRawElement decodes the provided BSON element into a []string. // This internally calls StringSliceFromRawValue on the element's value. The // error conditions outlined in that function's documentation apply for this // function as well. func stringSliceFromRawElement(element bson.RawElement) ([]string, error) { return bsonutil.StringSliceFromRawValue(element.Key(), element.Value()) }	14.88 KB	2025-05-25 12:27
📄 server_kind.go // 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 description // ServerKind represents the type of a single server in a topology. type ServerKind uint32 // These constants are the possible types of servers. const ( Standalone ServerKind = 1 RSMember ServerKind = 2 RSPrimary ServerKind = 4 + RSMember RSSecondary ServerKind = 8 + RSMember RSArbiter ServerKind = 16 + RSMember RSGhost ServerKind = 32 + RSMember Mongos ServerKind = 256 LoadBalancer ServerKind = 512 ) // String returns a stringified version of the kind or "Unknown" if the kind is invalid. func (kind ServerKind) String() string { switch kind { case Standalone: return "Standalone" case RSMember: return "RSOther" case RSPrimary: return "RSPrimary" case RSSecondary: return "RSSecondary" case RSArbiter: return "RSArbiter" case RSGhost: return "RSGhost" case Mongos: return "Mongos" case LoadBalancer: return "LoadBalancer" } return "Unknown" }	1.17 KB	2025-05-25 12:27
📄 server_selector.go // 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 description import ( "encoding/json" "fmt" "math" "time" "go.mongodb.org/mongo-driver/mongo/readpref" "go.mongodb.org/mongo-driver/tag" ) // ServerSelector is an interface implemented by types that can perform server selection given a topology description // and list of candidate servers. The selector should filter the provided candidates list and return a subset that // matches some criteria. type ServerSelector interface { SelectServer(Topology, []Server) ([]Server, error) } // ServerSelectorFunc is a function that can be used as a ServerSelector. type ServerSelectorFunc func(Topology, []Server) ([]Server, error) // SelectServer implements the ServerSelector interface. func (ssf ServerSelectorFunc) SelectServer(t Topology, s []Server) ([]Server, error) { return ssf(t, s) } // serverSelectorInfo contains metadata concerning the server selector for the // purpose of publication. type serverSelectorInfo struct { Type string Data string `json:",omitempty"` Selectors []serverSelectorInfo `json:",omitempty"` } // String returns the JSON string representation of the serverSelectorInfo. func (sss serverSelectorInfo) String() string { bytes, _ := json.Marshal(sss) return string(bytes) } // serverSelectorInfoGetter is an interface that defines an info() method to // get the serverSelectorInfo. type serverSelectorInfoGetter interface { info() serverSelectorInfo } type compositeSelector struct { selectors []ServerSelector } func (cs compositeSelector) info() serverSelectorInfo { csInfo := serverSelectorInfo{Type: "compositeSelector"} for _, sel := range cs.selectors { if getter, ok := sel.(serverSelectorInfoGetter); ok { csInfo.Selectors = append(csInfo.Selectors, getter.info()) } } return csInfo } // String returns the JSON string representation of the compositeSelector. func (cs compositeSelector) String() string { return cs.info().String() } // CompositeSelector combines multiple selectors into a single selector by applying them in order to the candidates // list. // // For example, if the initial candidates list is [s0, s1, s2, s3] and two selectors are provided where the first // matches s0 and s1 and the second matches s1 and s2, the following would occur during server selection: // // 1. firstSelector([s0, s1, s2, s3]) -> [s0, s1] // 2. secondSelector([s0, s1]) -> [s1] // // The final list of candidates returned by the composite selector would be [s1]. func CompositeSelector(selectors []ServerSelector) ServerSelector { return &compositeSelector{selectors: selectors} } func (cs compositeSelector) SelectServer(t Topology, candidates []Server) ([]Server, error) { var err error for _, sel := range cs.selectors { candidates, err = sel.SelectServer(t, candidates) if err != nil { return nil, err } } return candidates, nil } type latencySelector struct { latency time.Duration } // LatencySelector creates a ServerSelector which selects servers based on their average RTT values. func LatencySelector(latency time.Duration) ServerSelector { return &latencySelector{latency: latency} } func (latencySelector) info() serverSelectorInfo { return serverSelectorInfo{Type: "latencySelector"} } func (selector latencySelector) String() string { return selector.info().String() } func (selector latencySelector) SelectServer(t Topology, candidates []Server) ([]Server, error) { if selector.latency < 0 { return candidates, nil } if t.Kind == LoadBalanced { // In LoadBalanced mode, there should only be one server in the topology and it must be selected. return candidates, nil } switch len(candidates) { case 0, 1: return candidates, nil default: min := time.Duration(math.MaxInt64) for _, candidate := range candidates { if candidate.AverageRTTSet { if candidate.AverageRTT < min { min = candidate.AverageRTT } } } if min == math.MaxInt64 { return candidates, nil } max := min + selector.latency viableIndexes := make([]int, 0, len(candidates)) for i, candidate := range candidates { if candidate.AverageRTTSet { if candidate.AverageRTT <= max { viableIndexes = append(viableIndexes, i) } } } if len(viableIndexes) == len(candidates) { return candidates, nil } result := make([]Server, len(viableIndexes)) for i, idx := range viableIndexes { result[i] = candidates[idx] } return result, nil } } type writeServerSelector struct{} // WriteSelector selects all the writable servers. func WriteSelector() ServerSelector { return writeServerSelector{} } func (writeServerSelector) info() serverSelectorInfo { return serverSelectorInfo{Type: "writeSelector"} } func (selector writeServerSelector) String() string { return selector.info().String() } func (writeServerSelector) SelectServer(t Topology, candidates []Server) ([]Server, error) { switch t.Kind { case Single, LoadBalanced: return candidates, nil default: // Determine the capacity of the results slice. selected := 0 for _, candidate := range candidates { switch candidate.Kind { case Mongos, RSPrimary, Standalone: selected++ } } // Append candidates to the results slice. result := make([]Server, 0, selected) for _, candidate := range candidates { switch candidate.Kind { case Mongos, RSPrimary, Standalone: result = append(result, candidate) } } return result, nil } } type readPrefServerSelector struct { rp readpref.ReadPref isOutputAggregate bool } // ReadPrefSelector selects servers based on the provided read preference. func ReadPrefSelector(rp readpref.ReadPref) ServerSelector { return readPrefServerSelector{ rp: rp, isOutputAggregate: false, } } func (selector readPrefServerSelector) info() serverSelectorInfo { return serverSelectorInfo{ Type: "readPrefSelector", Data: selector.rp.String(), } } func (selector readPrefServerSelector) String() string { return selector.info().String() } func (selector readPrefServerSelector) SelectServer(t Topology, candidates []Server) ([]Server, error) { if t.Kind == LoadBalanced { // In LoadBalanced mode, there should only be one server in the topology and it must be selected. We check // this before checking MaxStaleness support because there's no monitoring in this mode, so the candidate // server wouldn't have a wire version set, which would result in an error. return candidates, nil } switch t.Kind { case Single: return candidates, nil case ReplicaSetNoPrimary, ReplicaSetWithPrimary: return selectForReplicaSet(selector.rp, selector.isOutputAggregate, t, candidates) case Sharded: return selectByKind(candidates, Mongos), nil } return nil, nil } // OutputAggregateSelector selects servers based on the provided read preference // given that the underlying operation is aggregate with an output stage. func OutputAggregateSelector(rp readpref.ReadPref) ServerSelector { return readPrefServerSelector{ rp: rp, isOutputAggregate: true, } } func selectForReplicaSet(rp readpref.ReadPref, isOutputAggregate bool, t Topology, candidates []Server) ([]Server, error) { if err := verifyMaxStaleness(rp, t); err != nil { return nil, err } // If underlying operation is an aggregate with an output stage, only apply read preference // if all candidates are 5.0+. Otherwise, operate under primary read preference. if isOutputAggregate { for _, s := range candidates { if s.WireVersion.Max < 13 { return selectByKind(candidates, RSPrimary), nil } } } switch rp.Mode() { case readpref.PrimaryMode: return selectByKind(candidates, RSPrimary), nil case readpref.PrimaryPreferredMode: selected := selectByKind(candidates, RSPrimary) if len(selected) == 0 { selected = selectSecondaries(rp, candidates) return selectByTagSet(selected, rp.TagSets()), nil } return selected, nil case readpref.SecondaryPreferredMode: selected := selectSecondaries(rp, candidates) selected = selectByTagSet(selected, rp.TagSets()) if len(selected) > 0 { return selected, nil } return selectByKind(candidates, RSPrimary), nil case readpref.SecondaryMode: selected := selectSecondaries(rp, candidates) return selectByTagSet(selected, rp.TagSets()), nil case readpref.NearestMode: selected := selectByKind(candidates, RSPrimary) selected = append(selected, selectSecondaries(rp, candidates)...) return selectByTagSet(selected, rp.TagSets()), nil } return nil, fmt.Errorf("unsupported mode: %d", rp.Mode()) } func selectSecondaries(rp readpref.ReadPref, candidates []Server) []Server { secondaries := selectByKind(candidates, RSSecondary) if len(secondaries) == 0 { return secondaries } if maxStaleness, set := rp.MaxStaleness(); set { primaries := selectByKind(candidates, RSPrimary) if len(primaries) == 0 { baseTime := secondaries[0].LastWriteTime for i := 1; i < len(secondaries); i++ { if secondaries[i].LastWriteTime.After(baseTime) { baseTime = secondaries[i].LastWriteTime } } var selected []Server for _, secondary := range secondaries { estimatedStaleness := baseTime.Sub(secondary.LastWriteTime) + secondary.HeartbeatInterval if estimatedStaleness <= maxStaleness { selected = append(selected, secondary) } } return selected } primary := primaries[0] var selected []Server for _, secondary := range secondaries { estimatedStaleness := secondary.LastUpdateTime.Sub(secondary.LastWriteTime) - primary.LastUpdateTime.Sub(primary.LastWriteTime) + secondary.HeartbeatInterval if estimatedStaleness <= maxStaleness { selected = append(selected, secondary) } } return selected } return secondaries } func selectByTagSet(candidates []Server, tagSets []tag.Set) []Server { if len(tagSets) == 0 { return candidates } for _, ts := range tagSets { // If this tag set is empty, we can take a fast path because the empty list is a subset of all tag sets, so // all candidate servers will be selected. if len(ts) == 0 { return candidates } var results []Server for _, s := range candidates { // ts is non-empty, so only servers with a non-empty set of tags need to be checked. if len(s.Tags) > 0 && s.Tags.ContainsAll(ts) { results = append(results, s) } } if len(results) > 0 { return results } } return []Server{} } func selectByKind(candidates []Server, kind ServerKind) []Server { // Record the indices of viable candidates first and then append those to the returned slice // to avoid appending costly Server structs directly as an optimization. viableIndexes := make([]int, 0, len(candidates)) for i, s := range candidates { if s.Kind == kind { viableIndexes = append(viableIndexes, i) } } if len(viableIndexes) == len(candidates) { return candidates } result := make([]Server, len(viableIndexes)) for i, idx := range viableIndexes { result[i] = candidates[idx] } return result } func verifyMaxStaleness(rp readpref.ReadPref, t Topology) error { maxStaleness, set := rp.MaxStaleness() if !set { return nil } if maxStaleness < 90time.Second { return fmt.Errorf("max staleness (%s) must be greater than or equal to 90s", maxStaleness) } if len(t.Servers) < 1 { // Maybe we should return an error here instead? return nil } // we'll assume all candidates have the same heartbeat interval. s := t.Servers[0] idleWritePeriod := 10 time.Second if maxStaleness < s.HeartbeatInterval+idleWritePeriod { return fmt.Errorf( "max staleness (%s) must be greater than or equal to the heartbeat interval (%s) plus idle write period (%s)", maxStaleness, s.HeartbeatInterval, idleWritePeriod, ) } return nil }	11.74 KB	2025-05-25 12:27
📄 server_test.go // 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 description import ( "errors" "testing" "time" "go.mongodb.org/mongo-driver/bson/primitive" "go.mongodb.org/mongo-driver/internal/assert" "go.mongodb.org/mongo-driver/mongo/address" "go.mongodb.org/mongo-driver/tag" ) func TestServer(t testing.T) { int64ToPtr := func(i64 int64) int64 { return &i64 } t.Run("equals", func(t testing.T) { defaultServer := Server{} // Only some of the Server fields affect equality testCases := []struct { name string server Server equal bool }{ {"empty", Server{}, true}, {"address", Server{Addr: address.Address("foo")}, true}, {"arbiters", Server{Arbiters: []string{"foo"}}, false}, {"rtt", Server{AverageRTT: time.Second}, true}, {"compression", Server{Compression: []string{"foo"}}, true}, {"canonicalAddr", Server{CanonicalAddr: address.Address("foo")}, false}, {"electionID", Server{ElectionID: primitive.NewObjectID()}, false}, {"heartbeatInterval", Server{HeartbeatInterval: time.Second}, true}, {"hosts", Server{Hosts: []string{"foo"}}, false}, {"lastError", Server{LastError: errors.New("foo")}, false}, {"lastUpdateTime", Server{LastUpdateTime: time.Now()}, true}, {"lastWriteTime", Server{LastWriteTime: time.Now()}, true}, {"maxBatchCount", Server{MaxBatchCount: 1}, true}, {"maxDocumentSize", Server{MaxDocumentSize: 1}, true}, {"maxMessageSize", Server{MaxMessageSize: 1}, true}, {"members", Server{Members: []address.Address{address.Address("foo")}}, true}, {"passives", Server{Passives: []string{"foo"}}, false}, {"passive", Server{Passive: true}, true}, {"primary", Server{Primary: address.Address("foo")}, false}, {"readOnly", Server{ReadOnly: true}, true}, { "sessionTimeoutMinutes", Server{ SessionTimeoutMinutesPtr: int64ToPtr(1), SessionTimeoutMinutes: 1, }, false, }, {"setName", Server{SetName: "foo"}, false}, {"setVersion", Server{SetVersion: 1}, false}, {"tags", Server{Tags: tag.Set{tag.Tag{"foo", "bar"}}}, false}, {"topologyVersion", Server{TopologyVersion: &TopologyVersion{primitive.NewObjectID(), 0}}, false}, {"kind", Server{Kind: Standalone}, false}, {"wireVersion", Server{WireVersion: &VersionRange{1, 2}}, false}, } for _, tc := range testCases { t.Run(tc.name, func(t testing.T) { actual := defaultServer.Equal(tc.server) assert.Equal(t, actual, tc.equal, "expected %v, got %v", tc.equal, actual) }) } }) }	2.66 KB	2025-05-25 12:27
📄 shared_spec_test.go // 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 description import ( "encoding/json" "io/ioutil" "path" "strconv" "testing" "time" "go.mongodb.org/mongo-driver/internal/require" "go.mongodb.org/mongo-driver/mongo/address" "go.mongodb.org/mongo-driver/mongo/readpref" "go.mongodb.org/mongo-driver/tag" ) type testCase struct { TopologyDescription topDesc `json:"topology_description"` Operation string `json:"operation"` ReadPreference readPref `json:"read_preference"` SuitableServers []serverDesc `json:"suitable_servers"` InLatencyWindow []serverDesc `json:"in_latency_window"` HeartbeatFrequencyMS int `json:"heartbeatFrequencyMS"` Error bool } type topDesc struct { Type string `json:"type"` Servers []serverDesc `json:"servers"` } type serverDesc struct { Address string `json:"address"` AverageRTTMS int `json:"avg_rtt_ms"` MaxWireVersion int32 `json:"maxWireVersion"` LastUpdateTime int `json:"lastUpdateTime"` LastWrite lastWriteDate `json:"lastWrite"` Type string `json:"type"` Tags map[string]string `json:"tags"` } type lastWriteDate struct { LastWriteDate lastWriteDateInner `json:"lastWriteDate"` } // TODO(GODRIVER-33): Use proper extended JSON parsing to eliminate the need for this struct. type lastWriteDateInner struct { Value string `json:"$numberLong"` } type readPref struct { MaxStaleness int `json:"maxStalenessSeconds"` Mode string `json:"mode"` TagSets []map[string]string `json:"tag_sets"` } func topologyKindFromString(t testing.T, s string) TopologyKind { t.Helper() switch s { case "Single": return Single case "ReplicaSet": return ReplicaSet case "ReplicaSetNoPrimary": return ReplicaSetNoPrimary case "ReplicaSetWithPrimary": return ReplicaSetWithPrimary case "Sharded": return Sharded case "LoadBalanced": return LoadBalanced case "Unknown": return Unknown default: t.Fatalf("unrecognized topology kind: %q", s) } return Unknown } func serverKindFromString(t testing.T, s string) ServerKind { t.Helper() switch s { case "Standalone": return Standalone case "RSOther": return RSMember case "RSPrimary": return RSPrimary case "RSSecondary": return RSSecondary case "RSArbiter": return RSArbiter case "RSGhost": return RSGhost case "Mongos": return Mongos case "LoadBalancer": return LoadBalancer case "PossiblePrimary", "Unknown": // Go does not have a PossiblePrimary server type and per the SDAM spec, this type is synonymous with Unknown. return Unknown default: t.Fatalf("unrecognized server kind: %q", s) } return Unknown } func findServerByAddress(servers []Server, address string) Server { for _, server := range servers { if server.Addr.String() == address { return server } } return Server{} } func anyTagsInSets(sets []tag.Set) bool { for _, set := range sets { if len(set) > 0 { return true } } return false } func compareServers(t testing.T, expected []serverDesc, actual []Server) { require.Equal(t, len(expected), len(actual)) for _, expectedServer := range expected { actualServer := findServerByAddress(actual, expectedServer.Address) require.NotNil(t, actualServer) if expectedServer.AverageRTTMS != nil { require.Equal(t, expectedServer.AverageRTTMS, int(actualServer.AverageRTT/time.Millisecond)) } require.Equal(t, expectedServer.Type, actualServer.Kind.String()) require.Equal(t, len(expectedServer.Tags), len(actualServer.Tags)) for _, actualTag := range actualServer.Tags { expectedTag, ok := expectedServer.Tags[actualTag.Name] require.True(t, ok) require.Equal(t, expectedTag, actualTag.Value) } } } func selectServers(t testing.T, test testCase) error { servers := make([]Server, 0, len(test.TopologyDescription.Servers)) // Times in the JSON files are given as offsets from an unspecified time, but the driver // stores the lastWrite field as a timestamp, so we arbitrarily choose the current time // as the base to offset from. baseTime := time.Now() for _, serverDescription := range test.TopologyDescription.Servers { server := Server{ Addr: address.Address(serverDescription.Address), Kind: serverKindFromString(t, serverDescription.Type), } if serverDescription.AverageRTTMS != nil { server.AverageRTT = time.Duration(serverDescription.AverageRTTMS) * time.Millisecond server.AverageRTTSet = true } if test.HeartbeatFrequencyMS != nil { server.HeartbeatInterval = time.Duration(test.HeartbeatFrequencyMS) time.Millisecond } if serverDescription.LastUpdateTime != nil { ms := int64(serverDescription.LastUpdateTime) server.LastUpdateTime = time.Unix(ms/1e3, ms%1e3/1e6) } if serverDescription.LastWrite != nil { i, err := strconv.ParseInt(serverDescription.LastWrite.LastWriteDate.Value, 10, 64) if err != nil { return err } timeWithOffset := baseTime.Add(time.Duration(i) time.Millisecond) server.LastWriteTime = timeWithOffset } if serverDescription.MaxWireVersion != nil { versionRange := NewVersionRange(0, serverDescription.MaxWireVersion) server.WireVersion = &versionRange } if serverDescription.Tags != nil { server.Tags = tag.NewTagSetFromMap(serverDescription.Tags) } if test.ReadPreference.MaxStaleness != nil && server.WireVersion == nil { server.WireVersion = &VersionRange{Max: 21} } servers = append(servers, server) } c := Topology{ Kind: topologyKindFromString(t, test.TopologyDescription.Type), Servers: servers, } if len(test.ReadPreference.Mode) == 0 { test.ReadPreference.Mode = "Primary" } readprefMode, err := readpref.ModeFromString(test.ReadPreference.Mode) if err != nil { return err } options := make([]readpref.Option, 0, 1) tagSets := tag.NewTagSetsFromMaps(test.ReadPreference.TagSets) if anyTagsInSets(tagSets) { options = append(options, readpref.WithTagSets(tagSets...)) } if test.ReadPreference.MaxStaleness != nil { s := time.Duration(test.ReadPreference.MaxStaleness) * time.Second options = append(options, readpref.WithMaxStaleness(s)) } rp, err := readpref.New(readprefMode, options...) if err != nil { return err } selector := ReadPrefSelector(rp) if test.Operation == "write" { selector = CompositeSelector( []ServerSelector{WriteSelector(), selector}, ) } result, err := selector.SelectServer(c, c.Servers) if err != nil { return err } compareServers(t, test.SuitableServers, result) latencySelector := LatencySelector(time.Duration(15) * time.Millisecond) selector = CompositeSelector( []ServerSelector{selector, latencySelector}, ) result, err = selector.SelectServer(c, c.Servers) if err != nil { return err } compareServers(t, test.InLatencyWindow, result) return nil } func runTest(t testing.T, testsDir string, directory string, filename string) { filepath := path.Join(testsDir, directory, filename) content, err := ioutil.ReadFile(filepath) require.NoError(t, err) // Remove ".json" from filename. filename = filename[:len(filename)-5] testName := directory + "/" + filename + ":" t.Run(testName, func(t testing.T) { var test testCase require.NoError(t, json.Unmarshal(content, &test)) err := selectServers(t, &test) if test.Error == nil \|\| !*test.Error { require.NoError(t, err) } else { require.Error(t, err) } }) }	7.6 KB	2025-05-25 12:27
📄 topology.go // 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 description import ( "fmt" "go.mongodb.org/mongo-driver/mongo/readpref" ) // Topology contains information about a MongoDB cluster. type Topology struct { Servers []Server SetName string Kind TopologyKind // Deprecated: Use SessionTimeoutMinutesPtr instead. SessionTimeoutMinutes uint32 SessionTimeoutMinutesPtr *int64 CompatibilityErr error } // String implements the Stringer interface. func (t Topology) String() string { var serversStr string for _, s := range t.Servers { serversStr += "{ " + s.String() + " }, " } return fmt.Sprintf("Type: %s, Servers: [%s]", t.Kind, serversStr) } // Equal compares two topology descriptions and returns true if they are equal. func (t Topology) Equal(other Topology) bool { if t.Kind != other.Kind { return false } topoServers := make(map[string]Server) for _, s := range t.Servers { topoServers[s.Addr.String()] = s } otherServers := make(map[string]Server) for _, s := range other.Servers { otherServers[s.Addr.String()] = s } if len(topoServers) != len(otherServers) { return false } for _, server := range topoServers { otherServer := otherServers[server.Addr.String()] if !server.Equal(otherServer) { return false } } return true } // HasReadableServer returns true if the topology contains a server suitable for reading. // // If the Topology's kind is Single or Sharded, the mode parameter is ignored and the function contains true if any of // the servers in the Topology are of a known type. // // For replica sets, the function returns true if the cluster contains a server that matches the provided read // preference mode. func (t Topology) HasReadableServer(mode readpref.Mode) bool { switch t.Kind { case Single, Sharded: return hasAvailableServer(t.Servers, 0) case ReplicaSetWithPrimary: return hasAvailableServer(t.Servers, mode) case ReplicaSetNoPrimary, ReplicaSet: if mode == readpref.PrimaryMode { return false } // invalid read preference if !mode.IsValid() { return false } return hasAvailableServer(t.Servers, mode) } return false } // HasWritableServer returns true if a topology has a server available for writing. // // If the Topology's kind is Single or Sharded, this function returns true if any of the servers in the Topology are of // a known type. // // For replica sets, the function returns true if the replica set contains a primary. func (t Topology) HasWritableServer() bool { return t.HasReadableServer(readpref.PrimaryMode) } // hasAvailableServer returns true if any servers are available based on the read preference. func hasAvailableServer(servers []Server, mode readpref.Mode) bool { switch mode { case readpref.PrimaryMode: for _, s := range servers { if s.Kind == RSPrimary { return true } } return false case readpref.PrimaryPreferredMode, readpref.SecondaryPreferredMode, readpref.NearestMode: for _, s := range servers { if s.Kind == RSPrimary \|\| s.Kind == RSSecondary { return true } } return false case readpref.SecondaryMode: for _, s := range servers { if s.Kind == RSSecondary { return true } } return false } // read preference is not specified for _, s := range servers { switch s.Kind { case Standalone, RSMember, RSPrimary, RSSecondary, RSArbiter, RSGhost, Mongos: return true } } return false }	3.58 KB	2025-05-25 12:27
📄 topology_kind.go // 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 description // TopologyKind represents a specific topology configuration. type TopologyKind uint32 // These constants are the available topology configurations. const ( Single TopologyKind = 1 ReplicaSet TopologyKind = 2 ReplicaSetNoPrimary TopologyKind = 4 + ReplicaSet ReplicaSetWithPrimary TopologyKind = 8 + ReplicaSet Sharded TopologyKind = 256 LoadBalanced TopologyKind = 512 ) // String implements the fmt.Stringer interface. func (kind TopologyKind) String() string { switch kind { case Single: return "Single" case ReplicaSet: return "ReplicaSet" case ReplicaSetNoPrimary: return "ReplicaSetNoPrimary" case ReplicaSetWithPrimary: return "ReplicaSetWithPrimary" case Sharded: return "Sharded" case LoadBalanced: return "LoadBalanced" } return "Unknown" }	1.1 KB	2025-05-25 12:27
📄 topology_version.go // 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 description import ( "fmt" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/bson/primitive" ) // TopologyVersion represents a software version. type TopologyVersion struct { ProcessID primitive.ObjectID Counter int64 } // NewTopologyVersion creates a TopologyVersion based on doc func NewTopologyVersion(doc bson.Raw) (TopologyVersion, error) { elements, err := doc.Elements() if err != nil { return nil, err } var tv TopologyVersion var ok bool for _, element := range elements { switch element.Key() { case "processId": tv.ProcessID, ok = element.Value().ObjectIDOK() if !ok { return nil, fmt.Errorf("expected 'processId' to be a objectID but it's a BSON %s", element.Value().Type) } case "counter": tv.Counter, ok = element.Value().Int64OK() if !ok { return nil, fmt.Errorf("expected 'counter' to be an int64 but it's a BSON %s", element.Value().Type) } } } return &tv, nil } // CompareToIncoming compares the receiver, which represents the currently known TopologyVersion for a server, to an // incoming TopologyVersion extracted from a server command response. // // This returns -1 if the receiver version is less than the response, 0 if the versions are equal, and 1 if the // receiver version is greater than the response. This comparison is not commutative. func (tv TopologyVersion) CompareToIncoming(responseTV *TopologyVersion) int { if tv == nil \|\| responseTV == nil { return -1 } if tv.ProcessID != responseTV.ProcessID { return -1 } if tv.Counter == responseTV.Counter { return 0 } if tv.Counter < responseTV.Counter { return -1 } return 1 }	1.88 KB	2025-05-25 12:27
📄 version_range.go // 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 description import "fmt" // VersionRange represents a range of versions. type VersionRange struct { Min int32 Max int32 } // NewVersionRange creates a new VersionRange given a min and a max. func NewVersionRange(min, max int32) VersionRange { return VersionRange{Min: min, Max: max} } // Includes returns a bool indicating whether the supplied integer is included // in the range. func (vr VersionRange) Includes(v int32) bool { return v >= vr.Min && v <= vr.Max } // Equals returns a bool indicating whether the supplied VersionRange is equal. func (vr VersionRange) Equals(other VersionRange) bool { if vr == nil && other == nil { return true } if vr == nil \|\| other == nil { return false } return vr.Min == other.Min && vr.Max == other.Max } // String implements the fmt.Stringer interface. func (vr VersionRange) String() string { return fmt.Sprintf("[%d, %d]", vr.Min, vr.Max) }	1.16 KB	2025-05-25 12:27
📄 version_range_test.go // 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 description import "testing" func TestRange_Includes(t *testing.T) { t.Parallel() subject := NewVersionRange(1, 3) tests := []struct { n int32 expected bool }{ {0, false}, {1, true}, {2, true}, {3, true}, {4, false}, {10, false}, } for _, test := range tests { actual := subject.Includes(test.n) if actual != test.expected { t.Fatalf("expected %v to be %t", test.n, test.expected) } } }	709 B	2025-05-25 12:27

Terminal