server.go

Documentation: go.mongodb.org/mongo-driver/mongo/description

     1  // Copyright (C) MongoDB, Inc. 2017-present.
     2  //
     3  // Licensed under the Apache License, Version 2.0 (the "License"); you may
     4  // not use this file except in compliance with the License. You may obtain
     5  // a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
     6  
     7  package description
     8  
     9  import (
    10  	"errors"
    11  	"fmt"
    12  	"time"
    13  
    14  	"go.mongodb.org/mongo-driver/bson"
    15  	"go.mongodb.org/mongo-driver/bson/primitive"
    16  	"go.mongodb.org/mongo-driver/internal/bsonutil"
    17  	"go.mongodb.org/mongo-driver/internal/handshake"
    18  	"go.mongodb.org/mongo-driver/internal/ptrutil"
    19  	"go.mongodb.org/mongo-driver/mongo/address"
    20  	"go.mongodb.org/mongo-driver/tag"
    21  )
    22  
    23  // SelectedServer augments the Server type by also including the TopologyKind of the topology that includes the server.
    24  // This type should be used to track the state of a server that was selected to perform an operation.
    25  type SelectedServer struct {
    26  	Server
    27  	Kind TopologyKind
    28  }
    29  
    30  // Server contains information about a node in a cluster. This is created from hello command responses. If the value
    31  // of the Kind field is LoadBalancer, only the Addr and Kind fields will be set. All other fields will be set to the
    32  // zero value of the field's type.
    33  type Server struct {
    34  	Addr address.Address
    35  
    36  	Arbiters          []string
    37  	AverageRTT        time.Duration
    38  	AverageRTTSet     bool
    39  	Compression       []string // compression methods returned by server
    40  	CanonicalAddr     address.Address
    41  	ElectionID        primitive.ObjectID
    42  	HeartbeatInterval time.Duration
    43  	HelloOK           bool
    44  	Hosts             []string
    45  	IsCryptd          bool
    46  	LastError         error
    47  	LastUpdateTime    time.Time
    48  	LastWriteTime     time.Time
    49  	MaxBatchCount     uint32
    50  	MaxDocumentSize   uint32
    51  	MaxMessageSize    uint32
    52  	Members           []address.Address
    53  	Passives          []string
    54  	Passive           bool
    55  	Primary           address.Address
    56  	ReadOnly          bool
    57  	ServiceID         *primitive.ObjectID // Only set for servers that are deployed behind a load balancer.
    58  	// Deprecated: Use SessionTimeoutMinutesPtr instead.
    59  	SessionTimeoutMinutes    uint32
    60  	SessionTimeoutMinutesPtr *int64
    61  	SetName                  string
    62  	SetVersion               uint32
    63  	Tags                     tag.Set
    64  	TopologyVersion          *TopologyVersion
    65  	Kind                     ServerKind
    66  	WireVersion              *VersionRange
    67  }
    68  
    69  // NewServer creates a new server description from the given hello command response.
    70  func NewServer(addr address.Address, response bson.Raw) Server {
    71  	desc := Server{Addr: addr, CanonicalAddr: addr, LastUpdateTime: time.Now().UTC()}
    72  	elements, err := response.Elements()
    73  	if err != nil {
    74  		desc.LastError = err
    75  		return desc
    76  	}
    77  	var ok bool
    78  	var isReplicaSet, isWritablePrimary, hidden, secondary, arbiterOnly bool
    79  	var msg string
    80  	var versionRange VersionRange
    81  	for _, element := range elements {
    82  		switch element.Key() {
    83  		case "arbiters":
    84  			var err error
    85  			desc.Arbiters, err = stringSliceFromRawElement(element)
    86  			if err != nil {
    87  				desc.LastError = err
    88  				return desc
    89  			}
    90  		case "arbiterOnly":
    91  			arbiterOnly, ok = element.Value().BooleanOK()
    92  			if !ok {
    93  				desc.LastError = fmt.Errorf("expected 'arbiterOnly' to be a boolean but it's a BSON %s", element.Value().Type)
    94  				return desc
    95  			}
    96  		case "compression":
    97  			var err error
    98  			desc.Compression, err = stringSliceFromRawElement(element)
    99  			if err != nil {
   100  				desc.LastError = err
   101  				return desc
   102  			}
   103  		case "electionId":
   104  			desc.ElectionID, ok = element.Value().ObjectIDOK()
   105  			if !ok {
   106  				desc.LastError = fmt.Errorf("expected 'electionId' to be a objectID but it's a BSON %s", element.Value().Type)
   107  				return desc
   108  			}
   109  		case "iscryptd":
   110  			desc.IsCryptd, ok = element.Value().BooleanOK()
   111  			if !ok {
   112  				desc.LastError = fmt.Errorf("expected 'iscryptd' to be a boolean but it's a BSON %s", element.Value().Type)
   113  				return desc
   114  			}
   115  		case "helloOk":
   116  			desc.HelloOK, ok = element.Value().BooleanOK()
   117  			if !ok {
   118  				desc.LastError = fmt.Errorf("expected 'helloOk' to be a boolean but it's a BSON %s", element.Value().Type)
   119  				return desc
   120  			}
   121  		case "hidden":
   122  			hidden, ok = element.Value().BooleanOK()
   123  			if !ok {
   124  				desc.LastError = fmt.Errorf("expected 'hidden' to be a boolean but it's a BSON %s", element.Value().Type)
   125  				return desc
   126  			}
   127  		case "hosts":
   128  			var err error
   129  			desc.Hosts, err = stringSliceFromRawElement(element)
   130  			if err != nil {
   131  				desc.LastError = err
   132  				return desc
   133  			}
   134  		case "isWritablePrimary":
   135  			isWritablePrimary, ok = element.Value().BooleanOK()
   136  			if !ok {
   137  				desc.LastError = fmt.Errorf("expected 'isWritablePrimary' to be a boolean but it's a BSON %s", element.Value().Type)
   138  				return desc
   139  			}
   140  		case handshake.LegacyHelloLowercase:
   141  			isWritablePrimary, ok = element.Value().BooleanOK()
   142  			if !ok {
   143  				desc.LastError = fmt.Errorf("expected legacy hello to be a boolean but it's a BSON %s", element.Value().Type)
   144  				return desc
   145  			}
   146  		case "isreplicaset":
   147  			isReplicaSet, ok = element.Value().BooleanOK()
   148  			if !ok {
   149  				desc.LastError = fmt.Errorf("expected 'isreplicaset' to be a boolean but it's a BSON %s", element.Value().Type)
   150  				return desc
   151  			}
   152  		case "lastWrite":
   153  			lastWrite, ok := element.Value().DocumentOK()
   154  			if !ok {
   155  				desc.LastError = fmt.Errorf("expected 'lastWrite' to be a document but it's a BSON %s", element.Value().Type)
   156  				return desc
   157  			}
   158  			dateTime, err := lastWrite.LookupErr("lastWriteDate")
   159  			if err == nil {
   160  				dt, ok := dateTime.DateTimeOK()
   161  				if !ok {
   162  					desc.LastError = fmt.Errorf("expected 'lastWriteDate' to be a datetime but it's a BSON %s", dateTime.Type)
   163  					return desc
   164  				}
   165  				desc.LastWriteTime = time.Unix(dt/1000, dt%1000*1000000).UTC()
   166  			}
   167  		case "logicalSessionTimeoutMinutes":
   168  			i64, ok := element.Value().AsInt64OK()
   169  			if !ok {
   170  				desc.LastError = fmt.Errorf("expected 'logicalSessionTimeoutMinutes' to be an integer but it's a BSON %s", element.Value().Type)
   171  				return desc
   172  			}
   173  
   174  			desc.SessionTimeoutMinutes = uint32(i64)
   175  			desc.SessionTimeoutMinutesPtr = &i64
   176  		case "maxBsonObjectSize":
   177  			i64, ok := element.Value().AsInt64OK()
   178  			if !ok {
   179  				desc.LastError = fmt.Errorf("expected 'maxBsonObjectSize' to be an integer but it's a BSON %s", element.Value().Type)
   180  				return desc
   181  			}
   182  			desc.MaxDocumentSize = uint32(i64)
   183  		case "maxMessageSizeBytes":
   184  			i64, ok := element.Value().AsInt64OK()
   185  			if !ok {
   186  				desc.LastError = fmt.Errorf("expected 'maxMessageSizeBytes' to be an integer but it's a BSON %s", element.Value().Type)
   187  				return desc
   188  			}
   189  			desc.MaxMessageSize = uint32(i64)
   190  		case "maxWriteBatchSize":
   191  			i64, ok := element.Value().AsInt64OK()
   192  			if !ok {
   193  				desc.LastError = fmt.Errorf("expected 'maxWriteBatchSize' to be an integer but it's a BSON %s", element.Value().Type)
   194  				return desc
   195  			}
   196  			desc.MaxBatchCount = uint32(i64)
   197  		case "me":
   198  			me, ok := element.Value().StringValueOK()
   199  			if !ok {
   200  				desc.LastError = fmt.Errorf("expected 'me' to be a string but it's a BSON %s", element.Value().Type)
   201  				return desc
   202  			}
   203  			desc.CanonicalAddr = address.Address(me).Canonicalize()
   204  		case "maxWireVersion":
   205  			versionRange.Max, ok = element.Value().AsInt32OK()
   206  			if !ok {
   207  				desc.LastError = fmt.Errorf("expected 'maxWireVersion' to be an integer but it's a BSON %s", element.Value().Type)
   208  				return desc
   209  			}
   210  		case "minWireVersion":
   211  			versionRange.Min, ok = element.Value().AsInt32OK()
   212  			if !ok {
   213  				desc.LastError = fmt.Errorf("expected 'minWireVersion' to be an integer but it's a BSON %s", element.Value().Type)
   214  				return desc
   215  			}
   216  		case "msg":
   217  			msg, ok = element.Value().StringValueOK()
   218  			if !ok {
   219  				desc.LastError = fmt.Errorf("expected 'msg' to be a string but it's a BSON %s", element.Value().Type)
   220  				return desc
   221  			}
   222  		case "ok":
   223  			okay, ok := element.Value().AsInt32OK()
   224  			if !ok {
   225  				desc.LastError = fmt.Errorf("expected 'ok' to be a boolean but it's a BSON %s", element.Value().Type)
   226  				return desc
   227  			}
   228  			if okay != 1 {
   229  				desc.LastError = errors.New("not ok")
   230  				return desc
   231  			}
   232  		case "passives":
   233  			var err error
   234  			desc.Passives, err = stringSliceFromRawElement(element)
   235  			if err != nil {
   236  				desc.LastError = err
   237  				return desc
   238  			}
   239  		case "passive":
   240  			desc.Passive, ok = element.Value().BooleanOK()
   241  			if !ok {
   242  				desc.LastError = fmt.Errorf("expected 'passive' to be a boolean but it's a BSON %s", element.Value().Type)
   243  				return desc
   244  			}
   245  		case "primary":
   246  			primary, ok := element.Value().StringValueOK()
   247  			if !ok {
   248  				desc.LastError = fmt.Errorf("expected 'primary' to be a string but it's a BSON %s", element.Value().Type)
   249  				return desc
   250  			}
   251  			desc.Primary = address.Address(primary)
   252  		case "readOnly":
   253  			desc.ReadOnly, ok = element.Value().BooleanOK()
   254  			if !ok {
   255  				desc.LastError = fmt.Errorf("expected 'readOnly' to be a boolean but it's a BSON %s", element.Value().Type)
   256  				return desc
   257  			}
   258  		case "secondary":
   259  			secondary, ok = element.Value().BooleanOK()
   260  			if !ok {
   261  				desc.LastError = fmt.Errorf("expected 'secondary' to be a boolean but it's a BSON %s", element.Value().Type)
   262  				return desc
   263  			}
   264  		case "serviceId":
   265  			oid, ok := element.Value().ObjectIDOK()
   266  			if !ok {
   267  				desc.LastError = fmt.Errorf("expected 'serviceId' to be an ObjectId but it's a BSON %s", element.Value().Type)
   268  			}
   269  			desc.ServiceID = &oid
   270  		case "setName":
   271  			desc.SetName, ok = element.Value().StringValueOK()
   272  			if !ok {
   273  				desc.LastError = fmt.Errorf("expected 'setName' to be a string but it's a BSON %s", element.Value().Type)
   274  				return desc
   275  			}
   276  		case "setVersion":
   277  			i64, ok := element.Value().AsInt64OK()
   278  			if !ok {
   279  				desc.LastError = fmt.Errorf("expected 'setVersion' to be an integer but it's a BSON %s", element.Value().Type)
   280  				return desc
   281  			}
   282  			desc.SetVersion = uint32(i64)
   283  		case "tags":
   284  			m, err := decodeStringMap(element, "tags")
   285  			if err != nil {
   286  				desc.LastError = err
   287  				return desc
   288  			}
   289  			desc.Tags = tag.NewTagSetFromMap(m)
   290  		case "topologyVersion":
   291  			doc, ok := element.Value().DocumentOK()
   292  			if !ok {
   293  				desc.LastError = fmt.Errorf("expected 'topologyVersion' to be a document but it's a BSON %s", element.Value().Type)
   294  				return desc
   295  			}
   296  
   297  			desc.TopologyVersion, err = NewTopologyVersion(doc)
   298  			if err != nil {
   299  				desc.LastError = err
   300  				return desc
   301  			}
   302  		}
   303  	}
   304  
   305  	for _, host := range desc.Hosts {
   306  		desc.Members = append(desc.Members, address.Address(host).Canonicalize())
   307  	}
   308  
   309  	for _, passive := range desc.Passives {
   310  		desc.Members = append(desc.Members, address.Address(passive).Canonicalize())
   311  	}
   312  
   313  	for _, arbiter := range desc.Arbiters {
   314  		desc.Members = append(desc.Members, address.Address(arbiter).Canonicalize())
   315  	}
   316  
   317  	desc.Kind = Standalone
   318  
   319  	if isReplicaSet {
   320  		desc.Kind = RSGhost
   321  	} else if desc.SetName != "" {
   322  		if isWritablePrimary {
   323  			desc.Kind = RSPrimary
   324  		} else if hidden {
   325  			desc.Kind = RSMember
   326  		} else if secondary {
   327  			desc.Kind = RSSecondary
   328  		} else if arbiterOnly {
   329  			desc.Kind = RSArbiter
   330  		} else {
   331  			desc.Kind = RSMember
   332  		}
   333  	} else if msg == "isdbgrid" {
   334  		desc.Kind = Mongos
   335  	}
   336  
   337  	desc.WireVersion = &versionRange
   338  
   339  	return desc
   340  }
   341  
   342  // NewDefaultServer creates a new unknown server description with the given address.
   343  func NewDefaultServer(addr address.Address) Server {
   344  	return NewServerFromError(addr, nil, nil)
   345  }
   346  
   347  // NewServerFromError creates a new unknown server description with the given parameters.
   348  func NewServerFromError(addr address.Address, err error, tv *TopologyVersion) Server {
   349  	return Server{
   350  		Addr:            addr,
   351  		LastError:       err,
   352  		Kind:            Unknown,
   353  		TopologyVersion: tv,
   354  	}
   355  }
   356  
   357  // SetAverageRTT sets the average round trip time for this server description.
   358  func (s Server) SetAverageRTT(rtt time.Duration) Server {
   359  	s.AverageRTT = rtt
   360  	s.AverageRTTSet = true
   361  	return s
   362  }
   363  
   364  // DataBearing returns true if the server is a data bearing server.
   365  func (s Server) DataBearing() bool {
   366  	return s.Kind == RSPrimary ||
   367  		s.Kind == RSSecondary ||
   368  		s.Kind == Mongos ||
   369  		s.Kind == Standalone
   370  }
   371  
   372  // LoadBalanced returns true if the server is a load balancer or is behind a load balancer.
   373  func (s Server) LoadBalanced() bool {
   374  	return s.Kind == LoadBalancer || s.ServiceID != nil
   375  }
   376  
   377  // String implements the Stringer interface
   378  func (s Server) String() string {
   379  	str := fmt.Sprintf("Addr: %s, Type: %s",
   380  		s.Addr, s.Kind)
   381  	if len(s.Tags) != 0 {
   382  		str += fmt.Sprintf(", Tag sets: %s", s.Tags)
   383  	}
   384  
   385  	if s.AverageRTTSet {
   386  		str += fmt.Sprintf(", Average RTT: %d", s.AverageRTT)
   387  	}
   388  
   389  	if s.LastError != nil {
   390  		str += fmt.Sprintf(", Last error: %s", s.LastError)
   391  	}
   392  	return str
   393  }
   394  
   395  func decodeStringMap(element bson.RawElement, name string) (map[string]string, error) {
   396  	doc, ok := element.Value().DocumentOK()
   397  	if !ok {
   398  		return nil, fmt.Errorf("expected '%s' to be a document but it's a BSON %s", name, element.Value().Type)
   399  	}
   400  	elements, err := doc.Elements()
   401  	if err != nil {
   402  		return nil, err
   403  	}
   404  	m := make(map[string]string)
   405  	for _, element := range elements {
   406  		key := element.Key()
   407  		value, ok := element.Value().StringValueOK()
   408  		if !ok {
   409  			return nil, fmt.Errorf("expected '%s' to be a document of strings, but found a BSON %s", name, element.Value().Type)
   410  		}
   411  		m[key] = value
   412  	}
   413  	return m, nil
   414  }
   415  
   416  // Equal compares two server descriptions and returns true if they are equal
   417  func (s Server) Equal(other Server) bool {
   418  	if s.CanonicalAddr.String() != other.CanonicalAddr.String() {
   419  		return false
   420  	}
   421  
   422  	if !sliceStringEqual(s.Arbiters, other.Arbiters) {
   423  		return false
   424  	}
   425  
   426  	if !sliceStringEqual(s.Hosts, other.Hosts) {
   427  		return false
   428  	}
   429  
   430  	if !sliceStringEqual(s.Passives, other.Passives) {
   431  		return false
   432  	}
   433  
   434  	if s.Primary != other.Primary {
   435  		return false
   436  	}
   437  
   438  	if s.SetName != other.SetName {
   439  		return false
   440  	}
   441  
   442  	if s.Kind != other.Kind {
   443  		return false
   444  	}
   445  
   446  	if s.LastError != nil || other.LastError != nil {
   447  		if s.LastError == nil || other.LastError == nil {
   448  			return false
   449  		}
   450  		if s.LastError.Error() != other.LastError.Error() {
   451  			return false
   452  		}
   453  	}
   454  
   455  	if !s.WireVersion.Equals(other.WireVersion) {
   456  		return false
   457  	}
   458  
   459  	if len(s.Tags) != len(other.Tags) || !s.Tags.ContainsAll(other.Tags) {
   460  		return false
   461  	}
   462  
   463  	if s.SetVersion != other.SetVersion {
   464  		return false
   465  	}
   466  
   467  	if s.ElectionID != other.ElectionID {
   468  		return false
   469  	}
   470  
   471  	if ptrutil.CompareInt64(s.SessionTimeoutMinutesPtr, other.SessionTimeoutMinutesPtr) != 0 {
   472  		return false
   473  	}
   474  
   475  	// If TopologyVersion is nil for both servers, CompareToIncoming will return -1 because it assumes that the
   476  	// incoming response is newer. We want the descriptions to be considered equal in this case, though, so an
   477  	// explicit check is required.
   478  	if s.TopologyVersion == nil && other.TopologyVersion == nil {
   479  		return true
   480  	}
   481  	return s.TopologyVersion.CompareToIncoming(other.TopologyVersion) == 0
   482  }
   483  
   484  func sliceStringEqual(a []string, b []string) bool {
   485  	if len(a) != len(b) {
   486  		return false
   487  	}
   488  	for i, v := range a {
   489  		if v != b[i] {
   490  			return false
   491  		}
   492  	}
   493  	return true
   494  }
   495  
   496  // stringSliceFromRawElement decodes the provided BSON element into a []string.
   497  // This internally calls StringSliceFromRawValue on the element's value. The
   498  // error conditions outlined in that function's documentation apply for this
   499  // function as well.
   500  func stringSliceFromRawElement(element bson.RawElement) ([]string, error) {
   501  	return bsonutil.StringSliceFromRawValue(element.Key(), element.Value())
   502  }
   503
View as plain text