pool

Documentation

Index

• Constants
• Variables
• func PeekFirstByte(conn net.Conn) (net.Conn, byte, error)
• func PeekForTLS(conn net.Conn) (net.Conn, bool, error)
• type Conn
  • func (c *Conn) Close() error
• type ConnPool
  • func (p *ConnPool) DialTimeout(dc string, nodeName string, addr net.Addr, actualRPCType RPCType) (net.Conn, HalfCloser, error)
  • func (p *ConnPool) Ping(dc string, nodeName string, addr net.Addr) (bool, error)
  • func (p *ConnPool) RPC(dc string, nodeName string, addr net.Addr, method string, args interface{}, ...) error
  • func (p *ConnPool) Shutdown() error
• type HalfCloser
  • func DialRPCViaMeshGateway(ctx context.Context, dc string, nodeName string, srcAddr *net.TCPAddr, ...) (net.Conn, HalfCloser, error)
• type RPCType
  • func (t RPCType) ALPNString() string
• type StreamClient
  • func (sc *StreamClient) Close()

Constants

const (
	// regular old rpc (note there is no equivalent of RPCMultiplex, RPCTLS, or RPCTLSInsecure)
	ALPN_RPCConsul      = "consul/rpc-single"   // RPCConsul
	ALPN_RPCRaft        = "consul/raft"         // RPCRaft
	ALPN_RPCMultiplexV2 = "consul/rpc-multi"    // RPCMultiplexV2
	ALPN_RPCSnapshot    = "consul/rpc-snapshot" // RPCSnapshot
	ALPN_RPCGossip      = "consul/rpc-gossip"   // RPCGossip
	ALPN_RPCGRPC        = "consul/rpc-grpc"     // RPCGRPC
	// wan federation additions
	ALPN_WANGossipPacket = "consul/wan-gossip/packet"
	ALPN_WANGossipStream = "consul/wan-gossip/stream"
)

const DefaultDialTimeout = 10 * time.Second

Variables

var RPCNextProtos = []string{
	ALPN_RPCConsul,
	ALPN_RPCRaft,
	ALPN_RPCMultiplexV2,
	ALPN_RPCSnapshot,
	ALPN_RPCGossip,
	ALPN_RPCGRPC,
	ALPN_WANGossipPacket,
	ALPN_WANGossipStream,
}

Functions

func PeekFirstByte

func PeekFirstByte(conn net.Conn) (net.Conn, byte, error)

PeekFirstByte will read the first byte on the conn.

This function does not close the conn on an error.

The returned conn has the initial read buffered internally for the purposes of not consuming the first byte. After that buffer is drained the conn is a pass through to the original conn.

func PeekForTLS

func PeekForTLS(conn net.Conn) (net.Conn, bool, error)

PeekForTLS will read the first byte on the conn to determine if the client request is a TLS connection request or a consul-specific framed rpc request.

This function does not close the conn on an error.

The returned conn has the initial read buffered internally for the purposes of not consuming the first byte. After that buffer is drained the conn is a pass through to the original conn.

The TLS record layer governs the very first byte. The available options start at 20 as per:

• v1.2: https://tools.ietf.org/html/rfc5246#appendix-A.1
• v1.3: https://tools.ietf.org/html/rfc8446#appendix-B.1

Note: this indicates that '0' is 'invalid'. Given that we only care about the first byte of a long-lived connection this is irrelevant, since you must always start out with a client hello handshake which is '22'.

Types

type Conn

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

Conn is a pooled connection to a Consul server

func (*Conn) Close

func (c *Conn) Close() error

type ConnPool

type ConnPool struct {
	// SrcAddr is the source address for outgoing connections.
	SrcAddr *net.TCPAddr

	// Logger passed to yamux
	// TODO: consider refactoring to accept a full yamux.Config instead of a logger
	Logger *log.Logger

	// The maximum time to keep a connection open
	MaxTime time.Duration

	// The maximum number of open streams to keep
	MaxStreams int

	// TLSConfigurator
	TLSConfigurator *tlsutil.Configurator

	// GatewayResolver is a function that returns a suitable random mesh
	// gateway address for dialing servers in a given DC. This is only
	// needed if wan federation via mesh gateways is enabled.
	GatewayResolver func(string) string

	// Datacenter is the datacenter of the current agent.
	Datacenter string

	// Server should be set to true if this connection pool is configured in a
	// server instead of a client.
	Server bool

	sync.Mutex
	// contains filtered or unexported fields
}

