ctrl/requests.go

// The structure of how to add new method types to the system.
// -----------------------------------------------------------
// All methods need 3 things:
//  - A type definition
//  - The type needs a getKind method
//  - The type needs a handler method
// Overall structure example shown below.
//
// ---
// type methodCommandCLICommandRequest struct {
// 	commandOrEvent CommandOrEvent
// }
//
// func (m methodCommandCLICommandRequest) getKind() CommandOrEvent {
// 	return m.commandOrEvent
// }
//
// func (m methodCommandCLICommandRequest) handler(s *server, message Message, node string) ([]byte, error) {
//  ...
//  ...
// 	ackMsg := []byte(fmt.Sprintf("confirmed from node: %v: messageID: %v\n---\n%s---", node, message.ID, out))
// 	return ackMsg, nil
// }
//
// ---
// You also need to make a constant for the Method, and add
// that constant as the key in the map, where the value is
// the actual type you want to map it to with a handler method.
// You also specify if it is a Command or Event, and if it is
// ACK or NACK.
// Check out the existing code below for more examples.

package steward

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"time"

	"github.com/hpcloud/tail"
	"github.com/prometheus/client_golang/prometheus"
)

// Method is used to specify the actual function/method that
// is represented in a typed manner.
type Method string

// ------------------------------------------------------------
// The constants that will be used throughout the system for
// when specifying what kind of Method to send or work with.
const (
	// Initial parent method used to start other processes.
	REQInitial Method = "REQInitial"
	// Command for client operation request of the system. The op
	// command to execute shall be given in the data field of the
	// message as string value. For example "ps".
	REQOpCommand Method = "REQOpCommand"
	// Execute a CLI command in for example bash or cmd.
	// This is an event type, where a message will be sent to a
	// node with the command to execute and an ACK will be replied
	// if it was delivered succesfully. The output of the command
	// ran will be delivered back to the node where it was initiated
	// as a new message.
	// The data field is a slice of strings where the first string
	// value should be the command, and the following the arguments.
	REQCliCommand Method = "REQCliCommand"
	// REQnCliCommandCont same as normal Cli command, but can be used
	// when running a command that will take longer time and you want
	// to send the output of the command continually back as it is
	// generated, and not just when the command is finished.
	REQnCliCommandCont Method = "REQnCliCommandCont"
	// Send text to be logged to the console.
	// The data field is a slice of strings where the first string
	// value should be the command, and the following the arguments.
	REQToConsole Method = "REQToConsole"
	// Send text logging to some host by appending the output to a
	// file, if the file do not exist we create it.
	// A file with the full subject+hostName will be created on
	// the receiving end.
	// The data field is a slice of strings where the values of the
	// slice will be written to the log file.
	REQToFileAppend Method = "REQToFileAppend"
	// Send text to some host by overwriting the existing content of
	// the fileoutput to a file. If the file do not exist we create it.
	// A file with the full subject+hostName will be created on
	// the receiving end.
	// The data field is a slice of strings where the values of the
	// slice will be written to the file.
	REQToFile Method = "REQToFile"
	// Send Hello I'm here message.
	REQHello Method = "REQHello"
	// Error log methods to centralError node.
	REQErrorLog Method = "REQErrorLog"
	// Echo request will ask the subscriber for a
	// reply generated as a new message, and sent back to where
	// the initial request was made.
	REQPing Method = "REQPing"
	// Will generate a reply for a ECHORequest
	REQPong Method = "REQPong"
	// Http Get
	REQHttpGet Method = "REQHttpGet"
	// Tail file
	REQTailFile Method = "REQTailFile"
	// Write to steward socket
	REQToSocket Method = "REQToSocket"
)

// The mapping of all the method constants specified, what type
// it references, and the kind if it is an Event or Command, and
// if it is ACK or NACK.
//  Allowed values for the commandOrEvent field are:
//   - CommandACK
//   - CommandNACK
//   - EventACK
//   - EventNack
func (m Method) GetMethodsAvailable() MethodsAvailable {

	// Command, Used to make a request to perform an action
	// Event, Used to communicate that something have happened.
	ma := MethodsAvailable{
		Methodhandlers: map[Method]methodHandler{
			REQInitial: methodREQInitial{
				commandOrEvent: CommandACK,
			},
			REQOpCommand: methodREQOpCommand{
				commandOrEvent: CommandACK,
			},
			REQCliCommand: methodREQCliCommand{
				commandOrEvent: CommandACK,
			},
			REQnCliCommandCont: methodREQnCliCommandCont{
				commandOrEvent: CommandACK,
			},
			REQToConsole: methodREQToConsole{
				commandOrEvent: EventACK,
			},
			REQToFileAppend: methodREQToFileAppend{
				commandOrEvent: EventACK,
			},
			REQToFile: methodREQToFile{
				commandOrEvent: EventACK,
			},
			REQHello: methodREQHello{
				commandOrEvent: EventNACK,
			},
			REQErrorLog: methodREQErrorLog{
				commandOrEvent: EventACK,
			},
			REQPing: methodREQPing{
				commandOrEvent: EventACK,
			},
			REQPong: methodREQPong{
				commandOrEvent: EventACK,
			},
			REQHttpGet: methodREQHttpGet{
				commandOrEvent: EventACK,
			},
			REQTailFile: methodREQTailFile{
				commandOrEvent: EventACK,
			},
			REQToSocket: methodREQToSocket{
				commandOrEvent: EventACK,
			},
		},
	}

	return ma
}

