mirror of
https://github.com/postmannen/ctrl.git
synced 2024-12-14 12:37:31 +00:00
169 lines
7 KiB
Go
169 lines
7 KiB
Go
package steward
|
|
|
|
import (
|
|
"fmt"
|
|
"log"
|
|
"os"
|
|
)
|
|
|
|
// --- Message
|
|
|
|
type Message struct {
|
|
_ struct{} `cbor:",toarray"`
|
|
// The node to send the message to.
|
|
ToNode Node `json:"toNode" yaml:"toNode"`
|
|
// ToNodes to specify several hosts to send message to in the
|
|
// form of an slice/array.
|
|
// The ToNodes field is only a concept that exists when messages
|
|
// are injected f.ex. on a socket, and there they are directly
|
|
//converted into separate node messages for each node, and from
|
|
// there the ToNodes field is not used any more within the system.
|
|
// With other words, a message that exists within Steward is always
|
|
// for just for a single node.
|
|
ToNodes []Node `json:"toNodes,omitempty" yaml:"toNodes,omitempty"`
|
|
// The Unique ID of the message
|
|
ID int `json:"id" yaml:"id"`
|
|
// The actual data in the message. This is typically where we
|
|
// specify the cli commands to execute on a node, and this is
|
|
// also the field where we put the returned data in a reply
|
|
// message.
|
|
Data []byte `json:"data" yaml:"data"`
|
|
// Method, what request type to use, like REQCliCommand, REQHttpGet..
|
|
Method Method `json:"method" yaml:"method"`
|
|
// Additional arguments that might be needed when executing the
|
|
// method. Can be f.ex. an ip address if it is a tcp sender, or the
|
|
// shell command to execute in a cli session.
|
|
MethodArgs []string `json:"methodArgs" yaml:"methodArgs"`
|
|
// ArgSignature is the ed25519 signature of the methodArgs.
|
|
ArgSignature []byte `json:"argSignature" yaml:"argSignature"`
|
|
// ReplyMethod, is the method to use for the reply message.
|
|
// By default the reply method will be set to log to file, but
|
|
// you can override it setting your own here.
|
|
ReplyMethod Method `json:"replyMethod" yaml:"replyMethod"`
|
|
// Additional arguments that might be needed when executing the reply
|
|
// method. Can be f.ex. an ip address if it is a tcp sender, or the
|
|
// shell command to execute in a cli session.
|
|
ReplyMethodArgs []string `json:"replyMethodArgs" yaml:"replyMethodArgs"`
|
|
// IsReply are used to tell that this is a reply message. By default
|
|
// the system sends the output of a request method back to the node
|
|
// the message originated from. If it is a reply method we want the
|
|
// result of the reply message to be sent to the central server, so
|
|
// we can use this value if set to swap the toNode, and fromNode
|
|
// fields.
|
|
IsReply bool `json:"isReply" yaml:"isReply"`
|
|
// From what node the message originated
|
|
FromNode Node `json:"fromNode" yaml:"fromNode"`
|
|
// ACKTimeout for waiting for an ack message
|
|
ACKTimeout int `json:"ACKTimeout" yaml:"ACKTimeout"`
|
|
// RetryWait specified the time in seconds to wait between retries.
|
|
RetryWait int `json:"retryWait" yaml:"retryWait"`
|
|
// IsSubPublishedMsg enables timeout of publishing process, and is used together with process.isSubProcess to be able to terminate the sub processes publishers.
|
|
IsSubPublishedMsg bool `json:"isSubPublishedMsg" yaml:"isSubPublishedMsg"`
|
|
// Resend retries
|
|
Retries int `json:"retries" yaml:"retries"`
|
|
// The ACK timeout of the new message created via a request event.
|
|
ReplyACKTimeout int `json:"replyACKTimeout" yaml:"replyACKTimeout"`
|
|
// The retries of the new message created via a request event.
|
|
ReplyRetries int `json:"replyRetries" yaml:"replyRetries"`
|
|
// Timeout for long a process should be allowed to operate
|
|
MethodTimeout int `json:"methodTimeout" yaml:"methodTimeout"`
|
|
// Timeout for long a process should be allowed to operate
|
|
ReplyMethodTimeout int `json:"replyMethodTimeout" yaml:"replyMethodTimeout"`
|
|
// Directory is a string that can be used to create the
|
|
//directory structure when saving the result of some method.
|
|
// For example "syslog","metrics", or "metrics/mysensor"
|
|
// The type is typically used in the handler of a method.
|
|
Directory string `json:"directory" yaml:"directory"`
|
|
// FileName is used to be able to set a wanted name
|
|
// on a file being saved as the result of data being handled
|
|
// by a method handler.
|
|
FileName string `json:"fileName" yaml:"fileName"`
|
|
// PreviousMessage are used for example if a reply message is
|
|
// generated and we also need a copy of the details of the the
|
|
// initial request message.
|
|
PreviousMessage *Message
|
|
// Schedule
|
|
Schedule []int `json:"schedule" yaml:"schedule"`
|
|
|
|
// done is used to signal when a message is fully processed.
|
|
// This is used for signaling back to the ringbuffer that we are
|
|
// done with processing a message, and the message can be removed
|
|
// from the ringbuffer and into the time series log.
|
|
done chan struct{}
|
|
|
|
// ctx for the specifix message. Used for for example canceling
|
|
// scheduled messages.
|
|
// NB: Commented out this field for specific message context
|
|
// to be used within handlers, since it will override the structure
|
|
// we have today. Keeping the code for a bit incase it makes sense
|
|
// to implement later.
|
|
//ctx context.Context
|
|
}
|
|
|
|
// --- Subject
|
|
|
|
// Node is the type definition for the node who receive or send a message.
|
|
type Node string
|
|
|
|
// subject contains the representation of a subject to be used with one
|
|
// specific process
|
|
type Subject struct {
|
|
// node, the name of the node to receive the message.
|
|
ToNode string `json:"node" yaml:"toNode"`
|
|
// Event, event type like EventACK or EventNACK.
|
|
Event Event `json:"event" yaml:"event"`
|
|
// method, what is this message doing, etc. CLICommand, Syslog, etc.
|
|
Method Method `json:"method" yaml:"method"`
|
|
// messageCh is used by publisher kind processes to read new messages
|
|
// to be published. The content on this channel have been routed here
|
|
// from routeMessagesToPublish in *server.
|
|
// This channel is only used for publishing processes.
|
|
messageCh chan Message
|
|
}
|
|
|
|
// newSubject will return a new variable of the type subject, and insert
|
|
// all the values given as arguments. It will create the channel
|
|
// to receive new messages on the specific subject.
|
|
// The function will also verify that there is a methodHandler defined
|
|
// for the Request type.
|
|
func newSubject(method Method, node string) Subject {
|
|
// Get the Event type for the Method.
|
|
ma := method.GetMethodsAvailable()
|
|
mh, ok := ma.CheckIfExists(method)
|
|
//mh, ok := ma.Methodhandlers[method]
|
|
if !ok {
|
|
log.Printf("error: newSubject: no Event type specified for the method: %v\n", method)
|
|
os.Exit(1)
|
|
}
|
|
|
|
return Subject{
|
|
ToNode: node,
|
|
Event: mh.getKind(),
|
|
Method: method,
|
|
messageCh: make(chan Message),
|
|
}
|
|
}
|
|
|
|
// newSubjectNoVerifyHandler will return a new variable of the type subject, and insert
|
|
// all the values given as arguments. It will create the channel
|
|
// to receive new messages on the specific subject.
|
|
// The function will not verify that there is a methodHandler defined
|
|
// for the Request type.
|
|
func newSubjectNoVerifyHandler(method Method, node string) Subject {
|
|
// Get the Event type for the Method.
|
|
|
|
return Subject{
|
|
ToNode: node,
|
|
Event: EventACK,
|
|
Method: method,
|
|
messageCh: make(chan Message),
|
|
}
|
|
}
|
|
|
|
// subjectName is the complete representation of a subject
|
|
type subjectName string
|
|
|
|
// Return a value of the subjectName for the subject as used with nats subject.
|
|
func (s Subject) name() subjectName {
|
|
return subjectName(fmt.Sprintf("%s.%s.%s", s.ToNode, s.Method, s.Event))
|
|
}
|