GODRIVER-2572 Add log messages to Server selection spec

internal/logger/component.go

@@ -28,6 +28,10 @@ const (
 	ConnectionCheckoutFailed         = "Connection checkout failed"
 	ConnectionCheckedOut             = "Connection checked out"
 	ConnectionCheckedIn              = "Connection checked in"
+	ServerSelectionFailed            = "Server selection failed"
+	ServerSelectionStarted           = "Server selection started"
+	ServerSelectionSucceeded         = "Server selection succeeded"
+	ServerSelectionWaiting           = "Waiting for suitable server to become available"
 	TopologyClosed                   = "Stopped topology monitoring"
 	TopologyDescriptionChanged       = "Topology description changed"
 	TopologyOpening                  = "Starting topology monitoring"
@@ -53,19 +57,32 @@ const (
 	KeyMessage             = "message"
 	KeyMinPoolSize         = "minPoolSize"
 	KeyNewDescription      = "newDescription"
+	KeyOperation           = "operation"
 	KeyOperationID         = "operationId"
 	KeyPreviousDescription = "previousDescription"
+	KeyRemainingTimeMS     = "remainingTimeMS"
 	KeyReason              = "reason"
 	KeyReply               = "reply"
 	KeyRequestID           = "requestId"
+	KeySelector            = "selector"
 	KeyServerConnectionID  = "serverConnectionId"
 	KeyServerHost          = "serverHost"
 	KeyServerPort          = "serverPort"
 	KeyServiceID           = "serviceId"
 	KeyTimestamp           = "timestamp"
+	KeyTopologyDescription = "topologyDescription"
 	KeyTopologyID          = "topologyId"
 )
 
+// ContextKey is a custom type used to prevent key collisions when using the
+// context package.
+type ContextKey string
+
+const (
+	ContextKeyOperation   ContextKey = KeyOperation
+	ContextKeyOperationID ContextKey = KeyOperationID
+)
+
 type KeyValues []interface{}
 
 func (kvs *KeyValues) Add(key string, value interface{}) {
@@ -252,6 +269,36 @@ func SerializeServer(srv Server, extraKV ...interface{}) KeyValues {
 	return keysAndValues
 }
 
+// ServerSelection contains data that all server selection messages MUST
+// contain.
+type ServerSelection struct {
+	Selector            string
+	OperationID         interface{}
+	Operation           interface{}
+	TopologyDescription string
+}
+
+// SerializeServerSelection serializes a Topology message into a slice of keys
+// and values that can be passed to a logger.
+func SerializeServerSelection(srvSelection ServerSelection, extraKV ...interface{}) KeyValues {
+	keysAndValues := KeyValues{
+		KeySelector, srvSelection.Selector,
+		KeyOperation, srvSelection.Operation,
+		KeyTopologyDescription, srvSelection.TopologyDescription,
+	}
+
+	if srvSelection.OperationID != nil {
+		keysAndValues.Add(KeyOperationID, srvSelection.OperationID)
+	}
+
+	// Add the optional keys and values.
+	for i := 0; i < len(extraKV); i += 2 {
+		keysAndValues.Add(extraKV[i].(string), extraKV[i+1])
+	}
+
+	return keysAndValues
+}
+
 // Topology contains data that all topology messages MAY contain.
 type Topology struct {
 	ID      primitive.ObjectID // Driver's unique ID for this topology

mongo/collection.go

@@ -1868,9 +1868,26 @@ func (coll *Collection) drop(ctx context.Context) error {
 	return nil
 }
 
-// makePinnedSelector makes a selector for a pinned session with a pinned server. Will attempt to do server selection on
-// the pinned server but if that fails it will go through a list of default selectors
-func makePinnedSelector(sess *session.Client, defaultSelector description.ServerSelector) description.ServerSelectorFunc {
+type pinnedServerSelector struct {
+	stringer fmt.Stringer
+	fn       description.ServerSelectorFunc
+}
+
+func (pss pinnedServerSelector) String() string {
+	return pss.stringer.String()
+}
+
+func (pss pinnedServerSelector) SelectServer(
+	t description.Topology,
+	s []description.Server,
+) ([]description.Server, error) {
+	return pss.fn(t, s)
+}
+
+// pinnedSelectorFunc makes a selector for a pinned session with a pinned
+// server. It attempts to do server selection on the pinned server, but if that
+// fails, it will go through a list of default selectors.
+func pinnedSelectorFunc(sess *session.Client, defaultSelector description.ServerSelector) description.ServerSelectorFunc {
 	return func(t description.Topology, svrs []description.Server) ([]description.Server, error) {
 		if sess != nil && sess.PinnedServer != nil {
 			// If there is a pinned server, try to find it in the list of candidates.
@@ -1887,7 +1904,18 @@ func makePinnedSelector(sess *session.Client, defaultSelector description.Server
 	}
 }
 
-func makeReadPrefSelector(sess *session.Client, selector description.ServerSelector, localThreshold time.Duration) description.ServerSelectorFunc {
+func makePinnedSelector(sess *session.Client, defaultSelector description.ServerSelector) description.ServerSelector {
+	if srvSelectorStringer, ok := defaultSelector.(fmt.Stringer); ok {
+		return pinnedServerSelector{
+			stringer: srvSelectorStringer,
+			fn:       pinnedSelectorFunc(sess, defaultSelector),
+		}
+	}
+
+	return pinnedSelectorFunc(sess, defaultSelector)
+}
+
+func makeReadPrefSelector(sess *session.Client, selector description.ServerSelector, localThreshold time.Duration) description.ServerSelector {
 	if sess != nil && sess.TransactionRunning() {
 		selector = description.CompositeSelector([]description.ServerSelector{
 			description.ReadPrefSelector(sess.CurrentRp),
@@ -1898,7 +1926,7 @@ func makeReadPrefSelector(sess *session.Client, selector description.ServerSelec
 	return makePinnedSelector(sess, selector)
 }
 
-func makeOutputAggregateSelector(sess *session.Client, rp *readpref.ReadPref, localThreshold time.Duration) description.ServerSelectorFunc {
+func makeOutputAggregateSelector(sess *session.Client, rp *readpref.ReadPref, localThreshold time.Duration) description.ServerSelector {
 	if sess != nil && sess.TransactionRunning() {
 		// Use current transaction's read preference if available
 		rp = sess.CurrentRp

mongo/description/server_selector.go

@@ -7,6 +7,7 @@
 package description
 
 import (
+	"encoding/json"
 	"fmt"
 	"math"
 	"time"
@@ -30,10 +31,52 @@ func (ssf ServerSelectorFunc) SelectServer(t Topology, s []Server) ([]Server, er
 	return ssf(t, s)
 }
 
+// serverSelectorInfo contains metadata concerning the server selector for the
+// purpose of publication.
+type serverSelectorInfo struct {
+	Type           string
+	CustomSelector bool
+	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 {
+	cssInfo := &serverSelectorInfo{
+		Type:           "compositeSelector",
+		CustomSelector: true,
+	}
+
+	for _, sel := range cs.selectors {
+		if getter, ok := sel.(serverSelectorInfoGetter); ok {
+			cssInfo.Selectors = append(cssInfo.Selectors, getter.info())
+		}
+	}
+
+	return *cssInfo
+}
+
+// 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.
 //
@@ -63,13 +106,21 @@ type latencySelector struct {
 	latency time.Duration
 }
 
+func (latencySelector) info() serverSelectorInfo {
+	return serverSelectorInfo{Type: "latencySelector", CustomSelector: true}
+}
+
+func (selector latencySelector) String() string {
+	return selector.info().String()
+}
+
 // LatencySelector creates a ServerSelector which selects servers based on their average RTT values.
 func LatencySelector(latency time.Duration) ServerSelector {
 	return &latencySelector{latency: latency}
 }
 
-func (ls *latencySelector) SelectServer(t Topology, candidates []Server) ([]Server, error) {
-	if ls.latency < 0 {
+func (selector *latencySelector) SelectServer(t Topology, candidates []Server) ([]Server, error) {
+	if selector.latency < 0 {
 		return candidates, nil
 	}
 	if t.Kind == LoadBalanced {
@@ -94,7 +145,7 @@ func (ls *latencySelector) SelectServer(t Topology, candidates []Server) ([]Serv
 			return candidates, nil
 		}
 
-		max := min + ls.latency
+		max := min + selector.latency
 
 		viableIndexes := make([]int, 0, len(candidates))
 		for i, candidate := range candidates {
@@ -115,37 +166,66 @@ func (ls *latencySelector) SelectServer(t Topology, candidates []Server) ([]Serv
 	}
 }
 
-// WriteSelector selects all the writable servers.
-func WriteSelector() ServerSelector {
-	return ServerSelectorFunc(func(t Topology, candidates []Server) ([]Server, error) {
-		switch t.Kind {
-		case Single, LoadBalanced:
-			return candidates, nil
-		default:
-			result := []Server{}
-			for _, candidate := range candidates {
-				switch candidate.Kind {
-				case Mongos, RSPrimary, Standalone:
-					result = append(result, candidate)
-				}
+type writeServerSelector struct {
+	fn ServerSelectorFunc
+}
+
+func (writeServerSelector) info() serverSelectorInfo {
+	return serverSelectorInfo{Type: "writeSelector", CustomSelector: true}
+}
+
+func (selector writeServerSelector) String() string {
+	return selector.info().String()
+}
+
+func (selector writeServerSelector) SelectServer(t Topology, s []Server) ([]Server, error) {
+	return selector.fn(t, s)
+}
+
+func writeSelectorFunc(t Topology, candidates []Server) ([]Server, error) {
+	switch t.Kind {
+	case Single, LoadBalanced:
+		return candidates, nil
+	default:
+		result := []Server{}
+		for _, candidate := range candidates {
+			switch candidate.Kind {
+			case Mongos, RSPrimary, Standalone:
+				result = append(result, candidate)
 			}
-			return result, nil
 		}
-	})
+		return result, nil
+	}
 }
 
-// ReadPrefSelector selects servers based on the provided read preference.
-func ReadPrefSelector(rp *readpref.ReadPref) ServerSelector {
-	return readPrefSelector(rp, false)
+// WriteSelector selects all the writable servers.
+func WriteSelector() ServerSelector {
+	return writeServerSelector{
+		fn: writeSelectorFunc,
+	}
 }
 
-// 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 readPrefSelector(rp, true)
+type readPrefServerSelector struct {
+	stringer fmt.Stringer
+	fn       ServerSelectorFunc
+}
+
+func (selector readPrefServerSelector) info() serverSelectorInfo {
+	return serverSelectorInfo{
+		Type: "readPrefSelector",
+		Data: selector.stringer.String(),
+	}
+}
+
+func (selector readPrefServerSelector) String() string {
+	return selector.info().String()
 }
 
-func readPrefSelector(rp *readpref.ReadPref, isOutputAggregate bool) ServerSelector {
+func (selector readPrefServerSelector) SelectServer(t Topology, s []Server) ([]Server, error) {
+	return selector.fn(t, s)
+}
+
+func readPrefSelectorFunc(rp *readpref.ReadPref, isOutputAggregate bool) ServerSelectorFunc {
 	return ServerSelectorFunc(func(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
@@ -167,6 +247,24 @@ func readPrefSelector(rp *readpref.ReadPref, isOutputAggregate bool) ServerSelec
 	})
 }
 
+// ReadPrefSelector selects servers based on the provided read preference.
+func ReadPrefSelector(rp *readpref.ReadPref) ServerSelector {
+	return readPrefServerSelector{
+		stringer: rp,
+		fn:       readPrefSelectorFunc(rp, false),
+	}
+
+}
+
+// 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{
+		stringer: rp,
+		fn:       readPrefSelectorFunc(rp, true),
+	}
+}
+
 func selectForReplicaSet(rp *readpref.ReadPref, isOutputAggregate bool, t Topology, candidates []Server) ([]Server, error) {
 	if err := verifyMaxStaleness(rp, t); err != nil {
 		return nil, err