topology

package
v2.0.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Dec 10, 2024 License: Apache-2.0 Imports: 37 Imported by: 0

Documentation

Overview

Package topology is intended for internal use only. It is made available to facilitate use cases that require access to internal MongoDB driver functionality and state. The API of this package is not stable and there is no backward compatibility guarantee.

WARNING: THIS PACKAGE IS EXPERIMENTAL AND MAY BE MODIFIED OR REMOVED WITHOUT NOTICE! USE WITH EXTREME CAUTION!

Package topology contains types that handles the discovery, monitoring, and selection of servers. This package is designed to expose enough inner workings of service discovery and monitoring to allow low level applications to have fine grained control, while hiding most of the detailed implementation of the algorithms.

Example (ClusterMonitoring) 
package main

import (
	"log"

	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/topology"
)

func main() {
	topo, err := topology.New(nil)
	if err != nil {
		log.Fatalf("could not create topology: %v", err)
	}
	err = topo.Connect()
	if err != nil {
		log.Fatalf("could not create topology: %v", err)
	}

	sub, err := topo.Subscribe()
	if err != nil {
		log.Fatalf("could not subscribe to topology: %v", err)
	}

	for desc := range sub.Updates {
		log.Printf("%#v", desc)
	}
}

Output:

Index

Examples

Constants

This section is empty.

Variables

View Source 
var (
	// MinSupportedMongoDBVersion is the version string for the lowest MongoDB version supported by the driver.
	MinSupportedMongoDBVersion = "3.6"

	// SupportedWireVersions is the range of wire versions supported by the driver.
	SupportedWireVersions = driverutil.NewVersionRange(driverutil.MinWireVersion, driverutil.MaxWireVersion)
)
View Source 
var (
	// BGReadTimeout is the maximum amount of the to wait when trying to read
	// the server reply on a connection after an operation timed out. The
	// default is 400ms.
	//
	// Deprecated: BGReadTimeout is intended for internal use only and may be
	// removed or modified at any time.
	BGReadTimeout = 400 * time.Millisecond

	// BGReadCallback is a callback for monitoring the behavior of the
	// background-read-on-timeout connection preserving mechanism.
	//
	// Deprecated: BGReadCallback is intended for internal use only and may be
	// removed or modified at any time.
	BGReadCallback func(addr string, start, read time.Time, errs []error, connClosed bool)
)
View Source 
var (
	// ErrServerClosed occurs when an attempt to Get a connection is made after
	// the server has been closed.
	ErrServerClosed = errors.New("server is closed")
	// ErrServerConnected occurs when at attempt to Connect is made after a server
	// has already been connected.
	ErrServerConnected = errors.New("server is connected")
)
View Source 
var ErrConnectionClosed = ConnectionError{ConnectionID: "<closed>", /* contains filtered or unexported fields */}

ErrConnectionClosed is returned from an attempt to use an already closed connection.

View Source 
var ErrPoolClosed = PoolError("attempted to check out a connection from closed connection pool")

ErrPoolClosed is returned when attempting to check out a connection from a closed pool.

View Source 
var ErrPoolNotPaused = PoolError("only a paused pool can be marked ready")

ErrPoolNotPaused is returned when attempting to mark a connection pool "ready" that is not currently "paused".

View Source 
var ErrSubscribeAfterClosed = errors.New("cannot subscribe after closeConnection")

ErrSubscribeAfterClosed is returned when a user attempts to subscribe to a closed Server or Topology.

View Source 
var ErrTopologyClosed = errors.New("topology is closed")

ErrTopologyClosed is returned when a user attempts to call a method on a closed Topology.

View Source 
var ErrTopologyConnected = errors.New("topology is connected or connecting")

ErrTopologyConnected is returned whena user attempts to Connect to an already connected Topology.

View Source 
var ErrWrongPool = PoolError("connection does not belong to this pool")

ErrWrongPool is return when a connection is returned to a pool it doesn't belong to.

Functions

func ConvertCreds 

func ConvertCreds(cred *options.Credential) *driver.Cred

