1
0
Fork 0
mirror of https://github.com/postmannen/ctrl.git synced 2024-12-14 12:37:31 +00:00
ctrl/process.go

418 lines
14 KiB
Go
Raw Normal View History

2021-03-03 13:14:32 +00:00
package steward
import (
"bytes"
"encoding/gob"
"fmt"
"log"
"time"
"github.com/nats-io/nats.go"
)
// processKind are either kindSubscriber or kindPublisher, and are
// used to distinguish the kind of process to spawn and to know
// the process kind put in the process map.
type processKind string
const (
processKindSubscriber processKind = "subscriber"
processKindPublisher processKind = "publisher"
)
// process are represent the communication to one individual host
type process struct {
messageID int
// the subject used for the specific process. One process
// can contain only one sender on a message bus, hence
// also one subject
subject Subject
// Put a node here to be able know the node a process is at.
// NB: Might not be needed later on.
node node
// The processID for the current process
processID int
2021-03-12 08:38:19 +00:00
// errorCh is the same channel the errorKernel uses to
// read incomming errors. By having this channel available
// within a process we can send errors to the error kernel,
// the EK desided what to do, and sends the action about
// what to do back to the process where the error came from.
2021-03-03 13:14:32 +00:00
errorCh chan errProcess
processKind processKind
// Who are we allowed to receive from ?
allowedReceivers map[node]struct{}
// methodsAvailable
methodsAvailable MethodsAvailable
// Helper or service function that can do some kind of work
// for the process.
// The idea is that this can hold for example the map of the
// the hello nodes to limit shared resources in the system as
// a whole for sharing a map from the *server level.
2021-03-10 13:14:09 +00:00
procFunc procFunc
// The channel to send a messages to the procFunc go routine.
// This is typically used within the methodHandler.
procFuncCh chan Message
2021-03-08 13:09:14 +00:00
// copy of the configuration from server
configuration *Configuration
// The new messages channel copied from *Server
toRingbufferCh chan<- []subjectAndMessage
// The structure who holds all processes information
processes *processes
2021-03-03 13:14:32 +00:00
}
// prepareNewProcess will set the the provided values and the default
// values for a process.
func newProcess(processes *processes, toRingbufferCh chan<- []subjectAndMessage, configuration *Configuration, subject Subject, errCh chan errProcess, processKind processKind, allowedReceivers []node, procFunc func() error) process {
2021-03-03 13:14:32 +00:00
// create the initial configuration for a sessions communicating with 1 host process.
processes.lastProcessID++
2021-03-03 13:14:32 +00:00
// make the slice of allowedReceivers into a map value for easy lookup.
m := make(map[node]struct{})
for _, a := range allowedReceivers {
m[a] = struct{}{}
}
var method Method
2021-03-03 13:14:32 +00:00
proc := process{
messageID: 0,
subject: subject,
node: node(subject.ToNode),
processID: processes.lastProcessID,
2021-03-03 13:14:32 +00:00
errorCh: errCh,
processKind: processKind,
allowedReceivers: m,
methodsAvailable: method.GetMethodsAvailable(),
toRingbufferCh: toRingbufferCh,
configuration: configuration,
processes: processes,
2021-03-03 13:14:32 +00:00
}
return proc
}
2021-03-10 13:14:09 +00:00
// procFunc is a helper function that will do some extra work for
// a message received for a process. This allows us to ACK back
// to the publisher that the message was received, but we can let
// the processFunc keep on working.
// This can also be used to wrap in other types which we want to
// work with that come from the outside. An example can be handling
// of metrics which the message have no notion of, but a procFunc
// can have that wrapped in from when it was constructed.
type procFunc func() error
2021-03-03 13:14:32 +00:00
// The purpose of this function is to check if we should start a
// publisher or subscriber process, where a process is a go routine
// that will handle either sending or receiving messages on one
// subject.
//
// It will give the process the next available ID, and also add the
// process to the processes map in the server structure.
func (p process) spawnWorker(s *server) {
// We use the full name of the subject to identify a unique
// process. We can do that since a process can only handle
// one message queue.
var pn processName
if p.processKind == processKindPublisher {
pn = processNameGet(p.subject.name(), processKindPublisher)
}
if p.processKind == processKindSubscriber {
pn = processNameGet(p.subject.name(), processKindSubscriber)
}
// Add information about the new process to the started processes map.
s.processes.mu.Lock()
s.processes.active[pn] = p
s.processes.mu.Unlock()
2021-03-03 13:14:32 +00:00
// Start a publisher worker, which will start a go routine (process)
// That will take care of all the messages for the subject it owns.
if p.processKind == processKindPublisher {
2021-03-25 13:30:39 +00:00
2021-03-09 06:43:55 +00:00
// If there is a procFunc for the process, start it.
if p.procFunc != nil {
2021-03-25 13:30:39 +00:00
2021-03-09 06:43:55 +00:00
// Start the procFunc in it's own anonymous func so we are able
// to get the return error.
go func() {
err := p.procFunc()
if err != nil {
2021-03-12 08:38:19 +00:00
er := fmt.Errorf("error: spawnWorker: procFunc failed: %v", err)
sendErrorLogMessage(p.toRingbufferCh, node(p.node), er)
2021-03-09 06:43:55 +00:00
}
}()
}
go p.publishMessages(s.natsConn, s.processes)
2021-03-03 13:14:32 +00:00
}
// Start a subscriber worker, which will start a go routine (process)
// That will take care of all the messages for the subject it owns.
if p.processKind == processKindSubscriber {
// If there is a procFunc for the process, start it.
if p.procFunc != nil {
2021-03-25 13:30:39 +00:00
// Start the procFunc in it's own anonymous func so we are able
// to get the return error.
go func() {
err := p.procFunc()
if err != nil {
2021-03-12 08:38:19 +00:00
er := fmt.Errorf("error: spawnWorker: procFunc failed: %v", err)
sendErrorLogMessage(p.toRingbufferCh, node(p.node), er)
}
}()
}
2021-03-05 12:10:46 +00:00
2021-03-03 13:14:32 +00:00
p.subscribeMessages(s)
}
}
// messageDeliverNats will take care of the delivering the message
// as converted to gob format as a nats.Message. It will also take
// care of checking timeouts and retries specified for the message.
2021-03-09 06:43:55 +00:00
func (p process) messageDeliverNats(natsConn *nats.Conn, message Message) {
2021-03-03 13:14:32 +00:00
retryAttempts := 0
for {
dataPayload, err := gobEncodeMessage(message)
if err != nil {
2021-03-12 08:38:19 +00:00
er := fmt.Errorf("error: createDataPayload: %v", err)
sendErrorLogMessage(p.toRingbufferCh, node(p.node), er)
2021-03-12 08:38:19 +00:00
continue
2021-03-03 13:14:32 +00:00
}
msg := &nats.Msg{
2021-03-09 06:43:55 +00:00
Subject: string(p.subject.name()),
2021-03-03 13:14:32 +00:00
// Subject: fmt.Sprintf("%s.%s.%s", proc.node, "command", "CLICommand"),
// Structure of the reply message are:
// reply.<nodename>.<message type>.<method>
2021-03-09 06:43:55 +00:00
Reply: fmt.Sprintf("reply.%s", p.subject.name()),
2021-03-03 13:14:32 +00:00
Data: dataPayload,
}
// The SubscribeSync used in the subscriber, will get messages that
// are sent after it started subscribing, so we start a publisher
// that sends out a message every second.
//
// Create a subscriber for the reply message.
2021-03-09 06:43:55 +00:00
subReply, err := natsConn.SubscribeSync(msg.Reply)
2021-03-03 13:14:32 +00:00
if err != nil {
2021-03-12 08:38:19 +00:00
er := fmt.Errorf("error: nc.SubscribeSync failed: failed to create reply message: %v", err)
sendErrorLogMessage(p.toRingbufferCh, node(p.node), er)
2021-03-03 13:14:32 +00:00
continue
}
// Publish message
2021-03-09 06:43:55 +00:00
err = natsConn.PublishMsg(msg)
2021-03-03 13:14:32 +00:00
if err != nil {
2021-03-12 08:38:19 +00:00
er := fmt.Errorf("error: publish failed: %v", err)
sendErrorLogMessage(p.toRingbufferCh, node(p.node), er)
2021-03-03 13:14:32 +00:00
continue
}
// If the message is an ACK type of message we must check that a
// reply, and if it is not we don't wait here at all.
2021-03-25 13:30:39 +00:00
// fmt.Printf("info: messageDeliverNats: preparing to send message: %v\n", message)
2021-03-09 06:43:55 +00:00
if p.subject.CommandOrEvent == CommandACK || p.subject.CommandOrEvent == EventACK {
2021-03-03 13:14:32 +00:00
// Wait up until timeout specified for a reply,
// continue and resend if noo reply received,
// or exit if max retries for the message reached.
msgReply, err := subReply.NextMsg(time.Second * time.Duration(message.Timeout))
if err != nil {
er := fmt.Errorf("error: subReply.NextMsg failed for node=%v, subject=%v: %v", p.node, p.subject.name(), err)
sendErrorLogMessage(p.toRingbufferCh, message.FromNode, er)
2021-03-03 13:14:32 +00:00
// did not receive a reply, decide what to do..
retryAttempts++
fmt.Printf("Retry attempts:%v, retries: %v, timeout: %v\n", retryAttempts, message.Retries, message.Timeout)
switch {
case message.Retries == 0:
// 0 indicates unlimited retries
continue
case retryAttempts >= message.Retries:
// max retries reached
er := fmt.Errorf("info: max retries for message reached, breaking out: %v", message)
sendErrorLogMessage(p.toRingbufferCh, message.FromNode, er)
2021-03-03 13:14:32 +00:00
return
2021-03-03 13:14:32 +00:00
default:
// none of the above matched, so we've not reached max retries yet
continue
}
}
2021-03-11 05:34:36 +00:00
log.Printf("<--- publisher: received ACK from:%v, for: %v, data: %s\n", message.ToNode, message.Method, msgReply.Data)
2021-03-03 13:14:32 +00:00
}
return
}
}
// subscriberHandler will deserialize the message when a new message is
// received, check the MessageType field in the message to decide what
// kind of message it is and then it will check how to handle that message type,
// and then call the correct method handler for it.
//
// This handler function should be started in it's own go routine,so
// one individual handler is started per message received so we can keep
// the state of the message being processed, and then reply back to the
// correct sending process's reply, meaning so we ACK back to the correct
// publisher.
func (p process) subscriberHandler(natsConn *nats.Conn, thisNode string, msg *nats.Msg, s *server) {
message := Message{}
// Create a buffer to decode the gob encoded binary data back
// to it's original structure.
buf := bytes.NewBuffer(msg.Data)
gobDec := gob.NewDecoder(buf)
err := gobDec.Decode(&message)
if err != nil {
2021-03-12 08:38:19 +00:00
er := fmt.Errorf("error: gob decoding failed: %v", err)
sendErrorLogMessage(s.toRingbufferCh, node(thisNode), er)
2021-03-03 13:14:32 +00:00
}
switch {
case p.subject.CommandOrEvent == CommandACK || p.subject.CommandOrEvent == EventACK:
2021-03-09 06:43:55 +00:00
mh, ok := p.methodsAvailable.CheckIfExists(message.Method)
2021-03-03 13:14:32 +00:00
if !ok {
2021-03-12 08:38:19 +00:00
er := fmt.Errorf("error: subscriberHandler: method type not available: %v", p.subject.CommandOrEvent)
sendErrorLogMessage(s.toRingbufferCh, node(thisNode), er)
2021-03-03 13:14:32 +00:00
}
out := []byte("not allowed from " + message.FromNode)
2021-03-12 08:38:19 +00:00
//var err error
2021-03-03 13:14:32 +00:00
// Check if we are allowed to receive from that host
_, arOK1 := p.allowedReceivers[message.FromNode]
_, arOK2 := p.allowedReceivers["*"]
if arOK1 || arOK2 {
// Start the method handler for that specific subject type.
// The handler started here is what actually doing the action
// that executed a CLI command, or writes to a log file on
// the node who received the message.
2021-03-09 06:43:55 +00:00
out, err = mh.handler(p, message, thisNode)
2021-03-03 13:14:32 +00:00
if err != nil {
2021-03-12 08:38:19 +00:00
er := fmt.Errorf("error: subscriberHandler: failed to execute event: %v", err)
sendErrorLogMessage(s.toRingbufferCh, node(thisNode), er)
2021-03-03 13:14:32 +00:00
}
} else {
er := fmt.Errorf("info: we don't allow receiving from: %v, %v", message.FromNode, p.subject)
sendErrorLogMessage(s.toRingbufferCh, node(thisNode), er)
2021-03-03 13:14:32 +00:00
}
// Send a confirmation message back to the publisher
natsConn.Publish(msg.Reply, out)
case p.subject.CommandOrEvent == CommandNACK || p.subject.CommandOrEvent == EventNACK:
mf, ok := p.methodsAvailable.CheckIfExists(message.Method)
2021-03-03 13:14:32 +00:00
if !ok {
2021-03-12 08:38:19 +00:00
er := fmt.Errorf("error: subscriberHandler: method type not available: %v", p.subject.CommandOrEvent)
sendErrorLogMessage(s.toRingbufferCh, node(thisNode), er)
2021-03-03 13:14:32 +00:00
}
// Check if we are allowed to receive from that host
_, arOK1 := p.allowedReceivers[message.FromNode]
_, arOK2 := p.allowedReceivers["*"]
2021-03-03 13:14:32 +00:00
if arOK1 || arOK2 {
2021-03-25 13:30:39 +00:00
// Start the method handler for that specific subject type.
// The handler started here is what actually doing the action
// that executed a CLI command, or writes to a log file on
// the node who received the message.
//
// since we don't send a reply for a NACK message, we don't care about the
// out return when calling mf.handler
_, err := mf.handler(p, message, thisNode)
if err != nil {
er := fmt.Errorf("error: subscriberHandler: failed to execute event: %v", err)
sendErrorLogMessage(s.toRingbufferCh, node(thisNode), er)
}
} else {
er := fmt.Errorf("info: we don't allow receiving from: %v, %v", message.FromNode, p.subject)
sendErrorLogMessage(s.toRingbufferCh, node(thisNode), er)
2021-03-03 13:14:32 +00:00
}
// ---
2021-03-03 13:14:32 +00:00
default:
2021-03-12 08:38:19 +00:00
er := fmt.Errorf("info: did not find that specific type of command: %#v", p.subject.CommandOrEvent)
sendErrorLogMessage(s.toRingbufferCh, node(thisNode), er)
2021-03-12 08:38:19 +00:00
2021-03-03 13:14:32 +00:00
}
}
// Subscribe will start up a Go routine under the hood calling the
// callback function specified when a new message is received.
func (p process) subscribeMessages(s *server) {
subject := string(p.subject.name())
_, err := s.natsConn.Subscribe(subject, func(msg *nats.Msg) {
2021-03-25 13:30:39 +00:00
2021-03-03 13:14:32 +00:00
// We start one handler per message received by using go routines here.
// This is for being able to reply back the current publisher who sent
// the message.
go p.subscriberHandler(s.natsConn, s.nodeName, msg, s)
})
if err != nil {
log.Printf("error: Subscribe failed: %v\n", err)
}
}
2021-03-09 06:43:55 +00:00
// publishMessages will do the publishing of messages for one single
// process.
func (p process) publishMessages(natsConn *nats.Conn, processes *processes) {
2021-03-03 13:14:32 +00:00
for {
2021-03-12 08:38:19 +00:00
var err error
2021-03-03 13:14:32 +00:00
// Wait and read the next message on the message channel
m := <-p.subject.messageCh
2021-03-09 06:43:55 +00:00
// Get the process name so we can look up the process in the
// processes map, and increment the message counter.
2021-03-03 13:14:32 +00:00
pn := processNameGet(p.subject.name(), processKindPublisher)
2021-03-09 06:43:55 +00:00
m.ID = processes.active[pn].messageID
p.messageDeliverNats(natsConn, m)
// Signaling back to the ringbuffer that we are done with the
// current message, and it can remove it from the ringbuffer.
2021-03-03 13:14:32 +00:00
m.done <- struct{}{}
// Increment the counter for the next message to be sent.
p.messageID++
2021-03-09 06:43:55 +00:00
processes.mu.Lock()
processes.active[pn] = p
processes.mu.Unlock()
2021-03-12 08:38:19 +00:00
// Handle the error.
//
2021-03-12 08:38:19 +00:00
// NOTE: None of the processes above generate an error, so the the
// if clause will never be triggered. But keeping it here as an example
// for now for how to handle errors.
if err != nil {
// Create an error type which also creates a channel which the
// errorKernel will send back the action about what to do.
ep := errProcess{
infoText: "process failed",
process: p,
message: m,
errorActionCh: make(chan errorAction),
}
p.errorCh <- ep
// Wait for the response action back from the error kernel, and
// decide what to do. Should we continue, quit, or .... ?
switch <-ep.errorActionCh {
case errActionContinue:
// Just log and continue
log.Printf("The errAction was continue...so we're continuing\n")
case errActionKill:
log.Printf("The errAction was kill...so we're killing\n")
// ....
default:
log.Printf("Info: publishMessages: The errAction was not defined, so we're doing nothing\n")
}
}
2021-03-03 13:14:32 +00:00
}
}