Documentation
¶
Overview ¶
Package messaging implements the duplex transmission between the Executer and its document worker.
Index ¶
- Constants
- func CreateDatagram(t MessageType, content interface{}) (string, error)
- func GetLatestVersion() string
- func IsMachineShuttingDown() bool
- func Messaging(log log.T, ipc filewatcherbasedipc.IPCChannel, backend MessagingBackend, ...) (err error)
- func PrepareForShutdown(logger log.T)
- func RewindPatchDocumentForReboot(logger log.T, docState *contracts.DocumentState) bool
- func SetMachineShuttingDown()
- type ExecuterBackend
- func (p *ExecuterBackend) Accept() <-chan string
- func (p *ExecuterBackend) Close()
- func (p *ExecuterBackend) CloseStop()
- func (p *ExecuterBackend) ForceQuit()
- func (p *ExecuterBackend) GetBackendState() int32
- func (p *ExecuterBackend) Process(datagram string) error
- func (p *ExecuterBackend) Stop() <-chan int
- type Message
- type MessageType
- type MessagingBackend
- type PluginRunner
- type WorkerBackend
Constants ¶
const ( // Backend states with potential future addition to further manage backend // Currently with Init and Processing statuses to avoid idle process leak. BackendStateInit int32 = 0 BackendStateProc int32 = 1 )
const ( MessageTypePluginConfig = "pluginconfig" MessageTypeComplete = "complete" MessageTypeReply = "reply" MessageTypeCancel = "cancel" )
Message types
Variables ¶
This section is empty.
Functions ¶
func CreateDatagram ¶
func CreateDatagram(t MessageType, content interface{}) (string, error)
CreateDatagram marshals a given arbitrary object to raw json string Message schema is determined by the current version, content struct is indicated by type field TODO add version handling
func GetLatestVersion ¶
func GetLatestVersion() string
GetLatestVersion retrieves the current latest message version of the agent build
func IsMachineShuttingDown ¶
func IsMachineShuttingDown() bool
IsMachineShuttingDown reports whether a machine shutdown was signaled to this process, which is what makes a Failed patch result attributable to the reboot rather than to the patching itself.
func Messaging ¶
func Messaging(log log.T, ipc filewatcherbasedipc.IPCChannel, backend MessagingBackend, stopTimer chan bool) (err error)
Messaging implements the duplex transmission between master and worker, it send datagram it received to data backend, TODO ipc should not be destroyed within this worker, destroying ipc object should be done in its caller: Executer
func PrepareForShutdown ¶
PrepareForShutdown records that core relayed a machine-shutdown notification, so a patch document that fails or is cut short from here on is treated as reboot-interrupted rather than failed. Runs while core blocks the OS in the SCM preshutdown window, ahead of any worker being killed.
Setting a flag is all it does, and that is the design. The decision belongs where the false failure is otherwise produced, and there are exactly two such places -- Process, for a result the worker did report before it died, and the Executer's messaging error path, for a run the shutdown killed before it could report anything. Both run on the messaging goroutine, so neither needs to be reached from here and nothing needs to guard docState against a concurrent writer.
func RewindPatchDocumentForReboot ¶
func RewindPatchDocumentForReboot(logger log.T, docState *contracts.DocumentState) bool
RewindPatchDocumentForReboot rewinds a patch document whose run a machine shutdown cut short, and reports whether it did. Called from the Executer's messaging error path: when the shutdown kills the document worker before it can report anything, that path is what would otherwise synthesize a terminal failure for a document that has not actually failed.
Nothing is persisted here. The Executer's end-of-run save writes whatever this leaves behind, and if the machine goes down before that save runs, the state already on disk is the resumable one written at the start of the run -- a run only ever persists at its start and its end, so a Failed cannot reach disk while a shutdown is in progress.
func SetMachineShuttingDown ¶
func SetMachineShuttingDown()
SetMachineShuttingDown marks that a machine shutdown is underway.
Types ¶
type ExecuterBackend ¶
type ExecuterBackend struct {
// contains filtered or unexported fields
}
Executer backend formulate the run request to the worker, and collect back the responses from worker
func NewExecuterBackend ¶
func NewExecuterBackend(log log.T, output chan contracts.DocumentResult, docState *contracts.DocumentState, cancelFlag task.CancelFlag) *ExecuterBackend
func (*ExecuterBackend) Accept ¶
func (p *ExecuterBackend) Accept() <-chan string
func (*ExecuterBackend) Close ¶
func (p *ExecuterBackend) Close()
func (*ExecuterBackend) CloseStop ¶
func (p *ExecuterBackend) CloseStop()
func (*ExecuterBackend) ForceQuit ¶
func (p *ExecuterBackend) ForceQuit()
func (*ExecuterBackend) GetBackendState ¶
func (p *ExecuterBackend) GetBackendState() int32
func (*ExecuterBackend) Process ¶
func (p *ExecuterBackend) Process(datagram string) error
TODO handle error and logging, when err, ask messaging to stop TODO version handling?
func (*ExecuterBackend) Stop ¶
func (p *ExecuterBackend) Stop() <-chan int
type Message ¶
type Message struct {
Version string `json:"version"`
Type MessageType `json:"type"`
Content string `json:"content"`
}
type MessageType ¶
type MessageType string
func ParseDatagram ¶
func ParseDatagram(datagram string) (MessageType, string)
TODO add version and error handling
type MessagingBackend ¶
type MessagingBackend interface {
Accept() <-chan string
Stop() <-chan int
// Process a given datagram, should not be blocked
Process(string) error
// Sets input channel to nil.
Close()
// Sets stop channel to nil.
CloseStop()
// Get backend state
GetBackendState() int32
// Force Quit backend and exits the messaging block.
ForceQuit()
}
MessagingBackend defines an asycn message in/out processing pipeline
type PluginRunner ¶
type PluginRunner func( context context.T, docState contracts.DocumentState, resChan chan contracts.PluginResult, cancelFlag task.CancelFlag, )
type WorkerBackend ¶
type WorkerBackend struct {
// contains filtered or unexported fields
}
worker backend receives request messages from master, controls a pluginRunner based off the request and send reponses to Executer
func NewWorkerBackend ¶
func NewWorkerBackend(ctx context.T, runner PluginRunner) *WorkerBackend
func (*WorkerBackend) Accept ¶
func (p *WorkerBackend) Accept() <-chan string
func (*WorkerBackend) Close ¶
func (p *WorkerBackend) Close()
func (*WorkerBackend) CloseStop ¶
func (p *WorkerBackend) CloseStop()
func (*WorkerBackend) ForceQuit ¶
func (p *WorkerBackend) ForceQuit()
func (*WorkerBackend) GetBackendState ¶
func (p *WorkerBackend) GetBackendState() int32
func (*WorkerBackend) Process ¶
func (p *WorkerBackend) Process(datagram string) error
func (*WorkerBackend) Stop ¶
func (p *WorkerBackend) Stop() <-chan int
Source Files