ConvertCreds takes an options.Credential and returns the equivalent driver.Cred.

func ConvertToDriverAPIOptions 

func ConvertToDriverAPIOptions(opts *options.ServerAPIOptions) *driver.ServerAPIOptions

ConvertToDriverAPIOptions converts a given ServerAPIOptions object from the options package to a ServerAPIOptions object from the driver package.

func ServerAPIFromServerOptions 

func ServerAPIFromServerOptions(connectTimeout time.Duration, opts []ServerOption) *driver.ServerAPIOptions

ServerAPIFromServerOptions will return the server API options if they have been functionally set on the ServerOption slice.

Types

type Config 

type Config struct {
	Mode                   MonitorMode
	ReplicaSetName         string
	SeedList               []string
	ServerOpts             []ServerOption
	URI                    string
	ConnectTimeout         time.Duration
	Timeout                *time.Duration
	ServerSelectionTimeout time.Duration
	ServerMonitor          *event.ServerMonitor
	SRVMaxHosts            int
	SRVServiceName         string
	LoadBalanced           bool
	// contains filtered or unexported fields
}

Config is used to construct a topology.

func NewConfig 

func NewConfig(opts *options.ClientOptions, clock *session.ClusterClock) (*Config, error)

NewConfig will translate data from client options into a topology config for building non-default deployments. Server and topology options are not honored if a custom deployment is used.

func NewConfigFromOptionsWithAuthenticator 

func NewConfigFromOptionsWithAuthenticator(opts *options.ClientOptions, clock *session.ClusterClock, authenticator driver.Authenticator) (*Config, error)

NewConfigFromOptionsWithAuthenticator will translate data from client options into a topology config for building non-default deployments. Server and topology options are not honored if a custom deployment is used. It uses a passed in authenticator to authenticate the connection.

type Connection 

type Connection struct {
	// contains filtered or unexported fields
}

Connection implements the driver.Connection interface to allow reading and writing wire messages and the driver.Expirable interface to allow expiring. It wraps an underlying topology.connection to make it more goroutine-safe and nil-safe.

func (*Connection) Address 

func (c *Connection) Address() address.Address

Address returns the address of this connection.

func (*Connection) Alive 

func (c *Connection) Alive() bool

Alive returns if the connection is still alive.

func (*Connection) Close 

func (c *Connection) Close() error

Close returns this connection to the connection pool. This method may not closeConnection the underlying socket.

func (*Connection) CompressWireMessage 

func (c *Connection) CompressWireMessage(src, dst []byte) ([]byte, error)

CompressWireMessage handles compressing the provided wire message using the underlying connection's compressor. The dst parameter will be overwritten with the new wire message. If there is no compressor set on the underlying connection, then no compression will be performed.

func (*Connection) Description 

func (c *Connection) Description() description.Server

Description returns the server description of the server this connection is connected to.

func (*Connection) DriverConnectionID 

func (c *Connection) DriverConnectionID() int64

DriverConnectionID returns the driver connection ID.

func (*Connection) Expire 

func (c *Connection) Expire() error

Expire closes this connection and will closeConnection the underlying socket.

func (*Connection) ID 

func (c *Connection) ID() string

ID returns the ID of this connection.

func (*Connection) LocalAddress 

func (c *Connection) LocalAddress() address.Address

LocalAddress returns the local address of the connection

func (*Connection) OIDCTokenGenID 

func (c *Connection) OIDCTokenGenID() uint64

OIDCTokenGenID returns the OIDC token generation ID.

func (*Connection) PinToCursor 

func (c *Connection) PinToCursor() error

PinToCursor updates this connection to reflect that it is pinned to a cursor.

func (*Connection) PinToTransaction 

func (c *Connection) PinToTransaction() error

PinToTransaction updates this connection to reflect that it is pinned to a transaction.

func (*Connection) Read 

func (c *Connection) Read(ctx context.Context) ([]byte, error)

ReadWireMessage handles reading a wire message from the underlying connection. The dst parameter will be overwritten with the new wire message.