// Reply methods. The slice generated here is primarily used within
// the Stew client for knowing what of the req types are generally
// used as reply methods.
func (m Method) GetReplyMethods() []Method {
	rm := []Method{REQToConsole, REQToFile, REQToFileAppend, REQToSocket}
	return rm
}

// getHandler will check the methodsAvailable map, and return the
// method handler for the method given
// as input argument.
func (m Method) getHandler(method Method) methodHandler {
	ma := m.GetMethodsAvailable()
	mh := ma.Methodhandlers[method]

	return mh
}

// The structure that works as a reference for all the methods and if
// they are of the command or event type, and also if it is a ACK or
// NACK message.

// ----

// Initial parent method used to start other processes.
type methodREQInitial struct {
	commandOrEvent CommandOrEvent
}

func (m methodREQInitial) getKind() CommandOrEvent {
	return m.commandOrEvent
}

func (m methodREQInitial) handler(proc process, message Message, node string) ([]byte, error) {
	// proc.procFuncCh <- message
	ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
	return ackMsg, nil
}

// ----

// MethodsAvailable holds a map of all the different method types and the
// associated handler to that method type.
type MethodsAvailable struct {
	Methodhandlers map[Method]methodHandler
}

// Check if exists will check if the Method is defined. If true the bool
// value will be set to true, and the methodHandler function for that type
// will be returned.
func (ma MethodsAvailable) CheckIfExists(m Method) (methodHandler, bool) {
	mFunc, ok := ma.Methodhandlers[m]
	if ok {
		// fmt.Printf("******THE TOPIC EXISTS: %v******\n", m)
		return mFunc, true
	} else {
		// fmt.Printf("******THE TOPIC DO NOT EXIST: %v******\n", m)
		return nil, false
	}
}

// Create a new message for the reply containing the output of the
// action executed put in outData, and put it on the ringbuffer to
// be published.
// The method to use for the reply message should initially be
// specified within the first message as the replyMethod, and we will
// pick up that value here, and use it as the method for the new
// request message. If no replyMethod is set we default to the
// REQToFileAppend method type.
func newReplyMessage(proc process, message Message, outData []byte) {

	// If no replyMethod is set we default to writing to writing to
	// a log file.
	if message.ReplyMethod == "" {
		message.ReplyMethod = REQToFileAppend
	}

	// Create a new message for the reply, and put it on the
	// ringbuffer to be published.
	newMsg := Message{
		ToNode:        message.FromNode,
		Data:          []string{string(outData)},
		Method:        message.ReplyMethod,
		MethodArgs:    message.ReplyMethodArgs,
		MethodTimeout: message.ReplyMethodTimeout,
		ACKTimeout:    message.ReplyACKTimeout,
		Retries:       message.ReplyRetries,

		// Put in a copy of the initial request message, so we can use it's properties if
		// needed to for example create the file structure naming on the subscriber.
		PreviousMessage: &message,
	}

	sam, err := newSubjectAndMessage(newMsg)
	if err != nil {
		// In theory the system should drop the message before it reaches here.
		er := fmt.Errorf("error: newReplyMessage : %v, message: %v", err, message)
		sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
		log.Printf("%v\n", er)
	}
	proc.toRingbufferCh <- []subjectAndMessage{sam}
}

// selectFileNaming will figure out the correct naming of the file
// structure to use for the reply data.
// It will return the filename, and the tree structure for the folders
// to create.
func selectFileNaming(message Message, proc process) (string, string) {
	var fileName string
	var folderTree string

	switch {
	case message.PreviousMessage == nil:
		// If this was a direct request there are no previous message to take
		// information from, so we use the one that are in the current mesage.
		fileName = message.FileName
		folderTree = filepath.Join(proc.configuration.SubscribersDataFolder, message.Directory, string(message.ToNode))
	case message.PreviousMessage.ToNode != "":
		fileName = message.PreviousMessage.FileName
		folderTree = filepath.Join(proc.configuration.SubscribersDataFolder, message.PreviousMessage.Directory, string(message.PreviousMessage.ToNode))
	case message.PreviousMessage.ToNode == "":
		fileName = message.PreviousMessage.FileName
		folderTree = filepath.Join(proc.configuration.SubscribersDataFolder, message.PreviousMessage.Directory, string(message.FromNode))
	}

	return fileName, folderTree
}

