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

427 lines
12 KiB
Go
Raw Normal View History

package steward
import (
"bytes"
"encoding/json"
"fmt"
"io"
2022-01-26 08:23:02 +00:00
"io/ioutil"
"log"
"net"
2021-09-10 03:26:16 +00:00
"net/http"
"os"
2022-01-26 08:23:02 +00:00
"path/filepath"
)
2022-01-26 08:23:02 +00:00
// readStartupFolder will check the <workdir>/startup folder when Steward
// starts for messages to process.
// The purpose of the startup folder is that we can define messages on a
// node that will be run when Steward starts up.
// Messages defined in the startup folder should have the toNode set to
// self, and the from node set to where we want the answer sent. The reason
// for this is that all replies normally pick up the host from the original
// first message, but here we inject it on an end node so we need to specify
// the fromNode to get the reply back to the node we want.
func (s *server) readStartupFolder() {
// Get the names of all the files in the startup folder.
const startupFolder = "startup"
filePaths, err := s.getFilePaths(startupFolder)
if err != nil {
er := fmt.Errorf("error: readStartupFolder: unable to get filenames: %v", err)
s.errorKernel.errSend(s.processInitial, Message{}, er)
return
}
for _, filePath := range filePaths {
// Read the content of each file.
readBytes, err := func(filePath string) ([]byte, error) {
fh, err := os.Open(filePath)
if err != nil {
er := fmt.Errorf("error: failed to open file in startup folder: %v", err)
return nil, er
}
defer fh.Close()
b, err := io.ReadAll(fh)
if err != nil {
er := fmt.Errorf("error: failed to read file in startup folder: %v", err)
return nil, er
}
return b, nil
}(filePath)
if err != nil {
s.errorKernel.errSend(s.processInitial, Message{}, err)
continue
}
readBytes = bytes.Trim(readBytes, "\x00")
// unmarshal the JSON into a struct
sams, err := s.convertBytesToSAMs(readBytes)
if err != nil {
er := fmt.Errorf("error: startup folder: malformed json read: %v", err)
s.errorKernel.errSend(s.processInitial, Message{}, er)
continue
}
// Check if fromNode field is specified, and remove the message if blank.
for i := range sams {
if sams[i].Message.FromNode == "" {
sams = append(sams[:i], sams[i+1:]...)
er := fmt.Errorf(" error: missing from field in startup message")
s.errorKernel.errSend(s.processInitial, Message{}, er)
2022-01-26 08:23:02 +00:00
}
// Bounds check.
if i == len(sams)-1 {
break
}
}
// Send the SAM struct to be picked up by the ring buffer.
s.ringBufferBulkInCh <- sams
}
}
// getFilePaths will get the names of all the messages in
// the folder specified from current working directory.
func (s *server) getFilePaths(dirName string) ([]string, error) {
dirPath, err := os.Getwd()
if err != nil {
return nil, fmt.Errorf("error: startup folder: unable to get the working directory %v: %v", dirPath, err)
}
dirPath = filepath.Join(dirPath, dirName)
// Check if the startup folder exist.
if _, err := os.Stat(dirPath); os.IsNotExist(err) {
err := os.MkdirAll(dirPath, 0700)
if err != nil {
er := fmt.Errorf("error: failed to create startup folder: %v", err)
return nil, er
}
}
fInfo, err := ioutil.ReadDir(dirPath)
if err != nil {
er := fmt.Errorf("error: failed to get filenames in startup folder: %v", err)
return nil, er
}
filePaths := []string{}
for _, v := range fInfo {
realpath := filepath.Join(dirPath, v.Name())
filePaths = append(filePaths, realpath)
}
return filePaths, nil
}
// readSocket will read the .sock file specified.
// It will take a channel of []byte as input, and it is in this
// channel the content of a file that has changed is returned.
2021-08-25 08:16:55 +00:00
func (s *server) readSocket() {
// Loop, and wait for new connections.
for {
2021-08-09 12:41:31 +00:00
conn, err := s.StewardSocket.Accept()
2021-02-05 06:25:12 +00:00
if err != nil {
er := fmt.Errorf("error: failed to accept conn on socket: %v", err)
s.processes.errorKernel.errSend(s.processInitial, Message{}, er)
2021-02-05 06:25:12 +00:00
}
go func(conn net.Conn) {
defer conn.Close()
var readBytes []byte
for {
b := make([]byte, 1500)
_, err = conn.Read(b)
if err != nil && err != io.EOF {
er := fmt.Errorf("error: failed to read data from tcp listener: %v", err)
s.processes.errorKernel.errSend(s.processInitial, Message{}, er)
return
}
readBytes = append(readBytes, b...)
if err == io.EOF {
break
}
}
readBytes = bytes.Trim(readBytes, "\x00")
// unmarshal the JSON into a struct
2021-08-26 04:35:54 +00:00
sams, err := s.convertBytesToSAMs(readBytes)
if err != nil {
2021-09-23 06:19:53 +00:00
er := fmt.Errorf("error: malformed json received on socket: %v", err)
s.processes.errorKernel.errSend(s.processInitial, Message{}, er)
return
}
2021-08-25 08:16:55 +00:00
for i := range sams {
// Fill in the value for the FromNode field, so the receiver
// can check this field to know where it came from.
2021-08-25 08:16:55 +00:00
sams[i].Message.FromNode = Node(s.nodeName)
}
2021-02-05 06:25:12 +00:00
// Send the SAM struct to be picked up by the ring buffer.
s.ringBufferBulkInCh <- sams
}(conn)
2021-02-05 06:25:12 +00:00
}
}
// readTCPListener wait and read messages delivered on the TCP
// port if started.
// It will take a channel of []byte as input, and it is in this
// channel the content of a file that has changed is returned.
func (s *server) readTCPListener() {
ln, err := net.Listen("tcp", s.configuration.TCPListener)
if err != nil {
log.Printf("error: readTCPListener: failed to start tcp listener: %v\n", err)
os.Exit(1)
}
// Loop, and wait for new connections.
for {
conn, err := ln.Accept()
if err != nil {
er := fmt.Errorf("error: failed to accept conn on socket: %v", err)
s.processes.errorKernel.errSend(s.processInitial, Message{}, er)
continue
}
go func(conn net.Conn) {
defer conn.Close()
var readBytes []byte
for {
b := make([]byte, 1500)
_, err = conn.Read(b)
if err != nil && err != io.EOF {
er := fmt.Errorf("error: failed to read data from tcp listener: %v", err)
s.processes.errorKernel.errSend(s.processInitial, Message{}, er)
return
}
readBytes = append(readBytes, b...)
if err == io.EOF {
break
}
}
readBytes = bytes.Trim(readBytes, "\x00")
// unmarshal the JSON into a struct
2021-08-26 04:35:54 +00:00
sam, err := s.convertBytesToSAMs(readBytes)
if err != nil {
2021-09-23 06:19:53 +00:00
er := fmt.Errorf("error: malformed json received on tcp listener: %v", err)
s.processes.errorKernel.errSend(s.processInitial, Message{}, er)
return
}
for i := range sam {
// Fill in the value for the FromNode field, so the receiver
// can check this field to know where it came from.
sam[i].Message.FromNode = Node(s.nodeName)
}
// Send the SAM struct to be picked up by the ring buffer.
s.ringBufferBulkInCh <- sam
}(conn)
}
}
2021-09-10 03:26:16 +00:00
func (s *server) readHTTPlistenerHandler(w http.ResponseWriter, r *http.Request) {
2021-09-10 04:06:43 +00:00
var readBytes []byte
for {
b := make([]byte, 1500)
_, err := r.Body.Read(b)
if err != nil && err != io.EOF {
er := fmt.Errorf("error: failed to read data from tcp listener: %v", err)
s.processes.errorKernel.errSend(s.processInitial, Message{}, er)
2021-09-10 04:06:43 +00:00
return
}
readBytes = append(readBytes, b...)
if err == io.EOF {
break
}
}
readBytes = bytes.Trim(readBytes, "\x00")
// unmarshal the JSON into a struct
sam, err := s.convertBytesToSAMs(readBytes)
2021-09-10 03:26:16 +00:00
if err != nil {
2021-09-23 06:19:53 +00:00
er := fmt.Errorf("error: malformed json received on HTTPListener: %v", err)
s.processes.errorKernel.errSend(s.processInitial, Message{}, er)
2021-09-10 04:06:43 +00:00
return
}
for i := range sam {
// Fill in the value for the FromNode field, so the receiver
// can check this field to know where it came from.
sam[i].Message.FromNode = Node(s.nodeName)
2021-09-10 03:26:16 +00:00
}
2021-09-10 04:06:43 +00:00
// Send the SAM struct to be picked up by the ring buffer.
s.ringBufferBulkInCh <- sam
2021-09-10 03:26:16 +00:00
}
func (s *server) readHttpListener() {
go func() {
n, err := net.Listen("tcp", s.configuration.HTTPListener)
if err != nil {
log.Printf("error: startMetrics: failed to open prometheus listen port: %v\n", err)
os.Exit(1)
}
mux := http.NewServeMux()
2021-09-10 03:45:09 +00:00
mux.HandleFunc("/", s.readHTTPlistenerHandler)
2021-09-10 03:26:16 +00:00
err = http.Serve(n, mux)
if err != nil {
log.Printf("error: startMetrics: failed to start http.Serve: %v\n", err)
os.Exit(1)
}
}()
}
2021-08-16 11:01:12 +00:00
// The subject are made up of different parts of the message field.
// To make things easier and to avoid figuring out what the subject
// is in all places we've created the concept of subjectAndMessage
// (sam) where we get the subject for the message once, and use the
// sam structure with subject alongside the message instead.
type subjectAndMessage struct {
2021-02-04 10:46:58 +00:00
Subject `json:"subject" yaml:"subject"`
Message `json:"message" yaml:"message"`
}
2021-08-25 08:16:55 +00:00
// convertBytesToSAMs will range over the byte representing a message given in
// json format. For each element found the Message type will be converted into
// a SubjectAndMessage type value and appended to a slice, and the slice is
// returned to the caller.
2021-08-26 04:35:54 +00:00
func (s *server) convertBytesToSAMs(b []byte) ([]subjectAndMessage, error) {
MsgSlice := []Message{}
err := json.Unmarshal(b, &MsgSlice)
if err != nil {
2021-02-04 10:46:58 +00:00
return nil, fmt.Errorf("error: unmarshal of file failed: %#v", err)
}
2021-08-26 05:02:36 +00:00
// Check for toNode and toNodes field.
2021-08-26 04:35:54 +00:00
MsgSlice = s.checkMessageToNodes(MsgSlice)
s.metrics.promUserMessagesTotal.Add(float64(len(MsgSlice)))
2021-08-26 04:35:54 +00:00
sam := []subjectAndMessage{}
// Range over all the messages parsed from json, and create a subject for
// each message.
for _, m := range MsgSlice {
sm, err := newSubjectAndMessage(m)
if err != nil {
er := fmt.Errorf("error: newSubjectAndMessage: %v", err)
s.processes.errorKernel.errSend(s.processInitial, m, er)
2021-08-26 04:35:54 +00:00
continue
}
sam = append(sam, sm)
}
return sam, nil
}
// checkMessageToNodes will check that either toHost or toHosts are
// specified in the message. If not specified it will drop the message
// and send an error.
// if toNodes is specified, the original message will be used, and
// and an individual message will be created with a toNode field for
// each if the toNodes specified.
func (s *server) checkMessageToNodes(MsgSlice []Message) []Message {
msgs := []Message{}
2021-08-26 04:08:39 +00:00
for _, v := range MsgSlice {
switch {
// if toNode specified, we don't care about the toHosts.
case v.ToNode != "":
2021-08-26 04:35:54 +00:00
msgs = append(msgs, v)
2021-08-26 04:08:39 +00:00
continue
// if toNodes specified, we use the original message, and
// create new node messages for each of the nodes specified.
case len(v.ToNodes) != 0:
for _, n := range v.ToNodes {
m := v
2021-08-26 04:35:54 +00:00
// Set the toNodes field to nil since we're creating
// an individual toNode message for each of the toNodes
// found, and hence we no longer need that field.
2021-08-26 04:08:39 +00:00
m.ToNodes = nil
m.ToNode = n
2021-08-26 04:35:54 +00:00
msgs = append(msgs, m)
2021-08-26 04:08:39 +00:00
}
continue
// No toNode or toNodes specified. Drop the message by not appending it to
// the slice since it is not valid.
default:
2021-09-23 06:19:53 +00:00
er := fmt.Errorf("error: no toNode or toNodes where specified in the message, dropping message: %v", v)
s.processes.errorKernel.errSend(s.processInitial, v, er)
2021-08-26 04:08:39 +00:00
continue
}
}
2021-08-26 04:35:54 +00:00
return msgs
}
2021-08-25 06:56:44 +00:00
// newSubjectAndMessage will look up the correct values and value types to
2021-08-16 11:01:12 +00:00
// be used in a subject for a Message (sam), and return the a combined structure
2021-03-10 06:11:14 +00:00
// of type subjectAndMessage.
2021-08-25 06:56:44 +00:00
func newSubjectAndMessage(m Message) (subjectAndMessage, error) {
2021-03-10 06:11:14 +00:00
// We need to create a tempory method type to look up the kind for the
// real method for the message.
var mt Method
tmpH := mt.getHandler(m.Method)
if tmpH == nil {
return subjectAndMessage{}, fmt.Errorf("error: newSubjectAndMessage: no such request type defined: %v", m.Method)
}
switch {
case m.ToNode == "":
return subjectAndMessage{}, fmt.Errorf("error: newSubjectAndMessage: ToNode empty: %+v", m)
case m.Method == "":
return subjectAndMessage{}, fmt.Errorf("error: newSubjectAndMessage: Method empty: %v", m)
}
2021-03-10 06:11:14 +00:00
sub := Subject{
ToNode: string(m.ToNode),
Event: tmpH.getKind(),
Method: m.Method,
2021-03-10 06:11:14 +00:00
}
2021-08-25 06:56:44 +00:00
sam := subjectAndMessage{
2021-03-10 06:11:14 +00:00
Subject: sub,
Message: m,
}
2021-08-25 06:56:44 +00:00
return sam, nil
2021-03-10 06:11:14 +00:00
}