func (*Connection) ServerConnectionID 

func (c *Connection) ServerConnectionID() *int64

ServerConnectionID returns the server connection ID of this connection.

func (*Connection) SetOIDCTokenGenID 

func (c *Connection) SetOIDCTokenGenID(genID uint64)

SetOIDCTokenGenID sets the OIDC token generation ID.

func (*Connection) Stale 

func (c *Connection) Stale() bool

Stale returns if the connection is stale.

func (*Connection) UnpinFromCursor 

func (c *Connection) UnpinFromCursor() error

UnpinFromCursor updates this connection to reflect that it is no longer pinned to a cursor.

func (*Connection) UnpinFromTransaction 

func (c *Connection) UnpinFromTransaction() error

UnpinFromTransaction updates this connection to reflect that it is no longer pinned to a transaction.

func (*Connection) Write 

func (c *Connection) Write(ctx context.Context, wm []byte) error

WriteWireMessage handles writing a wire message to the underlying connection.

type ConnectionError 

type ConnectionError struct {
	ConnectionID string
	Wrapped      error
	// contains filtered or unexported fields
}

ConnectionError represents a connection error.

func (ConnectionError) Error 

func (e ConnectionError) Error() string

Error implements the error interface.

func (ConnectionError) Unwrap 

func (e ConnectionError) Unwrap() error

Unwrap returns the underlying error.

type ConnectionOption 

type ConnectionOption func(*connectionConfig)

ConnectionOption is used to configure a connection.

func WithCompressors 

func WithCompressors(fn func([]string) []string) ConnectionOption

WithCompressors sets the compressors that can be used for communication.

func WithConnectionLoadBalanced 

func WithConnectionLoadBalanced(fn func(bool) bool) ConnectionOption

WithConnectionLoadBalanced specifies whether or not the connection is to a server behind a load balancer.

func WithDialer 

func WithDialer(fn func(Dialer) Dialer) ConnectionOption

WithDialer configures the Dialer to use when making a new connection to MongoDB.

func WithDisableOCSPEndpointCheck 

func WithDisableOCSPEndpointCheck(fn func(bool) bool) ConnectionOption

WithDisableOCSPEndpointCheck specifies whether or the driver should perform non-stapled OCSP verification. If set to true, the driver will only check stapled responses and will continue the connection without reaching out to OCSP responders.

func WithHTTPClient 

func WithHTTPClient(fn func(*http.Client) *http.Client) ConnectionOption

WithHTTPClient configures the HTTP client for a connection.

func WithHandshaker 

func WithHandshaker(fn func(Handshaker) Handshaker) ConnectionOption

WithHandshaker configures the Handshaker that wll be used to initialize newly dialed connections.

func WithIdleTimeout 

func WithIdleTimeout(fn func(time.Duration) time.Duration) ConnectionOption

WithIdleTimeout configures the maximum idle time to allow for a connection.

func WithMonitor 

func WithMonitor(fn func(*event.CommandMonitor) *event.CommandMonitor) ConnectionOption

WithMonitor configures a event for command monitoring.

func WithOCSPCache 

func WithOCSPCache(fn func(ocsp.Cache) ocsp.Cache) ConnectionOption

WithOCSPCache specifies a cache to use for OCSP verification.

func WithTLSConfig 

func WithTLSConfig(fn func(*tls.Config) *tls.Config) ConnectionOption

WithTLSConfig configures the TLS options for a connection.

func WithZlibLevel 

func WithZlibLevel(fn func(*int) *int) ConnectionOption

WithZlibLevel sets the zLib compression level.

func WithZstdLevel 

func WithZstdLevel(fn func(*int) *int) ConnectionOption

WithZstdLevel sets the zstd compression level.

type Dialer 

type Dialer interface {
	DialContext(ctx context.Context, network, address string) (net.Conn, error)
}

Dialer is used to make network connections. 

var DefaultDialer Dialer = &net.Dialer{}

