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

Changed to upercase const values for command and events, and added choosing of prometheus port

Browse source
This commit is contained in:
postmannen 2021-02-22 09:33:37 +01:00
parent 25de8982b2
commit e19d76a5fc
13 changed files with 55 additions and 70 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
README.md
View file

@ -81,20 +81,20 @@ Subject naming are case sensitive, and can not contain the space are the tab cha
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. 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. `command` or `event`. Description of the differences are mentioned earlier.\ 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. 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 `shellcommand` or `syslogforwarding` Method: Are the functionality the message provide. Example could be `Shellcommand` or `Syslogforwarding`
##### Complete subject example ##### Complete subject example
For syslog of type event to a host named "ship1" For syslog of type event to a host named "ship1"
`ship1.event.syslogforwarding` `ship1.EventACK.Syslogforwarding`
and for a shell command of type command to a host named "ship2" and for a shell command of type command to a host named "ship2"
`ship2.command.shellcommand` `ship2.CommandACK.Shellcommand`
## TODO ## TODO
@ -146,6 +146,8 @@ One the nodes out there
`./steward --node="ship1"` & `./steward --node="ship1"` and so on. `./steward --node="ship1"` & `./steward --node="ship1"` and so on.
Use the `--help` flag to get all possibilities.
### Sending messages with commands or events ### Sending messages with commands or events
Right now there are to types of messages. Right now there are to types of messages.
@ -169,8 +171,8 @@ Example JSON for pasting a message of type command into the `inmsg.txt` file
"toNode": "ship1", "toNode": "ship1",
"data": ["bash","-c","ls -l ../"], "data": ["bash","-c","ls -l ../"],
"commandOrEvent":"commandACK", "commandOrEvent":"CommandACK",
"method":"shellCommand" "method":"ShellCommand"
} }
] ]
@ -184,16 +186,16 @@ To send specify more messages at once do
"toNode": "ship1", "toNode": "ship1",
"data": ["bash","-c","ls -l ../"], "data": ["bash","-c","ls -l ../"],
"commandOrEvent":"commandACK", "commandOrEvent":"CommandACK",
"method":"shellCommand" "method":"ShellCommand"
}, },
{ {
"toNode": "ship2", "toNode": "ship2",
"data": ["bash","-c","ls -l ../"], "data": ["bash","-c","ls -l ../"],
"commandOrEvent":"commandACK", "commandOrEvent":"CommandACK",
"method":"shellCommand" "method":"ShellCommand"
} }
] ]
@ -212,8 +214,8 @@ The content of `inmsg.txt` will be erased as messages a processed.
{ {
"toNode": "central", "toNode": "central",
"data": ["some message sent from a ship for testing\n"], "data": ["some message sent from a ship for testing\n"],
"commandOrEvent":"eventACK", "commandOrEvent":"EventACK",
"method":"textLogging" "method":"TextLogging"
} }
] ]
``` ```

3
cmd/main.go
View file