// ------------------------------------------------------------
// Subscriber method handlers
// ------------------------------------------------------------

// The methodHandler interface.
type methodHandler interface {
	handler(proc process, message Message, node string) ([]byte, error)
	getKind() CommandOrEvent
}

// -----

type methodREQOpCommand struct {
	commandOrEvent CommandOrEvent
}

func (m methodREQOpCommand) getKind() CommandOrEvent {
	return m.commandOrEvent
}

type OpCmdStartProc struct {
	Method       Method `json:"method"`
	AllowedNodes []Node `json:"allowedNodes"`
}

type OpCmdStopProc struct {
	RecevingNode Node        `json:"receivingNode"`
	Method       Method      `json:"method"`
	Kind         processKind `json:"kind"`
	ID           int         `json:"id"`
}

// handler to run a CLI command with timeout context. The handler will
// return the output of the command run back to the calling publisher
// in the ack message.
func (m methodREQOpCommand) handler(proc process, message Message, nodeName string) ([]byte, error) {
	proc.processes.wg.Add(1)
	go func() {
		defer proc.processes.wg.Done()

		out := []byte{}

		// unmarshal the json.RawMessage field called OpArgs.
		//
		// Dst interface is the generic type to Unmarshal OpArgs into, and we will
		// set the type it should contain depending on the value specified in Cmd.
		var dst interface{}

		switch message.Operation.OpCmd {
		case "ps":
			proc.processes.mu.Lock()
			// Loop the the processes map, and find all that is active to
			// be returned in the reply message.
			for _, idMap := range proc.processes.active {
				for _, v := range idMap {
					s := fmt.Sprintf("%v, proc: %v, id: %v, name: %v, allowed from: %s\n", time.Now().Format("Mon Jan _2 15:04:05 2006"), v.processKind, v.processID, v.subject.name(), v.allowedReceivers)
					sb := []byte(s)
					out = append(out, sb...)
				}

			}
			proc.processes.mu.Unlock()

		case "startProc":
			// Set the empty interface type dst to &OpStart.
			dst = &OpCmdStartProc{}

			err := json.Unmarshal(message.Operation.OpArg, &dst)
			if err != nil {
				er := fmt.Errorf("error: methodREQOpCommand startProc json.Umarshal failed : %v, message: %v", err, message)
				sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
				log.Printf("%v\n", er)
			}

			// Assert it into the correct non pointer value.
			arg := *dst.(*OpCmdStartProc)

			//fmt.Printf(" ** Content of Arg = %#v\n", arg)

			if len(arg.AllowedNodes) == 0 {
				er := fmt.Errorf("error: startProc: no allowed publisher nodes specified: %v" + fmt.Sprint(message))
				sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
				log.Printf("%v\n", er)
				return
			}

			if arg.Method == "" {
				er := fmt.Errorf("error: startProc: no method specified: %v" + fmt.Sprint(message))
				sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
				log.Printf("%v\n", er)
				return
			}

			// Create the process and start it.
			sub := newSubject(arg.Method, proc.configuration.NodeName)
			procNew := newProcess(proc.ctx, proc.processes.metrics, proc.natsConn, proc.processes, proc.toRingbufferCh, proc.configuration, sub, proc.errorCh, processKindSubscriber, nil)
			go procNew.spawnWorker(proc.processes, proc.natsConn)

			er := fmt.Errorf("info: startProc: started id: %v, subject: %v: node: %v", procNew.processID, sub, message.ToNode)
			sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)

		case "stopProc":
			// Set the interface type dst to &OpStart.
			dst = &OpCmdStopProc{}

			err := json.Unmarshal(message.Operation.OpArg, &dst)
			if err != nil {
				er := fmt.Errorf("error: methodREQOpCommand stopProc json.Umarshal failed : %v, message: %v", err, message)
				sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
				log.Printf("%v\n", er)
			}

			// Assert it into the correct non pointer value.
			arg := *dst.(*OpCmdStopProc)

			// Based on the arg values received in the message we create a
			// processName structure as used in naming the real processes.
			// We can then use this processName to get the real values for the
			// actual process we want to stop.
			sub := newSubject(arg.Method, string(arg.RecevingNode))
			processName := processNameGet(sub.name(), arg.Kind)

			// Check if the message contains an id.
			err = func() error {
				if arg.ID == 0 {
					er := fmt.Errorf("error: stopProc: did not find process to stop: %v on %v", sub, message.ToNode)
					sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
					return er
				}
				return nil
			}()

			if err != nil {
				er := fmt.Errorf("error: stopProc: err was not nil: %v : %v on %v", err, sub, message.ToNode)
				sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
				log.Printf("%v\n", er)
				return
			}

			proc.processes.mu.Lock()

			// Remove the process from the processes active map if found.
			toStopProc, ok := proc.processes.active[processName][arg.ID]
			if ok {
				// Delete the process from the processes map
				delete(proc.processes.active, processName)
				// Stop started go routines that belong to the process.
				toStopProc.ctxCancel()
				// Stop subscribing for messages on the process's subject.
				err := toStopProc.natsSubscription.Unsubscribe()
				if err != nil {
					er := fmt.Errorf("error: methodREQOpCommand, toStopProc, failed to stop nats.Subscription: %v, message: %v", err, message)
					sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
					log.Printf("%v\n", er)
				}

				// Remove the prometheus label
				proc.processes.metrics.promProcessesAllRunning.Delete(prometheus.Labels{"processName": string(processName)})

				er := fmt.Errorf("info: stopProc: stopped %v on %v", sub, message.ToNode)
				sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
				log.Printf("%v\n", er)

				newReplyMessage(proc, message, []byte(er.Error()))

			} else {
				er := fmt.Errorf("error: stopProc: methodREQOpCommand, did not find process to stop: %v on %v", sub, message.ToNode)
				sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
				log.Printf("%v\n", er)

				newReplyMessage(proc, message, []byte(er.Error()))
			}

			proc.processes.mu.Unlock()

		}

		newReplyMessage(proc, message, out)
	}()

	ackMsg := []byte(fmt.Sprintf("confirmed from node: %v: messageID: %v\n---\n", proc.node, message.ID))
	return ackMsg, nil
}

