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

306 lines
8.3 KiB
Go
Raw Normal View History

// Notes:
2021-02-01 10:13:38 +00:00
package steward
2021-01-25 14:23:00 +00:00
import (
"bytes"
"encoding/gob"
"fmt"
"log"
2021-01-28 13:58:16 +00:00
"sync"
2021-01-25 14:23:00 +00:00
"time"
"github.com/nats-io/nats.go"
)
2021-01-28 13:58:16 +00:00
var mu sync.Mutex
2021-01-25 14:50:44 +00:00
type messageType int
// TODO: Figure it makes sense to have these types at all.
// It might make more sense to implement these as two
// individual subjects.
2021-01-25 14:50:44 +00:00
const (
2021-01-27 08:45:52 +00:00
// shellCommand, command that will just wait for an
// ack, and nothing of the output of the command are
// delivered back in the reply ack message.
// The message should contain the unique ID of the
// command.
2021-01-27 13:02:57 +00:00
commandReturnOutput messageType = iota
2021-01-25 14:50:44 +00:00
// shellCommand, wait for and return the output
// of the command in the ACK message. This means
// that the command should be executed immediately
// and that we should get the confirmation that it
// was successful or not.
2021-01-27 13:02:57 +00:00
eventReturnAck messageType = iota
2021-01-25 14:50:44 +00:00
// eventCommand, just wait for the ACK that the
// message is received. What action happens on the
// receiving side is up to the received to decide.
)
2021-01-25 14:23:00 +00:00
type Message struct {
2021-01-25 14:50:44 +00:00
// The Unique ID of the message
ID int
// The actual data in the message
2021-01-25 15:50:21 +00:00
// TODO: Change this to a slice instead...or maybe use an
// interface type here to handle several data types ?
2021-01-27 08:45:52 +00:00
Data []string
2021-01-25 14:50:44 +00:00
// The type of the message being sent
MessageType messageType
2021-01-25 14:23:00 +00:00
}
// server is the structure that will hold the state about spawned
// processes on a local instance.
2021-01-27 13:02:57 +00:00
type server struct {
natsConn *nats.Conn
// TODO: sessions should probably hold a slice/map of processes ?
processes map[subjectName]process
// The last processID created
lastProcessID int
nodeName string
2021-01-27 13:02:57 +00:00
}
// newServer will prepare and return a server type
2021-02-01 10:13:38 +00:00
func NewServer(brokerAddress string, nodeName string) (*server, error) {
conn, err := nats.Connect(brokerAddress, nil)
if err != nil {
log.Printf("error: nats.Connect failed: %v\n", err)
}
s := &server{
nodeName: nodeName,
natsConn: conn,
processes: make(map[subjectName]process),
}
2021-01-28 13:58:16 +00:00
go func() {
for {
for k := range s.processes {
select {
case e := <-s.processes[k].errorCh:
fmt.Printf("*** %v\n", e)
default:
time.Sleep(time.Millisecond * 100)
}
}
}
}()
return s, nil
}
func (s *server) PublisherStart() {
// start the checking of files for input messages
fileReadCh := make((chan []byte))
go getMessagesFromFile("./", "inmsg.txt", fileReadCh)
// TODO: For now we just print content of the files read.
// Replace this whit a broker function that will know how
// send it on to the correct publisher.
go func() {
for b := range fileReadCh {
// Check if there are new content read from file input
fmt.Printf("received: %s\n", b)
}
}()
{
sub := subject{
node: "btship1",
messageType: "command",
method: "shellcommand",
domain: "shell",
}
proc := s.processPrepareNew(sub)
// fmt.Printf("*** %#v\n", proc)
go s.processSpawn(proc)
}
{
sub := subject{
node: "btship2",
messageType: "command",
method: "shellcommand",
domain: "shell",
}
proc := s.processPrepareNew(sub)
// fmt.Printf("*** %#v\n", proc)
go s.processSpawn(proc)
}
select {}
2021-01-28 13:58:16 +00:00
}
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
node string
// messageType, command/event
messageType string
// method, what is this message doing, etc. shellcommand, syslog, etc.
method string
// domain is used to differentiate services. Like there can be more
// logging services, but rarely more logging services for the same
// thing. Domain is here used to differentiate the the services and
// tell with one word what it is for.
domain string
}
type subjectName string
func (s subject) name() subjectName {
return subjectName(fmt.Sprintf("%s.%s.%s.%s", s.node, s.messageType, s.method, s.domain))
}
2021-01-28 13:58:16 +00:00
// process are represent the communication to one individual host
type process struct {
messageID int
2021-02-03 07:28:21 +00:00
// the subject used for the specific process. One process
// can contain only one sender on a message bus, hence
// also one subject
subject subject
2021-01-28 13:58:16 +00:00
// 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
// errorCh is used to report errors from a process
// NB: Implementing this as an int to report for testing
errorCh chan string
// messageCh are the channel where we put the message we want
// a process to send
messageCh chan Message
}
// prepareNewProcess will set the the provided values and the default
// values for a process.
func (s *server) processPrepareNew(subject subject) process {
2021-02-03 08:06:37 +00:00
// create the initial configuration for a sessions communicating with 1 host process.
s.lastProcessID++
proc := process{
2021-01-27 13:02:57 +00:00
messageID: 0,
subject: subject,
node: node(subject.node),
processID: s.lastProcessID,
2021-01-28 13:58:16 +00:00
errorCh: make(chan string),
messageCh: make(chan Message),
2021-01-27 13:02:57 +00:00
}
return proc
}
// spawnProcess will spawn a new process. It will give the process
// the next available ID, and also add the process to the processes
// map.
func (s *server) processSpawn(proc process) {
2021-01-28 13:58:16 +00:00
mu.Lock()
s.processes[proc.subject.name()] = proc
2021-01-28 13:58:16 +00:00
mu.Unlock()
2021-01-27 13:02:57 +00:00
for k, v := range s.processes {
fmt.Printf("DEBUG: k=%v, v=%v \n", k, v)
}
2021-01-27 13:02:57 +00:00
// Loop creating one new message every second to simulate getting new
// messages to deliver.
//
// TODO: I think it makes most sense that the messages would come to
// here from some other message-pickup-process, and that process will
// give the message to the correct publisher process. A channel that
// is listened on in the for loop below could be used to receive the
// messages from the message-pickup-process.
2021-01-27 13:02:57 +00:00
for {
m := getMessageToDeliver()
m.ID = s.processes[proc.subject.name()].messageID
messageDeliver(proc, m, s.natsConn)
2021-01-27 08:45:52 +00:00
// Increment the counter for the next message to be sent.
proc.messageID++
s.processes[proc.subject.name()] = proc
2021-01-27 08:45:52 +00:00
time.Sleep(time.Second * 1)
2021-01-28 13:58:16 +00:00
// NB: simulate that we get an error, and that we can send that
2021-01-28 13:58:16 +00:00
// out of the process and receive it in another thread.
2021-02-03 11:59:08 +00:00
s.processes[proc.subject.name()].errorCh <- "received an error from process: " + fmt.Sprintf("ID=%v, subjectName=%v\n", proc.processID, proc.subject.name())
2021-01-28 13:58:16 +00:00
//fmt.Printf("%#v\n", s.processes[proc.node])
2021-01-27 08:45:52 +00:00
}
}
2021-01-27 13:02:57 +00:00
// get MessageToDeliver will pick up the next message to be created.
// TODO: read this from local file or rest or....?
func getMessageToDeliver() Message {
return Message{
Data: []string{"bash", "-c", "uname -a"},
2021-01-27 13:02:57 +00:00
MessageType: eventReturnAck,
}
}
func messageDeliver(proc process, message Message, natsConn *nats.Conn) {
2021-01-27 08:45:52 +00:00
for {
2021-01-27 13:02:57 +00:00
dataPayload, err := gobEncodePayload(message)
2021-01-27 08:45:52 +00:00
if err != nil {
log.Printf("error: createDataPayload: %v\n", err)
}
msg := &nats.Msg{
Subject: string(proc.subject.name()),
// Subject: fmt.Sprintf("%s.%s.%s", proc.node, "command", "shellcommand"),
// Structure of the reply message are:
// reply.<nodename>.<message type>.<method>
Reply: fmt.Sprintf("reply.%s", proc.subject.name()),
Data: dataPayload,
2021-01-25 14:23:00 +00:00
}
2021-01-27 08:45:52 +00:00
// 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.
subReply, err := natsConn.SubscribeSync(msg.Reply)
if err != nil {
log.Printf("error: nc.SubscribeSync failed: %v\n", err)
continue
}
// Publish message
err = natsConn.PublishMsg(msg)
if err != nil {
log.Printf("error: publish failed: %v\n", err)
continue
}
// Wait up until 10 seconds for a reply,
// continue and resend if to reply received.
msgReply, err := subReply.NextMsg(time.Second * 10)
if err != nil {
log.Printf("error: subRepl.NextMsg failed for node=%v, subject=%v: %v\n", proc.node, proc.subject.name(), err)
2021-01-27 08:45:52 +00:00
// did not receive a reply, continuing from top again
continue
}
fmt.Printf("publisher: received: %s\n", msgReply.Data)
return
}
}
2021-01-27 13:02:57 +00:00
// gobEncodePayload will encode the message structure along with its
// valued in gob binary format.
// TODO: Check if it adds value to compress with gzip.
2021-01-27 13:02:57 +00:00
func gobEncodePayload(m Message) ([]byte, error) {
2021-01-27 08:45:52 +00:00
var buf bytes.Buffer
gobEnc := gob.NewEncoder(&buf)
err := gobEnc.Encode(m)
if err != nil {
return nil, fmt.Errorf("error: gob.Enode failed: %v", err)
}
2021-01-25 14:23:00 +00:00
2021-01-27 08:45:52 +00:00
return buf.Bytes(), nil
2021-01-25 14:23:00 +00:00
}