tunserver

Documentation

Index

• type DataCallback
• type FindHandle
• type GatewayFinder
  • func NewGatewayFinder(outgoingCtx context.Context, log *slog.Logger, ...) GatewayFinder
• type PollGatewayURLsCallback
• type PollingGatewayURLQuerier
• type ReadyGateway
• type Router
  • func (r *Router) RegisterTunclientAPI(desc *grpc.ServiceDesc)
• type RouterPlugin
• type RoutingClientConn
  • 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 *slog.Logger, rpcAPI modshared.RPCAPI, incomingStream grpc.ServerStream, ...) error

Types

type DataCallback

type DataCallback interface {
	Header(map[string]*prototool.HeaderValues) error
	Message([]byte) error
	Trailer(map[string]*prototool.HeaderValues) 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 *slog.Logger, gatewayPool grpctool.PoolInterface,
	gatewayQuerier PollingGatewayURLQuerier, api modshared.API, fullMethod string,
	ownPrivateAPIURL string, agentID int64, pollConfig retry.PollConfigFactory, tryNewGatewayInterval time.Duration) GatewayFinder

type PollGatewayURLsCallback

type PollGatewayURLsCallback func(gatewayURLs []string)

PollGatewayURLsCallback is called periodically with found gateway tunserver 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 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 modshared.RPCAPIFromContext(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 {
	// FindReadyGateway finds a ready gateway tunserver to route the request through.
	// It returns an error, compatible with gRPC status package.
	FindReadyGateway(ctx context.Context, log *slog.Logger, method string) (ReadyGateway, *slog.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, *slog.Logger, FindHandle, error)
	// PrepareStreamForForwarding is used to perform any actions, like wrapping, on the stream just before
	// forwarding it to tunclient.
	// Returned error is compatible with gRPC-status error.
	PrepareStreamForForwarding(stream grpc.ServerStream) (grpc.ServerStream, error)
}

type RoutingClientConn

type RoutingClientConn struct {
	Log    *slog.Logger
	API    modshared.API
	Plugin RouterPlugin
}

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 *slog.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       grpc.BidiStreamingServer[rpc.ConnectRequest, rpc.ConnectResponse]
	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 *slog.Logger, rpcAPI modshared.RPCAPI, incomingStream grpc.ServerStream, cb DataCallback) error