//----

type methodREQToFileAppend struct {
	commandOrEvent CommandOrEvent
}

func (m methodREQToFileAppend) getKind() CommandOrEvent {
	return m.commandOrEvent
}

// Handle appending data to file.
func (m methodREQToFileAppend) handler(proc process, message Message, node string) ([]byte, error) {

	// If it was a request type message we want to check what the initial messages
	// method, so we can use that in creating the file name to store the data.
	fileName, folderTree := selectFileNaming(message, proc)

	// Check if folder structure exist, if not create it.
	if _, err := os.Stat(folderTree); os.IsNotExist(err) {
		err := os.MkdirAll(folderTree, 0700)
		if err != nil {
			er := fmt.Errorf("error: methodREQToFileAppend failed to create toFileAppend directory tree:%v, %v, message: %v", folderTree, err, message)
			sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
			log.Printf("%v\n", er)
		}

		log.Printf("info: Creating subscribers data folder at %v\n", folderTree)
	}

	// Open file and write data.
	file := filepath.Join(folderTree, fileName)
	f, err := os.OpenFile(file, os.O_APPEND|os.O_RDWR|os.O_CREATE|os.O_SYNC, 0600)
	if err != nil {
		er := fmt.Errorf("error: methodREQToFileAppend.handler: failed to open file : %v, message: %v", err, message)
		sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
		log.Printf("%v\n", er)
		return nil, err
	}
	defer f.Close()

	for _, d := range message.Data {
		_, err := f.Write([]byte(d))
		f.Sync()
		if err != nil {
			er := fmt.Errorf("error: methodEventTextLogging.handler: failed to write to file : %v, message: %v", err, message)
			sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
			log.Printf("%v\n", er)
		}
	}

	ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
	return ackMsg, nil
}

// -----

type methodREQToFile struct {
	commandOrEvent CommandOrEvent
}

func (m methodREQToFile) getKind() CommandOrEvent {
	return m.commandOrEvent
}

// Handle writing to a file. Will truncate any existing data if the file did already
// exist.
func (m methodREQToFile) handler(proc process, message Message, node string) ([]byte, error) {

	// If it was a request type message we want to check what the initial messages
	// method, so we can use that in creating the file name to store the data.
	fileName, folderTree := selectFileNaming(message, proc)

	// Check if folder structure exist, if not create it.
	if _, err := os.Stat(folderTree); os.IsNotExist(err) {
		err := os.MkdirAll(folderTree, 0700)
		if err != nil {
			er := fmt.Errorf("error: methodREQToFile failed to create toFile directory tree %v: %v, %v", folderTree, err, message)
			sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
			log.Printf("%v\n", er)

			return nil, er
		}

		log.Printf("info: Creating subscribers data folder at %v\n", folderTree)
	}

	// Open file and write data.
	file := filepath.Join(folderTree, fileName)
	f, err := os.OpenFile(file, os.O_CREATE|os.O_RDWR|os.O_TRUNC, 0755)
	if err != nil {
		er := fmt.Errorf("error: methodREQToFile.handler: failed to open file, check that you've specified a value for fileName in the message: %v", err)
		sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
		log.Printf("%v\n", er)
		return nil, err
	}
	defer f.Close()

	for _, d := range message.Data {
		_, err := f.Write([]byte(d))
		f.Sync()
		if err != nil {
			er := fmt.Errorf("error: methodEventTextLogging.handler: failed to write to file: %v, %v", err, message)
			sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
			log.Printf("%v\n", er)
		}
	}

	ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
	return ackMsg, nil
}