@ -15,6 +15,7 @@ func main() {
nodeName := flag.String("node", "0", "some unique string to identify this Edge unit") nodeName := flag.String("node", "0", "some unique string to identify this Edge unit")
brokerAddress := flag.String("brokerAddress", "0", "the address of the message broker") brokerAddress := flag.String("brokerAddress", "0", "the address of the message broker")
profilingPort := flag.String("profilingPort", "", "The number of the profiling port") profilingPort := flag.String("profilingPort", "", "The number of the profiling port")
promHostAndPort := flag.String("promHostAndPort", ":2112", "host and port for prometheus listener, e.g. localhost:2112")
//isCentral := flag.Bool("isCentral", false, "used to indicate that this is the central master that will subscribe to error message subjects") //isCentral := flag.Bool("isCentral", false, "used to indicate that this is the central master that will subscribe to error message subjects")
flag.Parse() flag.Parse()
@ -27,7 +28,7 @@ func main() {
} }
s, err := steward.NewServer(*brokerAddress, *nodeName) s, err := steward.NewServer(*brokerAddress, *nodeName, *promHostAndPort)
if err != nil { if err != nil {
log.Printf("error: failed to connect to broker: %v\n", err) log.Printf("error: failed to connect to broker: %v\n", err)
os.Exit(1) os.Exit(1)

8
commandOrEventType.go
View file

@ -41,17 +41,17 @@ const (
// delivered back in the reply ack message. // delivered back in the reply ack message.
// The message should contain the unique ID of the // The message should contain the unique ID of the
// command. // command.
CommandACK CommandOrEvent = "commandACK" CommandACK CommandOrEvent = "CommandACK"
// Same as above, but No ACK. // Same as above, but No ACK.
CommandNACK CommandOrEvent = "commandNACK" CommandNACK CommandOrEvent = "CommandNACK"
// Same as above, but No ACK // Same as above, but No ACK
// Event, wait for and return the ACK message. This means // Event, wait for and return the ACK message. This means
// that the command should be executed immediately // that the command should be executed immediately
// and that we should get the confirmation if it // and that we should get the confirmation if it
// was successful or not. // was successful or not.
EventACK CommandOrEvent = "eventACK" EventACK CommandOrEvent = "EventACK"
// Same as above, but No ACK. // Same as above, but No ACK.
EventNACK CommandOrEvent = "eventNACK" EventNACK CommandOrEvent = "EventNACK"
// eventCommand, just wait for the ACK that the // eventCommand, just wait for the ACK that the
// message is received. What action happens on the // message is received. What action happens on the
// receiving side is up to the received to decide. // receiving side is up to the received to decide.

30
example/orig.json
View file

@ -1,30 +0,0 @@
[
{
"subject":
{
"node":"ship1",
"messageType":"command",
"method":"shellcommand",
"domain":"shell"
},
"message":
{
"data": ["bash","-c","uname -a"],
"messageType":"Command"
}
},
{
"subject":
{
"node":"ship2",
"messageType":"command",
"method":"shellcommand",
"domain":"shell"
},
"message":
{
"data": ["bash","-c","uname -a"],
"messageType":"Command"
}
}
]

4
example/orig-toCentral-sayHello.json → example/toCentral-sayHello.json
View file

@ -2,7 +2,7 @@
{ {
"toNode": "central", "toNode": "central",
"data": [""], "data": [""],
"commandOrEvent":"eventNACK", "commandOrEvent":"EventNACK",
"method":"sayHello" "method":"SayHello"
} }
] ]

4
example/orig-toCentral.json → example/toCentral.json
View file

@ -2,7 +2,7 @@
{ {
"toNode": "central", "toNode": "central",
"data": ["some message sent from a ship for testing\n"], "data": ["some message sent from a ship for testing\n"],
"commandOrEvent":"eventACK", "commandOrEvent":"EventACK",
"method":"textLogging" "method":"TextLogging"
} }
] ]

4
example/orig-toShip1.json → example/toShip1.json
View file

@ -3,8 +3,8 @@
"toNode": "ship1", "toNode": "ship1",
"data": ["bash","-c","netstat -an|grep -i listen"], "data": ["bash","-c","netstat -an|grep -i listen"],
"commandOrEvent":"commandACK", "commandOrEvent":"CommandACK",
"method":"shellCommand" "method":"ShellCommand"
} }
] ]

4
example/orig-toShip1and2.json → example/toShip1and2.json
View file

@ -2,8 +2,8 @@
{ {
"toNode": "*", "toNode": "*",
"data": ["bash","-c","tree ../"], "data": ["bash","-c","tree ../"],
"commandOrEvent":"commandACK", "commandOrEvent":"CommandACK",
"method":"shellCommand" "method":"ShellCommand"
} }
] ]

4
example/orig-toShip2.json → example/toShip2.json
View file

@ -2,8 +2,8 @@
{ {
"toNode": "ship2", "toNode": "ship2",
"data": ["bash","-c","tree ../"], "data": ["bash","-c","tree ../"],
"commandOrEvent":"commandACK", "commandOrEvent":"CommandACK",
"method":"shellCommand" "method":"ShellCommand"
} }
] ]

BIN
incommmingBuffer.db
View file

Binary file not shown.

18
prometheus.go
View file

@ -2,6 +2,8 @@ package steward
import ( import (
"fmt" "fmt"
"log"
"net"
"net/http" "net/http"
"github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus"
@ -19,14 +21,17 @@ type metrics struct {
sayHelloNodes map[node]struct{} sayHelloNodes map[node]struct{}
// The channel to pass metrics that should be processed // The channel to pass metrics that should be processed
metricsCh chan metricType metricsCh chan metricType
// host and port where prometheus metrics will be exported
hostAndPort string
} }
// HERE: // HERE:
func newMetrics() *metrics { func newMetrics(hostAndPort string) *metrics {
m := metrics{ m := metrics{
sayHelloNodes: make(map[node]struct{}), sayHelloNodes: make(map[node]struct{}),
} }
m.metricsCh = make(chan metricType) m.metricsCh = make(chan metricType)
m.hostAndPort = hostAndPort
return &m return &m
} }
@ -100,6 +105,13 @@ func (s *server) startMetrics() {
} }
}() }()
http.Handle("/metrics", promhttp.Handler()) //http.Handle("/metrics", promhttp.Handler())
http.ListenAndServe(":2112", nil) //http.ListenAndServe(":2112", nil)
n, err := net.Listen("tcp", s.metrics.hostAndPort)
if err != nil {
log.Printf("error: failed to open prometheus listen port: %v\n", err)
}
m := http.NewServeMux()
m.Handle("/metrics", promhttp.Handler())
http.Serve(n, m)
} }

