mgmt

type Client ¶

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

Client is a synchronous client for the management socket. It is not safe for concurrent use - callers should open one per goroutine.

func Dial(socketPath string) (*Client, error)

func (c *Client) Call(t ipc.MsgType, payload any) (ipc.Header, []byte, error)

func (c *Client) Close() error

func (c *Client) ReadFrame() (ipc.Header, []byte, error)

func (c *Client) Stream(t ipc.MsgType, payload any) error

type EventsStreamPayload ¶

type EventsStreamPayload struct {
	ProcessID string `msgpack:"pid"`
	Type      string `msgpack:"t"`
	Timestamp int64  `msgpack:"ts"`
	Summary   string `msgpack:"sum"`
}

EventsStreamPayload carries a serialized event summary.

type EventsSubscribePayload ¶

type EventsSubscribePayload struct {
	ProcessIDs []string `msgpack:"pids"`
}

EventsSubscribePayload is sent by a client to subscribe to a stream of lifecycle events. An empty ProcessIDs slice subscribes to every process.

type ListEntry ¶

type ListEntry struct {
	ID       string `msgpack:"id"`
	Name     string `msgpack:"name"`
	State    string `msgpack:"state"`
	ChildPID int    `msgpack:"pid"`
	RSS      uint64 `msgpack:"rss"`
}

ListEntry is one row in a list response.

type ListResponsePayload ¶

type ListResponsePayload struct {
	Entries []ListEntry `msgpack:"entries"`
}

ListResponsePayload is returned for MsgListResponse.

type LogsRequestPayload ¶

type LogsRequestPayload struct {
	ID     string `msgpack:"id"`
	Stream uint8  `msgpack:"s"`
	Follow bool   `msgpack:"f"`
}

LogsRequestPayload asks for log output from a process.

If Follow is false (the default), the server returns a single MsgLogsStream frame with EOF=true containing the current ring-buffer snapshot. If Follow is true, the server first sends the snapshot with EOF=false and then keeps the connection open, pushing additional MsgLogsStream frames (also EOF=false) as new chunks arrive, until the client disconnects or the manager shuts down.

type LogsStreamPayload ¶

type LogsStreamPayload struct {
	ID     string `msgpack:"id"`
	Stream uint8  `msgpack:"s"`
	Data   []byte `msgpack:"d"`
	EOF    bool   `msgpack:"eof"`
}

LogsStreamPayload carries a single chunk of streamed log bytes.

type ManagerAPI ¶

type ManagerAPI interface {
	List() []ListEntry
	StatusOf(id string) (ListEntry, error)
	Stop(ctx context.Context, id string) error
	LogsSnapshot(id string, stream uint8) []byte
	// WatchLogs returns a channel that delivers every log chunk
	// emitted for processID after the subscription is registered,
	// plus an unregister function the caller MUST invoke exactly
	// once. The channel closes when the manager shuts down.
	WatchLogs(processID string) (<-chan cwtypes.LogChunk, func())
	Events() event.Bus
}

ManagerAPI is the subset of Manager functionality the server exposes. Using an interface keeps tests decoupled from the concrete Manager.

type Server ¶

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

Server listens on a Unix domain socket and handles management requests.

func NewServer(opts ServerOptions) (*Server, error)

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

func (s *Server) Start() error

type ServerOptions ¶

type ServerOptions struct {
	SocketPath string
	Manager    ManagerAPI
	SpillerDir string
}

ServerOptions configures the management server.

type StartRequestPayload ¶

type StartRequestPayload struct {
	ID string `msgpack:"id"`
}

StartRequestPayload requests that a registered process be started.

type StatusRequestPayload ¶

type StatusRequestPayload struct {
	ID string `msgpack:"id"`
}

StatusRequestPayload asks for a specific process's status.

type StatusResponsePayload ¶

type StatusResponsePayload struct {
	Entry ListEntry `msgpack:"entry"`
	Err   string    `msgpack:"err"`
}

StatusResponsePayload summarizes one process.

type StopRequestPayload ¶

type StopRequestPayload struct {
	ID string `msgpack:"id"`
}

StopRequestPayload requests that a running process be stopped.