grpcconn

Documentation

Index

• Variables
• func WithLogger[T any](log *logrus.Entry) option[T]
• type Config
• type Conn
  • func (c *Conn[T]) Client() T
  • func (c *Conn[T]) ID() string
  • func (c *Conn[T]) Target() string
• type ConnFactory
• type ConnMgr
  • func NewConnMgr[T any](cfg *Config, httpClient *http.Client, connFactory ConnFactory[T], ...) *ConnMgr[T]
  • func (cm *ConnMgr[T]) AcquireConn(ctx context.Context) (_ *Conn[T], err error)
  • func (cm *ConnMgr[T]) Close()
  • func (cm *ConnMgr[T]) ReleaseConn(conn *Conn[T], err error) bool
  • func (cm *ConnMgr[T]) TransCountInTests() int
• type GenericResponse
• type GetGRPCEndpointsRs
• type ID
  • func (id ID) String() string
• type IDPool
  • func NewIDPool() *IDPool
  • func (i *IDPool) Allocate() ID
  • func (i *IDPool) Release(id ID)
• type ServerSpec

Variables

var (
	ErrConnMgrClosed = errors.New("connection manager closed")

	MetricGRPCConnections = prometheus.NewGaugeVec(prometheus.GaugeOpts{
		Name: "grpcconn_connections",
		Help: "The number of gRPC connections used by grpcconn.",
	}, []string{"remote_service", "peer"})
)

Functions

func WithLogger

func WithLogger[T any](log *logrus.Entry) option[T]

Types

type Config

type Config struct {
	RPCTimeout         clock.Duration
	BackOffTimeout     clock.Duration
	Zone               string
	OverrideHostHeader string

	// NumConnections is the number of client connections to establish
	// per target endpoint
	//
	// NOTE: A single GRPC client opens a maximum of 100 HTTP/2 Connections
	// to an endpoint. Once those connections are saturated, it will queue
	// requests to be delivered once there is availability. The recommended
	// method of overcoming this limitation is establishing multiple GPRC client
	// connections. See https://grpc.io/docs/guides/performance/
	NumConnections int
}

type Conn

type Conn[T any] struct {
	// contains filtered or unexported fields
}

func (*Conn[T]) Client

func (c *Conn[T]) Client() T

func (*Conn[T]) ID

func (c *Conn[T]) ID() string

func (*Conn[T]) Target

func (c *Conn[T]) Target() string

type ConnFactory

type ConnFactory[T any] interface {
	NewGRPCClient(cc grpc.ClientConnInterface) T
	GetServerListURL() string
	ServiceName() string
	ShouldDisposeOfConn(err error) bool
}

ConnFactory creates gRPC client objects.

type ConnMgr

type ConnMgr[T any] struct {
	// contains filtered or unexported fields
}

ConnMgr automates gRPC `Connection` pooling. This is necessary for use cases requiring frequent stream creation and high stream concurrency to avoid reaching the default 100 stream per connection limit. ConnMgr resolves gRPC instance endpoints and connects to them. Both resolution and connection is performed on the background allowing any number of concurrent AcquireConn to result in only one reconnect event.

func NewConnMgr

func NewConnMgr[T any](cfg *Config, httpClient *http.Client, connFactory ConnFactory[T], opts ...option[T]) *ConnMgr[T]

NewConnMgr instantiates a connection manager that maintains a gRPC connection pool.

func (*ConnMgr[T]) AcquireConn

func (cm *ConnMgr[T]) AcquireConn(ctx context.Context) (_ *Conn[T], err error)

func (*ConnMgr[T]) Close

func (cm *ConnMgr[T]) Close()

func (*ConnMgr[T]) ReleaseConn

func (cm *ConnMgr[T]) ReleaseConn(conn *Conn[T], err error) bool

func (*ConnMgr[T]) TransCountInTests

func (cm *ConnMgr[T]) TransCountInTests() int

TransCountInTests returns the total number of pending read/write operations. It is only supposed to be used in tests, hence it is not exposed in Client interface.

type GenericResponse

type GenericResponse struct {
	Msg string `json:"message"`
}

type GetGRPCEndpointsRs

type GetGRPCEndpointsRs struct {
	Servers []ServerSpec `json:"servers"`
	TTL     int          `json:"ttl"`
	// FIXME: Remove the following fields once all clients are upgraded.
	Endpoint string `json:"grpc_endpoint"`
	Zone     string `json:"zone"`
}

type ID

type ID int

func (ID) String

func (id ID) String() string

type IDPool

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

IDPool maintains a pool of ID values that can be released and reused. This is handy to keep cardinality low as gRPC connections are periodically released and reconnected and require an id. This solves the problem of infinitely incrementing ids used in Prometheus metric labels causing infinite growth of historical metric values.

func NewIDPool

func NewIDPool() *IDPool

func (*IDPool) Allocate

func (i *IDPool) Allocate() ID

func (*IDPool) Release

func (i *IDPool) Release(id ID)

type ServerSpec

type ServerSpec struct {
	Endpoint  string     `json:"endpoint"`
	Zone      string     `json:"zone"`
	Timestamp clock.Time `json:"timestamp"`
}