tunserver

Documentation

Index

• Constants
• type AgentFinder
• type DataCallback
• type FindHandle
• type GatewayFinder
  • func NewGatewayFinder(outgoingCtx context.Context, log *zap.Logger, ...) GatewayFinder
• type PollGatewayURLsCallback
• type PollingGatewayURLQuerier
• type ReadyGateway
• type RecentlyConnectedAgentNotFound
  • func (e *RecentlyConnectedAgentNotFound) Error() string
• type Router
  • func (r *Router) RegisterTunclientAPI(desc *grpc.ServiceDesc)
• type RouterPlugin
• type RoutingClientConn
  • func NewRoutingClientConn(log *zap.Logger, api modshared.API, plugin RouterPlugin, ...) (*RoutingClientConn, error)
  • func (c *RoutingClientConn) Invoke(ctx context.Context, method string, args any, reply any, ...) error
  • func (c *RoutingClientConn) NewStream(ctx context.Context, desc *grpc.StreamDesc, method string, ...) (grpc.ClientStream, error)
• type StateType
• type Tunnel
• type TunnelImpl
  • func (t *TunnelImpl) Done(ctx context.Context)
  • func (t *TunnelImpl) ForwardStream(log *zap.Logger, rpcAPI modshared.RPCAPI, incomingStream grpc.ServerStream, ...) error

Constants

const (
	// RoutingHopPrefix is a metadata key prefix that is used for metadata keys that should be consumed by
	// the gateway kas instances and not passed along to agentk.
	RoutingHopPrefix = "kas-hop-"
)

Types

type AgentFinder

type AgentFinder interface {
	AgentLastConnected(ctx context.Context, agentID int64) (time.Time, error)
}

type DataCallback

type DataCallback interface {
	Header(map[string]*prototool.Values) error
	Message([]byte) error
	Trailer(map[string]*prototool.Values) error
	Error(*statuspb.Status) error
}

type FindHandle

type FindHandle interface {
	// Get finds a tunnel to an agentk.
	// It waits for a matching tunnel to proxy a connection through. When a matching tunnel is found, it is returned.
	// It returns gRPC status errors only, ready to return from RPC handler.
	Get(ctx context.Context) (Tunnel, error)
	// Done must be called to free resources of this FindHandle instance.
	// ctx is used for tracing only.
	Done(ctx context.Context)
}

type GatewayFinder

type GatewayFinder interface {
	Find(ctx context.Context) (ReadyGateway, error)
}

func NewGatewayFinder

func NewGatewayFinder(outgoingCtx context.Context, log *zap.Logger, gatewayPool grpctool.PoolInterface,
	gatewayQuerier PollingGatewayURLQuerier, agentFinder AgentFinder, api modshared.API, fullMethod string,
	ownPrivateAPIURL string, agentID int64, pollConfig retry.PollConfigFactory, tryNewGatewayInterval time.Duration) GatewayFinder

type PollGatewayURLsCallback

type PollGatewayURLsCallback func(kasURLs []string)

PollGatewayURLsCallback is called periodically with found kas URLs for a particular agent id.

type PollingGatewayURLQuerier

type PollingGatewayURLQuerier interface {
	PollGatewayURLs(ctx context.Context, agentID int64, cb PollGatewayURLsCallback)
	CachedGatewayURLs(agentID int64) []string
}

type ReadyGateway

type ReadyGateway struct {
	URL          string
	Stream       grpc.ClientStream
	StreamCancel context.CancelFunc
	Codec        encoding.Codec
}

type RecentlyConnectedAgentNotFound

type RecentlyConnectedAgentNotFound struct {
	LastConnectedAt time.Time
}

func (*RecentlyConnectedAgentNotFound) Error

func (e *RecentlyConnectedAgentNotFound) Error() string

type Router

type Router struct {
	Plugin RouterPlugin
	// PrivateAPIServer is the gRPC server that other kas instances can talk to.
	// Request handlers can obtain the per-request logger using grpctool.LoggerFromContext(requestContext).
	PrivateAPIServer grpc.ServiceRegistrar
}

Router routes traffic from kas to another kas to agentk. routing kas -> gateway kas -> agentk

func (*Router) RegisterTunclientAPI

func (r *Router) RegisterTunclientAPI(desc *grpc.ServiceDesc)

type RouterPlugin

type RouterPlugin interface {
	// GatewayFinder finds a gateway tunserver to route the request through.
	// It returns an error, compatible with gRPC status package.
	GatewayFinder(ctx context.Context, log *zap.Logger, method string) (GatewayFinder, *zap.Logger, int64, error)
	// FindTunnel starts searching for a tunnel to a matching tunclient.
	// Tunnel found boolean indicates whether a suitable tunnel is immediately available from the
	// returned FindHandle object.
	FindTunnel(grpc.ServerStream, modshared.RPCAPI) (bool, *zap.Logger, FindHandle, error)
}

type RoutingClientConn

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

func NewRoutingClientConn

func NewRoutingClientConn(log *zap.Logger, api modshared.API, plugin RouterPlugin, tp trace.TracerProvider,
	dm otelmetric.Meter, tunnelFindTimeout time.Duration) (*RoutingClientConn, error)

func (*RoutingClientConn) Invoke

func (c *RoutingClientConn) Invoke(ctx context.Context, method string, args any, reply any, opts ...grpc.CallOption) error

func (*RoutingClientConn) NewStream

func (c *RoutingClientConn) NewStream(ctx context.Context, desc *grpc.StreamDesc, method string, opts ...grpc.CallOption) (grpc.ClientStream, error)

type StateType

type StateType int

const (

	// StateReady - tunnel is owned by the registry and is ready to be found and used for forwarding.
	StateReady StateType
	// StateFound - tunnel is not owned by registry, was found and about to be used for forwarding.
	StateFound
	// StateForwarding - tunnel is not owned by registry, is being used for forwarding.
	StateForwarding
	// StateDone - tunnel is not owned by anyone, it has been used for forwarding, Done() has been called.
	StateDone
	// StateContextDone - tunnel is not owned by anyone, reverse tunnel's context signaled done in HandleTunnel().
	StateContextDone
)

type Tunnel

type Tunnel interface {
	// ForwardStream performs bi-directional message forwarding between incomingStream and the tunnel.
	// cb is called with header, messages and trailer coming from the tunnel. It's the callers
	// responsibility to forward them into the incomingStream.
	ForwardStream(log *zap.Logger, rpcAPI modshared.RPCAPI, incomingStream grpc.ServerStream, cb DataCallback) error
	// Done must be called when the caller is done with the Tunnel.
	// ctx is used for tracing only.
	Done(ctx context.Context)
}

type TunnelImpl

type TunnelImpl struct {
	Tunnel       rpc.ReverseTunnel_ConnectServer
	TunnelRetErr chan<- error
	AgentID      int64
	Descriptor   *info.APIDescriptor
	State        StateType

	OnForward func(*TunnelImpl) error
	OnDone    func(context.Context, *TunnelImpl)
}

func (*TunnelImpl) Done

func (t *TunnelImpl) Done(ctx context.Context)

func (*TunnelImpl) ForwardStream

func (t *TunnelImpl) ForwardStream(log *zap.Logger, rpcAPI modshared.RPCAPI, incomingStream grpc.ServerStream, cb DataCallback) error