upstream

Documentation

Index

• type ConnUpstream
  • func NewConnUpstream(endpointID string, sess *yamux.Session) *ConnUpstream
  • func (u *ConnUpstream) Dial() (net.Conn, error)
  • func (u *ConnUpstream) EndpointID() string
  • func (u *ConnUpstream) Forward() bool
• type LoadBalancedManager
  • func NewLoadBalancedManager(cluster *cluster.State) *LoadBalancedManager
  • func (m *LoadBalancedManager) AddConn(u Upstream)
  • func (m *LoadBalancedManager) Endpoints() map[string]int
  • func (m *LoadBalancedManager) Metrics() *Metrics
  • func (m *LoadBalancedManager) RemoveConn(u Upstream)
  • func (m *LoadBalancedManager) Select(endpointID string, allowRemote bool) (Upstream, bool)
  • func (m *LoadBalancedManager) Usage() *Usage
• type Manager
• type Metrics
  • func NewMetrics() *Metrics
  • func (m *Metrics) Register(registry *prometheus.Registry)
• type NodeUpstream
  • func NewNodeUpstream(endpointID string, node *cluster.Node) *NodeUpstream
  • func (u *NodeUpstream) Dial() (net.Conn, error)
  • func (u *NodeUpstream) EndpointID() string
  • func (u *NodeUpstream) Forward() bool
• type Server
  • func NewServer(upstreams Manager, verifier auth.Verifier, tlsConfig *tls.Config, ...) *Server
  • func (s *Server) Serve(ln net.Listener) error
  • func (s *Server) Shutdown(ctx context.Context) error
• type Status
  • func NewStatus(manager *LoadBalancedManager) *Status
  • func (s *Status) Register(group *gin.RouterGroup)
• type Upstream
• type Usage

Types

type ConnUpstream

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

ConnUpstream represents a connection to an upstream service thats connected to the local node.

func NewConnUpstream

func NewConnUpstream(endpointID string, sess *yamux.Session) *ConnUpstream

func (*ConnUpstream) Dial

func (u *ConnUpstream) Dial() (net.Conn, error)

func (*ConnUpstream) EndpointID

func (u *ConnUpstream) EndpointID() string

func (*ConnUpstream) Forward

func (u *ConnUpstream) Forward() bool

type LoadBalancedManager

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

func NewLoadBalancedManager

func NewLoadBalancedManager(cluster *cluster.State) *LoadBalancedManager

func (*LoadBalancedManager) AddConn

func (m *LoadBalancedManager) AddConn(u Upstream)

func (*LoadBalancedManager) Endpoints

func (m *LoadBalancedManager) Endpoints() map[string]int

func (*LoadBalancedManager) Metrics

func (m *LoadBalancedManager) Metrics() *Metrics

func (*LoadBalancedManager) RemoveConn

func (m *LoadBalancedManager) RemoveConn(u Upstream)

func (*LoadBalancedManager) Select

func (m *LoadBalancedManager) Select(endpointID string, allowRemote bool) (Upstream, bool)

func (*LoadBalancedManager) Usage

func (m *LoadBalancedManager) Usage() *Usage

type Manager

type Manager interface {
	// Select looks up an upstream for the given endpoint ID.
	//
	// This will first look for an upstream connected to the local node, and
	// load balance among the available connected upstreams.
	//
	// If there are no upstreams connected for the endpoint, and 'allowForward'
	// is true, it will look for another node in the cluster that has an
	// upstream connection for the endpoint and use that node as the upstream.
	Select(endpointID string, allowForward bool) (Upstream, bool)

	// AddConn adds a local upstream connection.
	AddConn(u Upstream)

	// RemoveConn removes a local upstream connection.
	RemoveConn(u Upstream)
}

Manager manages the upstream routes for each endpoint.

This includes upstreams connected to the local node, or other server nodes in the cluster with a connected upstream for the target endpoint.

type Metrics

type Metrics struct {
	// ConnectedUpstreams is the number of upstreams connected to this node.
	ConnectedUpstreams prometheus.Gauge

	// RegisteredEndpoints is the number of endpoints registered to this node.
	RegisteredEndpoints prometheus.Gauge

	// UpstreamRequestsTotal is the number of requests sent to an
	// upstream connected to the local node.
	UpstreamRequestsTotal prometheus.Counter

	// RemoteRequestsTotal is the number of requests sent to another node.
	// Labelled by target node ID.
	RemoteRequestsTotal *prometheus.CounterVec
}

func NewMetrics

func NewMetrics() *Metrics

func (*Metrics) Register

func (m *Metrics) Register(registry *prometheus.Registry)

type NodeUpstream

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

NodeUpstream represents a remote Piko server node.

func NewNodeUpstream

func NewNodeUpstream(endpointID string, node *cluster.Node) *NodeUpstream

func (*NodeUpstream) Dial

func (u *NodeUpstream) Dial() (net.Conn, error)

func (*NodeUpstream) EndpointID

func (u *NodeUpstream) EndpointID() string

func (*NodeUpstream) Forward

func (u *NodeUpstream) Forward() bool

type Server

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

Server accepts connections from upstream services.

func NewServer

func NewServer(
	upstreams Manager,
	verifier auth.Verifier,
	tlsConfig *tls.Config,
	logger log.Logger,
) *Server

func (*Server) Serve

func (s *Server) Serve(ln net.Listener) error

func (*Server) Shutdown

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

Shutdown attempts to gracefully shutdown the server by waiting for pending requests to complete.

type Status

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

func NewStatus

func NewStatus(manager *LoadBalancedManager) *Status

func (*Status) Register

func (s *Status) Register(group *gin.RouterGroup)

type Upstream

type Upstream interface {
	EndpointID() string
	Dial() (net.Conn, error)
	// Forward indicates whether the upstream is forwarding traffic to a remote
	// node rather than a client listener.
	Forward() bool
}

Upstream represents an upstream for a given endpoint.

An upstream may be an upstream service connected to the local node, or another Piko server node.

type Usage

type Usage struct {
	Requests  *atomic.Uint64
	Upstreams *atomic.Uint64
}