14
publisher.go
View file

@ -64,7 +64,7 @@ type server struct {
} }
// newServer will prepare and return a server type // newServer will prepare and return a server type
func NewServer(brokerAddress string, nodeName string) (*server, error) { func NewServer(brokerAddress string, nodeName string, promHostAndPort string) (*server, error) {
conn, err := nats.Connect(brokerAddress, nil) conn, err := nats.Connect(brokerAddress, nil)
if err != nil { if err != nil {
log.Printf("error: nats.Connect failed: %v\n", err) log.Printf("error: nats.Connect failed: %v\n", err)
@ -82,7 +82,7 @@ func NewServer(brokerAddress string, nodeName string) (*server, error) {
logCh: make(chan []byte), logCh: make(chan []byte),
methodsAvailable: m.GetMethodsAvailable(), methodsAvailable: m.GetMethodsAvailable(),
commandOrEventAvailable: co.GetCommandOrEventAvailable(), commandOrEventAvailable: co.GetCommandOrEventAvailable(),
metrics: newMetrics(), metrics: newMetrics(promHostAndPort),
} }
return s, nil return s, nil
@ -112,7 +112,7 @@ func (s *server) Start() {
// Start a subscriber for shellCommand messages // Start a subscriber for shellCommand messages
{ {
fmt.Printf("Starting shellCommand subscriber: %#v\n", s.nodeName) fmt.Printf("Starting shellCommand subscriber: %#v\n", s.nodeName)
sub := newSubject("shellCommand", CommandACK, s.nodeName) sub := newSubject(ShellCommand, CommandACK, s.nodeName)
proc := s.processPrepareNew(sub, s.errorCh, processKindSubscriber) proc := s.processPrepareNew(sub, s.errorCh, processKindSubscriber)
// fmt.Printf("*** %#v\n", proc) // fmt.Printf("*** %#v\n", proc)
go s.processSpawnWorker(proc) go s.processSpawnWorker(proc)
@ -121,16 +121,16 @@ func (s *server) Start() {
// Start a subscriber for textLogging messages // Start a subscriber for textLogging messages
{ {
fmt.Printf("Starting textlogging subscriber: %#v\n", s.nodeName) fmt.Printf("Starting textlogging subscriber: %#v\n", s.nodeName)
sub := newSubject("textLogging", EventACK, s.nodeName) sub := newSubject(TextLogging, EventACK, s.nodeName)
proc := s.processPrepareNew(sub, s.errorCh, processKindSubscriber) proc := s.processPrepareNew(sub, s.errorCh, processKindSubscriber)
// fmt.Printf("*** %#v\n", proc) // fmt.Printf("*** %#v\n", proc)
go s.processSpawnWorker(proc) go s.processSpawnWorker(proc)
} }
// Start a subscriber for sayHello messages // Start a subscriber for SayHello messages
{ {
fmt.Printf("Starting sayHello subscriber: %#v\n", s.nodeName) fmt.Printf("Starting SayHello subscriber: %#v\n", s.nodeName)
sub := newSubject("sayHello", EventNACK, s.nodeName) sub := newSubject(SayHello, EventNACK, s.nodeName)
proc := s.processPrepareNew(sub, s.errorCh, processKindSubscriber) proc := s.processPrepareNew(sub, s.errorCh, processKindSubscriber)
// fmt.Printf("*** %#v\n", proc) // fmt.Printf("*** %#v\n", proc)
go s.processSpawnWorker(proc) go s.processSpawnWorker(proc)

6
subscriberMethodTypes.go
View file

@ -34,11 +34,11 @@ func (m Method) GetMethodsAvailable() MethodsAvailable {
const ( const (
// Shell command to be executed via f.ex. bash // Shell command to be executed via f.ex. bash
ShellCommand Method = "shellCommand" ShellCommand Method = "ShellCommand"
// Send text logging to some host // Send text logging to some host
TextLogging Method = "textLogging" TextLogging Method = "TextLogging"
// Send Hello I'm here message // Send Hello I'm here message
SayHello Method = "sayHello" SayHello Method = "SayHello"
) )
type MethodsAvailable struct { type MethodsAvailable struct {