mirrors/ctrl
mirrors/ctrl
1
0
Fork 0
mirror of https://github.com/postmannen/ctrl.git synced 2024-12-15 17:51:15 +00:00
Code Activity

updated comments and README

Browse source
This commit is contained in:
postmannen 2021-03-11 17:14:43 +01:00
parent f5b0d79981
commit 80096efe24
3 changed files with 210 additions and 125 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

260
README.md
View file

@ -1,39 +1,24 @@
# steward # steward
## What is it ?
Command And Control anything asynchronously. Command And Control anything asynchronously.
Send shell commands to control your servers by passing a message that will have guaranteed delivery if/when the subsribing node is available. Or for example send logs or metrics from an end node back to a central log subscriber. Send shell commands to control your servers by passing a message that will have guaranteed delivery if/when the subsribing node is available. Or for example send logs or metrics from an end node back to a central log subscriber.
The idea is to build and use a pure message passing architecture for the commands back and forth from nodes, with guaranteed delivery. A node can be a server running any host operating system, a container living in the cloud somewhere, a rapsberry pi, or something else that needs to be controlled that have an operating system installed . The message passing backend used is <https://nats.io> The idea is to build and use a pure message passing architecture for the commands back and forth from nodes, where delivery is guaranteed, and where all of the processes in the system are running concurrently so if something breaks or some process is slow it will not affect the handling and delivery of the other messages in the system.
```text By default the system guarantees that the order of the messages are handled by the subscriber in the order they where sent. There have also been implemented a special type `NOSEQ` which will allow messages to be handled within that process in a not sequential manner. This is handy for jobs that will run for a long time, and where other messages are not dependent on it's result.
┌─────────────────┐
│ │
│ │
│ ┌────────────┐ │
│ │ Edge Unit │ │
┌─────────────────┐ ┌─────────────────┐ │ └────────────┘ │
│ │ │ │ ────Event────▶ │ │
│ ┌────────────┐ │ ─────Event────▶ │ ┌───────────┐ │ ◀────ACK ───── │ │
│ │ Management │ │ │ │ Message │ │ └─────────────────┘
│ │ station │ │ │ │ broker │ │ ┌─────────────────┐
│ │ │ │ │ └───────────┘ │ │ │
│ └────────────┘ │ ◀─────ACK ───── │ │ ────Event────▶ │ │
│ │ │ │ ◀────ACK ───── │ ┌────────────┐ │
└─────────────────┘ └─────────────────┘ │ │ Edge Unit │ │
│ └────────────┘ │
│ │
│ │
└─────────────────┘
```
Why ? A node can be a server running any host operating system, a container living in the cloud somewhere, a rapsberry pi, or something else that needs to be controlled that have an operating system installed . The message passing backend used is <https://nats.io>
## Why ?
With existing solutions there is often either a push or a pull kind of setup. With existing solutions there is often either a push or a pull kind of setup.
In a push setup the commands to be executed is pushed to the receiver, but if a command fails because for example a broken network link it is up to you as an administrator to detect those failures and retry them at a later time until it is executed successfully. In a push setup the commands to be executed is pushed to the receiver, but if a command fails because for example a broken network link it is up to you as an administrator to detect those failures and retry them at a later time until it is executed successfully.
In a pull setup an agent is installed at the Edge unit, and the configuration or commands to execute locally are pulled from a central repository. With this kind of setup you can be pretty certain that sometime in the future the Edge unit will reach it's desired state, but you don't know when. And if you want to know the current state you will need to have some second service which gives you that. In a pull setup an agent is installed at the Edge unit, and the configuration or commands to execute locally are pulled from a central repository. With this kind of setup you can be pretty certain that sometime in the future the node will reach it's desired state, but you don't know when. And if you want to know the current state you will need to have some second service which gives you that information.
In it's simplest form the idea about using an event driven system as the core for management of Edge units is that the sender/publisher are fully decoupled from the receiver/subscriber. We can get an acknowledge if a message is received or not, and with this functionality we will at all times know the current state of the receiving end. We can also add information in the ACK message if the command sent to the receiver was successful or not by appending the actual output of the command. In it's simplest form the idea about using an event driven system as the core for management of Edge units is that the sender/publisher are fully decoupled from the receiver/subscriber. We can get an acknowledge if a message is received or not, and with this functionality we will at all times know the current state of the receiving end. We can also add information in the ACK message if the command sent to the receiver was successful or not by appending the actual output of the command.
@ -41,23 +26,34 @@ In it's simplest form the idea about using an event driven system as the core fo
All code in this repository are to be concidered not-production-ready. The code are the attempt to concretize the idea of a purely async management system where the controlling unit is decoupled from the receiving unit, and that that we know the state of all the receiving units at all times. All code in this repository are to be concidered not-production-ready. The code are the attempt to concretize the idea of a purely async management system where the controlling unit is decoupled from the receiving unit, and that that we know the state of all the receiving units at all times.
## Terminology
- Node: Something with an operating system that have network available. This can be a server, a cloud instance, a container, or other.
- Process: One message handler running in it's own thread with 1 subject for sending and 1 for reply.
- Message:
- Command: Something to be executed on the message received. An example can be a shell command.
- Event: Something that have happened. An example can be transfer of syslog data from a host.
## Features ## Features
- Send a message with some command or event to some node and be sure that the message is delivered. If for example the network link is down, the message will be retried until succesful. - By default the system guarantees that the order of the messages are handled by the subscriber in the order they where sent. So if a network link is down when the message is being sent, it will automatically be rescheduled at the specified interval with the given number of retries.
- Error messages can or will be sent back to the central upon failure. - There have been implemented a special type `NOSEQ` which will allow messages to be handled within that process in a not sequential manner. This is handy for jobs that will run for a long time, and where other messages are not dependent on it's result.
- The handling of all messages is done by spawning up a process for the handling in it's own thread, this allows us for individually keep the state for each message both in regards to ACK an Errors handled all within the thread that was spawned for owning the message (if that made sense ??). - Error messages will be sent back to the central error handler upon failure on a node.
- Processes for handling messages on a host can and might be automatically restarted upon failure, or asked to just terminate and send a message back to the operator that something have gone seriously wrong. This is right now just partially implemented to test that the concept works. - The handling of all messages is done by spawning up a process for the handling the message in it's own thread. This allows us to individually down to the message level keep the state for each message both in regards to ACK's, error handling, send retries, and rerun of a method for a message if the first run was not successful.
- Processes for handling messages on a host can be restarted upon failure, or asked to just terminate and send a message back to the operator that something have gone seriously wrong. This is right now just partially implemented to test that the concept works.
- Processes on the publishing node for handling incomming messages for new nodes will automatically be spawned when needed if it does not already exist. - Processes on the publishing node for handling incomming messages for new nodes will automatically be spawned when needed if it does not already exist.
- Publishing processes will potentially be able to send to all nodes. It is the subscribing nodes who will limit from where and what they will receive from. - Publishing processes will potentially be able to send to all nodes. It is the subscribing nodes who will limit from where and what they will receive from.
- Messages not fully processed or not started yet will be automatically handled in chronological order if the service is restarted. - Messages not fully processed or not started yet will be automatically handled in chronological order if the service is restarted since the current state of all the messages being processed are stored on the local node in a key value store until they are finished.
- All messages processed by a publisher will be written to a log file as they are processed, with all the information needed to recreate the same message. - All messages processed by a publisher will be written to a log file as they are processed, with all the information needed to recreate the same message if needed, or it can be used for auditing.
- All handling down to the process and message level are handled concurrently. So if there are problems handling one message sent to a node on a subject it will not affect the messages being sent to other nodes, or other messages sent on other subjects to the same host. - All handling down to the process and message level are handled concurrently. So if there are problems handling one message sent to a node on a subject it will not affect the messages being sent to other nodes, or other messages sent on other subjects to the same host.
@ -68,66 +64,12 @@ All code in this repository are to be concidered not-production-ready. The code
- Message types of both ACK and NACK, so we can decide if we want or don't want an Acknowledge if a message was delivered succesfully. - Message types of both ACK and NACK, so we can decide if we want or don't want an Acknowledge if a message was delivered succesfully.
Example: We probably want an ACK when sending some CLICommand to be executed, but we don't care for an acknowledge (NACK) when we send an "hello I'm here" event. Example: We probably want an ACK when sending some CLICommand to be executed, but we don't care for an acknowledge (NACK) when we send an "hello I'm here" event.
- Prometheus exporters for Metrics - Prometheus exporters for Metrics
- More will come. In active development. - More will come. In active development.
## Concepts/Ideas
### Terminology
- Node: Something with an operating system that have network available. This can be a server, a cloud instance, a container, or other.
- Process: One message handler running in it's own thread with 1 subject for sending and 1 for reply.
- Message:
- Command: Something to be executed on the message received. An example can be a shell command.
- Event: Something that have happened. An example can be transfer of syslog data from a host.
### Naming
#### Subject
Subject naming are case sensitive, and can not contain the space are the tab character.
`<nodename>.<command/event>.<method>`
Nodename: Are the hostname of the device. This do not have to be resolvable via DNS, it is just a unique name for the host to receive the message.
Command/Event: Are type of message sent. `CommandACK`/`EventACK`/`CommandNACK`/`EventNACK`. Description of the differences are mentioned earlier.\
Info: The command/event which is called a MessageType are present in both the Subject structure and the Message structure. The reason for this is that it is used both in the naming of a subject, and in the message for knowing what kind of message it is and how to handle it.
Method: Are the functionality the message provide. Example could be `CLICommand` or `Syslogforwarding`
##### Complete subject example
For syslog of type event to a host named "ship1"
`ship1.EventACK.Syslogforwarding`
and for a shell command of type command to a host named "ship2"
`ship2.CommandACK.CLICommand
`
## TODO
- FIX so it can handle multiple slices of input for inmsg.txt
- Add config file options to use when starting up the program for options.
- Rename CLICommand to cli
- Make a scraper that first send an EventACK, and the content of the scraping is returned by a node as a new EventACK back the where the initial event originated.
- Go through all processes and check that the error is handled correctly, and also reported back on the error subject to the master supervisor.
- Implement a log scraper method in `tail -f` style ?
- Implement a web scraper method ?
- Encryption between Node instances and brokers.
- Authentication between node instances and brokers.
## Howto ## Howto
### Build and Run ### Build and Run
@ -165,35 +107,82 @@ One the nodes out there
Use the `--help` flag to get all possibilities. Use the `--help` flag to get all possibilities.
### Sending messages with commands or events ### Message fields explanation
Right now there are to types of messages. ```go
// The node to send the message to
- Commands, are some command you typically want to run on a node, wait for it to finish, get an ACK back with the result contained in the ACK message. Think like running a shell command on some remote host. toNode
- Events, are something you just want to deliver, and wait to get an ACK back that it was delivered succesfully. Think like forwarding logs to some host, you just want to be sure it was delivered. // The Unique ID of the message
data
Both Commands and Events can be either ACK or NACK // method, what is this message doing, etc. CLI, syslog, etc.
method
The types are // Normal Reply wait timeout
timeout
```text // Normal Resend retries
CommandACK retries
CommandNACK // The timeout of the new message created via a request event.
EventACK requestTimeout
EventNACK // The retries of the new message created via a request event.
requestRetries
// Timeout for long a process should be allowed to operate
methodTimeout
``` ```
### How to send a Message ### How to send a Message
Right now the API for sending a message from one node to another node is by pasting a structured JSON object into a file called `inmsg.txt` living alongside the binary. This file will be watched continously, and when updated the content will be picked up, parsed, and if OK it will be sent a message to the node specified. Right now the API for sending a message from one node to another node is by pasting a structured JSON object into a file called `inmsg.txt` living alongside the binary. This file will be watched continously, and when updated the content will be picked up, umarshaled, and if OK it will be sent a message to the node specified in the `toNode` field.
Currently there is one Command message type for running shell comands and one event message type for sending logs implemented. More will come. The `method` is what defines what the event will do. The preconfigured methods are:
NB: The keys and the predefined values like `CommandACK` are case sensitive. ```go
// Execute a CLI command in for example bash or cmd.
// This is a command type, so the output of the command executed
// will directly showed in the ACK message received.
CLICommand Method = "CLICommand"
// 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.
CLICommandRequest Method = "CLICommandRequest"
// 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 NOSEQ method will process messages as they are recived,
// and the reply back will be sent as soon as the process is
// done. No order are preserved.
CLICommandRequestNOSEQ Method = "CLICommandRequestNOSEQ"
// Will generate a reply for a CLICommandRequest.
// This type is normally not used by the user when creating
// a message. It is used in creating the reply message with
// request messages. It is also used when defining a process to
// start up for receiving the CLICommand request messages.
CLICommandReply Method = "CLICommandReply"
// Send text logging to some host.
// A file with the full subject+hostName will be created on
// the receiving end.
TextLogging Method = "TextLogging"
// Send Hello I'm here message.
SayHello Method = "SayHello"
// Error log methods to centralError node.
ErrorLog Method = "ErrorLog"
// Echo request will ask the subscriber for a
// reply generated as a new message, and sent back to where
// the initial request was made.
ECHORequest Method = "ECHORequest"
// Will generate a reply for a ECHORequest
ECHOReply Method = "ECHOReply"
```
NB: Both the keys and the values used are case sensitive.
#### Sending a command from one Node to Another Node #### Sending a command from one Node to Another Node
Example JSON for pasting a message of type command into the `inmsg.txt` file Example JSON for appending a message of type command into the `inmsg.txt` file
```json ```json
[ [
@ -272,6 +261,69 @@ You can save the content to myfile.JSON and append it to `inmsg.txt`
The content of `inmsg.txt` will be erased as messages a processed. The content of `inmsg.txt` will be erased as messages a processed.
## Concepts/Ideas
### Naming
#### Subject
`<nodename>.<command/event>.<method>`
Nodename: Are the hostname of the device. This do not have to be resolvable via DNS, it is just a unique name for the host to receive the message.
Command/Event: Are type of message sent. `CommandACK`/`EventACK`/`CommandNACK`/`EventNACK`. Description of the differences are mentioned earlier.\
Info: The command/event which is called a MessageType are present in both the Subject structure and the Message structure. The reason for this is that it is used both in the naming of a subject, and in the message for knowing what kind of message it is and how to handle it.
Method: Are the functionality the message provide. Example could be `CLICommand` or `Syslogforwarding`
##### Complete subject example
For syslog of type event to a host named "ship1"
`ship1.EventACK.Syslogforwarding`
and for a shell command of type command to a host named "ship2"
`ship2.CommandACK.CLICommand`
## TODO
- FIX so it can handle multiple slices of input for inmsg.txt
- Add config file options to use when starting up the program for options.
- Rename CLICommand to cli
- Make a scraper that first send an EventACK, and the content of the scraping is returned by a node as a new EventACK back the where the initial event originated.
- Go through all processes and check that the error is handled correctly, and also reported back on the error subject to the master supervisor.
- Implement a log scraper method in `tail -f` style ?
- Implement a web scraper method ?
- Encryption between Node instances and brokers.
- Authentication between node instances and brokers.
## Overview ## Overview
All parts of the system like processes, method handlers, messages, error handling are running concurrently.
If one process hangs on a long running message method it will not affect the rest of the system.
### Publisher
- A message in valid format is appended to the in pipe.
- The message is picked up by the system and put on a FIFO ringbuffer.
- The method type of the message is checked, a subject is created based on the content of the message, and a process to handle that message type for that specific receiving node is started if it does not exist.
- The message is then serialized to binary format, and sent to the subscriber on the receiving node.
- If the message is expected to be ACK'ed by the subcriber then the publisher will wait for an ACK if the message was delivered. If an ACK was not received within the defined timeout the message will be resent. The amount of retries are defined within the message.
### Subscriber
- The subscriber will need to have a listener started on the wanted subject to be able to receive messages.
- When a message have been deserialized, it will lookup the correct handler for the method type specified within the message, and execute that handler.
- If the output of the method called is supposed to be returned to the publiser it will do so, or else it will be finish, and pick up the next message in the queue.
![overview](steward.svg) ![overview](steward.svg)

12
message-and-subject.go
View file

@ -9,6 +9,7 @@ import (
// --- Message // --- Message
type Message struct { type Message struct {
// The node to send the message to
ToNode node `json:"toNode" yaml:"toNode"` ToNode node `json:"toNode" yaml:"toNode"`
// The Unique ID of the message // The Unique ID of the message
ID int `json:"id" yaml:"id"` ID int `json:"id" yaml:"id"`
@ -17,14 +18,19 @@ type Message struct {
// method, what is this message doing, etc. CLI, syslog, etc. // method, what is this message doing, etc. CLI, syslog, etc.
Method Method `json:"method" yaml:"method"` Method Method `json:"method" yaml:"method"`
FromNode node FromNode node
// Reply wait timeout // Normal Reply wait timeout
Timeout int `json:"timeout" yaml:"timeout"` Timeout int `json:"timeout" yaml:"timeout"`
// Resend retries // Normal Resend retries
Retries int `json:"retries" yaml:"retries"` Retries int `json:"retries" yaml:"retries"`
// The timeout of the new message created via a request event.
RequestTimeout int `json:"requestTimeout" yaml:"requestTimeout"`
// The retries of the new message created via a request event.
RequestRetries int `json:"requestRetries" yaml:"requestRetries"`
// Timeout for long a process should be allowed to operate
MethodTimeout int `json:"methodTimeout" yaml:"methodTimeout"`
// done is used to signal when a message is fully processed. // done is used to signal when a message is fully processed.
// This is used when choosing when to move the message from // This is used when choosing when to move the message from
// the ringbuffer into the time series log. // the ringbuffer into the time series log.
MethodTimeout int `json:"methodTimeout" yaml:"methodTimeout"`
done chan struct{} done chan struct{}
} }

61
subscriberMethodTypes.go
View file

@ -41,38 +41,67 @@ import (
"time" "time"
) )
// 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 // The constants that will be used throughout the system for
// when specifying what kind of Method to send or work with. // when specifying what kind of Method to send or work with.
const ( const (
// Shell command to be executed via f.ex. bash // Execute a CLI command in for example bash or cmd.
// This is a command type, so the output of the command executed
// will directly showed in the ACK message received.
// The data field is a slice of strings where the first string
// value should be the command, and the following the arguments.
CLICommand Method = "CLICommand" CLICommand Method = "CLICommand"
// Shell command to be executed via f.ex. bash // 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.
CLICommandRequest Method = "CLICommandRequest" CLICommandRequest Method = "CLICommandRequest"
// Shell command to be executed via f.ex. bash // 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 NOSEQ method will process messages as they are recived, // The NOSEQ method will process messages as they are recived,
// and the reply back will be sent as soon as the process is // and the reply back will be sent as soon as the process is
// done. No order are preserved. // done. No order are preserved.
// The data field is a slice of strings where the first string
// value should be the command, and the following the arguments.
CLICommandRequestNOSEQ Method = "CLICommandRequestNOSEQ" CLICommandRequestNOSEQ Method = "CLICommandRequestNOSEQ"
// Will generate a reply for a CLICommandRequest // Will generate a reply for a CLICommandRequest.
// This type is normally not used by the user when creating
// a message. It is used in creating the reply message with
// request messages. It is also used when defining a process to
// start up for receiving the CLICommand request messages.
// The data field is a slice of strings where the first string
// value should be the command, and the following the arguments.
CLICommandReply Method = "CLICommandReply" CLICommandReply Method = "CLICommandReply"
// Send text logging to some host // Send text logging to some host.
// 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.
TextLogging Method = "TextLogging" TextLogging Method = "TextLogging"
// Send Hello I'm here message // Send Hello I'm here message.
SayHello Method = "SayHello" SayHello Method = "SayHello"
// Error log methods to centralError // Error log methods to centralError node.
ErrorLog Method = "ErrorLog" ErrorLog Method = "ErrorLog"
// Echo request will ask the subscriber for a // Echo request will ask the subscriber for a
// reply generated as a new message // reply generated as a new message, and sent back to where
// the initial request was made.
ECHORequest Method = "ECHORequest" ECHORequest Method = "ECHORequest"
// Will generate a reply for a ECHORequest // Will generate a reply for a ECHORequest
ECHOReply Method = "ECHOReply" ECHOReply Method = "ECHOReply"
) )
// Method is used to specify the actual function/method that
// is represented in a typed manner.
type Method string
// The mapping of all the method constants specified, what type // The mapping of all the method constants specified, what type
// it references, and the kind if it is an Event or Command, and // it references, and the kind if it is an Event or Command, and
// if it is ACK or NACK. // if it is ACK or NACK.
@ -127,6 +156,9 @@ func (m Method) getHandler(method Method) methodHandler {
return mh 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.
type MethodsAvailable struct { type MethodsAvailable struct {
methodhandlers map[Method]methodHandler methodhandlers map[Method]methodHandler
} }
@ -165,9 +197,6 @@ func (m methodCLICommand) getKind() CommandOrEvent {
} }
func (m methodCLICommand) handler(proc process, message Message, node string) ([]byte, error) { func (m methodCLICommand) handler(proc process, message Message, node string) ([]byte, error) {
// Since the command to execute is at the first position in the
// slice we need to slice it out. The arguments are at the
// remaining positions.
c := message.Data[0] c := message.Data[0]
a := message.Data[1:] a := message.Data[1:]
cmd := exec.Command(c, a...) cmd := exec.Command(c, a...)
@ -212,8 +241,6 @@ func (m methodTextLogging) handler(proc process, message Message, node string) (
if err != nil { if err != nil {
log.Printf("error: methodEventTextLogging.handler: failed to write to file: %v\n", err) log.Printf("error: methodEventTextLogging.handler: failed to write to file: %v\n", err)
} }
//s.subscriberServices.logCh <- []byte(d)
} }
ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID)) ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))