mirrors/ctrl
1
0
Fork 0
mirror of https://github.com/postmannen/ctrl.git synced 2025-03-31 01:24:31 +00:00
Code Activity

cleaning up restructuring

Browse source
This commit is contained in:
postmannen 2021-02-24 10:58:02 +01:00
parent af6a282b6a
commit f812c4307e
12 changed files with 289 additions and 254 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
errorkernel.go
View file

@ -19,12 +19,15 @@ type errorKernel struct {
// of error-state which is a map of all the processes,
// how many times a process have failed over the same
// message etc...
// errorCh is used to report errors from a process
errorCh chan errProcess
}
// newErrorKernel will initialize and return a new error kernel
func newErrorKernel() *errorKernel {
return &errorKernel{
// ringBuffer: newringBuffer(),
errorCh: make(chan errProcess, 2),
}
}
@ -37,13 +40,13 @@ 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(errorCh chan errProcess) {
func (e *errorKernel) startErrorKernel() {
// TODO: For now it will just print the error messages to the
// console.
go func() {
for {
er := <-errorCh
er := <-e.errorCh
// We should be able to handle each error individually and
// also concurrently, so the handler is started in it's

0
example/toCentral.json → example/toCentral-textlogging.json
View file

BIN
incommmingBuffer.db
View file

Binary file not shown.

15
prometheus.go
View file

@ -1,7 +1,6 @@
package steward
import (
"fmt"
"log"
"net"
"net/http"
@ -13,25 +12,18 @@ import (
// metrics are generally used to hold the structure around metrics
// handling
type metrics struct {
// sayHelloNodes are the register where the register where nodes
// who have sent an sayHello are stored. Since the sayHello
// subscriber is a handler that will be just be called when a
// hello message is received we need to store the metrics somewhere
// else, that is why we store it here....at least for now.
sayHelloNodes map[node]struct{}
// The channel to pass metrics that should be processed
metricsCh chan metricType
// host and port where prometheus metrics will be exported
hostAndPort string
}
// HERE:
// newMetrics will prepare and return a *metrics
func newMetrics(hostAndPort string) *metrics {
m := metrics{
sayHelloNodes: make(map[node]struct{}),
metricsCh: make(chan metricType),
hostAndPort: hostAndPort,
}
m.metricsCh = make(chan metricType)
m.hostAndPort = hostAndPort
return &m
}
@ -76,7 +68,6 @@ func (s *server) startMetrics() {
go func() {
for {
for f := range s.metrics.metricsCh {
fmt.Printf("********** RANGED A METRIC = %v, %v\n", f.metric, f.value)
// // Try to register the metric of the interface type prometheus.Collector
// prometheus.Register(f.metric)

124
publisher.go Normal file
View file

@ -0,0 +1,124 @@
package steward
import (
"log"
"time"
)
// processNewMessages takes a database name and an input channel as
// it's input arguments.
// The database will be used as the persistent store for the work queue
// which is implemented as a ring buffer.
// The input channel are where we read new messages to publish.
// Incomming messages will be routed to the correct subject process, where
// the handling of each nats subject is handled within it's own separate
// worker process.
// It will also handle the process of spawning more worker processes
// for publisher subjects if it does not exist.
func (s *server) processNewMessages(dbFileName string, newSAM chan []subjectAndMessage) {
// Prepare and start a new ring buffer
const bufferSize int = 1000
rb := newringBuffer(bufferSize, dbFileName)
inCh := make(chan subjectAndMessage)
ringBufferOutCh := make(chan samDBValue)
// start the ringbuffer.
rb.start(inCh, ringBufferOutCh)
// Start reading new fresh messages received on the incomming message
// pipe/file requested, and fill them into the buffer.
go func() {
for samSlice := range newSAM {
for _, sam := range samSlice {
inCh <- sam
}
}
close(inCh)
}()
// Process the messages that are in the ring buffer. Check and
// send if there are a specific subject for it, and no subject
// exist throw an error.
go func() {
for samTmp := range ringBufferOutCh {
sam := samTmp.Data
// Check if the format of the message is correct.
// TODO: Send a message to the error kernel here that
// it was unable to process the message with the reason
// why ?
if _, ok := s.methodsAvailable.CheckIfExists(sam.Message.Method); !ok {
continue
}
if !s.commandOrEventAvailable.CheckIfExists(sam.Message.CommandOrEvent) {
continue
}
redo:
// Adding a label here so we are able to redo the sending
// of the last message if a process with specified subject
// is not present. The process will then be created, and
// the code will loop back to the redo: label.
m := sam.Message
subjName := sam.Subject.name()
// DEBUG: fmt.Printf("** handleNewOperatorMessages: message: %v, ** subject: %#v\n", m, sam.Subject)
_, ok := s.processes[subjName]
// Are there already a process for that subject, put the
// message on that processes incomming message channel.
if ok {
log.Printf("info: found the specific subject: %v\n", subjName)
s.processes[subjName].subject.messageCh <- m
// 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
// 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)
sub := newSubject(sam.Subject.Method, sam.Subject.CommandOrEvent, sam.Subject.ToNode)
proc := s.processPrepareNew(sub, s.errorKernel.errorCh, processKindPublisher)
// fmt.Printf("*** %#v\n", proc)
go s.spawnWorkerProcess(proc)
time.Sleep(time.Millisecond * 500)
s.printProcessesMap()
// Now when the process is spawned we jump back to the redo: label,
// and send the message to that new process.
goto redo
}
}
}()
}
func (s *server) publishMessages(proc process) {
for {
// 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)
m.done <- struct{}{}
// Increment the counter for the next message to be sent.
proc.messageID++
s.processes[proc.subject.name()] = proc
time.Sleep(time.Second * 1)
// NB: simulate that we get an error, and that we can send that
// out of the process and receive it in another thread.
ep := errProcess{
infoText: "process failed",
process: proc,
message: m,
errorActionCh: make(chan errorAction),
}
s.errorKernel.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:
log.Printf("The errAction was continue...so we're continuing\n")
}
}
}

10
ringbuffer.go
View file

@ -40,8 +40,8 @@ type ringBuffer struct {
}
// newringBuffer is a push/pop storage for values.
func newringBuffer(size int) *ringBuffer {
db, err := bolt.Open("./incommmingBuffer.db", 0600, nil)
func newringBuffer(size int, dbFileName string) *ringBuffer {
db, err := bolt.Open(dbFileName, 0600, nil)
if err != nil {
log.Printf("error: failed to open db: %v\n", err)
}
@ -186,14 +186,12 @@ func (r *ringBuffer) processBufferMessages(samValueBucket string, outCh chan sam
// 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.
<-v.Data.done
fmt.Println("-----------------------------------------------------------")
fmt.Printf("### DONE WITH THE MESSAGE\n")
fmt.Println("-----------------------------------------------------------")
log.Printf("info: done with message %v\n", v.ID)
// Since we are now done with the specific message we can delete
// it out of the K/V Store.
r.deleteKeyFromBucket(samValueBucket, strconv.Itoa(v.ID))
fmt.Printf("******* DELETED KEY %v FROM BUCKET*******\n", v.ID)
log.Printf("info: deleting key %v from bucket\n", v.ID)
r.permStore <- fmt.Sprintf("%v : %+v\n", time.Now().UTC(), v)

210
server.go
View file

@ -43,17 +43,15 @@ type server struct {
lastProcessID int
// The name of the node
nodeName string
mu sync.Mutex
// The channel where we receive new messages from the outside to
// insert into the system for being processed
// Mutex for locking when writing to the process map
mu sync.Mutex
// The channel where we receive new messages read from file.
// We can than range this channel for new messages to process.
inputFromFileCh chan []subjectAndMessage
// errorCh is used to report errors from a process
// NB: Implementing this as an int to report for testing
errorCh chan errProcess
// errorKernel
// 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.
errorKernel *errorKernel
// TODO: replace this with some structure to hold the logCh value
logCh chan []byte
// used to check if the methods specified in message is valid
methodsAvailable MethodsAvailable
// Map who holds the command and event types available.
@ -61,6 +59,12 @@ type server struct {
commandOrEventAvailable CommandOrEventAvailable
// metric exporter
metrics *metrics
// subscriberServices are where we find the services and the API to
// use services needed by subscriber.
// For example, this can be a service that knows
// how to forward the data for a received message of type log to a
// central logger.
subscriberServices *subscriberServices
}
// newServer will prepare and return a server type
@ -71,33 +75,32 @@ func NewServer(brokerAddress string, nodeName string, promHostAndPort string) (*
}
var m Method
var co CommandOrEvent
var coe CommandOrEvent
s := &server{
nodeName: nodeName,
natsConn: conn,
processes: make(map[subjectName]process),
inputFromFileCh: make(chan []subjectAndMessage),
errorCh: make(chan errProcess, 2),
logCh: make(chan []byte),
methodsAvailable: m.GetMethodsAvailable(),
commandOrEventAvailable: co.GetCommandOrEventAvailable(),
commandOrEventAvailable: coe.GetCommandOrEventAvailable(),
metrics: newMetrics(promHostAndPort),
subscriberServices: newSubscriberServices(),
}
return s, nil
}
// Start will spawn up all the defined subscriber processes.
// Start will spawn up all the predefined subscriber processes.
// Spawning of publisher processes is done on the fly by checking
// if there is publisher process for a given message subject. This
// checking is also started here in Start by calling handleMessagesToPublish.
// if there is publisher process for a given message subject, and
// not exist it will spawn one.
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.errorCh)
s.errorKernel.startErrorKernel()
// Start collecting the metrics
go s.startMetrics()
@ -106,40 +109,20 @@ func (s *server) Start() {
go s.getMessagesFromFile("./", "inmsg.txt", s.inputFromFileCh)
// Start the textLogging service that will run on the subscribers
// TODO: Figure out how to structure event services like these
go s.startTextLogging(s.logCh)
// TODO: This should only be started if the flag value provided when
// starting asks to subscribe to TextLogging events.
go s.subscriberServices.startTextLogging()
// Start a subscriber for shellCommand messages
{
fmt.Printf("Starting shellCommand subscriber: %#v\n", s.nodeName)
sub := newSubject(ShellCommand, CommandACK, s.nodeName)
proc := s.processPrepareNew(sub, s.errorCh, processKindSubscriber)
// fmt.Printf("*** %#v\n", proc)
go s.processSpawnWorker(proc)
}
// Start a subscriber for textLogging messages
{
fmt.Printf("Starting textlogging subscriber: %#v\n", s.nodeName)
sub := newSubject(TextLogging, EventACK, s.nodeName)
proc := s.processPrepareNew(sub, s.errorCh, processKindSubscriber)
// fmt.Printf("*** %#v\n", proc)
go s.processSpawnWorker(proc)
}
// Start a subscriber for SayHello messages
{
fmt.Printf("Starting SayHello subscriber: %#v\n", s.nodeName)
sub := newSubject(SayHello, EventNACK, s.nodeName)
proc := s.processPrepareNew(sub, s.errorCh, processKindSubscriber)
// fmt.Printf("*** %#v\n", proc)
go s.processSpawnWorker(proc)
}
// Start up the predefined subscribers.
// TODO: What to subscribe on should be handled via flags, or config
// files.
s.subscribersStart()
time.Sleep(time.Second * 2)
s.printProcessesMap()
s.handleMessagesInRingbuffer()
// Start the processing of new messaging from an input channel.
s.processNewMessages("./incommmingBuffer.db", s.inputFromFileCh)
select {}
@ -163,86 +146,6 @@ func (s *server) printProcessesMap() {
fmt.Println("--------------------------------------------------------------------------------------------")
}
// handleNewOperatorMessages will handle all the new operator messages
// given to the system, and route them to the correct subject queue.
// It will also handle the process of spawning more worker processes
// for publisher subjects if it does not exist.
func (s *server) handleMessagesInRingbuffer() {
// Prepare and start a new ring buffer
const bufferSize int = 1000
rb := newringBuffer(bufferSize)
inCh := make(chan subjectAndMessage)
ringBufferOutCh := make(chan samDBValue)
// start the ringbuffer.
rb.start(inCh, ringBufferOutCh)
// Start reading new messages received on the incomming message
// pipe requested by operator, and fill them into the buffer.
go func() {
for samSlice := range s.inputFromFileCh {
for _, sam := range samSlice {
inCh <- sam
}
}
close(inCh)
}()
// Process the messages that are in the ring buffer. Check and
// send if there are a specific subject for it, and no subject
// exist throw an error.
go func() {
for samTmp := range ringBufferOutCh {
sam := samTmp.Data
// Check if the format of the message is correct.
// TODO: Send a message to the error kernel here that
// it was unable to process the message with the reason
// why ?
if _, ok := s.methodsAvailable.CheckIfExists(sam.Message.Method); !ok {
continue
}
if !s.commandOrEventAvailable.CheckIfExists(sam.Message.CommandOrEvent) {
continue
}
redo:
// Adding a label here so we are able to redo the sending
// of the last message if a process with specified subject
// is not present. The process will then be created, and
// the code will loop back to the redo: label.
m := sam.Message
subjName := sam.Subject.name()
// DEBUG: fmt.Printf("** handleNewOperatorMessages: message: %v, ** subject: %#v\n", m, sam.Subject)
_, ok := s.processes[subjName]
// Are there already a process for that subject, put the
// message on that processes incomming message channel.
if ok {
log.Printf("info: found the specific subject: %v\n", subjName)
s.processes[subjName].subject.messageCh <- m
// 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
// 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)
sub := newSubject(sam.Subject.Method, sam.Subject.CommandOrEvent, sam.Subject.ToNode)
proc := s.processPrepareNew(sub, s.errorCh, processKindPublisher)
// fmt.Printf("*** %#v\n", proc)
go s.processSpawnWorker(proc)
time.Sleep(time.Millisecond * 500)
s.printProcessesMap()
// Now when the process is spawned we jump back to the redo: label,
// and send the message to that new process.
goto redo
}
}
}()
}
type node string
// subject contains the representation of a subject to be used with one
@ -323,10 +226,11 @@ func (s *server) processPrepareNew(subject Subject, errCh chan errProcess, proce
return proc
}
// processSpawnWorker 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) processSpawnWorker(proc process) {
// spawnWorkerProcess will spawn take care of spawning both publisher
// and subscriber proesses.
//It will give the process the next available ID, and also add the
// process to the processes map.
func (s *server) spawnWorkerProcess(proc process) {
s.mu.Lock()
// We use the full name of the subject to identify a unique
// process. We can do that since a process can only handle
@ -342,52 +246,12 @@ func (s *server) processSpawnWorker(proc process) {
//
// Handle publisher workers
if proc.processKind == processKindPublisher {
for {
// 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)
m.done <- struct{}{}
// Increment the counter for the next message to be sent.
proc.messageID++
s.processes[proc.subject.name()] = proc
time.Sleep(time.Second * 1)
// NB: simulate that we get an error, and that we can send that
// out of the process and receive it in another thread.
ep := errProcess{
infoText: "process failed",
process: proc,
message: m,
errorActionCh: make(chan errorAction),
}
s.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:
log.Printf("The errAction was continue...so we're continuing\n")
}
}
s.publishMessages(proc)
}
// handle subscriber workers
if proc.processKind == processKindSubscriber {
subject := string(proc.subject.name())
// Subscribe will start up a Go routine under the hood calling the
// callback function specified when a new message is received.
_, err := s.natsConn.Subscribe(subject, func(msg *nats.Msg) {
// 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 s.subscriberHandler(s.natsConn, s.nodeName, msg)
})
if err != nil {
log.Printf("error: Subscribe failed: %v\n", err)
}
s.subscribeMessages(proc)
}
}
@ -437,7 +301,7 @@ func messageDeliver(proc process, message Message, natsConn *nats.Conn) {
// did not receive a reply, continuing from top again
continue
}
log.Printf("publisher: received ACK: %s\n", msgReply.Data)
log.Printf("info: publisher: received ACK for message: %s\n", msgReply.Data)
}
return
}

58
subscriber-services.go Normal file
View file

@ -0,0 +1,58 @@
package steward
import (
"fmt"
"log"
"os"
)
// subscriberServices will hold all the helper services needed for
// the different subcribers. Example of a help service can be a log
// subscriber needs a way to write logs locally or send them to some
// other central logging system.
type subscriberServices struct {
// Where we should put the data to write to a log file
logCh chan []byte
// sayHelloNodes are the register where the register where nodes
// who have sent an sayHello are stored. Since the sayHello
// subscriber is a handler that will be just be called when a
// hello message is received we need to store the metrics somewhere
// else, that is why we store it here....at least for now.
sayHelloNodes map[node]struct{}
}
//newSubscriberServices will prepare and return a *subscriberServices
func newSubscriberServices() *subscriberServices {
s := subscriberServices{
logCh: make(chan []byte),
sayHelloNodes: make(map[node]struct{}),
}
return &s
}
// ---
// startTextLogging will open a file ready for writing log messages to,
// and the input for writing to the file is given via the logCh argument.
func (s *subscriberServices) startTextLogging() {
fileName := "./textlogging.log"
f, err := os.OpenFile(fileName, os.O_APPEND|os.O_RDWR|os.O_CREATE, os.ModeAppend)
if err != nil {
log.Printf("Failed to open file %v\n", err)
return
}
defer f.Close()
for b := range s.logCh {
fmt.Printf("***** Trying to write to file : %s\n\n", b)
_, err := f.Write(b)
f.Sync()
if err != nil {
log.Printf("Failed to open file %v\n", err)
}
}
}

53
subscriber.go Normal file
View file

@ -0,0 +1,53 @@
package steward
import (
"fmt"
"log"
"github.com/nats-io/nats.go"
)
func (s *server) subscribeMessages(proc process) {
subject := string(proc.subject.name())
// Subscribe will start up a Go routine under the hood calling the
// callback function specified when a new message is received.
_, err := s.natsConn.Subscribe(subject, func(msg *nats.Msg) {
// 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 s.subscriberHandler(s.natsConn, s.nodeName, msg)
})
if err != nil {
log.Printf("error: Subscribe failed: %v\n", err)
}
}
func (s *server) subscribersStart() {
// Start a subscriber for shellCommand messages
{
fmt.Printf("Starting shellCommand subscriber: %#v\n", s.nodeName)
sub := newSubject(ShellCommand, CommandACK, s.nodeName)
proc := s.processPrepareNew(sub, s.errorKernel.errorCh, processKindSubscriber)
// fmt.Printf("*** %#v\n", proc)
go s.spawnWorkerProcess(proc)
}
// Start a subscriber for textLogging messages
{
fmt.Printf("Starting textlogging subscriber: %#v\n", s.nodeName)
sub := newSubject(TextLogging, EventACK, s.nodeName)
proc := s.processPrepareNew(sub, s.errorKernel.errorCh, processKindSubscriber)
// fmt.Printf("*** %#v\n", proc)
go s.spawnWorkerProcess(proc)
}
// Start a subscriber for SayHello messages
{
fmt.Printf("Starting SayHello subscriber: %#v\n", s.nodeName)
sub := newSubject(SayHello, EventNACK, s.nodeName)
proc := s.processPrepareNew(sub, s.errorKernel.errorCh, processKindSubscriber)
// fmt.Printf("*** %#v\n", proc)
go s.spawnWorkerProcess(proc)
}
}

6
subscriberMethodTypes.go
View file

@ -92,7 +92,7 @@ type methodEventTextLogging struct{}
func (m methodEventTextLogging) handler(s *server, message Message, node string) ([]byte, error) {
for _, d := range message.Data {
s.logCh <- []byte(d)
s.subscriberServices.logCh <- []byte(d)
}
outMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
@ -107,7 +107,7 @@ func (m methodEventSayHello) handler(s *server, message Message, node string) ([
log.Printf("################## Received hello from %v ##################\n", message.FromNode)
// Since the handler is only called to handle a specific type of message we need
// to store it elsewhere, and choice for now is under s.metrics.sayHelloNodes
s.metrics.sayHelloNodes[message.FromNode] = struct{}{}
s.subscriberServices.sayHelloNodes[message.FromNode] = struct{}{}
// update the prometheus metrics
s.metrics.metricsCh <- metricType{
@ -115,7 +115,7 @@ func (m methodEventSayHello) handler(s *server, message Message, node string) ([
Name: "hello_nodes",
Help: "The current number of total nodes who have said hello",
}),
value: float64(len(s.metrics.sayHelloNodes)),
value: float64(len(s.subscriberServices.sayHelloNodes)),
}
outMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
return outMsg, nil

29
textlogging.go
View file

@ -1,30 +1 @@
package steward
import (
"fmt"
"log"
"os"
)
// startTextLogging will open a file ready for writing log messages to,
// and the input for writing to the file is given via the logCh argument.
func (s *server) startTextLogging(logCh chan []byte) {
fileName := "./textlogging.log"
f, err := os.OpenFile(fileName, os.O_APPEND|os.O_RDWR|os.O_CREATE, os.ModeAppend)
if err != nil {
log.Printf("Failed to open file %v\n", err)
return
}
defer f.Close()
for b := range logCh {
fmt.Printf("***** Trying to write to file : %s\n\n", b)
_, err := f.Write(b)
f.Sync()
if err != nil {
log.Printf("Failed to open file %v\n", err)
}
}
}

29
textlogging.log
View file

@ -1,31 +1,4 @@
some message sent from a shipsome message sent from a shipsome message sent from a shipsome message sent from a shipsome message sent from a shipsome message sent from a shipsome message sent from a ship
some message sent from a ship
some message sent from a ship
some message sent from a ship
some message sent from a ship
some message sent from a ship
some message sent from a ship
some message sent from a ship
some message sent from a ship
some message sent from a ship
some message sent from a ship
some message sent from a ship for testing
some message sent from a ship for testing
some message sent from a ship for testing
some message sent from a ship for testing
some message sent from a ship for testing
some message sent from a ship for testing
some message sent from a ship for testing
some message sent from a ship for testing
some message sent from a ship for testing
some message sent from a ship for testing
some message sent from a ship for testing
some message sent from a ship for testing
some message sent from a ship for testing
some message sent from a ship for testing
some message sent from a ship for testing
some message sent from a ship for testing
some message sent from a ship for testing
some message sent from a ship for testing
some message sent from a ship for testing
some message sent from a ship for testing