controllers

Documentation

Index

• Constants
• type AgentHandler
  • func (ah *AgentHandler) Stop()
• type AgentTopicListener
  • func NewAgentTopicListener(agtMgr agent.Manager, tpMgr *tracepoint.Manager, sendMsgFn SendMessageFn) (*AgentTopicListener, error)
  • func (a *AgentTopicListener) HandleMessage(msg *nats.Msg) error
  • func (a *AgentTopicListener) Initialize() error
  • func (a *AgentTopicListener) SendMessageToAgent(agentID uuid.UUID, msg messagespb.VizierMessage) error
  • func (a *AgentTopicListener) Stop()
  • func (a *AgentTopicListener) StopAgent(agentID uuid.UUID)
• type EtcdManager
  • func NewEtcdManager(client *clientv3.Client) *EtcdManager
  • func (m *EtcdManager) IsDefragging() bool
  • func (m *EtcdManager) Run()
  • func (m *EtcdManager) Stop()
• type MessageBusController
  • func NewMessageBusController(conn *nats.Conn, agtMgr agent.Manager, tpMgr *tracepoint.Manager, ...) (*MessageBusController, error)
  • func (mc *MessageBusController) Close()
• type SendMessageFn
• type Server
  • func NewServer(env metadataenv.MetadataEnv, ds datastore.MultiGetterSetterDeleterCloser, ...) *Server
  • func (s *Server) ConvertLabelsToPods(tp *logicalpb.TracepointDeployment) error
  • func (s *Server) GetAgentInfo(ctx context.Context, req *metadatapb.AgentInfoRequest) (*metadatapb.AgentInfoResponse, error)
  • func (s *Server) GetAgentUpdates(req *metadatapb.AgentUpdatesRequest, ...) error
  • func (s *Server) GetSchemas(ctx context.Context, req *metadatapb.SchemaRequest) (*metadatapb.SchemaResponse, error)
  • func (s *Server) GetTracepointInfo(ctx context.Context, req *metadatapb.GetTracepointInfoRequest) (*metadatapb.GetTracepointInfoResponse, error)
  • func (s *Server) GetWithPrefixKey(ctx context.Context, req *metadatapb.WithPrefixKeyRequest) (*metadatapb.WithPrefixKeyResponse, error)
  • func (s *Server) RegisterTracepoint(ctx context.Context, req *metadatapb.RegisterTracepointRequest) (*metadatapb.RegisterTracepointResponse, error)
  • func (s *Server) RemoveTracepoint(ctx context.Context, req *metadatapb.RemoveTracepointRequest) (*metadatapb.RemoveTracepointResponse, error)
  • func (s *Server) UpdateConfig(ctx context.Context, req *metadatapb.UpdateConfigRequest) (*metadatapb.UpdateConfigResponse, error)
• type TopicListener

Constants

const UnhealthyAgentThreshold = 30 * time.Second

UnhealthyAgentThreshold is the amount of time where an agent is considered unhealthy if its last heartbeat is greater than this value.

Types

type AgentHandler

type AgentHandler struct {
	MsgChannel chan *nats.Msg
	// contains filtered or unexported fields
}

AgentHandler is responsible for handling messages for a specific agent.

func (*AgentHandler) Stop

func (ah *AgentHandler) Stop()

Stop immediately stops the agent handler from listening to any messages. It blocks until the agent is cleaned up.

type AgentTopicListener

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

AgentTopicListener is responsible for listening to and handling messages on the agent topic.

func NewAgentTopicListener

func NewAgentTopicListener(agtMgr agent.Manager, tpMgr *tracepoint.Manager,
	sendMsgFn SendMessageFn) (*AgentTopicListener, error)

NewAgentTopicListener creates a new agent topic listener.

func (*AgentTopicListener) HandleMessage

func (a *AgentTopicListener) HandleMessage(msg *nats.Msg) error

HandleMessage handles a message on the agent topic.

func (*AgentTopicListener) Initialize

func (a *AgentTopicListener) Initialize() error

Initialize loads in the current agent state into the agent topic listener.

func (*AgentTopicListener) SendMessageToAgent

func (a *AgentTopicListener) SendMessageToAgent(agentID uuid.UUID, msg messagespb.VizierMessage) error

SendMessageToAgent sends the given message to the agent over the NATS agent channel.

func (*AgentTopicListener) Stop

func (a *AgentTopicListener) Stop()

Stop stops processing any agent messagespb.

func (*AgentTopicListener) StopAgent

func (a *AgentTopicListener) StopAgent(agentID uuid.UUID)

