state

Documentation

Index

• type InterceptFinalizer
• type SessionState
• type State
  • func NewState(ctx context.Context) *State
  • func (s *State) AddAgent(agent *rpc.AgentInfo, now time.Time) string
  • func (s *State) AddClient(client *rpc.ClientInfo, now time.Time) string
  • func (s *State) AddIntercept(sessionID, clusterID, apiKey string, client *rpc.ClientInfo, ...) (*rpc.InterceptInfo, error)
  • func (s *State) AddInterceptFinalizer(interceptID string, finalizer InterceptFinalizer) error
  • func (s *State) AgentsLookup(ctx context.Context, clientSessionID string, request *rpc.LookupHostRequest) (iputil.IPs, int, error)
  • func (s *State) CountAllClients() int
  • func (s *State) ExpireSessions(ctx context.Context, clientMoment, agentMoment time.Time)
  • func (s *State) GetAgent(sessionID string) *rpc.AgentInfo
  • func (s *State) GetAgentsByName(name, namespace string) map[string]*rpc.AgentInfo
  • func (s *State) GetAllAgents() map[string]*rpc.AgentInfo
  • func (s *State) GetAllClients() map[string]*rpc.ClientInfo
  • func (s *State) GetClient(sessionID string) *rpc.ClientInfo
  • func (s *State) GetIntercept(interceptID string) (*rpc.InterceptInfo, bool)
  • func (s *State) InitialTempLogLevel() *rpc.LogLevelRequest
  • func (s *State) MarkSession(req *rpc.RemainRequest, now time.Time) (ok bool)
  • func (s *State) PostLookupResponse(response *rpc.LookupHostAgentResponse)
  • func (s *State) PrepareIntercept(ctx context.Context, cr *managerrpc.CreateInterceptRequest) (*managerrpc.PreparedIntercept, error)
  • func (s *State) RemoveIntercept(interceptID string) bool
  • func (s *State) RemoveSession(ctx context.Context, sessionID string)
  • func (s *State) SessionDone(id string) (<-chan struct{}, error)
  • func (s *State) SetTempLogLevel(ctx context.Context, logLevelRequest *rpc.LogLevelRequest)
  • func (s *State) Tunnel(ctx context.Context, stream tunnel.Stream) error
  • func (s *State) UpdateIntercept(interceptID string, apply func(*rpc.InterceptInfo)) *rpc.InterceptInfo
  • func (s *State) WaitForTempLogLevel(stream rpc.Manager_WatchLogLevelServer) error
  • func (s *State) WatchAgents(ctx context.Context, filter func(sessionID string, agent *rpc.AgentInfo) bool) <-chan watchable.Snapshot[*rpc.AgentInfo]
  • func (s *State) WatchClients(ctx context.Context, ...) <-chan watchable.Snapshot[*rpc.ClientInfo]
  • func (s *State) WatchDial(sessionID string) <-chan *rpc.DialRequest
  • func (s *State) WatchIntercepts(ctx context.Context, ...) <-chan watchable.Snapshot[*rpc.InterceptInfo]
  • func (s *State) WatchLookupHost(agentSessionID string) <-chan *rpc.LookupHostRequest

Types

type InterceptFinalizer

type InterceptFinalizer func(ctx context.Context, interceptInfo *managerrpc.InterceptInfo) error

type SessionState

type SessionState interface {
	Cancel()
	Done() <-chan struct{}
	LastMarked() time.Time
	SetLastMarked(lastMarked time.Time)
	Dials() <-chan *rpc.DialRequest
	EstablishBidiPipe(context.Context, tunnel.Stream) (tunnel.Endpoint, error)
	OnConnect(context.Context, tunnel.Stream) (tunnel.Endpoint, error)
}

type State

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

State is the total state of the Traffic Manager. A zero State is invalid; you must call NewState.

func NewState

func NewState(ctx context.Context) *State

func (*State) AddAgent

func (s *State) AddAgent(agent *rpc.AgentInfo, now time.Time) string

func (*State) AddClient

func (s *State) AddClient(client *rpc.ClientInfo, now time.Time) string

func (*State) AddIntercept

func (s *State) AddIntercept(sessionID, clusterID, apiKey string, client *rpc.ClientInfo, spec *rpc.InterceptSpec) (*rpc.InterceptInfo, error)

func (*State) AddInterceptFinalizer

func (s *State) AddInterceptFinalizer(interceptID string, finalizer InterceptFinalizer) error

func (*State) AgentsLookup

func (s *State) AgentsLookup(ctx context.Context, clientSessionID string, request *rpc.LookupHostRequest) (iputil.IPs, int, error)

AgentsLookup will send the given request to all agents currently intercepted by the client identified with the clientSessionID, it will then wait for results to arrive, collect those results, and return them as a unique and sorted slice together with a count of how many agents that replied.

func (*State) CountAllClients

func (s *State) CountAllClients() int

func (*State) ExpireSessions

func (s *State) ExpireSessions(ctx context.Context, clientMoment, agentMoment time.Time)