// ----
type methodREQHello struct {
	commandOrEvent CommandOrEvent
}

func (m methodREQHello) getKind() CommandOrEvent {
	return m.commandOrEvent
}

// Handler for receiving hello messages.
func (m methodREQHello) handler(proc process, message Message, node string) ([]byte, error) {
	data := fmt.Sprintf("%v, Received hello from %#v\n", time.Now().Format("Mon Jan _2 15:04:05 2006"), message.FromNode)

	fileName := message.FileName
	folderTree := filepath.Join(proc.configuration.SubscribersDataFolder, message.Directory, string(message.FromNode))

	// Check if folder structure exist, if not create it.
	if _, err := os.Stat(folderTree); os.IsNotExist(err) {
		err := os.MkdirAll(folderTree, 0700)
		if err != nil {
			return nil, fmt.Errorf("error: failed to create errorLog directory tree %v: %v", folderTree, err)
		}

		log.Printf("info: Creating subscribers data folder at %v\n", folderTree)
	}

	// Open file and write data.
	file := filepath.Join(folderTree, fileName)
	f, err := os.OpenFile(file, os.O_APPEND|os.O_RDWR|os.O_CREATE|os.O_SYNC, 0600)

	if err != nil {
		log.Printf("error: methodREQHello.handler: failed to open file: %v\n", err)
		return nil, err
	}
	defer f.Close()

	_, err = f.Write([]byte(data))
	f.Sync()
	if err != nil {
		log.Printf("error: methodEventTextLogging.handler: failed to write to file: %v\n", err)
	}

	// --------------------------

	// send the message to the procFuncCh which is running alongside the process
	// and can hold registries and handle special things for an individual process.
	proc.procFuncCh <- message

	ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
	return ackMsg, nil
}

// ---

type methodREQErrorLog struct {
	commandOrEvent CommandOrEvent
}

func (m methodREQErrorLog) getKind() CommandOrEvent {
	return m.commandOrEvent
}

// Handle the writing of error logs.
func (m methodREQErrorLog) handler(proc process, message Message, node string) ([]byte, error) {
	proc.processes.metrics.promErrorMessagesReceivedTotal.Inc()

	// If it was a request type message we want to check what the initial messages
	// method, so we can use that in creating the file name to store the data.
	fileName, folderTree := selectFileNaming(message, proc)

	// Check if folder structure exist, if not create it.
	if _, err := os.Stat(folderTree); os.IsNotExist(err) {
		err := os.MkdirAll(folderTree, 0700)
		if err != nil {
			return nil, fmt.Errorf("error: failed to create errorLog directory tree %v: %v", folderTree, err)
		}

		log.Printf("info: Creating subscribers data folder at %v\n", folderTree)
	}

	// Open file and write data.
	file := filepath.Join(folderTree, fileName)
	f, err := os.OpenFile(file, os.O_APPEND|os.O_RDWR|os.O_CREATE|os.O_SYNC, 0600)
	if err != nil {
		log.Printf("error: methodREQErrorLog.handler: failed to open file: %v\n", err)
		return nil, err
	}
	defer f.Close()

	for _, d := range message.Data {
		_, err := f.Write([]byte(d))
		f.Sync()
		if err != nil {
			log.Printf("error: methodEventTextLogging.handler: failed to write to file: %v\n", err)
		}
	}

	ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
	return ackMsg, nil
}

// ---

type methodREQPing struct {
	commandOrEvent CommandOrEvent
}

func (m methodREQPing) getKind() CommandOrEvent {
	return m.commandOrEvent
}