ConnPool is used to maintain a connection pool to other Consul servers. This is used to reduce the latency of RPC requests between servers. It is only used to pool connections in the rpcConsul mode. Raft connections are pooled separately. Maintain at most one connection per host, for up to MaxTime. When MaxTime connection reaping is disabled. MaxStreams is used to control the number of idle streams allowed. If TLS settings are provided outgoing connections use TLS.

func (*ConnPool) DialTimeout

func (p *ConnPool) DialTimeout(
	dc string,
	nodeName string,
	addr net.Addr,
	actualRPCType RPCType,
) (net.Conn, HalfCloser, error)

DialTimeout is used to establish a raw connection to the given server, with given connection timeout. It also writes RPCTLS as the first byte.

func (*ConnPool) Ping

func (p *ConnPool) Ping(dc string, nodeName string, addr net.Addr) (bool, error)

Ping sends a Status.Ping message to the specified server and returns true if healthy, false if an error occurred

func (*ConnPool) RPC

func (p *ConnPool) RPC(
	dc string,
	nodeName string,
	addr net.Addr,
	method string,
	args interface{},
	reply interface{},
) error

RPC is used to make an RPC call to a remote host

func (*ConnPool) Shutdown

func (p *ConnPool) Shutdown() error

Shutdown is used to close the connection pool

type HalfCloser

type HalfCloser interface {
	CloseWrite() error
}

HalfCloser is an interface that exposes a TCP half-close without exposing the underlying TLS or raw TCP connection.

func DialRPCViaMeshGateway

func DialRPCViaMeshGateway(
	ctx context.Context,
	dc string,
	nodeName string,
	srcAddr *net.TCPAddr,
	alpnWrapper tlsutil.ALPNWrapper,
	nextProto string,
	dialingFromServer bool,
	gatewayResolver func(string) string,
) (net.Conn, HalfCloser, error)

DialRPCViaMeshGateway dials the destination node and sets up the connection to be the correct RPC type using ALPN. This currently is exclusively used to dial other servers in foreign datacenters via mesh gateways.

type RPCType

type RPCType byte

const (
	// keep numbers unique.
	RPCConsul      RPCType = 0
	RPCRaft        RPCType = 1
	RPCMultiplex   RPCType = 2 // Old Muxado byte, no longer supported.
	RPCTLS         RPCType = 3
	RPCMultiplexV2 RPCType = 4
	RPCSnapshot    RPCType = 5
	RPCGossip      RPCType = 6
	// RPCTLSInsecure is used to flag RPC calls that require verify
	// incoming to be disabled, even when it is turned on in the
	// configuration. At the time of writing there is only AutoEncrypt.Sign
	// that is supported and it might be the only one there
	// ever is.
	RPCTLSInsecure RPCType = 7
	RPCGRPC        RPCType = 8

	// RPCMaxTypeValue is the maximum rpc type byte value currently used for the
	// various protocols riding over our "rpc" port.
	//
	// Currently our 0-8 values are mutually exclusive with any valid first byte
	// of a TLS header.  The first TLS header byte will begin with a TLS content
	// type and the values 0-19 are all explicitly unassigned and marked as
	// requiring coordination. RFC 7983 does the marking and goes into some
	// details about multiplexing connections and identifying TLS.
	//
	// We use this value to determine if the incoming request is actual real
	// native TLS (where we can de-multiplex based on ALPN protocol) or our older
	// type-byte system when new connections are established.
	//
	// NOTE: if you add new RPCTypes beyond this value, you must similarly bump
	// this value.
	RPCMaxTypeValue = 8
)

func (RPCType) ALPNString

func (t RPCType) ALPNString() string

type StreamClient

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

streamClient is used to wrap a stream with an RPC client

func (*StreamClient) Close

func (sc *StreamClient) Close()