DefaultDialer is the Dialer implementation that is used by this package. Changing this will also change the Dialer used for this package. This should only be changed why all of the connections being made need to use a different Dialer. Most of the time, using a WithDialer option is more appropriate than changing this variable.

type DialerFunc 

type DialerFunc func(ctx context.Context, network, address string) (net.Conn, error)

DialerFunc is a type implemented by functions that can be used as a Dialer.

func (DialerFunc) DialContext 

func (df DialerFunc) DialContext(ctx context.Context, network, address string) (net.Conn, error)

DialContext implements the Dialer interface.

type Handshaker 

type Handshaker = driver.Handshaker

Handshaker is the interface implemented by types that can perform a MongoDB handshake over a provided driver.Connection. This is used during connection initialization. Implementations must be goroutine safe.

type MonitorMode 

type MonitorMode uint8

MonitorMode represents the way in which a server is monitored. 

const (
	AutomaticMode MonitorMode = iota
	SingleMode
)

These constants are the available monitoring modes.

type PoolError 

type PoolError string

PoolError is an error returned from a Pool method.

func (PoolError) Error 

func (pe PoolError) Error() string

type SelectedServer 

type SelectedServer struct {
	*Server

	Kind description.TopologyKind
}

SelectedServer represents a specific server that was selected during server selection. It contains the kind of the topology it was selected from.

func (*SelectedServer) Description 

func (ss *SelectedServer) Description() description.SelectedServer

Description returns a description of the server as of the last heartbeat.

type Server 

type Server struct {
	// contains filtered or unexported fields
}

Server is a single server within a topology.

func ConnectServer 

func ConnectServer(
	addr address.Address,
	updateCallback updateTopologyCallback,
	topologyID bson.ObjectID,
	connectTimeout time.Duration,
	opts ...ServerOption,
) (*Server, error)

ConnectServer creates a new Server and then initializes it using the Connect method.

func NewServer 

func NewServer(
	addr address.Address,
	topologyID bson.ObjectID,
	connectTimeout time.Duration,
	opts ...ServerOption,
) *Server

NewServer creates a new server. The mongodb server at the address will be monitored on an internal monitoring goroutine.

func (*Server) Connect 

func (s *Server) Connect(updateCallback updateTopologyCallback) error

Connect initializes the Server by starting background monitoring goroutines. This method must be called before a Server can be used.

func (*Server) Connection 

func (s *Server) Connection(ctx context.Context) (*mnet.Connection, error)

Connection gets a connection to the server.

func (*Server) Description 

func (s *Server) Description() description.Server

Description returns a description of the server as of the last heartbeat.

func (*Server) Disconnect 

func (s *Server) Disconnect(ctx context.Context) error

Disconnect closes sockets to the server referenced by this Server. Subscriptions to this Server will be closed. Disconnect will shutdown any monitoring goroutines, closeConnection the idle connection pool, and will wait until all the in use connections have been returned to the connection pool and are closed before returning. If the context expires via cancellation, deadline, or timeout before the in use connections have been returned, the in use connections will be closed, resulting in the failure of any in flight read or write operations. If this method returns with no errors, all connections associated with this Server have been closed.

func (*Server) OperationCount 

func (s *Server) OperationCount() int64

OperationCount returns the current number of in-progress operations for this server.

func (*Server) ProcessError 

func (s *Server) ProcessError(err error, describer mnet.Describer) driver.ProcessErrorResult

ProcessError handles SDAM error handling and implements driver.ErrorProcessor.

func (*Server) ProcessHandshakeError 

func (s *Server) ProcessHandshakeError(err error, startingGenerationNumber uint64, serviceID *bson.ObjectID)

ProcessHandshakeError implements SDAM error handling for errors that occur before a connection finishes handshaking.

func (*Server) RTTMonitor 

func (s *Server) RTTMonitor() driver.RTTMonitor

RTTMonitor returns this server's round-trip-time monitor.

func (*Server) RequestImmediateCheck 

func (s *Server) RequestImmediateCheck()