// Handle receving a ping.
func (m methodREQPing) handler(proc process, message Message, node string) ([]byte, error) {
	// Write to file that we received a ping

	// If it was a request type message we want to check what the initial messages
	// method, so we can use that in creating the file name to store the data.
	fileName, folderTree := selectFileNaming(message, proc)

	// Check if folder structure exist, if not create it.
	if _, err := os.Stat(folderTree); os.IsNotExist(err) {
		err := os.MkdirAll(folderTree, 0700)
		if err != nil {
			er := fmt.Errorf("error: methodREQPing.handler: failed to create toFile directory tree %v: %v, %v", folderTree, err, message)
			sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
			log.Printf("%v\n", er)

			return nil, er
		}

		log.Printf("info: Creating subscribers data folder at %v\n", folderTree)
	}

	// Open file.
	file := filepath.Join(folderTree, fileName)
	f, err := os.OpenFile(file, os.O_CREATE|os.O_RDWR|os.O_TRUNC, 0755)
	if err != nil {
		er := fmt.Errorf("error: methodREQPing.handler: failed to open file, check that you've specified a value for fileName in the message: %v", err)
		sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
		log.Printf("%v\n", er)
		return nil, err
	}
	defer f.Close()

	// And write the data
	d := fmt.Sprintf("%v, ping received from %v\n", time.Now().Format("Mon Jan _2 15:04:05 2006"), message.FromNode)
	_, err = f.Write([]byte(d))
	f.Sync()
	if err != nil {
		er := fmt.Errorf("error: methodREQPing.handler: failed to write to file: %v, %v", err, message)
		sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
		log.Printf("%v\n", er)
	}

	proc.processes.wg.Add(1)
	go func() {
		defer proc.processes.wg.Done()

		newReplyMessage(proc, message, nil)
	}()

	ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
	return ackMsg, nil
}

// ---

type methodREQPong struct {
	commandOrEvent CommandOrEvent
}

func (m methodREQPong) getKind() CommandOrEvent {
	return m.commandOrEvent
}

// Handle receiving a pong.
func (m methodREQPong) handler(proc process, message Message, node string) ([]byte, error) {
	// Write to file that we received a pong

	// If it was a request type message we want to check what the initial messages
	// method, so we can use that in creating the file name to store the data.
	fileName, folderTree := selectFileNaming(message, proc)

	// Check if folder structure exist, if not create it.
	if _, err := os.Stat(folderTree); os.IsNotExist(err) {
		err := os.MkdirAll(folderTree, 0700)
		if err != nil {
			er := fmt.Errorf("error: methodREQPong.handler: failed to create toFile directory tree %v: %v, %v", folderTree, err, message)
			sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
			log.Printf("%v\n", er)

			return nil, er
		}

		log.Printf("info: Creating subscribers data folder at %v\n", folderTree)
	}

	// Open file.
	file := filepath.Join(folderTree, fileName)
	f, err := os.OpenFile(file, os.O_CREATE|os.O_RDWR|os.O_TRUNC, 0755)
	if err != nil {
		er := fmt.Errorf("error: methodREQPong.handler: failed to open file, check that you've specified a value for fileName in the message: %v", err)
		sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
		log.Printf("%v\n", er)
		return nil, err
	}
	defer f.Close()

	// And write the data
	d := fmt.Sprintf("%v, pong received from %v\n", time.Now().Format("Mon Jan _2 15:04:05 2006"), message.PreviousMessage.ToNode)
	_, err = f.Write([]byte(d))
	f.Sync()
	if err != nil {
		er := fmt.Errorf("error: methodREQPong.handler: failed to write to file: %v, %v", err, message)
		sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
		log.Printf("%v\n", er)
	}

	ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
	return ackMsg, nil
}

// ---

type methodREQCliCommand struct {
	commandOrEvent CommandOrEvent
}

func (m methodREQCliCommand) getKind() CommandOrEvent {
	return m.commandOrEvent
}

// handler to run a CLI command with timeout context. The handler will
// return the output of the command run back to the calling publisher
// as a new message.
func (m methodREQCliCommand) handler(proc process, message Message, node string) ([]byte, error) {
	log.Printf("<--- CLICommandREQUEST received from: %v, containing: %v", message.FromNode, message.Data)

	// Execute the CLI command in it's own go routine, so we are able
	// to return immediately with an ack reply that the messag was
	// received, and we create a new message to send back to the calling
	// node for the out put of the actual command.
	proc.processes.wg.Add(1)
	go func() {
		defer proc.processes.wg.Done()

		c := message.MethodArgs[0]
		a := message.MethodArgs[1:]

		ctx, cancel := context.WithTimeout(proc.ctx, time.Second*time.Duration(message.MethodTimeout))

		outCh := make(chan []byte)

		proc.processes.wg.Add(1)
		go func() {
			defer proc.processes.wg.Done()

			cmd := exec.CommandContext(ctx, c, a...)

			var out bytes.Buffer
			var stderr bytes.Buffer
			cmd.Stdout = &out
			cmd.Stderr = &stderr
			err := cmd.Run()
			if err != nil {
				if err != nil {
					log.Printf("error: failed to io.ReadAll of stderr: %v\n", err)
				}

				er := fmt.Errorf("error: methodREQCliCommand: cmd.Output : %v, message: %v, error_output: %v", err, message, stderr.String())
				sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
				log.Printf("%v\n", er)
			}
			select {
			case outCh <- out.Bytes():
			case <-ctx.Done():
				return
			}
		}()

		select {
		case <-ctx.Done():
			cancel()
			er := fmt.Errorf("error: methodREQCliCommand: method timed out %v", message)
			sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
		case out := <-outCh:
			cancel()

			// Prepare and queue for sending a new message with the output
			// of the action executed.
			newReplyMessage(proc, message, out)
		}

	}()

	ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
	return ackMsg, nil
}