StopAgent should be called when an agent should be deleted and its message processing should be stopped.

type EtcdManager

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

EtcdManager manages state for the given etcd instance.

func NewEtcdManager

func NewEtcdManager(client *clientv3.Client) *EtcdManager

NewEtcdManager creates a new etcd manager.

func (*EtcdManager) IsDefragging

func (m *EtcdManager) IsDefragging() bool

IsDefragging returns whether etcd is currently running a defrag.

func (*EtcdManager) Run

func (m *EtcdManager) Run()

Run periodically checks the size of the etcd db and defrags if necessary.

func (*EtcdManager) Stop

func (m *EtcdManager) Stop()

Stop stops the etcd manager from periodically checking and defragging etcd.

type MessageBusController

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

MessageBusController handles and responds to any incoming NATS messages.

func NewMessageBusController

func NewMessageBusController(conn *nats.Conn, agtMgr agent.Manager,
	tpMgr *tracepoint.Manager, k8smetaHandler *k8smeta.Handler,
	isLeader *bool) (*MessageBusController, error)

NewMessageBusController creates a new controller for handling NATS messages.

func (*MessageBusController) Close

func (mc *MessageBusController) Close()

Close closes the subscription and NATS connection.

type SendMessageFn

type SendMessageFn func(string, []byte) error

SendMessageFn is the function the TopicListener uses to publish messages back to NATS.

type Server

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

Server defines an gRPC server type.

func NewServer

func NewServer(env metadataenv.MetadataEnv, ds datastore.MultiGetterSetterDeleterCloser, pls k8smeta.PodLabelStore, agtMgr agent.Manager, tpMgr *tracepoint.Manager) *Server

NewServer creates GRPC handlers.

func (*Server) ConvertLabelsToPods

func (s *Server) ConvertLabelsToPods(tp *logicalpb.TracepointDeployment) error

ConvertLabelsToPods fetches all the pods in the PodLabelStore that match the labels described in the input tp, and then convert the LabelSelector to a PodProcess.

func (*Server) GetAgentInfo

func (s *Server) GetAgentInfo(ctx context.Context, req *metadatapb.AgentInfoRequest) (*metadatapb.AgentInfoResponse, error)

GetAgentInfo returns information about registered agents.

func (*Server) GetAgentUpdates

func (s *Server) GetAgentUpdates(req *metadatapb.AgentUpdatesRequest, srv metadatapb.MetadataService_GetAgentUpdatesServer) error

GetAgentUpdates streams agent updates to the requestor periodically as they come in. It first sends the complete initial agent state in the beginning of the request, and then deltas after that. Note that as it is currently designed, it can only handle one stream at a time (to a single metadata server). That is because the agent manager tracks the deltas in the state in a single object, rather than per request.

func (*Server) GetSchemas

func (s *Server) GetSchemas(ctx context.Context, req *metadatapb.SchemaRequest) (*metadatapb.SchemaResponse, error)

GetSchemas returns the schemas in the system.

func (*Server) GetTracepointInfo

func (s *Server) GetTracepointInfo(ctx context.Context, req *metadatapb.GetTracepointInfoRequest) (*metadatapb.GetTracepointInfoResponse, error)

GetTracepointInfo is a request to check the status for the given tracepoint.

func (*Server) GetWithPrefixKey

func (s *Server) GetWithPrefixKey(ctx context.Context, req *metadatapb.WithPrefixKeyRequest) (*metadatapb.WithPrefixKeyResponse, error)

GetWithPrefixKey fetches all the metadata KVs with the given prefix. This is used for debug purposes.

func (*Server) RegisterTracepoint

func (s *Server) RegisterTracepoint(ctx context.Context, req *metadatapb.RegisterTracepointRequest) (*metadatapb.RegisterTracepointResponse, error)

RegisterTracepoint is a request to register the tracepoints specified in the TracepointDeployment on all agents.

func (*Server) RemoveTracepoint

func (s *Server) RemoveTracepoint(ctx context.Context, req *metadatapb.RemoveTracepointRequest) (*metadatapb.RemoveTracepointResponse, error)

RemoveTracepoint is a request to evict the given tracepoint on all agents.

func (*Server) UpdateConfig

func (s *Server) UpdateConfig(ctx context.Context, req *metadatapb.UpdateConfigRequest) (*metadatapb.UpdateConfigResponse, error)

UpdateConfig updates the config for the specified agent.

type TopicListener

type TopicListener interface {
	Initialize() error
	HandleMessage(*nats.Msg) error
	Stop()
}

TopicListener handles NATS messages for a specific topic.