ExpireSessions prunes any sessions that haven't had a MarkSession heartbeat since respective given 'moment'.

func (*State) GetAgent

func (s *State) GetAgent(sessionID string) *rpc.AgentInfo

func (*State) GetAgentsByName

func (s *State) GetAgentsByName(name, namespace string) map[string]*rpc.AgentInfo

func (*State) GetAllAgents

func (s *State) GetAllAgents() map[string]*rpc.AgentInfo

func (*State) GetAllClients

func (s *State) GetAllClients() map[string]*rpc.ClientInfo

func (*State) GetClient

func (s *State) GetClient(sessionID string) *rpc.ClientInfo

func (*State) GetIntercept

func (s *State) GetIntercept(interceptID string) (*rpc.InterceptInfo, bool)

func (*State) InitialTempLogLevel

func (s *State) InitialTempLogLevel() *rpc.LogLevelRequest

InitialTempLogLevel returns the temporary log-level if it exists, along with the remaining duration for it, which might be zero, in which case the log-level is valid until a new level is requested.

func (*State) MarkSession

func (s *State) MarkSession(req *rpc.RemainRequest, now time.Time) (ok bool)

MarkSession marks a session as being present at the indicated time. Returns true if everything goes OK, returns false if the given session ID does not exist.

func (*State) PostLookupResponse

func (s *State) PostLookupResponse(response *rpc.LookupHostAgentResponse)

PostLookupResponse receives lookup responses from an agent and places them in the channel that corresponds to the lookup request

func (*State) PrepareIntercept

func (s *State) PrepareIntercept(ctx context.Context, cr *managerrpc.CreateInterceptRequest) (*managerrpc.PreparedIntercept, error)

PrepareIntercept ensures that the given request can be matched against the intercept configuration of the workload that it references. It returns a PreparedIntercept where all intercepted ports have been qualified with a service name and a service port name.

The first step is to find the requested Workload and the agent config for that workload. This step will create the initial ConfigMap for the namespace if it doesn't exist yet, and also generate the actual intercept config if it doesn't exist.

The second step matches all ServicePortIdentifiers in the request to the intercepts of the agent config and creates a resulting PreparedIntercept with a services array that has the same size and positions as the ServicePortIdentifiers in the request.

It's expected that the client that makes the call will update any unqualified service port identifiers with the ones in the returned PreparedIntercept.

func (*State) RemoveIntercept

func (s *State) RemoveIntercept(interceptID string) bool

func (*State) RemoveSession

func (s *State) RemoveSession(ctx context.Context, sessionID string)

RemoveSession removes a session from the set of present session IDs.

func (*State) SessionDone

func (s *State) SessionDone(id string) (<-chan struct{}, error)

SessionDone returns a channel that is closed when the session with the given ID terminates. If there is no such currently-live session, then an already-closed channel is returned.

func (*State) SetTempLogLevel

func (s *State) SetTempLogLevel(ctx context.Context, logLevelRequest *rpc.LogLevelRequest)

SetTempLogLevel sets the temporary log-level for the traffic-manager and all agents and, if a duration is given, it also starts a timer that will reset the log-level once it fires.

func (*State) Tunnel

func (s *State) Tunnel(ctx context.Context, stream tunnel.Stream) error

func (*State) UpdateIntercept

func (s *State) UpdateIntercept(interceptID string, apply func(*rpc.InterceptInfo)) *rpc.InterceptInfo

UpdateIntercept applies a given mutator function to the stored intercept with interceptID; storing and returning the result. If the given intercept does not exist, then the mutator function is not run, and nil is returned.

This does not lock; but instead uses CAS and may therefore call the mutator function multiple times. So: it is safe to perform blocking operations in your mutator function, but you must take care that it is safe to call your mutator function multiple times.

func (*State) WaitForTempLogLevel

func (s *State) WaitForTempLogLevel(stream rpc.Manager_WatchLogLevelServer) error

WaitForTempLogLevel waits for a new temporary log-level request. It returns the values of the last request that was made.

func (*State) WatchAgents

func (s *State) WatchAgents(
	ctx context.Context,
	filter func(sessionID string, agent *rpc.AgentInfo) bool,
) <-chan watchable.Snapshot[*rpc.AgentInfo]

func (*State) WatchClients

func (s *State) WatchClients(
	ctx context.Context,
	filter func(sessionID string, client *rpc.ClientInfo) bool,
) <-chan watchable.Snapshot[*rpc.ClientInfo]

func (*State) WatchDial

func (s *State) WatchDial(sessionID string) <-chan *rpc.DialRequest

func (*State) WatchIntercepts

func (s *State) WatchIntercepts(
	ctx context.Context,
	filter func(sessionID string, intercept *rpc.InterceptInfo) bool,
) <-chan watchable.Snapshot[*rpc.InterceptInfo]

func (*State) WatchLookupHost

func (s *State) WatchLookupHost(agentSessionID string) <-chan *rpc.LookupHostRequest