// ---

type methodREQToConsole struct {
	commandOrEvent CommandOrEvent
}

func (m methodREQToConsole) getKind() CommandOrEvent {
	return m.commandOrEvent
}

// Handler to write directly to console.
func (m methodREQToConsole) handler(proc process, message Message, node string) ([]byte, error) {
	fmt.Printf("<--- methodCLICommandReply: %v\n", message.Data)

	ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
	return ackMsg, nil
}

// ---

type methodREQHttpGet struct {
	commandOrEvent CommandOrEvent
}

func (m methodREQHttpGet) getKind() CommandOrEvent {
	return m.commandOrEvent
}

// handler to do a Http Get.
func (m methodREQHttpGet) handler(proc process, message Message, node string) ([]byte, error) {
	log.Printf("<--- REQHttpGet received from: %v, containing: %v", message.FromNode, message.Data)

	proc.processes.wg.Add(1)
	go func() {
		defer proc.processes.wg.Done()

		url := message.MethodArgs[0]

		client := http.Client{
			Timeout: time.Second * 5,
		}

		ctx, cancel := context.WithTimeout(proc.ctx, time.Second*time.Duration(message.MethodTimeout))

		req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
		if err != nil {
			er := fmt.Errorf("error: methodREQHttpGet: NewRequest failed: %v, bailing out: %v", err, message)
			sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
			cancel()
			return
		}

		outCh := make(chan []byte)

		proc.processes.wg.Add(1)
		go func() {
			defer proc.processes.wg.Done()

			resp, err := client.Do(req)
			if err != nil {
				er := fmt.Errorf("error: methodREQHttpGet: client.Do failed: %v, bailing out: %v", err, message)
				sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
				return
			}
			defer resp.Body.Close()

			if resp.StatusCode != 200 {
				cancel()
				er := fmt.Errorf("error: methodREQHttpGet: not 200, where %#v, bailing out: %v", resp.StatusCode, message)
				sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
				return
			}

			body, err := io.ReadAll(resp.Body)
			if err != nil {
				er := fmt.Errorf("error: methodREQHttpGet: io.ReadAll failed : %v, message: %v", err, message)
				sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
				log.Printf("%v\n", er)
			}

			out := body

			select {
			case outCh <- out:
			case <-ctx.Done():
				return
			}
		}()

		select {
		case <-ctx.Done():
			cancel()
			er := fmt.Errorf("error: methodREQHttpGet: method timed out %v", message)
			sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
		case out := <-outCh:
			cancel()

			// Prepare and queue for sending a new message with the output
			// of the action executed.
			newReplyMessage(proc, message, out)
		}

	}()

	ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
	return ackMsg, nil
}

// --- methodREQTailFile

type methodREQTailFile struct {
	commandOrEvent CommandOrEvent
}

func (m methodREQTailFile) getKind() CommandOrEvent {
	return m.commandOrEvent
}

// handler to run a tailing of files with timeout context. The handler will
// return the output of the command run back to the calling publisher
// as a new message.
func (m methodREQTailFile) handler(proc process, message Message, node string) ([]byte, error) {
	log.Printf("<--- TailFile REQUEST received from: %v, containing: %v", message.FromNode, message.Data)

	proc.processes.wg.Add(1)
	go func() {
		defer proc.processes.wg.Done()

		fp := message.MethodArgs[0]

		var ctx context.Context
		var cancel context.CancelFunc

		if message.MethodTimeout != 0 {
			ctx, cancel = context.WithTimeout(proc.ctx, time.Second*time.Duration(message.MethodTimeout))
		} else {
			ctx, cancel = context.WithCancel(proc.ctx)
		}

		outCh := make(chan []byte)
		t, err := tail.TailFile(fp, tail.Config{Follow: true, Location: &tail.SeekInfo{
			Offset: 0,
			Whence: os.SEEK_END,
		}})
		if err != nil {
			er := fmt.Errorf("error: methodREQToTailFile: tailFile: %v", err)
			log.Printf("%v\n", er)
			sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
		}

		proc.processes.wg.Add(1)
		go func() {
			defer proc.processes.wg.Done()

			for {
				select {
				case line := <-t.Lines:
					outCh <- []byte(line.Text + "\n")
				case <-ctx.Done():
					return
				}

			}
		}()

		for {
			select {
			case <-ctx.Done():
				cancel()
				// Close the lines channel so we exit the reading lines
				// go routine.
				// close(t.Lines)
				er := fmt.Errorf("info: method timeout reached, canceling: %v", message)
				sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)

				return
			case out := <-outCh:

				// Prepare and queue for sending a new message with the output
				// of the action executed.
				newReplyMessage(proc, message, out)
			}
		}

	}()

	ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
	return ackMsg, nil
}