RequestImmediateCheck will cause the server to send a heartbeat immediately instead of waiting for the heartbeat timeout.

func (*Server) SelectedDescription 

func (s *Server) SelectedDescription() description.SelectedServer

SelectedDescription returns a description.SelectedServer with a Kind of Single. This can be used when performing tasks like monitoring a batch of servers and you want to run one off commands against those servers.

func (*Server) String 

func (s *Server) String() string

String implements the Stringer interface.

func (*Server) Subscribe 

func (s *Server) Subscribe() (*ServerSubscription, error)

Subscribe returns a ServerSubscription which has a channel on which all updated server descriptions will be sent. The channel will have a buffer size of one, and will be pre-populated with the current description.

type ServerOption 

type ServerOption func(*serverConfig)

ServerOption configures a server.

func WithClock 

func WithClock(fn func(clock *session.ClusterClock) *session.ClusterClock) ServerOption

WithClock configures the ClusterClock for the server to use.

func WithCompressionOptions 

func WithCompressionOptions(fn func(...string) []string) ServerOption

WithCompressionOptions configures the server's compressors.

func WithConnectionOptions 

func WithConnectionOptions(fn func(...ConnectionOption) []ConnectionOption) ServerOption

WithConnectionOptions configures the server's connections.

func WithConnectionPoolMaintainInterval 

func WithConnectionPoolMaintainInterval(fn func(time.Duration) time.Duration) ServerOption

WithConnectionPoolMaintainInterval configures the interval that the background connection pool maintenance goroutine runs.

func WithConnectionPoolMaxIdleTime 

func WithConnectionPoolMaxIdleTime(fn func(time.Duration) time.Duration) ServerOption

WithConnectionPoolMaxIdleTime configures the maximum time that a connection can remain idle in the connection pool before being removed. If connectionPoolMaxIdleTime is 0, then no idle time is set and connections will not be removed because of their age

func WithConnectionPoolMonitor 

func WithConnectionPoolMonitor(fn func(*event.PoolMonitor) *event.PoolMonitor) ServerOption

WithConnectionPoolMonitor configures the monitor for all connection pool actions

func WithHeartbeatInterval 

func WithHeartbeatInterval(fn func(time.Duration) time.Duration) ServerOption

WithHeartbeatInterval configures a server's heartbeat interval.

func WithMaxConnecting 

func WithMaxConnecting(fn func(uint64) uint64) ServerOption

WithMaxConnecting configures the maximum number of connections a connection pool may establish simultaneously. If maxConnecting is 0, the default value of 2 is used.

func WithMaxConnections 

func WithMaxConnections(fn func(uint64) uint64) ServerOption

WithMaxConnections configures the maximum number of connections to allow for a given server. If max is 0, then maximum connection pool size is not limited.

func WithMinConnections 

func WithMinConnections(fn func(uint64) uint64) ServerOption

WithMinConnections configures the minimum number of connections to allow for a given server. If min is 0, then there is no lower limit to the number of connections.

func WithOuterLibraryName 

func WithOuterLibraryName(fn func(string) string) ServerOption

WithOuterLibraryName configures the name for the outer library to include in the drivers section of the handshake metadata.

func WithOuterLibraryPlatform 

func WithOuterLibraryPlatform(fn func(string) string) ServerOption

WithOuterLibraryPlatform configures the platform for the outer library to include in the platform section of the handshake metadata.

func WithOuterLibraryVersion 

func WithOuterLibraryVersion(fn func(string) string) ServerOption

WithOuterLibraryVersion configures the version for the outer library to include in the drivers section of the handshake metadata.

func WithRegistry 

func WithRegistry(fn func(*bson.Registry) *bson.Registry) ServerOption

WithRegistry configures the registry for the server to use when creating cursors.

func WithServerAPI 

func WithServerAPI(fn func(serverAPI *driver.ServerAPIOptions) *driver.ServerAPIOptions) ServerOption

WithServerAPI configures the server API options for the server to use.

func WithServerAppName 

func WithServerAppName(fn func(string) string) ServerOption

