From 61e2343f189e57e8c9fb78087c86dd26ddd61b81 Mon Sep 17 00:00:00 2001
From: postmannen <postmannen@gmail.com>
Date: Wed, 24 Feb 2021 15:43:31 +0100
Subject: [PATCH] refactored, and implemented error messages to central

---
 cmd/main.go                     |   3 +-
 errorkernel.go                  |  20 +++++-
 incommmingBuffer.db             | Bin 32768 -> 32768 bytes
 message-and-subject.go          |  78 ++++++++++++++++++++
 publisher.go                    |   8 ++-
 ringbuffer.go                   |   2 +
 server.go                       | 123 +++++++++++---------------------
 subscriberMethodTypes.go        |  14 ++++
 subscriber.go => subscribers.go |  11 +++
 9 files changed, 174 insertions(+), 85 deletions(-)
 create mode 100644 message-and-subject.go
 rename subscriber.go => subscribers.go (82%)

diff --git a/cmd/main.go b/cmd/main.go
index 71c177a..6dab5cb 100644
--- a/cmd/main.go
+++ b/cmd/main.go
@@ -16,6 +16,7 @@ func main() {
 	brokerAddress := flag.String("brokerAddress", "0", "the address of the message broker")
 	profilingPort := flag.String("profilingPort", "", "The number of the profiling port")
 	promHostAndPort := flag.String("promHostAndPort", ":2112", "host and port for prometheus listener, e.g. localhost:2112")
+	centralErrorLogger := flag.Bool("centralErrorLogger", false, "seet to true if this is the node that should receive the error log's from other nodes")
 	//isCentral := flag.Bool("isCentral", false, "used to indicate that this is the central master that will subscribe to error message subjects")
 	flag.Parse()
 
@@ -28,7 +29,7 @@ func main() {
 
 	}
 
-	s, err := steward.NewServer(*brokerAddress, *nodeName, *promHostAndPort)
+	s, err := steward.NewServer(*brokerAddress, *nodeName, *promHostAndPort, *centralErrorLogger)
 	if err != nil {
 		log.Printf("error: failed to connect to broker: %v\n", err)
 		os.Exit(1)
diff --git a/errorkernel.go b/errorkernel.go
index 631848d..c640dc3 100644
--- a/errorkernel.go
+++ b/errorkernel.go
@@ -40,7 +40,7 @@ func newErrorKernel() *errorKernel {
 // process if it should continue or not based not based on how severe
 // the error where. This should be right after sending the error
 // sending in the process.
-func (e *errorKernel) startErrorKernel() {
+func (e *errorKernel) startErrorKernel(newMessagesCh chan<- []subjectAndMessage) {
 	// TODO: For now it will just print the error messages to the
 	// console.
 	go func() {
@@ -63,6 +63,24 @@ func (e *errorKernel) startErrorKernel() {
 
 				// log.Printf("*** error_kernel: %#v, type=%T\n", er, er)
 				log.Printf("TESTING, we received and error from the process, but we're telling the process back to continue\n")
+
+				// // TESTING: Creating an error message to send to errorCentral
+				// fmt.Printf(" --- Sending error message to central !!!!!!!!!!!!!!!!!!!!!!!!!!!!\n")
+				// sam := subjectAndMessage{
+				// 	Subject: Subject{
+				// 		ToNode:         "errorCentral",
+				// 		CommandOrEvent: EventNACK,
+				// 		Method:         ErrorLog,
+				// 	},
+				// 	Message: Message{
+				// 		ToNode:         "errorCentral",
+				// 		Data:           []string{"some tull here .............."},
+				// 		CommandOrEvent: EventNACK,
+				// 		Method:         ErrorLog,
+				// 	},
+				// }
+				// newMessagesCh <- []subjectAndMessage{sam}
+
 				er.errorActionCh <- errActionContinue
 			}()
 		}
diff --git a/incommmingBuffer.db b/incommmingBuffer.db
index 7a3a58bd10b32a0ea5c6a1884c4a9d659a095ef1..2dd7e376302d118764dfe520a778a7974877366a 100644
GIT binary patch
delta 170
zcmZo@U}|V!n&2Q%zyJX;ty0QThMWzX6%97<PZDqt$cIWSR`a-czxm=yh=hU;lM3Tx
zJq33rWAn-74pw}hfXa-`s+Bxll&p-+ChvDpU^JTiz#&M}NJ&R2GeybDKu0Mhu_RH+
zDq1Niu{Z-Ls+$ZXN{Uib74-D<m0~A9bTHez*fExGVuH{n108|MItrdZ(>IqZbn-6>
GFaQ87elNWM

delta 184
zcmZo@U}|V!n&2Q1!T<q37Py6qe=B>uS<zqv|0Dqifl#Q#%##P@Pt<QWf=DRnOxAJm
zWHPeYT<*}xzsW#HfYEHSu7U?pWOBKJ6(0*D1A~!ywUVccl9iG9<oyZ?jE0jRC<Iv>
zD(NU?rYKn%=qRNmmLw`!MJpvG7H0rOb(4WaUTR5kNn(kDZem_ddQobDf^MclPG)gQ
VYMxT;<cA7on<py9@+}H5002O+HuC@g

diff --git a/message-and-subject.go b/message-and-subject.go
new file mode 100644
index 0000000..0cd91c4
--- /dev/null
+++ b/message-and-subject.go
@@ -0,0 +1,78 @@
+package steward
+
+import (
+	"bytes"
+	"encoding/gob"
+	"fmt"
+)
+
+// --- Message
+
+type Message struct {
+	ToNode node `json:"toNode" yaml:"toNode"`
+	// The Unique ID of the message
+	ID int `json:"id" yaml:"id"`
+	// The actual data in the message
+	// TODO: Change this to a slice instead...or maybe use an
+	// interface type here to handle several data types ?
+	Data []string `json:"data" yaml:"data"`
+	// The type of the message being sent
+	CommandOrEvent CommandOrEvent `json:"commandOrEvent" yaml:"commandOrEvent"`
+	// method, what is this message doing, etc. shellCommand, syslog, etc.
+	Method   Method `json:"method" yaml:"method"`
+	FromNode node
+	// done is used to signal when a message is fully processed.
+	// This is used when choosing when to move the message from
+	// the ringbuffer into the time series log.
+	done chan struct{}
+}
+
+// gobEncodePayload will encode the message structure along with its
+// valued in gob binary format.
+// TODO: Check if it adds value to compress with gzip.
+func gobEncodeMessage(m Message) ([]byte, error) {
+	var buf bytes.Buffer
+	gobEnc := gob.NewEncoder(&buf)
+	err := gobEnc.Encode(m)
+	if err != nil {
+		return nil, fmt.Errorf("error: gob.Encode failed: %v", err)
+	}
+
+	return buf.Bytes(), nil
+}
+
+// --- Subject
+
+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
+	ToNode string `json:"node" yaml:"toNode"`
+	// messageType, command/event
+	CommandOrEvent CommandOrEvent `json:"commandOrEvent" yaml:"commandOrEvent"`
+	// method, what is this message doing, etc. shellCommand, syslog, etc.
+	Method Method `json:"method" yaml:"method"`
+	// messageCh is the channel for receiving new content to be sent
+	messageCh chan Message
+}
+
+// newSubject will return a new variable of the type subject, and insert
+// all the values given as arguments. It will also create the channel
+// to receive new messages on the specific subject.
+func newSubject(method Method, commandOrEvent CommandOrEvent, node string) Subject {
+	return Subject{
+		ToNode:         node,
+		CommandOrEvent: commandOrEvent,
+		Method:         method,
+		messageCh:      make(chan Message),
+	}
+}
+
+// subjectName is the complete representation of a subject
+type subjectName string
+
+func (s Subject) name() subjectName {
+	return subjectName(fmt.Sprintf("%s.%s.%s", s.Method, s.CommandOrEvent, s.ToNode))
+}
diff --git a/publisher.go b/publisher.go
index e6c1e5b..cd240f2 100644
--- a/publisher.go
+++ b/publisher.go
@@ -36,7 +36,7 @@ func (s *server) processNewMessages(dbFileName string, newSAM chan []subjectAndM
 	}()
 
 	// Process the messages that are in the ring buffer. Check and
-	// send if there are a specific subject for it, and no subject
+	// send if there are a specific subject for it, and if no subject
 	// exist throw an error.
 	go func() {
 		for samTmp := range ringBufferOutCh {
@@ -46,9 +46,11 @@ func (s *server) processNewMessages(dbFileName string, newSAM chan []subjectAndM
 			// it was unable to process the message with the reason
 			// why ?
 			if _, ok := s.methodsAvailable.CheckIfExists(sam.Message.Method); !ok {
+				log.Printf("error: the method do not exist: %v\n", sam.Message.Method)
 				continue
 			}
 			if !s.commandOrEventAvailable.CheckIfExists(sam.Message.CommandOrEvent) {
+				log.Printf("error: the command or evnt do not exist: %v\n", sam.Message.CommandOrEvent)
 				continue
 			}
 
@@ -72,7 +74,7 @@ func (s *server) processNewMessages(dbFileName string, newSAM chan []subjectAndM
 				// If no process to handle the specific subject exist,
 				// the we create and spawn one.
 			} else {
-				// If a publisher do not exist for the given subject, create it, and
+				// If a publisher process do not exist for the given subject, create it, and
 				// by using the goto at the end redo the process for this specific message.
 				log.Printf("info: did not find that specific subject, starting new process for subject: %v\n", subjName)
 
@@ -96,7 +98,7 @@ func (s *server) publishMessages(proc process) {
 		// Wait and read the next message on the message channel
 		m := <-proc.subject.messageCh
 		m.ID = s.processes[proc.subject.name()].messageID
-		messageDeliver(proc, m, s.natsConn)
+		s.messageDeliverNats(proc, m)
 		m.done <- struct{}{}
 
 		// Increment the counter for the next message to be sent.
diff --git a/ringbuffer.go b/ringbuffer.go
index 15a23a3..045a482 100644
--- a/ringbuffer.go
+++ b/ringbuffer.go
@@ -185,8 +185,10 @@ func (r *ringBuffer) processBufferMessages(samValueBucket string, outCh chan sam
 			// Listen on the done channel here , so a go routine handling the
 			// message will be able to signal back here that the message have
 			// been processed, and that we then can delete it out of the K/V Store.
+			fmt.Printf("#.#.#.#.#.#.#.# Before DONE: %v\n", v)
 			<-v.Data.done
 			log.Printf("info: done with message %v\n", v.ID)
+			fmt.Printf("#.#.#.#.#.#.#.# Got DONE: %v\n", v)
 
 			// Since we are now done with the specific message we can delete
 			// it out of the K/V Store.
diff --git a/server.go b/server.go
index da0276e..6ac5231 100644
--- a/server.go
+++ b/server.go
@@ -14,25 +14,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 )
 
-type Message struct {
-	ToNode node `json:"toNode" yaml:"toNode"`
-	// The Unique ID of the message
-	ID int `json:"id" yaml:"id"`
-	// The actual data in the message
-	// TODO: Change this to a slice instead...or maybe use an
-	// interface type here to handle several data types ?
-	Data []string `json:"data" yaml:"data"`
-	// The type of the message being sent
-	CommandOrEvent CommandOrEvent `json:"commandOrEvent" yaml:"commandOrEvent"`
-	// method, what is this message doing, etc. shellCommand, syslog, etc.
-	Method   Method `json:"method" yaml:"method"`
-	FromNode node
-	// done is used to signal when a message is fully processed.
-	// This is used when choosing when to move the message from
-	// the ringbuffer into the time series log.
-	done chan struct{}
-}
-
 // server is the structure that will hold the state about spawned
 // processes on a local instance.
 type server struct {
@@ -45,9 +26,11 @@ type server struct {
 	nodeName string
 	// Mutex for locking when writing to the process map
 	mu sync.Mutex
-	// The channel where we receive new messages read from file.
+	// The channel where we put new messages read from file,
+	// or some other process who wants to send something via the
+	// system
 	// We can than range this channel for new messages to process.
-	inputFromFileCh chan []subjectAndMessage
+	newMessagesCh chan []subjectAndMessage
 	// errorKernel is doing all the error handling like what to do if
 	// an error occurs.
 	// TODO: Will also send error messages to cental error subscriber.
@@ -65,10 +48,12 @@ type server struct {
 	// how to forward the data for a received message of type log to a
 	// central logger.
 	subscriberServices *subscriberServices
+	// Is this the central error logger ?
+	centralErrorLogger bool
 }
 
 // newServer will prepare and return a server type
-func NewServer(brokerAddress string, nodeName string, promHostAndPort string) (*server, error) {
+func NewServer(brokerAddress string, nodeName string, promHostAndPort string, centralErrorLogger bool) (*server, error) {
 	conn, err := nats.Connect(brokerAddress, nil)
 	if err != nil {
 		log.Printf("error: nats.Connect failed: %v\n", err)
@@ -81,11 +66,12 @@ func NewServer(brokerAddress string, nodeName string, promHostAndPort string) (*
 		nodeName:                nodeName,
 		natsConn:                conn,
 		processes:               make(map[subjectName]process),
-		inputFromFileCh:         make(chan []subjectAndMessage),
+		newMessagesCh:           make(chan []subjectAndMessage),
 		methodsAvailable:        m.GetMethodsAvailable(),
 		commandOrEventAvailable: coe.GetCommandOrEventAvailable(),
 		metrics:                 newMetrics(promHostAndPort),
 		subscriberServices:      newSubscriberServices(),
+		centralErrorLogger:      centralErrorLogger,
 	}
 
 	return s, nil
@@ -100,13 +86,13 @@ func (s *server) Start() {
 	// Start the error kernel that will do all the error handling
 	// not done within a process.
 	s.errorKernel = newErrorKernel()
-	s.errorKernel.startErrorKernel()
+	s.errorKernel.startErrorKernel(s.newMessagesCh)
 
 	// Start collecting the metrics
 	go s.startMetrics()
 
 	// Start the checking the input file for new messages from operator.
-	go s.getMessagesFromFile("./", "inmsg.txt", s.inputFromFileCh)
+	go s.getMessagesFromFile("./", "inmsg.txt", s.newMessagesCh)
 
 	// Start the textLogging service that will run on the subscribers
 	// TODO: This should only be started if the flag value provided when
@@ -122,7 +108,7 @@ func (s *server) Start() {
 	s.printProcessesMap()
 
 	// Start the processing of new messaging from an input channel.
-	s.processNewMessages("./incommmingBuffer.db", s.inputFromFileCh)
+	s.processNewMessages("./incommmingBuffer.db", s.newMessagesCh)
 
 	select {}
 
@@ -146,40 +132,6 @@ func (s *server) printProcessesMap() {
 	fmt.Println("--------------------------------------------------------------------------------------------")
 }
 
-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
-	ToNode string `json:"node" yaml:"toNode"`
-	// messageType, command/event
-	CommandOrEvent CommandOrEvent `json:"commandOrEvent" yaml:"commandOrEvent"`
-	// method, what is this message doing, etc. shellCommand, syslog, etc.
-	Method Method `json:"method" yaml:"method"`
-	// messageCh is the channel for receiving new content to be sent
-	messageCh chan Message
-}
-
-// newSubject will return a new variable of the type subject, and insert
-// all the values given as arguments. It will also create the channel
-// to receive new messages on the specific subject.
-func newSubject(method Method, commandOrEvent CommandOrEvent, node string) Subject {
-	return Subject{
-		ToNode:         node,
-		CommandOrEvent: commandOrEvent,
-		Method:         method,
-		messageCh:      make(chan Message),
-	}
-}
-
-// subjectName is the complete representation of a subject
-type subjectName string
-
-func (s Subject) name() subjectName {
-	return subjectName(fmt.Sprintf("%s.%s.%s", s.Method, s.CommandOrEvent, s.ToNode))
-}
-
 // 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.
@@ -255,7 +207,7 @@ func (s *server) spawnWorkerProcess(proc process) {
 	}
 }
 
-func messageDeliver(proc process, message Message, natsConn *nats.Conn) {
+func (s *server) messageDeliverNats(proc process, message Message) {
 	for {
 		dataPayload, err := gobEncodeMessage(message)
 		if err != nil {
@@ -276,7 +228,7 @@ func messageDeliver(proc process, message Message, natsConn *nats.Conn) {
 		// that sends out a message every second.
 		//
 		// Create a subscriber for the reply message.
-		subReply, err := natsConn.SubscribeSync(msg.Reply)
+		subReply, err := s.natsConn.SubscribeSync(msg.Reply)
 		if err != nil {
 			log.Printf("error: nc.SubscribeSync failed: %v\n", err)
 			os.Exit(1)
@@ -284,7 +236,7 @@ func messageDeliver(proc process, message Message, natsConn *nats.Conn) {
 		}
 
 		// Publish message
-		err = natsConn.PublishMsg(msg)
+		err = s.natsConn.PublishMsg(msg)
 		if err != nil {
 			log.Printf("error: publish failed: %v\n", err)
 			continue
@@ -292,6 +244,7 @@ func messageDeliver(proc process, message Message, natsConn *nats.Conn) {
 
 		// 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.
+		fmt.Printf("---- MESSAGE : %v\n", message)
 		if message.CommandOrEvent == CommandACK || message.CommandOrEvent == EventACK {
 			// Wait up until 10 seconds for a reply,
 			// continue and resend if to reply received.
@@ -307,20 +260,6 @@ func messageDeliver(proc process, message Message, natsConn *nats.Conn) {
 	}
 }
 
-// gobEncodePayload will encode the message structure along with its
-// valued in gob binary format.
-// TODO: Check if it adds value to compress with gzip.
-func gobEncodeMessage(m Message) ([]byte, error) {
-	var buf bytes.Buffer
-	gobEnc := gob.NewEncoder(&buf)
-	err := gobEnc.Encode(m)
-	if err != nil {
-		return nil, fmt.Errorf("error: gob.Encode failed: %v", err)
-	}
-
-	return buf.Bytes(), nil
-}
-
 // handler 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,
@@ -330,7 +269,7 @@ func gobEncodeMessage(m Message) ([]byte, error) {
 // 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 (s *server) subscriberHandler(natsConn *nats.Conn, node string, msg *nats.Msg) {
+func (s *server) subscriberHandler(natsConn *nats.Conn, thisNode string, msg *nats.Msg) {
 
 	message := Message{}
 
@@ -357,7 +296,7 @@ func (s *server) subscriberHandler(natsConn *nats.Conn, node string, msg *nats.M
 			log.Printf("error: subscriberHandler: method type not available: %v\n", message.CommandOrEvent)
 		}
 		fmt.Printf("*** DEBUG: BEFORE CALLING HANDLER: ACK\n")
-		out, err := mf.handler(s, message, node)
+		out, err := mf.handler(s, message, thisNode)
 
 		if err != nil {
 			// TODO: Send to error kernel ?
@@ -366,6 +305,8 @@ func (s *server) subscriberHandler(natsConn *nats.Conn, node string, msg *nats.M
 
 		// Send a confirmation message back to the publisher
 		natsConn.Publish(msg.Reply, out)
+
+		sendErrorMessage(s.newMessagesCh, node(thisNode))
 	case message.CommandOrEvent == CommandNACK || message.CommandOrEvent == EventNACK:
 		log.Printf("info: subscriberHandler: message.CommandOrEvent received was = %v, preparing to call handler\n", message.CommandOrEvent)
 		mf, ok := s.methodsAvailable.CheckIfExists(message.Method)
@@ -376,7 +317,7 @@ func (s *server) subscriberHandler(natsConn *nats.Conn, node string, msg *nats.M
 		// since we don't send a reply for a NACK message, we don't care about the
 		// out return when calling mf.handler
 		fmt.Printf("*** DEBUG: BEFORE CALLING HANDLER: NACK\n")
-		_, err := mf.handler(s, message, node)
+		_, err := mf.handler(s, message, thisNode)
 
 		if err != nil {
 			// TODO: Send to error kernel ?
@@ -386,3 +327,25 @@ func (s *server) subscriberHandler(natsConn *nats.Conn, node string, msg *nats.M
 		log.Printf("info: did not find that specific type of command: %#v\n", message.CommandOrEvent)
 	}
 }
+
+func sendErrorMessage(newMessagesCh chan<- []subjectAndMessage, FromNode node) {
+	// --- Testing
+
+	// TESTING: Creating an error message to send to errorCentral
+	fmt.Printf(" --- Sending error message to central !!!!!!!!!!!!!!!!!!!!!!!!!!!!\n")
+	sam := subjectAndMessage{
+		Subject: Subject{
+			ToNode:         "errorCentral",
+			CommandOrEvent: EventNACK,
+			Method:         ErrorLog,
+		},
+		Message: Message{
+			ToNode:         "errorCentral",
+			FromNode:       FromNode,
+			Data:           []string{"some tull here .............."},
+			CommandOrEvent: EventNACK,
+			Method:         ErrorLog,
+		},
+	}
+	newMessagesCh <- []subjectAndMessage{sam}
+}
diff --git a/subscriberMethodTypes.go b/subscriberMethodTypes.go
index 3a96784..7a8f73c 100644
--- a/subscriberMethodTypes.go
+++ b/subscriberMethodTypes.go
@@ -26,6 +26,7 @@ func (m Method) GetMethodsAvailable() MethodsAvailable {
 			ShellCommand: methodCommandShellCommand{},
 			TextLogging:  methodEventTextLogging{},
 			SayHello:     methodEventSayHello{},
+			ErrorLog:     methodEventErrorLog{},
 		},
 	}
 
@@ -39,6 +40,8 @@ const (
 	TextLogging Method = "TextLogging"
 	// Send Hello I'm here message
 	SayHello Method = "SayHello"
+	// Error log methods to centralError
+	ErrorLog Method = "ErrorLog"
 )
 
 type MethodsAvailable struct {
@@ -120,3 +123,14 @@ func (m methodEventSayHello) handler(s *server, message Message, node string) ([
 	outMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
 	return outMsg, nil
 }
+
+// ---
+
+type methodEventErrorLog struct{}
+
+func (m methodEventErrorLog) handler(s *server, message Message, node string) ([]byte, error) {
+	log.Printf("----------------------------------------------------------------------------..\n")
+	log.Printf("Received error from: %v, containing: %v", message.FromNode, message.Data)
+	log.Printf("----------------------------------------------------------------------------..\n")
+	return nil, nil
+}
diff --git a/subscriber.go b/subscribers.go
similarity index 82%
rename from subscriber.go
rename to subscribers.go
index b34e911..d3472ce 100644
--- a/subscriber.go
+++ b/subscribers.go
@@ -50,4 +50,15 @@ func (s *server) subscribersStart() {
 		// fmt.Printf("*** %#v\n", proc)
 		go s.spawnWorkerProcess(proc)
 	}
+
+	if s.centralErrorLogger {
+		// Start a subscriber for ErrorLog messages
+		{
+			fmt.Printf("Starting ErrorLog subscriber: %#v\n", s.nodeName)
+			sub := newSubject(ErrorLog, EventNACK, "errorCentral")
+			proc := s.processPrepareNew(sub, s.errorKernel.errorCh, processKindSubscriber)
+			// fmt.Printf("*** %#v\n", proc)
+			go s.spawnWorkerProcess(proc)
+		}
+	}
 }