// ---
type methodREQnCliCommandCont struct {
	commandOrEvent CommandOrEvent
}

func (m methodREQnCliCommandCont) getKind() CommandOrEvent {
	return m.commandOrEvent
}

// Handler to run REQnCliCommandCont, which is the same as normal
// Cli command, but can be used when running a command that will take
// longer time and you want to send the output of the command continually
// back as it is generated, and not just when the command is finished.
func (m methodREQnCliCommandCont) handler(proc process, message Message, node string) ([]byte, error) {
	log.Printf("<--- CLInCommandCont REQUEST received from: %v, containing: %v", message.FromNode, message.Data)

	// Execute the CLI command in it's own go routine, so we are able
	// to return immediately with an ack reply that the message was
	// received, and we create a new message to send back to the calling
	// node for the out put of the actual command.
	proc.processes.wg.Add(1)
	go func() {
		defer proc.processes.wg.Done()

		c := message.MethodArgs[0]
		a := message.MethodArgs[1:]

		ctx, cancel := context.WithTimeout(proc.ctx, time.Second*time.Duration(message.MethodTimeout))

		outCh := make(chan []byte)

		proc.processes.wg.Add(1)
		go func() {
			proc.processes.wg.Done()

			cmd := exec.CommandContext(ctx, c, a...)

			// Using cmd.StdoutPipe here so we are continuosly
			// able to read the out put of the command.
			outReader, err := cmd.StdoutPipe()
			if err != nil {
				er := fmt.Errorf("error: methodREQnCliCommandCont: cmd.StdoutPipe failed : %v, message: %v", err, message)
				sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
				log.Printf("%v\n", er)

				log.Printf("error: %v\n", err)
			}

			ErrorReader, err := cmd.StderrPipe()
			if err != nil {
				er := fmt.Errorf("error: methodREQnCliCommandCont: cmd.StderrPipe failed : %v, message: %v", err, message)
				sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
				log.Printf("%v\n", er)

				log.Printf("error: %v\n", err)
			}

			if err := cmd.Start(); err != nil {
				er := fmt.Errorf("error: methodREQnCliCommandCont: cmd.Start failed : %v, message: %v", err, message)
				sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
				log.Printf("%v\n", er)

			}

			// Also send error messages that might happen during the time
			// cmd.Start runs.
			// Putting the scanner.Text value on a channel so we can make
			// the scanner non-blocking, and also check context cancelation.
			go func() {
				errCh := make(chan string, 1)

				scanner := bufio.NewScanner(ErrorReader)
				for scanner.Scan() {
					select {
					case errCh <- scanner.Text():
						er := fmt.Errorf("error: methodREQnCliCommandCont: cmd.Start failed : %v, message: %v, error_output: %v", err, message, <-errCh)
						sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
						log.Printf("%v\n", er)
					case <-ctx.Done():
						return
					}
				}
			}()

			scanner := bufio.NewScanner(outReader)
			for scanner.Scan() {
				select {
				case outCh <- []byte(scanner.Text() + "\n"):
				case <-ctx.Done():
					return
				}

			}

		}()

		// Check if context timer or command output were received.
		for {
			select {
			case <-ctx.Done():
				cancel()
				er := fmt.Errorf("info: methodREQnCliCommandCont: method timeout reached, canceling: %v", message)
				sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
				return
			case out := <-outCh:
				// Prepare and queue for sending a new message with the output
				// of the action executed.
				newReplyMessage(proc, message, out)
			}
		}
	}()

	ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
	return ackMsg, nil
}

// ---

type methodREQToSocket struct {
	commandOrEvent CommandOrEvent
}

func (m methodREQToSocket) getKind() CommandOrEvent {
	return m.commandOrEvent
}

// TODO:
// Handler to write to unix socket file.
func (m methodREQToSocket) handler(proc process, message Message, node string) ([]byte, error) {

	for _, d := range message.Data {
		// Write the data to the socket here.
		fmt.Printf("Info: Data to write to socket: %v\n", d)
	}

	ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
	return ackMsg, nil
}

// ----