WithServerAppName configures the server's application name.

func WithServerLoadBalanced 

func WithServerLoadBalanced(fn func(bool) bool) ServerOption

WithServerLoadBalanced specifies whether or not the server is behind a load balancer.

func WithServerMonitor 

func WithServerMonitor(fn func(*event.ServerMonitor) *event.ServerMonitor) ServerOption

WithServerMonitor configures the monitor for all SDAM events for a server

type ServerSelectionError 

type ServerSelectionError struct {
	Desc    description.Topology
	Wrapped error
}

ServerSelectionError represents a Server Selection error.

func (ServerSelectionError) Error 

func (e ServerSelectionError) Error() string

Error implements the error interface.

func (ServerSelectionError) Unwrap 

func (e ServerSelectionError) Unwrap() error

Unwrap returns the underlying error.

type ServerSubscription 

type ServerSubscription struct {
	C <-chan description.Server
	// contains filtered or unexported fields
}

ServerSubscription represents a subscription to the description.Server updates for a specific server.

func (*ServerSubscription) Unsubscribe 

func (ss *ServerSubscription) Unsubscribe() error

Unsubscribe unsubscribes this ServerSubscription from updates and closes the subscription channel.

type Topology 

type Topology struct {
	// contains filtered or unexported fields
}

Topology represents a MongoDB deployment.

func New 

func New(cfg *Config) (*Topology, error)

New creates a new topology. A "nil" config is interpreted as the default configuration.

func (*Topology) Connect 

func (t *Topology) Connect() error

Connect initializes a Topology and starts the monitoring process. This function must be called to properly monitor the topology.

func (*Topology) Description 

func (t *Topology) Description() description.Topology

Description returns a description of the topology.

func (*Topology) Disconnect 

func (t *Topology) Disconnect(ctx context.Context) error

Disconnect closes the topology. It stops the monitoring thread and closes all open subscriptions.

func (*Topology) FindServer 

func (t *Topology) FindServer(selected description.Server) (*SelectedServer, error)

FindServer will attempt to find a server that fits the given server description. This method will return nil, nil if a matching server could not be found.

func (*Topology) GetServerSelectionTimeout 

func (t *Topology) GetServerSelectionTimeout() time.Duration

GetServerSelectionTimeout returns the server selection timeout defined on the client options.

func (*Topology) Kind 

func (t *Topology) Kind() description.TopologyKind

Kind returns the topology kind of this Topology.

func (*Topology) RequestImmediateCheck 

func (t *Topology) RequestImmediateCheck()

RequestImmediateCheck will send heartbeats to all the servers in the topology right away, instead of waiting for the heartbeat timeout.

func (*Topology) SelectServer 

func (t *Topology) SelectServer(ctx context.Context, ss description.ServerSelector) (driver.Server, error)

SelectServer selects a server with given a selector, returning the remaining computedServerSelectionTimeout.

func (*Topology) String 

func (t *Topology) String() string

String implements the Stringer interface

func (*Topology) Subscribe 

func (t *Topology) Subscribe() (*driver.Subscription, error)

Subscribe returns a Subscription on which all updated description.Topologys will be sent. The channel of the subscription will have a buffer size of one, and will be pre-populated with the current description.Topology. Subscribe implements the driver.Subscriber interface.

func (*Topology) Unsubscribe 

func (t *Topology) Unsubscribe(sub *driver.Subscription) error

Unsubscribe unsubscribes the given subscription from the topology and closes the subscription channel. Unsubscribe implements the driver.Subscriber interface.

type WaitQueueTimeoutError 

type WaitQueueTimeoutError struct {
	Wrapped error
	// contains filtered or unexported fields
}

WaitQueueTimeoutError represents a timeout when requesting a connection from the pool

func (WaitQueueTimeoutError) Error 

func (w WaitQueueTimeoutError) Error() string

Error implements the error interface.

func (WaitQueueTimeoutError) Unwrap 

func (w WaitQueueTimeoutError) Unwrap() error

Unwrap returns the underlying error.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.