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

513 lines
17 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 (
"context"
2021-01-25 14:23:00 +00:00
"fmt"
"log"
2021-03-30 08:37:16 +00:00
"net"
2021-08-23 10:47:33 +00:00
"net/http"
"os"
2021-05-12 07:50:03 +00:00
"path/filepath"
2021-01-25 14:23:00 +00:00
"time"
"github.com/nats-io/nats.go"
"github.com/prometheus/client_golang/prometheus"
2021-01-25 14:23:00 +00:00
)
2021-02-26 06:55:28 +00:00
type processName string
// Will return a process name made up of subjectName+processKind
2021-02-26 06:55:28 +00:00
func processNameGet(sn subjectName, pk processKind) processName {
pn := fmt.Sprintf("%s_%s", sn, pk)
return processName(pn)
}
// 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 {
// The main background context
ctx context.Context
// The CancelFunc for the main context
2021-08-11 10:23:37 +00:00
cancel context.CancelFunc
// Configuration options used for running the server
configuration *Configuration
// The nats connection to the broker
2021-01-27 13:02:57 +00:00
natsConn *nats.Conn
2021-07-02 06:38:44 +00:00
// net listener for communicating via the steward socket
2021-08-09 12:41:31 +00:00
StewardSocket net.Listener
2021-07-02 06:38:44 +00:00
// net listener for the communication with Stew
2021-08-09 12:41:31 +00:00
StewSocket net.Listener
// processes holds all the information about running processes
processes *processes
// The name of the node
nodeName string
// newMessagesCh are the channel where new messages to be handled
2021-11-09 12:18:58 +00:00
// by the system are put. So if any process want's to send a message
// like an error message you just put the message on the newMessagesCh.
//
// In general the ringbuffer will read this
// channel for messages to put on the buffer.
//
// Example:
// A message is read from the socket, then put on the newMessagesCh
// and then put on the ringbuffer.
newMessagesCh chan []subjectAndMessage
2021-02-24 09:58:02 +00:00
// errorKernel is doing all the error handling like what to do if
// an error occurs.
errorKernel *errorKernel
2021-09-15 06:39:34 +00:00
// Ring buffer
ringBuffer *ringBuffer
2021-02-18 11:29:14 +00:00
// metric exporter
metrics *metrics
// Version of package
version string
2021-01-27 13:02:57 +00:00
}
// newServer will prepare and return a server type
func NewServer(c *Configuration, version string) (*server, error) {
2021-08-11 10:23:37 +00:00
// Set up the main background context.
ctx, cancel := context.WithCancel(context.Background())
2021-04-19 19:06:37 +00:00
var opt nats.Option
2021-08-16 11:01:12 +00:00
2021-04-19 19:06:37 +00:00
if c.RootCAPath != "" {
opt = nats.RootCAs(c.RootCAPath)
}
2021-05-20 10:27:25 +00:00
if c.NkeySeedFile != "" {
var err error
2021-08-11 10:23:37 +00:00
2021-05-20 10:27:25 +00:00
opt, err = nats.NkeyOptionFromSeed(c.NkeySeedFile)
if err != nil {
cancel()
2021-05-20 10:27:25 +00:00
return nil, fmt.Errorf("error: failed to read nkey seed file: %v", err)
}
}
var conn *nats.Conn
2021-08-16 11:01:12 +00:00
// Connect to the nats server, and retry until succesful.
for {
var err error
// Setting MaxReconnects to -1 which equals unlimited.
conn, err = nats.Connect(c.BrokerAddress, opt, nats.MaxReconnects(-1))
// If no servers where available, we loop and retry until succesful.
if err != nil {
log.Printf("error: could not connect, waiting %v seconds, and retrying: %v\n", c.NatsConnectRetryInterval, err)
time.Sleep(time.Duration(time.Second * time.Duration(c.NatsConnectRetryInterval)))
continue
}
2021-02-01 10:13:38 +00:00
break
}
2021-08-11 10:23:37 +00:00
2021-07-02 06:38:44 +00:00
// Prepare the connection to the Steward socket file
2021-05-12 07:50:03 +00:00
// Check if socket folder exists, if not create it
if _, err := os.Stat(c.SocketFolder); os.IsNotExist(err) {
err := os.MkdirAll(c.SocketFolder, 0700)
if err != nil {
cancel()
2021-05-20 10:27:25 +00:00
return nil, fmt.Errorf("error: failed to create socket folder directory %v: %v", c.SocketFolder, err)
2021-05-12 07:50:03 +00:00
}
}
2021-08-11 10:23:37 +00:00
// Just as an extra check we eventually delete any existing Steward socket files if found.
2021-05-12 07:50:03 +00:00
socketFilepath := filepath.Join(c.SocketFolder, "steward.sock")
if _, err := os.Stat(socketFilepath); !os.IsNotExist(err) {
err = os.Remove(socketFilepath)
if err != nil {
er := fmt.Errorf("error: could not delete sock file: %v", err)
cancel()
2021-05-12 07:50:03 +00:00
return nil, er
}
2021-03-30 08:37:16 +00:00
}
2021-08-11 10:23:37 +00:00
// Open the socket.
2021-05-12 07:50:03 +00:00
nl, err := net.Listen("unix", socketFilepath)
2021-03-30 08:37:16 +00:00
if err != nil {
2021-04-16 11:18:10 +00:00
er := fmt.Errorf("error: failed to open socket: %v", err)
cancel()
2021-04-16 11:18:10 +00:00
return nil, er
2021-03-30 08:37:16 +00:00
}
2021-07-02 06:38:44 +00:00
// ---
// Prepare the connection to the Stew socket file
// Check if socket folder exists, if not create it
if _, err := os.Stat(c.SocketFolder); os.IsNotExist(err) {
err := os.MkdirAll(c.SocketFolder, 0700)
if err != nil {
cancel()
2021-07-02 06:38:44 +00:00
return nil, fmt.Errorf("error: failed to create socket folder directory %v: %v", c.SocketFolder, err)
}
}
stewSocketFilepath := filepath.Join(c.SocketFolder, "stew.sock")
2021-08-11 10:23:37 +00:00
// Just as an extra check we eventually delete any existing Stew socket files if found.
2021-07-02 06:38:44 +00:00
if _, err := os.Stat(stewSocketFilepath); !os.IsNotExist(err) {
err = os.Remove(stewSocketFilepath)
if err != nil {
er := fmt.Errorf("error: could not delete stew.sock file: %v", err)
cancel()
2021-07-02 06:38:44 +00:00
return nil, er
}
}
stewNL, err := net.Listen("unix", stewSocketFilepath)
if err != nil {
er := fmt.Errorf("error: failed to open stew socket: %v", err)
cancel()
2021-07-02 06:38:44 +00:00
return nil, er
}
// ---
metrics := newMetrics(c.PromHostAndPort)
s := &server{
ctx: ctx,
cancel: cancel,
configuration: c,
nodeName: c.NodeName,
natsConn: conn,
StewardSocket: nl,
StewSocket: stewNL,
processes: newProcesses(ctx, metrics),
newMessagesCh: make(chan []subjectAndMessage),
metrics: metrics,
version: version,
}
// Create the default data folder for where subscribers should
2021-03-25 11:50:58 +00:00
// write it's data, check if data folder exist, and create it if needed.
if _, err := os.Stat(c.SubscribersDataFolder); os.IsNotExist(err) {
if c.SubscribersDataFolder == "" {
return nil, fmt.Errorf("error: subscribersDataFolder value is empty, you need to provide the config or the flag value at startup %v: %v", c.SubscribersDataFolder, err)
}
err := os.Mkdir(c.SubscribersDataFolder, 0700)
if err != nil {
2021-05-20 10:27:25 +00:00
return nil, fmt.Errorf("error: failed to create data folder directory %v: %v", c.SubscribersDataFolder, err)
}
log.Printf("info: Creating subscribers data folder at %v\n", c.SubscribersDataFolder)
}
2021-02-05 06:25:12 +00:00
return s, nil
}
2021-02-24 09:58:02 +00:00
// Start will spawn up all the predefined subscriber processes.
2021-02-10 06:25:44 +00:00
// Spawning of publisher processes is done on the fly by checking
2021-02-24 09:58:02 +00:00
// if there is publisher process for a given message subject, and
2021-08-16 11:01:12 +00:00
// if it does not exist it will spawn one.
func (s *server) Start() {
2021-09-23 03:46:25 +00:00
log.Printf("Starting steward, version=%+v\n", s.version)
s.metrics.promVersion.With(prometheus.Labels{"version": string(s.version)})
// Start the error kernel that will do all the error handling
2021-08-03 11:57:29 +00:00
// that is not done within a process.
2021-08-04 08:37:24 +00:00
s.errorKernel = newErrorKernel(s.ctx)
2021-08-04 08:37:24 +00:00
go func() {
err := s.errorKernel.start(s.newMessagesCh)
2021-08-04 08:37:24 +00:00
if err != nil {
log.Printf("%v\n", err)
}
}()
2021-02-18 11:29:14 +00:00
// Start collecting the metrics
2021-08-03 11:57:29 +00:00
go func() {
2021-08-04 08:37:24 +00:00
err := s.metrics.start()
2021-08-03 11:57:29 +00:00
if err != nil {
log.Printf("%v\n", err)
os.Exit(1)
}
}()
2021-02-18 11:29:14 +00:00
// Start the checking the input socket for new messages from operator.
2021-08-25 08:16:55 +00:00
go s.readSocket()
2021-09-10 03:26:16 +00:00
// Check if we should start the tcp listener for new messages from operator.
if s.configuration.TCPListener != "" {
go s.readTCPListener()
}
2021-09-10 03:26:16 +00:00
// Check if we should start the http listener for new messages from operator.
if s.configuration.HTTPListener != "" {
go s.readHttpListener()
}
// Start up the predefined subscribers.
//
// Since all the logic to handle processes are tied to the process
// struct, we need to create an initial process to start the rest.
2021-08-11 10:23:37 +00:00
//
// NB: The context of the initial process are set in processes.Start.
2021-08-09 07:18:30 +00:00
sub := newSubject(REQInitial, s.nodeName)
p := newProcess(context.TODO(), s.metrics, s.natsConn, s.processes, s.newMessagesCh, s.configuration, sub, s.errorKernel.errorCh, "", nil)
// Start all wanted subscriber processes.
s.processes.Start(p)
2021-08-09 07:18:30 +00:00
time.Sleep(time.Second * 1)
s.processes.printProcessesMap()
2021-08-23 10:47:33 +00:00
// Start exposing the the data folder via HTTP if flag is set.
if s.configuration.ExposeDataFolder != "" {
log.Printf("info: Starting expose of data folder via HTTP\n")
go s.exposeDataFolder(s.ctx)
}
2021-08-09 07:18:30 +00:00
// Start the processing of new messages from an input channel.
2021-09-15 06:39:34 +00:00
// NB: We might need to create a sub context for the ringbuffer here
// so we can cancel this context last, and not use the server.
2021-08-25 06:50:24 +00:00
s.routeMessagesToProcess("./incomingBuffer.db")
2021-02-10 06:25:44 +00:00
2021-08-09 07:18:30 +00:00
}
2021-08-09 07:18:30 +00:00
// Will stop all processes started during startup.
func (s *server) Stop() {
// fmt.Printf(" * DEBUG100 * processMap \n")
// s.processes.printProcessesMap()
2021-09-13 05:02:14 +00:00
2021-08-09 07:18:30 +00:00
// Stop the started pub/sub message processes.
2021-08-11 10:23:37 +00:00
s.processes.Stop()
2021-08-09 07:18:30 +00:00
log.Printf("info: stopped all subscribers\n")
2021-07-02 06:38:44 +00:00
2021-08-09 07:18:30 +00:00
// Stop the errorKernel.
s.errorKernel.stop()
log.Printf("info: stopped the errorKernel\n")
2021-07-02 11:26:30 +00:00
2021-08-09 07:18:30 +00:00
// Stop the main context.
2021-08-11 10:23:37 +00:00
s.cancel()
2021-08-09 07:18:30 +00:00
log.Printf("info: stopped the main context\n")
2021-08-09 12:41:31 +00:00
// Delete the socket file when the program exits.
socketFilepath := filepath.Join(s.configuration.SocketFolder, "steward.sock")
if _, err := os.Stat(socketFilepath); !os.IsNotExist(err) {
err = os.Remove(socketFilepath)
if err != nil {
er := fmt.Errorf("error: could not delete sock file: %v", err)
log.Printf("%v\n", er)
}
}
}
2021-02-25 10:08:05 +00:00
// sendErrorMessage will put the error message directly on the channel that is
// read by the nats publishing functions.
func sendErrorLogMessage(conf *Configuration, metrics *metrics, newMessagesCh chan<- []subjectAndMessage, FromNode Node, theError error) {
2021-03-26 04:13:51 +00:00
// NB: Adding log statement here for more visuality during development.
log.Printf("%v\n", theError)
sam := createErrorMsgContent(conf, FromNode, theError)
2021-02-25 10:08:05 +00:00
newMessagesCh <- []subjectAndMessage{sam}
2021-08-26 10:26:08 +00:00
metrics.promErrorMessagesSentTotal.Inc()
2021-02-25 10:08:05 +00:00
}
2021-09-23 06:31:30 +00:00
// sendInfoMessage will put the error message directly on the channel that is
// read by the nats publishing functions.
func sendInfoLogMessage(conf *Configuration, metrics *metrics, newMessagesCh chan<- []subjectAndMessage, FromNode Node, theError error) {
// NB: Adding log statement here for more visuality during development.
log.Printf("%v\n", theError)
sam := createErrorMsgContent(conf, FromNode, theError)
newMessagesCh <- []subjectAndMessage{sam}
metrics.promInfoMessagesSentTotal.Inc()
}
2021-02-25 10:08:05 +00:00
// createErrorMsgContent will prepare a subject and message with the content
// of the error
func createErrorMsgContent(conf *Configuration, FromNode Node, theError error) subjectAndMessage {
// Add time stamp
2021-09-23 06:19:53 +00:00
er := fmt.Sprintf("%v, node: %v, %v\n", time.Now().Format("Mon Jan _2 15:04:05 2006"), FromNode, theError.Error())
sam := subjectAndMessage{
2021-04-06 05:56:49 +00:00
Subject: newSubject(REQErrorLog, "errorCentral"),
Message: Message{
Directory: "errorLog",
ToNode: "errorCentral",
FromNode: FromNode,
FileName: "error.log",
Data: []string{er},
Method: REQErrorLog,
ACKTimeout: conf.ErrorMessageTimeout,
Retries: conf.ErrorMessageRetries,
},
}
2021-02-25 10:08:05 +00:00
return sam
}
2021-03-09 06:43:55 +00:00
// Contains the sam value as it is used in the state DB, and also a
// delivered function to be called when this message is picked up, so
// we can control if messages gets stale at some point.
type samDBValueAndDelivered struct {
2021-07-02 16:32:01 +00:00
samDBValue samDBValue
delivered func()
2021-07-02 16:32:01 +00:00
}
2021-08-25 06:50:24 +00:00
// routeMessagesToProcess takes a database name it's input argument.
// The database will be used as the persistent k/v store for the work
// queue which is implemented as a ring buffer.
// The newMessagesCh are where we get new messages to publish.
2021-03-09 06:43:55 +00:00
// 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.
2021-08-25 06:50:24 +00:00
func (s *server) routeMessagesToProcess(dbFileName string) {
2021-03-09 06:43:55 +00:00
// Prepare and start a new ring buffer
const bufferSize int = 1000
const samValueBucket string = "samValueBucket"
const indexValueBucket string = "indexValueBucket"
2021-09-15 07:17:35 +00:00
s.ringBuffer = newringBuffer(s.ctx, s.metrics, s.configuration, bufferSize, dbFileName, Node(s.nodeName), s.newMessagesCh, samValueBucket, indexValueBucket)
2021-08-26 05:02:36 +00:00
ringBufferInCh := make(chan subjectAndMessage)
ringBufferOutCh := make(chan samDBValueAndDelivered)
2021-03-09 06:43:55 +00:00
// start the ringbuffer.
2021-09-15 08:46:30 +00:00
s.ringBuffer.start(s.ctx, ringBufferInCh, ringBufferOutCh, s.configuration.DefaultMessageTimeout, s.configuration.DefaultMessageRetries)
2021-03-09 06:43:55 +00:00
// Start reading new fresh messages received on the incomming message
// pipe/file requested, and fill them into the buffer.
go func() {
2021-08-25 08:16:55 +00:00
for sams := range s.newMessagesCh {
for _, sam := range sams {
ringBufferInCh <- sam
2021-03-09 06:43:55 +00:00
}
}
close(ringBufferInCh)
2021-03-09 06:43:55 +00:00
}()
// Process the messages that are in the ring buffer. Check and
// send if there are a specific subject for it, and if no subject
// exist throw an error.
var coe CommandOrEvent
coeAvailable := coe.GetCommandOrEventAvailable()
var method Method
methodsAvailable := method.GetMethodsAvailable()
go func() {
2021-11-09 12:18:58 +00:00
for samDBVal := range ringBufferOutCh {
// fmt.Printf(" * DEBUG1.1 * before signaling back to the ringbuffer that message was picked from ring buffer, samTmp.delivered: %#v\n", samTmp.samDBValue.ID)
// Signal back to the ringbuffer that message have been picked up.
2021-11-09 12:18:58 +00:00
samDBVal.delivered()
2021-09-13 05:02:14 +00:00
// fmt.Printf(" * DEBUG1.2 * after signaling back to the ringbuffer that message was picked from ring buffer, samTmp.delivered: %#v\n", samTmp.samDBValue.ID)
2021-11-09 12:18:58 +00:00
sam := samDBVal.samDBValue.Data
2021-03-09 06:43:55 +00:00
// Check if the format of the message is correct.
if _, ok := methodsAvailable.CheckIfExists(sam.Message.Method); !ok {
er := fmt.Errorf("error: routeMessagesToProcess: the method do not exist, message dropped: %v", sam.Message.Method)
sendErrorLogMessage(s.configuration, s.metrics, s.newMessagesCh, Node(s.nodeName), er)
2021-03-09 06:43:55 +00:00
continue
}
if !coeAvailable.CheckIfExists(sam.Subject.CommandOrEvent, sam.Subject) {
er := fmt.Errorf("error: routeMessagesToProcess: the command or event do not exist, message dropped: %v", sam.Message.Method)
sendErrorLogMessage(s.configuration, s.metrics, s.newMessagesCh, Node(s.nodeName), er)
2021-03-09 06:43:55 +00:00
continue
}
2021-09-21 21:29:42 +00:00
for {
// looping 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 here.
m := sam.Message
subjName := sam.Subject.name()
// DEBUG: fmt.Printf("** handleNewOperatorMessages: message: %v, ** subject: %#v\n", m, sam.Subject)
pn := processNameGet(subjName, processKindPublisher)
// Check if there is a map of type map[int]process registered
// for the processName, and if it exists then return it.
2021-10-08 10:07:10 +00:00
kv := s.processes.active.get(pn)
existingProcIDMap := kv.v
valueOK := kv.ok
2021-09-21 21:29:42 +00:00
// If found a map above, range it, and are there already a process
// for that subject, put the message on that processes incomming
// message channel.
2021-10-08 10:07:10 +00:00
if valueOK {
2021-09-21 21:29:42 +00:00
// fmt.Printf(" * DEBUG1.3 * MUTEX.LOCK before range existingProcIDMap, samDBValue.id: %#v, existingProcIDMap length: %v\n", samTmp.samDBValue.ID, len(existingProcIDMap))
2021-10-08 20:39:46 +00:00
2021-09-21 21:29:42 +00:00
// var pid int
var proc process
// forLoopCounter := 1
for _, existingProc := range existingProcIDMap {
// fmt.Printf(" * DEBUG1.3 * forLoopCounter: %v\n", forLoopCounter)
// pid = existingProc.processID
// log.Printf("info: processNewMessages: found the specific subject: %v, proc.ID: %v\n", subjName, existingProc.processID)
// fmt.Printf(" * DEBUG1.4 * before putting on channel to found process, process ch: %#v,existingproc.id: %v\n", &existingProc.subject.messageCh, existingProc.processID)
proc = existingProc
// fmt.Printf(" * DEBUG1.5 * after putting on channel to found process, process ch: %#v,existingproc.id: %v\n", &existingProc.subject.messageCh, existingProc.processID)
// forLoopCounter++
}
// We have found the process to route the message to, deliver it.
proc.subject.messageCh <- m
// fmt.Printf(" *** DEBUG1.6 * MUTEX.UNLOCK after range existing Proc ID Map, samDBValue.id: %#v, proc.id: %v\n", samTmp.samDBValue.ID, pid)
// If no process to handle the specific subject exist,
// the we create and spawn one.
break
} else {
// 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: processNewMessages: did not find that specific subject, starting new process for subject: %v\n", subjName)
sub := newSubject(sam.Subject.Method, sam.Subject.ToNode)
proc := newProcess(s.ctx, s.metrics, s.natsConn, s.processes, s.newMessagesCh, s.configuration, sub, s.errorKernel.errorCh, processKindPublisher, nil)
// fmt.Printf("*** %#v\n", proc)
proc.spawnWorker(s.processes, s.natsConn)
log.Printf("info: processNewMessages: new process started, subject: %v, processID: %v\n", subjName, proc.processID)
// Now when the process is spawned we jump back to the redo: label,
// and send the message to that new process.
continue
}
2021-03-09 06:43:55 +00:00
}
}
}()
}
2021-08-23 10:47:33 +00:00
func (s *server) exposeDataFolder(ctx context.Context) {
2021-08-23 15:05:56 +00:00
fileHandler := func(w http.ResponseWriter, r *http.Request) {
// w.Header().Set("Content-Type", "text/html")
http.FileServer(http.Dir(s.configuration.SubscribersDataFolder)).ServeHTTP(w, r)
}
2021-08-23 10:47:33 +00:00
//create a file server, and serve the files found in ./
2021-08-23 15:05:56 +00:00
//fd := http.FileServer(http.Dir(s.configuration.SubscribersDataFolder))
http.HandleFunc("/", fileHandler)
2021-08-23 10:47:33 +00:00
// we create a net.Listen type to use later with the http.Serve function.
nl, err := net.Listen("tcp", s.configuration.ExposeDataFolder)
if err != nil {
log.Println("error: starting net.Listen: ", err)
}
// start the web server with http.Serve instead of the usual http.ListenAndServe
err = http.Serve(nl, nil)
if err != nil {
log.Printf("Error: failed to start web server: %v\n", err)
}
os.Exit(1)
}