Skip to content

GODRIVER-1642 add unknown server description constructor #417

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Jun 10, 2020
Merged

GODRIVER-1642 add unknown server description constructor #417

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ff6142e
GODRIVER-1642 add unknown server description constructor
Jun 9, 2020
08a78a6
edits
Jun 9, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions x/mongo/driver/description/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -306,6 +306,20 @@ func NewServer(addr address.Address, response bsoncore.Document) Server {
return desc
}

// NewDefaultServer creates a new unknown server description with the given address.
func NewDefaultServer(addr address.Address) Server {
return NewServerFromError(addr, nil)
}

// NewServerFromError creates a new unknown server description with the given address and error.
func NewServerFromError(addr address.Address, err error) Server {
return Server{
Addr: addr,
LastError: err,
Kind: Unknown,
}
}

// SetAverageRTT sets the average round trip time for this server description.
func (s Server) SetAverageRTT(rtt time.Duration) Server {
s.AverageRTT = rtt
Expand Down
31 changes: 8 additions & 23 deletions x/mongo/driver/topology/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,7 +140,7 @@ func NewServer(addr address.Address, opts ...ServerOption) (*Server, error) {

subscribers: make(map[uint64]chan description.Server),
}
s.desc.Store(description.Server{Addr: addr})
s.desc.Store(description.NewDefaultServer(addr))

callback := func(desc description.Server) { s.updateDescription(desc) }
pc := poolConfig{
Expand All @@ -164,7 +164,7 @@ func (s *Server) Connect(updateCallback updateTopologyCallback) error {
if !atomic.CompareAndSwapInt32(&s.connectionstate, disconnected, connected) {
return ErrServerConnected
}
s.desc.Store(description.Server{Addr: s.address})
s.desc.Store(description.NewDefaultServer(s.address))
s.updateTopologyCallback.Store(updateCallback)
go s.update()
s.closewg.Add(1)
Expand Down Expand Up @@ -231,10 +231,7 @@ func (s *Server) Connection(ctx context.Context) (driver.Connection, error) {

// Since the only kind of ConnectionError we receive from pool.Get will be an initialization
// error, we should set the description.Server appropriately.
desc := description.Server{
Kind: description.Unknown,
LastError: wrappedConnErr,
}
desc := description.NewServerFromError(s.address, wrappedConnErr)
s.updateDescription(desc)
s.pool.clear()

Expand Down Expand Up @@ -299,14 +296,12 @@ func (s *Server) RequestImmediateCheck() {

// ProcessError handles SDAM error handling and implements driver.ErrorProcessor.
func (s *Server) ProcessError(err error) {
desc := s.Description()
// Invalidate server description if not master or node recovering error occurs.
// These errors can be reported as a command error or a write concern error.
if cerr, ok := err.(driver.Error); ok && (cerr.NodeIsRecovering() || cerr.NotMaster()) {
desc := s.Description()
desc.Kind = description.Unknown
desc.LastError = err
// updates description to unknown
s.updateDescription(desc)
s.updateDescription(description.NewServerFromError(s.address, err))
// If the node is shutting down or is older than 4.2, we synchronously clear the pool
if cerr.NodeIsShuttingDown() || desc.WireVersion == nil || desc.WireVersion.Max < 8 {
s.RequestImmediateCheck()
Expand All @@ -315,11 +310,8 @@ func (s *Server) ProcessError(err error) {
return
}
if wcerr, ok := err.(driver.WriteConcernError); ok && (wcerr.NodeIsRecovering() || wcerr.NotMaster()) {
desc := s.Description()
desc.Kind = description.Unknown
desc.LastError = err
// updates description to unknown
s.updateDescription(desc)
s.updateDescription(description.NewServerFromError(s.address, err))
// If the node is shutting down or is older than 4.2, we synchronously clear the pool
if wcerr.NodeIsShuttingDown() || desc.WireVersion == nil || desc.WireVersion.Max < 8 {
s.RequestImmediateCheck()
Expand All @@ -341,11 +333,8 @@ func (s *Server) ProcessError(err error) {
return
}

desc := s.Description()
desc.Kind = description.Unknown
desc.LastError = err
// updates description to unknown
s.updateDescription(desc)
s.updateDescription(description.NewServerFromError(s.address, err))
s.pool.clear()
}

Expand Down Expand Up @@ -547,11 +536,7 @@ func (s *Server) heartbeat(conn *connection) (description.Server, *connection) {
}

if !set {
desc = description.Server{
Addr: s.address,
LastError: saved,
Kind: description.Unknown,
}
desc = description.NewServerFromError(s.address, saved)
}

return desc, conn
Expand Down