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

Squashed commit of the following:

Browse source 
commit ec6ea8158d5051010c25ef848a7b7b91fa40efef
Author: postmannen <postmannen@gmail.com>
Date:   Tue Nov 19 03:48:03 2024 +0100

    Removed remaining use of REQ in naming

commit b80bb06a47cf564a88d684c1d1d4fe2e408c3c0f
Author: postmannen <postmannen@gmail.com>
Date:   Mon Nov 18 23:15:32 2024 +0100

    removed REQ in more variable names, including methods

commit af7f6fef3e0093312d2ca12740f7b27840797800
Author: postmannen <postmannen@gmail.com>
Date:   Mon Nov 18 22:54:40 2024 +0100

    renamed all the method type variables

commit e657ecb37f
Author: postmannen <postmannen@gmail.com>
Date:   Mon Nov 18 22:32:13 2024 +0100

    renamed methods constant values and removed REQ
This commit is contained in:
postmannen 2024-11-19 03:48:42 +01:00
parent debed0f791
commit b1a5406598
58 changed files with 593 additions and 6084 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

34
README.md
View file

@ -52,7 +52,7 @@ cat << EOF > .env
NODE_NAME="node1"
BROKER_ADDRESS="127.0.0,1:4444"
ENABLE_DEBUG=1
START_PUB_REQ_HELLO=60
START_PUB_HELLO=60
IS_CENTRAL_ERROR_LOGGER=0
EOF
```
@ -70,7 +70,7 @@ cat << EOF > msg.yaml
---
- toNodes:
- node1
method: REQCliCommand
method: cliCommand
methodArgs:
- "bash"
- "-c"
@ -80,7 +80,7 @@ cat << EOF > msg.yaml
echo "some config line" > /etc/my-service-config.3
systemctl restart my-service
replyMethod: REQNone
replyMethod: none
ACKTimeout: 0
EOF
@ -129,20 +129,20 @@ ctrl supports both the use of flags with env variables. An .env file can also be
| Method name| Description|
|------------|------------|
|REQOpProcessList | Get a list of the running processes|
|REQOpProcessStart | Start up a process|
|REQOpProcessStop | Stop a process|
|REQCliCommand | Will run the command given, and return the stdout output of the command when the command is done|
|REQCliCommandCont | Will run the command given, and return the stdout output of the command continously while the command runs|
|REQTailFile | Tail log files on some node, and get the result for each new line read sent back in a reply message|
|REQHttpGet | Scrape web url, and get the html sent back in a reply message|
|REQHello | Send Hello messages|
|REQCopySrc| Copy a file from one node to another node|
|REQErrorLog | Method for receiving error logs for Central error logger|
|REQNone | Don't send a reply message|
|REQToConsole | Print to stdout or stderr|
|REQToFileAppend | Append to file, can also write to unix sockets|
|REQToFile | Write to file, can also write to unix sockets|
|opProcessList | Get a list of the running processes|
|opProcessStart | Start up a process|
|opProcessStop | Stop a process|
|cliCommand | Will run the command given, and return the stdout output of the command when the command is done|
|cliCommandCont | Will run the command given, and return the stdout output of the command continously while the command runs|
|tailFile | Tail log files on some node, and get the result for each new line read sent back in a reply message|
|httpGet | Scrape web url, and get the html sent back in a reply message|
|hello | Send Hello messages|
|copySrc| Copy a file from one node to another node|
|errorLog | Method for receiving error logs for Central error logger|
|none | Don't send a reply message|
|console | Print to stdout or stderr|
|fileAppend | Append to file, can also write to unix sockets|
|file | Write to file, can also write to unix sockets|
## History

4
central_auth_key_handling.go
View file

@ -294,7 +294,7 @@ func (c *centralAuth) updateHash(proc process, message Message) {
b, err := cbor.Marshal(sortedNodesAndKeys)
if err != nil {
er := fmt.Errorf("error: methodREQKeysAllow, failed to marshal slice, and will not update hash for public keys: %v", err)
er := fmt.Errorf("error: methodKeysAllow, failed to marshal slice, and will not update hash for public keys: %v", err)
c.pki.errorKernel.errSend(proc, message, er, logError)
return
@ -307,7 +307,7 @@ func (c *centralAuth) updateHash(proc process, message Message) {
// Store the key to the db for persistence.
err = c.pki.dbUpdateHash(hash[:])
if err != nil {
er := fmt.Errorf("error: methodREQKeysAllow, failed to store the hash into the db: %v", err)
er := fmt.Errorf("error: methodKeysAllow, failed to store the hash into the db: %v", err)
c.pki.errorKernel.errSend(proc, message, er, logError)
return

174
configuration_flags.go
View file

@ -42,10 +42,10 @@ type Configuration struct {
NatsReconnectJitter int `comment:"NatsReconnectJitter in milliseconds"`
// NatsReconnectJitterTLS in seconds
NatsReconnectJitterTLS int `comment:"NatsReconnectJitterTLS in seconds"`
// REQKeysRequestUpdateInterval in seconds
REQKeysRequestUpdateInterval int `comment:"REQKeysRequestUpdateInterval in seconds"`
// REQAclRequestUpdateInterval in seconds
REQAclRequestUpdateInterval int `comment:"REQAclRequestUpdateInterval in seconds"`
// KeysUpdateInterval in seconds
KeysUpdateInterval int `comment:"KeysUpdateInterval in seconds"`
// AclUpdateInterval in seconds
AclUpdateInterval int `comment:"AclUpdateInterval in seconds"`
// The number of the profiling port
ProfilingPort string `comment:"The number of the profiling port"`
// Host and port for prometheus listener, e.g. localhost:2112
@ -102,8 +102,8 @@ type Configuration struct {
// it have not received any messages for the given amount of time.
KeepPublishersAliveFor int `comment:"KeepPublishersAliveFor number of seconds Timer that will be used for when to remove the sub process publisher. The timer is reset each time a message is published with the process, so the sub process publisher will not be removed until it have not received any messages for the given amount of time."`
// StartPubREQHello, sets the interval in seconds for how often we send hello messages to central server
StartPubREQHello int `comment:"StartPubREQHello, sets the interval in seconds for how often we send hello messages to central server"`
// StartPubHello, sets the interval in seconds for how often we send hello messages to central server
StartPubHello int `comment:"StartPubHello, sets the interval in seconds for how often we send hello messages to central server"`
// Enable the updates of public keys
EnableKeyUpdates bool `comment:"Enable the updates of public keys"`
@ -113,27 +113,25 @@ type Configuration struct {
// Start the central error logger.
IsCentralErrorLogger bool `comment:"Start the central error logger."`
// Start subscriber for hello messages
StartSubREQHello bool `comment:"Start subscriber for hello messages"`
StartSubHello bool `comment:"Start subscriber for hello messages"`
// Start subscriber for text logging
StartSubREQToFileAppend bool `comment:"Start subscriber for text logging"`
StartSubFileAppend bool `comment:"Start subscriber for text logging"`
// Start subscriber for writing to file
StartSubREQToFile bool `comment:"Start subscriber for writing to file"`
StartSubFile bool `comment:"Start subscriber for writing to file"`
// Start subscriber for reading files to copy
StartSubREQCopySrc bool `comment:"Start subscriber for reading files to copy"`
StartSubCopySrc bool `comment:"Start subscriber for reading files to copy"`
// Start subscriber for writing copied files to disk
StartSubREQCopyDst bool `comment:"Start subscriber for writing copied files to disk"`
StartSubCopyDst bool `comment:"Start subscriber for writing copied files to disk"`
// Start subscriber for Echo Request
StartSubREQCliCommand bool `comment:"Start subscriber for CLICommandRequest"`
// Start subscriber for REQToConsole
StartSubREQToConsole bool `comment:"Start subscriber for REQToConsole"`
// Start subscriber for REQHttpGet
StartSubREQHttpGet bool `comment:"Start subscriber for REQHttpGet"`
// Start subscriber for REQHttpGetScheduled
StartSubREQHttpGetScheduled bool `comment:"Start subscriber for REQHttpGetScheduled"`
StartSubCliCommand bool `comment:"Start subscriber for CLICommand"`
// Start subscriber for Console
StartSubConsole bool `comment:"Start subscriber for Console"`
// Start subscriber for HttpGet
StartSubHttpGet bool `comment:"Start subscriber for HttpGet"`
// Start subscriber for tailing log files
StartSubREQTailFile bool `comment:"Start subscriber for tailing log files"`
StartSubTailFile bool `comment:"Start subscriber for tailing log files"`
// Start subscriber for continously delivery of output from cli commands.
StartSubREQCliCommandCont bool `comment:"Start subscriber for continously delivery of output from cli commands."`
StartSubCliCommandCont bool `comment:"Start subscriber for continously delivery of output from cli commands."`
}
// NewConfiguration will return a *Configuration.
@ -157,8 +155,8 @@ func NewConfiguration() *Configuration {
flag.IntVar(&c.NatsConnectRetryInterval, "natsConnectRetryInterval", CheckEnv("NATS_CONNECT_RETRY_INTERVAL", c.NatsConnectRetryInterval).(int), "default nats retry connect interval in seconds.")
flag.IntVar(&c.NatsReconnectJitter, "natsReconnectJitter", CheckEnv("NATS_RECONNECT_JITTER", c.NatsReconnectJitter).(int), "default nats ReconnectJitter interval in milliseconds.")
flag.IntVar(&c.NatsReconnectJitterTLS, "natsReconnectJitterTLS", CheckEnv("NATS_RECONNECT_JITTER_TLS", c.NatsReconnectJitterTLS).(int), "default nats ReconnectJitterTLS interval in seconds.")
flag.IntVar(&c.REQKeysRequestUpdateInterval, "REQKeysRequestUpdateInterval", CheckEnv("REQ_KEYS_UPDATE_INTERVAL", c.REQKeysRequestUpdateInterval).(int), "default interval in seconds for asking the central for public keys")
flag.IntVar(&c.REQAclRequestUpdateInterval, "REQAclRequestUpdateInterval", CheckEnv("REQ_ACL_REQUEST_UPDATE_INTERVAL", c.REQAclRequestUpdateInterval).(int), "default interval in seconds for asking the central for acl updates")
flag.IntVar(&c.KeysUpdateInterval, "keysUpdateInterval", CheckEnv("KEYS_UPDATE_INTERVAL", c.KeysUpdateInterval).(int), "default interval in seconds for asking the central for public keys")
flag.IntVar(&c.AclUpdateInterval, "aclUpdateInterval", CheckEnv("ACL_UPDATE_INTERVAL", c.AclUpdateInterval).(int), "default interval in seconds for asking the central for acl updates")
flag.StringVar(&c.ProfilingPort, "profilingPort", CheckEnv("PROFILING_PORT", c.ProfilingPort).(string), "The number of the profiling port")
flag.StringVar(&c.PromHostAndPort, "promHostAndPort", CheckEnv("PROM_HOST_AND_PORT", c.PromHostAndPort).(string), "host and port for prometheus listener, e.g. localhost:2112")
flag.IntVar(&c.DefaultMessageTimeout, "defaultMessageTimeout", CheckEnv("DEFAULT_MESSAGE_TIMEOUT", c.DefaultMessageTimeout).(int), "default message timeout in seconds. This can be overridden on the message level")
@ -187,24 +185,23 @@ func NewConfiguration() *Configuration {
// Start of Request publishers/subscribers
flag.IntVar(&c.StartPubREQHello, "startPubREQHello", CheckEnv("START_PUB_REQ_HELLO", c.StartPubREQHello).(int), "Make the current node send hello messages to central at given interval in seconds")
flag.IntVar(&c.StartPubHello, "startPubHello", CheckEnv("START_PUB_HELLO", c.StartPubHello).(int), "Make the current node send hello messages to central at given interval in seconds")
flag.BoolVar(&c.EnableKeyUpdates, "EnableKeyUpdates", CheckEnv("ENABLE_KEY_UPDATES", c.EnableKeyUpdates).(bool), "true/false")
flag.BoolVar(&c.EnableAclUpdates, "EnableAclUpdates", CheckEnv("ENABLE_ACL_UPDATES", c.EnableAclUpdates).(bool), "true/false")
flag.BoolVar(&c.IsCentralErrorLogger, "isCentralErrorLogger", CheckEnv("IS_CENTRAL_ERROR_LOGGER", c.IsCentralErrorLogger).(bool), "true/false")
flag.BoolVar(&c.StartSubREQHello, "startSubREQHello", CheckEnv("START_SUB_REQ_HELLO", c.StartSubREQHello).(bool), "true/false")
flag.BoolVar(&c.StartSubREQToFileAppend, "startSubREQToFileAppend", CheckEnv("START_SUB_REQ_TO_FILE_APPEND", c.StartSubREQToFileAppend).(bool), "true/false")
flag.BoolVar(&c.StartSubREQToFile, "startSubREQToFile", CheckEnv("START_SUB_REQ_TO_FILE", c.StartSubREQToFile).(bool), "true/false")
flag.BoolVar(&c.StartSubREQCopySrc, "startSubREQCopySrc", CheckEnv("START_SUB_REQ_COPY_SRC", c.StartSubREQCopySrc).(bool), "true/false")
flag.BoolVar(&c.StartSubREQCopyDst, "startSubREQCopyDst", CheckEnv("START_SUB_REQ_COPY_DST", c.StartSubREQCopyDst).(bool), "true/false")
flag.BoolVar(&c.StartSubREQCliCommand, "startSubREQCliCommand", CheckEnv("START_SUB_REQ_CLI_COMMAND", c.StartSubREQCliCommand).(bool), "true/false")
flag.BoolVar(&c.StartSubREQToConsole, "startSubREQToConsole", CheckEnv("START_SUB_REQ_TO_CONSOLE", c.StartSubREQToConsole).(bool), "true/false")
flag.BoolVar(&c.StartSubREQHttpGet, "startSubREQHttpGet", CheckEnv("START_SUB_REQ_HTTP_GET", c.StartSubREQHttpGet).(bool), "true/false")
flag.BoolVar(&c.StartSubREQHttpGetScheduled, "startSubREQHttpGetScheduled", CheckEnv("START_SUB_REQ_HTTP_GET_SCHEDULED", c.StartSubREQHttpGetScheduled).(bool), "true/false")
flag.BoolVar(&c.StartSubREQTailFile, "startSubREQTailFile", CheckEnv("START_SUB_REQ_TAIL_FILE", c.StartSubREQTailFile).(bool), "true/false")
flag.BoolVar(&c.StartSubREQCliCommandCont, "startSubREQCliCommandCont", CheckEnv("START_SUB_REQ_CLI_COMMAND_CONT", c.StartSubREQCliCommandCont).(bool), "true/false")
flag.BoolVar(&c.StartSubHello, "startSubHello", CheckEnv("START_SUB_HELLO", c.StartSubHello).(bool), "true/false")
flag.BoolVar(&c.StartSubFileAppend, "startSubFileAppend", CheckEnv("START_SUB_FILE_APPEND", c.StartSubFileAppend).(bool), "true/false")
flag.BoolVar(&c.StartSubFile, "startSubFile", CheckEnv("START_SUB_FILE", c.StartSubFile).(bool), "true/false")
flag.BoolVar(&c.StartSubCopySrc, "startSubCopySrc", CheckEnv("START_SUB_COPY_SRC", c.StartSubCopySrc).(bool), "true/false")
flag.BoolVar(&c.StartSubCopyDst, "startSubCopyDst", CheckEnv("START_SUB_COPY_DST", c.StartSubCopyDst).(bool), "true/false")
flag.BoolVar(&c.StartSubCliCommand, "startSubCliCommand", CheckEnv("START_SUB_CLI_COMMAND", c.StartSubCliCommand).(bool), "true/false")
flag.BoolVar(&c.StartSubConsole, "startSubConsole", CheckEnv("START_SUB_CONSOLE", c.StartSubConsole).(bool), "true/false")
flag.BoolVar(&c.StartSubHttpGet, "startSubHttpGet", CheckEnv("START_SUB_HTTP_GET", c.StartSubHttpGet).(bool), "true/false")
flag.BoolVar(&c.StartSubTailFile, "startSubTailFile", CheckEnv("START_SUB_TAIL_FILE", c.StartSubTailFile).(bool), "true/false")
flag.BoolVar(&c.StartSubCliCommandCont, "startSubCliCommandCont", CheckEnv("START_SUB_CLI_COMMAND_CONT", c.StartSubCliCommandCont).(bool), "true/false")
// Check that mandatory flag values have been set.
switch {
@ -222,62 +219,61 @@ func NewConfiguration() *Configuration {
// Get a Configuration struct with the default values set.
func newConfigurationDefaults() Configuration {
c := Configuration{
ConfigFolder: "./etc/",
SocketFolder: "./tmp",
ReadFolder: "./readfolder",
EnableReadFolder: true,
TCPListener: "",
HTTPListener: "",
DatabaseFolder: "./var/lib",
NodeName: "",
BrokerAddress: "127.0.0.1:4222",
NatsConnOptTimeout: 20,
NatsConnectRetryInterval: 10,
NatsReconnectJitter: 100,
NatsReconnectJitterTLS: 1,
REQKeysRequestUpdateInterval: 60,
REQAclRequestUpdateInterval: 60,
ProfilingPort: "",
PromHostAndPort: "",
DefaultMessageTimeout: 10,
DefaultMessageRetries: 1,
DefaultMethodTimeout: 10,
SubscribersDataFolder: "./data",
CentralNodeName: "central",
RootCAPath: "",
NkeySeedFile: "",
NkeyFromED25519SSHKeyFile: "",
NkeySeed: "",
ExposeDataFolder: "",
ErrorMessageTimeout: 60,
ErrorMessageRetries: 10,
Compression: "z",
Serialization: "cbor",
SetBlockProfileRate: 0,
EnableSocket: true,
EnableSignatureCheck: false,
EnableAclCheck: false,
IsCentralAuth: false,
EnableDebug: false,
LogLevel: "debug",
LogConsoleTimestamps: false,
KeepPublishersAliveFor: 10,
ConfigFolder: "./etc/",
SocketFolder: "./tmp",
ReadFolder: "./readfolder",
EnableReadFolder: true,
TCPListener: "",
HTTPListener: "",
DatabaseFolder: "./var/lib",
NodeName: "",
BrokerAddress: "127.0.0.1:4222",
NatsConnOptTimeout: 20,
NatsConnectRetryInterval: 10,
NatsReconnectJitter: 100,
NatsReconnectJitterTLS: 1,
KeysUpdateInterval: 60,
AclUpdateInterval: 60,
ProfilingPort: "",
PromHostAndPort: "",
DefaultMessageTimeout: 10,
DefaultMessageRetries: 1,
DefaultMethodTimeout: 10,
SubscribersDataFolder: "./data",
CentralNodeName: "central",
RootCAPath: "",
NkeySeedFile: "",
NkeyFromED25519SSHKeyFile: "",
NkeySeed: "",
ExposeDataFolder: "",
ErrorMessageTimeout: 60,
ErrorMessageRetries: 10,
Compression: "z",
Serialization: "cbor",
SetBlockProfileRate: 0,
EnableSocket: true,
EnableSignatureCheck: false,
EnableAclCheck: false,
IsCentralAuth: false,
EnableDebug: false,
LogLevel: "debug",
LogConsoleTimestamps: false,
KeepPublishersAliveFor: 10,
StartPubREQHello: 30,
EnableKeyUpdates: false,
EnableAclUpdates: false,
IsCentralErrorLogger: false,
StartSubREQHello: true,
StartSubREQToFileAppend: true,
StartSubREQToFile: true,
StartSubREQCopySrc: true,
StartSubREQCopyDst: true,
StartSubREQCliCommand: true,
StartSubREQToConsole: true,
StartSubREQHttpGet: true,
StartSubREQHttpGetScheduled: true,
StartSubREQTailFile: true,
StartSubREQCliCommandCont: true,
StartPubHello: 30,
EnableKeyUpdates: false,
EnableAclUpdates: false,
IsCentralErrorLogger: false,
StartSubHello: true,
StartSubFileAppend: true,
StartSubFile: true,
StartSubCopySrc: true,
StartSubCopyDst: true,
StartSubCliCommand: true,
StartSubConsole: true,
StartSubHttpGet: true,
StartSubTailFile: true,
StartSubCliCommandCont: true,
}
return c
}

6
doc/example-messages/cli-command-cont-tail-log-file.json
View file

@ -4,10 +4,10 @@
"fileName": "test",
"toNodes": ["ship1"],
"relayViaNode": "central",
"relayReplyMethod": "REQToConsole",
"relayReplyMethod": "console",
"methodArgs": ["bash","-c","cat /etc/hostname && journalctl -u steward -n 5 --follow"],
"method":"REQCliCommandCont",
"replyMethod":"REQToFileAppend",
"method":"cliCommandCont",
"replyMethod":"fileAppend",
"ACKTimeout":5,
"retries":3,
"replyACKTimeout":5,

4
doc/example-messages/cli-command-tree.json
View file

@ -4,8 +4,8 @@
"fileName":"somefile.result",
"toNode": "ship1",
"data": ["bash","-c","tree ../"],
"replyMethod":"REQToFileAppend",
"method":"REQCliCommand",
"replyMethod":"fileAppend",
"method":"cliCommand",
"ACKTimeout":3,
"retries":3,
"methodTimeout": 10

4
doc/example-messages/cli-command.json
View file

@ -4,8 +4,8 @@
"fileName": "somefile.result",
"toNode": "ship1",
"data": ["bash","-c","sleep 3 & tree ./"],
"method":"REQCliCommand",
"replyMethod":"REQToFileAppend",
"method":"cliCommand",
"replyMethod":"fileAppend",
"ACKTimeout":5,
"retries":1,
"replyACKTimeout":5,

4
doc/example-messages/clicommand-to-console.json
View file

@ -1,13 +1,13 @@
[
{
"toNodes": ["ship1"],
"method": "REQCliCommand",
"method": "cliCommand",
"methodArgs": [
"bash",
"-c",
"hostname && curl -v http://debian.org"
],
"replyMethod": "REQToConsole",
"replyMethod": "console",
"ACKTimeout": 5,
"retries": 3,
"replyACKTimeout": 5,

4
doc/example-messages/copy-files.json
View file

@ -3,11 +3,11 @@
"directory": "copy",
"fileName": "copy.log",
"toNodes": ["central"],
"method":"REQCopySrc",
"method":"copySrc",
"methodArgs": ["./testbinary","ship1","./apekatt/testbinary-copied","900000","60","0700"],
"methodTimeout": 10,
"retryWait" : 5,
"retries" : 3,
"replyMethod":"REQToConsole"
"replyMethod":"console"
}
]

4
doc/example-messages/create-and-execute-bash-script-with-heredoc.yaml
View file

@ -2,7 +2,7 @@
- toNodes:
- ship1
- ship2
method: REQCliCommand
method: cliCommand
methodArgs:
- "bash"
- "-c"
@ -24,7 +24,7 @@
./test.sh WHOOPS.txt
replyMethod: REQToFile
replyMethod: file
ACKTimeout: 5
retries: 120
replyACKTimeout: 5

6
doc/example-messages/flush-dns-restart-wireguard.json
View file

@ -1,13 +1,13 @@
[
{
"toNodes": ["ship1","ship2"],
"method": "REQCliCommand",
"method": "cliCommand",
"methodArgs": [
"bash",
"-c",
"cat /etc/hostname && systemd-resolve --flush-caches &&wg-quick down wg0 && sleep 2 && wg-quick up wg0"
],
"replyMethod": "REQToFile",
"replyMethod": "file",
"ACKTimeout": 20,
"retries": 3,
"replyACKTimeout": 20,
@ -17,6 +17,6 @@
"directory": "system/wireguard",
"fileName": "wireguard-down-up.log",
"relayViaNode": "central",
"relayReplyMethod": "REQToConsole"
"relayReplyMethod": "console"
}
]

4
doc/example-messages/http-get-ex2.json
View file

@ -4,8 +4,8 @@
"fileName": "somefile.html",
"toNode": "ship2",
"data": ["http://erter.org"],
"method":"REQHttpGet",
"replyMethod":"REQToFile",
"method":"httpGet",
"replyMethod":"file",
"ACKTimeout":5,
"retries":3,
"methodTimeout": 5

4
doc/example-messages/http-get-scheduled.json
View file

@ -1,11 +1,11 @@
[
{
"toNodes": ["ship1"],
"method": "REQHttpGet",
"method": "httpGet",
"methodArgs": [
"https://vg.no","10","1"
],
"replyMethod": "REQToConsole",
"replyMethod": "console",
"ACKTimeout": 5,
"retries": 3,
"replyACKTimeout": 5,

4
doc/example-messages/http-get.json
View file

@ -4,8 +4,8 @@
"fileName": "somefile.html",
"toNode": "ship1",
"data": ["http://vg.no"],
"method":"REQHttpGet",
"replyMethod":"REQToFile",
"method":"httpGet",
"replyMethod":"file",
"ACKTimeout":5,
"retries":3,
"methodTimeout": 5

4
doc/example-messages/k3s-delete-deployment.yaml
View file

@ -1,14 +1,14 @@
---
- toNodes:
- vbox1
method: REQCliCommand
method: cliCommand
methodArgs:
- "bash"
- "-c"
- |
kubectl delete -f test.yaml
replyMethod: REQToConsole
replyMethod: console
ACKTimeout: 5
retries: 120
replyACKTimeout: 5

4
doc/example-messages/k3s-deploy.yaml
View file

@ -1,7 +1,7 @@
---
- toNodes:
- vbox1
method: REQCliCommand
method: cliCommand
methodArgs:
- "bash"
- "-c"
@ -71,7 +71,7 @@
kubectl apply -f test.yaml
replyMethod: REQToConsole
replyMethod: console
ACKTimeout: 5
retries: 120
replyACKTimeout: 5

4
doc/example-messages/k3s-install.yaml
View file

@ -1,14 +1,14 @@
---
- toNodes:
- vbox1
method: REQCliCommandCont
method: cliCommandCont
methodArgs:
- "bash"
- "-c"
- |
curl -sfL https://get.k3s.io | sh -
replyMethod: REQToConsole
replyMethod: toConsole
ACKTimeout: 5
retries: 3
replyACKTimeout: 5

4
doc/example-messages/k3s-uninstall.yaml
View file

@ -1,14 +1,14 @@
---
- toNodes:
- vbox1
method: REQCliCommandCont
method: cliCommandCont
methodArgs:
- "bash"
- "-c"
- |
k3s-uninstall.sh
replyMethod: REQToConsole
replyMethod: console
ACKTimeout: 5
retries: 3
replyACKTimeout: 5

6
doc/example-messages/operation-start-proc.json
View file

@ -4,8 +4,8 @@
"fileName": "startHttp.log",
"toNode": "",
"toNodes": ["central"],
"method":"REQOpProcessStart",
"methodArgs": ["REQHttpGet"],
"replyMethod":"REQToFile"
"method":"opProcessStart",
"methodArgs": ["httpGet"],
"replyMethod":"file"
}
]

6
doc/example-messages/operation-stop-proc.json
View file

@ -4,8 +4,8 @@
"fileName": "stopHttp.log",
"toNode": "ship1",
"toNodes": ["central"],
"method":"REQOpProcessStop",
"methodArgs": ["REQHttpGet","central","subscriber"],
"replyMethod":"REQToFile"
"method":"opProcessStop",
"methodArgs": ["httpGet","central","subscriber"],
"replyMethod":"file"
}
]

2
doc/example-messages/say-hello.json
View file

@ -2,6 +2,6 @@
{
"toNode": "central",
"data": [""],
"method":"REQHello"
"method":"hello"
}
]

2
doc/example-messages/tail-file.json
View file

@ -4,7 +4,7 @@
"fileName": "somefile.log",
"toNode": "ship2",
"data": ["/var/log/system.log"],
"method":"REQTailFile",
"method":"tailFile",
"ACKTimeout":5,
"retries":3,
"methodTimeout": 10

4
doc/example-messages/use-sed-to-edit-files.json
View file

@ -4,8 +4,8 @@
"fileName": "test.txt",
"toNodes": ["ship1","ship2"],
"methodArgs": ["bash","-c","sed -i 's/StartSubREQErrorLog =.*/StartSubREQErrorLog = false/g' /usr/local/steward/etc/config.toml && systemctl restart steward"],
"method":"REQCliCommand",
"replyMethod":"REQToFile",
"method":"cliCommand",
"replyMethod":"file",
"ACKTimeout":10,
"retries":1,
"replyACKTimeout":10,

34
doc/kubernetes/steward.yaml
View file

@ -72,9 +72,9 @@ spec:
value: "100"
- name: NATS_RECONNECT_JITTER_TLS
value: "1"
- name: REQ_KEYS_REQUEST_UPDATE_INTERVAL
- name: KEYS_REQUEST_UPDATE_INTERVAL
value: "60"
- name: REQ_ACL_REQUEST_UPDATE_INTERVAL
- name: ACL_REQUEST_UPDATE_INTERVAL
value: "60"
- name: PROFILING_PORT
value: ""
@ -118,7 +118,7 @@ spec:
value: "0"
- name: KEEP_PUBLISHERS_ALIVE_FOR
value: "10"
- name: START_PUB_REQ_HELLO
- name: START_PUB_HELLO
value: "60"
- name: ENABLE_KEY_UPDATES
value: "1"
@ -126,33 +126,33 @@ spec:
value: "1"
- name: IS_CENTRAL_ERROR_LOGGER
value: "1"
- name: START_SUB_REQ_HELLO
- name: START_SUB_HELLO
value: "true"
- name: START_SUB_REQ_TO_FILE_APPEND
- name: START_SUB_FILE_APPEND
value: "true"
- name: START_SUB_REQ_TO_FILE
- name: START_SUB_FILE
value: "true"
- name: START_SUB_REQ_TO_FILE_NACK
- name: START_SUB_FILE_NACK
value: "true"
- name: START_SUB_REQ_COPY_SRC
- name: START_SUB_COPY_SRC
value: "true"
- name: START_SUB_REQ_COPY_DST
- name: START_SUB_COPY_DST
value: "true"
- name: START_SUB_REQ_PING
- name: START_SUB_PING
value: "true"
- name: START_SUB_REQ_PONG
- name: START_SUB_PONG
value: "true"
- name: START_SUB_REQ_CLI_COMMAND
- name: START_SUB_CLI_COMMAND
value: "true"
- name: START_SUB_REQ_TO_CONSOLE
- name: START_SUB_TO_CONSOLE
value: "true"
- name: START_SUB_REQ_HTTP_GET
- name: START_SUB_HTTP_GET
value: "true"
- name: START_SUB_REQ_HTTP_GET_SCHEDULED
- name: START_SUB_HTTP_GET_SCHEDULED
value: "true"
- name: START_SUB_REQ_TAIL_FILE
- name: START_SUB_TAIL_FILE
value: "true"
- name: START_SUB_REQ_CLI_COMMAND_CONT
- name: START_SUB_CLI_COMMAND_CONT
value: "true"
- image: localhost:5000/stewardwriter:0.1.1

10
doc/src/core_messaging_ctrl_DATA.md
View file

@ -14,9 +14,9 @@ Example of usage:
"directory":"cli_command_test",
"fileName":"cli_command.result",
"toNode": "node2",
"method":"REQCliCommand",
"method":"cliCommand",
"methodArgs": ["bash","-c","tree"],
"replyMethod":"REQCliCommand",
"replyMethod":"cliCommand",
"replyMethodArgs": ["bash", "-c","echo \"{{ctrl_DATA}}\" > apekatt.txt"],
"replyMethodTimeOut": 10,
"ACKTimeout":3,
@ -28,7 +28,7 @@ Example of usage:
The above example, with steps explained:
- Send a message from **node1** to **node2** with a Request Method of type REQCliCommand.
- Send a message from **node1** to **node2** with a Request Method of type cliCommand.
- When received at **node2** we execute the Reqest Method with the arguments specified in the methodArgs.
- When the method on **node2** is done the result data of the method run will be stored in the variable {{ctrl_DATA}}. We can then use this variable when we craft the reply message method by embedding it into a new bash command.
- The reply message is then sent back to **node1**, the method will be executed, and all newlines in the result data will be removed, and all the data with the new lines removed will be stored in a file called `apekatt.txt`
@ -41,9 +41,9 @@ The same using bash's herestring:
"directory":"cli_command_test",
"fileName":"cli_command.result",
"toNode": "ship2",
"method":"REQCliCommand",
"method":"cliCommand",
"methodArgs": ["bash","-c","tree"],
"replyMethod":"REQCliCommand",
"replyMethod":"cliCommand",
"replyMethodArgs": ["bash", "-c","cat <<< {{ctrl_DATA}} > hest.txt"],
"replyMethodTimeOut": 10,
"ACKTimeout":3,

6
doc/src/core_messaging_message_fields.md
View file

@ -19,10 +19,10 @@ data : data here in byte format
The actual data in the message. This is the field where we put the returned data in a reply message. The data field are of type []byte.
```yaml
method : REQCliCommand
method : cliCommand
```
What request method type to use, like REQCliCommand, REQHttpGet..
What request method type to use, like cliCommand, httpGet..
```yaml
methodArgs :
@ -36,7 +36,7 @@ What request method type to use, like REQCliCommand, REQHttpGet..
Additional arguments that might be needed when executing the method. Can be f.ex. an ip address if it is a tcp sender, or the actual shell command to execute in a cli.
```yaml
replyMethod : REQToFile
replyMethod : file
```
ReplyMethod, is the method to use for the reply message. By default the reply method will be set to log to file, but you can override it setting your own here.

2
doc/src/core_messaging_overview.md
View file

@ -7,7 +7,7 @@ All messages processed by a publisher will be written to a log file after they a
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.
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 **REQCLICommand** to be executed, but we don't care for an acknowledge **NACK** when we send an **REQHello** 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** event.
If a message are **ACK** or **NACK** type are defined by the value of the **ACKTimeout** for each individual message:
- **ACKTimeout** set to 0 will make the message become a **NACK** message.

28
doc/src/core_request_methods.md
View file

@ -2,17 +2,17 @@
| Method name| Description|
|------------|------------|
|REQOpProcessList | Get a list of the running processes|
|REQOpProcessStart | Start up a process|
|REQOpProcessStop | Stop a process|
|REQCliCommand | Will run the command given, and return the stdout output of the command when the command is done|
|REQCliCommandCont | Will run the command given, and return the stdout output of the command continously while the command runs|
|REQTailFile | Tail log files on some node, and get the result for each new line read sent back in a reply message|
|REQHttpGet | Scrape web url, and get the html sent back in a reply message|
|REQHello | Send Hello messages|
|REQCopySrc| Copy a file from one node to another node|
|REQErrorLog | Method for receiving error logs for Central error logger|
|REQNone | Don't send a reply message|
|REQToConsole | Print to stdout or stderr|
|REQToFileAppend | Append to file, can also write to unix sockets|
|REQToFile | Write to file, can also write to unix sockets|
|opProcessList | Get a list of the running processes|
|opProcessStart | Start up a process|
|opProcessStop | Stop a process|
|cliCommand | Will run the command given, and return the stdout output of the command when the command is done|
|cliCommandCont | Will run the command given, and return the stdout output of the command continously while the command runs|
|tailFile | Tail log files on some node, and get the result for each new line read sent back in a reply message|
|httpGet | Scrape web url, and get the html sent back in a reply message|
|hello | Send Hello messages|
|copySrc| Copy a file from one node to another node|
|errorLog | Method for receiving error logs for Central error logger|
|none | Don't send a reply message|
|console | Print to stdout or stderr|
|fileAppend | Append to file, can also write to unix sockets|
|file | Write to file, can also write to unix sockets|

4
doc/src/core_startup_folder.md
View file

@ -25,13 +25,13 @@ Since messages used in startup folder are ment to be delivered locally we can si
{
"toNode": "local",
"fromNode": "central",
"method": "REQCliCommand",
"method": "cliCommand",
"methodArgs": [
"bash",
"-c",
"curl localhost:2111/metrics"
],
"replyMethod": "REQToConsole",
"replyMethod": "console",
"methodTimeout": 10
}
]

22
doc/src/example_standard_reqclicommand.md
View file

@ -1,4 +1,4 @@
# REQCliCommand
# cliCommand
In JSON.
@ -8,9 +8,9 @@ In JSON.
"directory":"system",
"fileName":"system.log",
"toNodes": ["node2"],
"method":"REQCliCommand",
"method":"cliCommand",
"methodArgs": ["bash","-c","rm -rf ./data & systemctl restart ctrl"],
"replyMethod":"REQToFileAppend",
"replyMethod":"fileAppend",
"ACKTimeout":30,
"retries":1,
"methodTimeout": 30
@ -24,14 +24,14 @@ In YAML.
---
- toNodes:
- node2
method: REQCliCommand
method: cliCommand
methodArgs:
- "bash"
- "-c"
- |
rm -rf ./data & systemctl restart ctrl
replyMethod: REQToFileAppend
replyMethod: fileAppend
ACKTimeout: 30
retries: 1
ACKTimeout: 30
@ -49,13 +49,13 @@ Will send a message to node2 to delete the ctrl data folder, and then restart ct
[
{
"toNode": "central",
"method": "REQCliCommand",
"method": "cliCommand",
"methodArgs": [
"bash",
"-c",
"curl localhost:2111/metrics"
],
"replyMethod": "REQToConsole",
"replyMethod": "console",
"methodTimeout": 10
}
]
@ -68,13 +68,13 @@ Will send a message to node2 to delete the ctrl data folder, and then restart ct
[
{
"toNode": "node1",
"method": "REQCliCommandCont",
"method": "cliCommandCont",
"methodArgs": [
"bash",
"-c",
"nc -lk localhost 8888"
],
"replyMethod": "REQToConsole",
"replyMethod": "toConsole",
"methodTimeout": 10,
}
]
@ -90,9 +90,9 @@ The netcat tcp listener will run for 10 seconds before the method timeout kicks
"directory":"some/cli/command",
"fileName":"cli.result",
"toNode": "node2",
"method":"REQnCliCommand",
"method":"cliCommand",
"methodArgs": ["bash","-c","docker ps -a"],
"replyMethod":"REQToFileAppend",
"replyMethod":"fileAppend",
}
]
```

8
doc/src/example_standard_reqclicommandcont.md
View file

@ -1,6 +1,6 @@
# REQCliCommandCont
# cliCommandCont
The **REQCliCommand** and the **REQCliCommandCont** are the same, except for one thing. **REQCliCommand** will wait until wether the method is finished or the methodTimeout kicks in to send the result as one single message. **REQCliCommand** will when a line is given to either stdout or stderr create messages with that single line in the data field, and send it back to the node where the message originated.
The **cliCommand** and the **cliCommandCont** are the same, except for one thing. **cliCommand** will wait until wether the method is finished or the methodTimeout kicks in to send the result as one single message. **cliCommand** will when a line is given to either stdout or stderr create messages with that single line in the data field, and send it back to the node where the message originated.
```json
[
@ -8,9 +8,9 @@ The **REQCliCommand** and the **REQCliCommandCont** are the same, except for one
"directory":"some/cli/command",
"fileName":"cli.result",
"toNode": "node2",
"method":"REQCliCommandCont",
"method":"cliCommandCont",
"methodArgs": ["bash","-c","tcpdump -nni any port 8080"],
"replyMethod":"REQToFileAppend",
"replyMethod":"fileAppend",
"methodTimeout":10,
}
]

8
doc/src/example_standard_reqcopysrc.md
View file

@ -1,4 +1,4 @@
# REQCopySrc
# copySrc
Copy a file from one node to another node.
@ -8,10 +8,10 @@ Copy a file from one node to another node.
"directory": "copy",
"fileName": "copy.log",
"toNodes": ["central"],
"method":"REQCopySrc",
"method":"copySrc",
"methodArgs": ["./testbinary","ship1","./testbinary-copied","500000","20","0770"],
"methodTimeout": 10,
"replyMethod":"REQToConsole"
"replyMethod":"console"
}
]
```
@ -25,4 +25,4 @@ Copy a file from one node to another node.
5. MaxTotalCopyTime, specifies the maximum allowed time the complete copy should take. Make sure you set this long enough to allow the transfer to complete.
6. FolderPermission, the permissions to set on the destination folder if it does not exist and needs to be created. Will default to 0755 if no value is set.
To copy from a remote node to the local node, you specify the remote nodeName in the toNode field, and the message will be forwarded to the remote node. The copying request will then be picked up by the remote node's **REQCopySrc** handler, and the copy session will then be handled from the remote node.
To copy from a remote node to the local node, you specify the remote nodeName in the toNode field, and the message will be forwarded to the remote node. The copying request will then be picked up by the remote node's **copySrc** handler, and the copy session will then be handled from the remote node.

10
doc/src/example_standard_reqhttpget.md
View file

@ -1,4 +1,4 @@
# REQHttpGet
# httpGet
In JSON.
@ -8,9 +8,9 @@ In JSON.
"directory": "httpget",
"fileName": "finn.no.html",
"toNodes": ["node1","node2"],
"method":"REQHttpGet",
"method":"httpGet",
"methodArgs": ["https://finn.no"],
"replyMethod":"REQToFile",
"replyMethod":"file",
"ACKTimeout":5,
"retries":3,
"methodTimeout": 5
@ -24,10 +24,10 @@ In YAML.
---
- toNodes:
- ["node1","node2"]
method: REQHttpGet
method: httpGet
methodArgs:
- "https://finn.no"
replyMethod: REQToFile
replyMethod: file
ACKTimeout: 5
retries: 3
methodTimeout: 5

10
doc/src/example_standard_reqtailfile.md
View file

@ -1,4 +1,4 @@
# REQTailFile
# tailFile
In JSON.
@ -8,7 +8,7 @@ In JSON.
"directory": "tails",
"fileName": "some.log",
"toNodes": "node1","node2","node3",
"method":"REQTailFile",
"method":"tailFile",
"methodArgs": ["/var/log/syslog"],
"ACKTimeout":5,
"retries":3,
@ -17,7 +17,7 @@ In JSON.
]
```
NB: If no replyMethod are specified, it will default to **REQToFile**
NB: If no replyMethod are specified, it will default to **file**
In YAML.
@ -25,10 +25,10 @@ In YAML.
---
- toNodes:
- ["node1","node2","node3"]
method: REQTailFile
method: tailFile
methodArgs:
- "/var/log/syslog"
replyMethod: REQToFile
replyMethod: file
ACKTimeout: 5
retries: 3
methodTimeout: 5

4
doc/src/example_standard_send_more_messages.md
View file

@ -6,7 +6,7 @@
"directory":"cli-command-executed-result",
"fileName": "some.log",
"toNode": "ship1",
"method":"REQCliCommand",
"method":"cliCommand",
"methodArgs": ["bash","-c","sleep 3 & tree ./"],
"ACKTimeout":10,
"retries":3,
@ -16,7 +16,7 @@
"directory":"cli-command-executed-result",
"fileName": "some.log",
"toNode": "ship2",
"method":"REQCliCommand",
"method":"cliCommand",
"methodArgs": ["bash","-c","sleep 3 & tree ./"],
"ACKTimeout":10,
"retries":3,

6
doc/src/user_guide_install_docker.md
View file

@ -28,7 +28,7 @@ cat << EOF > .env
NODE_NAME="node1"
BROKER_ADDRESS="127.0.0,1:4444"
ENABLE_DEBUG=1
START_PUB_REQ_HELLO=60
START_PUB_HELLO=60
IS_CENTRAL_ERROR_LOGGER=0
EOF
```
@ -46,7 +46,7 @@ cat << EOF > msg.yaml
---
- toNodes:
- node1
method: REQCliCommand
method: cliCommand
methodArgs:
- "bash"
- "-c"
@ -56,7 +56,7 @@ cat << EOF > msg.yaml
echo "some config line" > /etc/my-service-config.3
systemctl restart my-service
replyMethod: REQNone
replyMethod: none
ACKTimeout: 0
EOF

6
doc/src/user_guide_install_host.md
View file

@ -40,7 +40,7 @@ cat << EOF > .env
NODE_NAME="node1"
BROKER_ADDRESS="127.0.0,1:4444"
ENABLE_DEBUG=1
START_PUB_REQ_HELLO=60
START_PUB_HELLO=60
IS_CENTRAL_ERROR_LOGGER=0
EOF
```
@ -60,7 +60,7 @@ cat << EOF > msg.yaml
---
- toNodes:
- node1
method: REQCliCommand
method: cliCommand
methodArgs:
- "bash"
- "-c"
@ -70,7 +70,7 @@ cat << EOF > msg.yaml
echo "some config line" > /etc/my-service-config.3
systemctl restart my-service
replyMethod: REQNone
replyMethod: none
ACKTimeout: 0
EOF

2385
doc/steward-not-grouped.svg
View file

File diff suppressed because it is too large Load diff
Side by side

Before

Width:  |  Height:  |  Size: 202 KiB

3099
doc/steward.svg
View file

File diff suppressed because it is too large Load diff
Side by side

Before

Width:  |  Height:  |  Size: 266 KiB

4
errorkernel.go
View file

@ -126,13 +126,13 @@ func (e *errorKernel) start(ringBufferBulkInCh chan<- []subjectAndMessage) error
FromNode: errEvent.process.node,
FileName: "error.log",
Data: []byte(er),
Method: REQErrorLog,
Method: ErrorLog,
ACKTimeout: errEvent.process.configuration.ErrorMessageTimeout,
Retries: errEvent.process.configuration.ErrorMessageRetries,
}
sam := subjectAndMessage{
Subject: newSubject(REQErrorLog, "errorCentral"),
Subject: newSubject(ErrorLog, "errorCentral"),
Message: m,
}

4
node_auth.go
View file

@ -392,7 +392,7 @@ func (n *nodeAuth) readKeyFile(keyFile string) (ed2519key []byte, b64Key []byte,
// verifySignature
func (n *nodeAuth) verifySignature(m Message) bool {
// NB: Only enable signature checking for REQCliCommand for now.
if m.Method != REQCliCommand {
if m.Method != CliCommand {
er := fmt.Errorf("verifySignature: not REQCliCommand and will not do signature check, method: %v", m.Method)
n.errorKernel.logInfo(er)
return true
@ -429,7 +429,7 @@ func (n *nodeAuth) verifySignature(m Message) bool {
// verifyAcl
func (n *nodeAuth) verifyAcl(m Message) bool {
// NB: Only enable acl checking for REQCliCommand for now.
if m.Method != REQCliCommand {
if m.Method != CliCommand {
er := fmt.Errorf("verifyAcl: not REQCliCommand and will not do acl check, method: %v", m.Method)
n.errorKernel.logInfo(er)
return true

2
process.go
View file

@ -334,7 +334,7 @@ func (p process) messageDeliverNats(natsMsgPayload []byte, natsMsgHeader nats.He
// We do not want to send errorLogs for REQErrorLog type since
// it will just cause an endless loop.
if message.Method != REQErrorLog {
if message.Method != ErrorLog {
p.errorKernel.infoSend(p, message, er)
}

124
processes.go
View file

@ -92,28 +92,28 @@ func (p *processes) Start(proc process) {
// --- Subscriber services that can be started via flags
proc.startup.subscriber(proc, REQOpProcessList, nil)
proc.startup.subscriber(proc, REQOpProcessStart, nil)
proc.startup.subscriber(proc, REQOpProcessStop, nil)
proc.startup.subscriber(proc, REQTest, nil)
proc.startup.subscriber(proc, OpProcessList, nil)
proc.startup.subscriber(proc, OpProcessStart, nil)
proc.startup.subscriber(proc, OpProcessStop, nil)
proc.startup.subscriber(proc, Test, nil)
if proc.configuration.StartSubREQToFileAppend {
proc.startup.subscriber(proc, REQToFileAppend, nil)
if proc.configuration.StartSubFileAppend {
proc.startup.subscriber(proc, FileAppend, nil)
}
if proc.configuration.StartSubREQToFile {
proc.startup.subscriber(proc, REQToFile, nil)
if proc.configuration.StartSubFile {
proc.startup.subscriber(proc, File, nil)
}
if proc.configuration.StartSubREQCopySrc {
proc.startup.subscriber(proc, REQCopySrc, nil)
if proc.configuration.StartSubCopySrc {
proc.startup.subscriber(proc, CopySrc, nil)
}
if proc.configuration.StartSubREQCopyDst {
proc.startup.subscriber(proc, REQCopyDst, nil)
if proc.configuration.StartSubCopyDst {
proc.startup.subscriber(proc, CopyDst, nil)
}
if proc.configuration.StartSubREQHello {
if proc.configuration.StartSubHello {
// subREQHello is the handler that is triggered when we are receiving a hello
// message. To keep the state of all the hello's received from nodes we need
// to also start a procFunc that will live as a go routine tied to this process,
@ -149,24 +149,24 @@ func (p *processes) Start(proc process) {
}
}
proc.startup.subscriber(proc, REQHello, pf)
proc.startup.subscriber(proc, Hello, pf)
}
if proc.configuration.IsCentralErrorLogger {
proc.startup.subscriber(proc, REQErrorLog, nil)
proc.startup.subscriber(proc, ErrorLog, nil)
}
if proc.configuration.StartSubREQCliCommand {
proc.startup.subscriber(proc, REQCliCommand, nil)
if proc.configuration.StartSubCliCommand {
proc.startup.subscriber(proc, CliCommand, nil)
}
if proc.configuration.StartSubREQToConsole {
proc.startup.subscriber(proc, REQToConsole, nil)
if proc.configuration.StartSubConsole {
proc.startup.subscriber(proc, Console, nil)
}
if proc.configuration.StartPubREQHello != 0 {
if proc.configuration.StartPubHello != 0 {
pf := func(ctx context.Context, procFuncCh chan Message) error {
ticker := time.NewTicker(time.Second * time.Duration(p.configuration.StartPubREQHello))
ticker := time.NewTicker(time.Second * time.Duration(p.configuration.StartPubHello))
defer ticker.Stop()
for {
@ -180,7 +180,7 @@ func (p *processes) Start(proc process) {
ToNode: Node(p.configuration.CentralNodeName),
FromNode: Node(proc.node),
Data: []byte(d),
Method: REQHello,
Method: Hello,
ACKTimeout: proc.configuration.DefaultMessageTimeout,
Retries: 1,
}
@ -203,15 +203,15 @@ func (p *processes) Start(proc process) {
}
}
}
proc.startup.publisher(proc, REQHello, pf)
proc.startup.publisher(proc, Hello, pf)
}
if proc.configuration.EnableKeyUpdates {
// pubREQKeysRequestUpdate defines the startup of a publisher that will send REQREQKeysRequestUpdate
// Define the startup of a publisher that will send KeysRequestUpdate
// to central server and ask for publics keys, and to get them deliver back with a request
// of type pubREQKeysDeliverUpdate.
// of type KeysDeliverUpdate.
pf := func(ctx context.Context, procFuncCh chan Message) error {
ticker := time.NewTicker(time.Second * time.Duration(p.configuration.REQKeysRequestUpdateInterval))
ticker := time.NewTicker(time.Second * time.Duration(p.configuration.KeysUpdateInterval))
defer ticker.Stop()
for {
@ -220,7 +220,7 @@ func (p *processes) Start(proc process) {
// and update with new keys back.
proc.nodeAuth.publicKeys.mu.Lock()
er := fmt.Errorf(" ----> publisher REQKeysRequestUpdate: sending our current hash: %v", []byte(proc.nodeAuth.publicKeys.keysAndHash.Hash[:]))
er := fmt.Errorf(" ----> publisher KeysRequestUpdate: sending our current hash: %v", []byte(proc.nodeAuth.publicKeys.keysAndHash.Hash[:]))
p.errorKernel.logDebug(er)
m := Message{
@ -229,8 +229,8 @@ func (p *processes) Start(proc process) {
ToNode: Node(p.configuration.CentralNodeName),
FromNode: Node(proc.node),
Data: []byte(proc.nodeAuth.publicKeys.keysAndHash.Hash[:]),
Method: REQKeysRequestUpdate,
ReplyMethod: REQKeysDeliverUpdate,
Method: KeysRequestUpdate,
ReplyMethod: KeysDeliverUpdate,
ACKTimeout: proc.configuration.DefaultMessageTimeout,
Retries: 1,
}
@ -253,13 +253,13 @@ func (p *processes) Start(proc process) {
}
}
}
proc.startup.publisher(proc, REQKeysRequestUpdate, pf)
proc.startup.subscriber(proc, REQKeysDeliverUpdate, nil)
proc.startup.publisher(proc, KeysRequestUpdate, pf)
proc.startup.subscriber(proc, KeysDeliverUpdate, nil)
}
if proc.configuration.EnableAclUpdates {
pf := func(ctx context.Context, procFuncCh chan Message) error {
ticker := time.NewTicker(time.Second * time.Duration(p.configuration.REQAclRequestUpdateInterval))
ticker := time.NewTicker(time.Second * time.Duration(p.configuration.AclUpdateInterval))
defer ticker.Stop()
for {
@ -268,7 +268,7 @@ func (p *processes) Start(proc process) {
// and update with new keys back.
proc.nodeAuth.nodeAcl.mu.Lock()
er := fmt.Errorf(" ----> publisher REQAclRequestUpdate: sending our current hash: %v", []byte(proc.nodeAuth.nodeAcl.aclAndHash.Hash[:]))
er := fmt.Errorf(" ----> publisher AclRequestUpdate: sending our current hash: %v", []byte(proc.nodeAuth.nodeAcl.aclAndHash.Hash[:]))
p.errorKernel.logDebug(er)
m := Message{
@ -277,8 +277,8 @@ func (p *processes) Start(proc process) {
ToNode: Node(p.configuration.CentralNodeName),
FromNode: Node(proc.node),
Data: []byte(proc.nodeAuth.nodeAcl.aclAndHash.Hash[:]),
Method: REQAclRequestUpdate,
ReplyMethod: REQAclDeliverUpdate,
Method: AclRequestUpdate,
ReplyMethod: AclDeliverUpdate,
ACKTimeout: proc.configuration.DefaultMessageTimeout,
Retries: 1,
}
@ -302,45 +302,41 @@ func (p *processes) Start(proc process) {
}
}
}
proc.startup.publisher(proc, REQAclRequestUpdate, pf)
proc.startup.subscriber(proc, REQAclDeliverUpdate, nil)
proc.startup.publisher(proc, AclRequestUpdate, pf)
proc.startup.subscriber(proc, AclDeliverUpdate, nil)
}
if proc.configuration.IsCentralAuth {
proc.startup.subscriber(proc, REQKeysRequestUpdate, nil)
proc.startup.subscriber(proc, REQKeysAllow, nil)
proc.startup.subscriber(proc, REQKeysDelete, nil)
proc.startup.subscriber(proc, REQAclRequestUpdate, nil)
proc.startup.subscriber(proc, REQAclAddCommand, nil)
proc.startup.subscriber(proc, REQAclDeleteCommand, nil)
proc.startup.subscriber(proc, REQAclDeleteSource, nil)
proc.startup.subscriber(proc, REQAclGroupNodesAddNode, nil)
proc.startup.subscriber(proc, REQAclGroupNodesDeleteNode, nil)
proc.startup.subscriber(proc, REQAclGroupNodesDeleteGroup, nil)
proc.startup.subscriber(proc, REQAclGroupCommandsAddCommand, nil)
proc.startup.subscriber(proc, REQAclGroupCommandsDeleteCommand, nil)
proc.startup.subscriber(proc, REQAclGroupCommandsDeleteGroup, nil)
proc.startup.subscriber(proc, REQAclExport, nil)
proc.startup.subscriber(proc, REQAclImport, nil)
proc.startup.subscriber(proc, KeysRequestUpdate, nil)
proc.startup.subscriber(proc, KeysAllow, nil)
proc.startup.subscriber(proc, KeysDelete, nil)
proc.startup.subscriber(proc, AclRequestUpdate, nil)
proc.startup.subscriber(proc, AclAddCommand, nil)
proc.startup.subscriber(proc, AclDeleteCommand, nil)
proc.startup.subscriber(proc, AclDeleteSource, nil)
proc.startup.subscriber(proc, AclGroupNodesAddNode, nil)
proc.startup.subscriber(proc, AclGroupNodesDeleteNode, nil)
proc.startup.subscriber(proc, AclGroupNodesDeleteGroup, nil)
proc.startup.subscriber(proc, AclGroupCommandsAddCommand, nil)
proc.startup.subscriber(proc, AclGroupCommandsDeleteCommand, nil)
proc.startup.subscriber(proc, AclGroupCommandsDeleteGroup, nil)
proc.startup.subscriber(proc, AclExport, nil)
proc.startup.subscriber(proc, AclImport, nil)
}
if proc.configuration.StartSubREQHttpGet {
proc.startup.subscriber(proc, REQHttpGet, nil)
if proc.configuration.StartSubHttpGet {
proc.startup.subscriber(proc, HttpGet, nil)
}
if proc.configuration.StartSubREQHttpGetScheduled {
proc.startup.subscriber(proc, REQHttpGetScheduled, nil)
if proc.configuration.StartSubTailFile {
proc.startup.subscriber(proc, TailFile, nil)
}
if proc.configuration.StartSubREQTailFile {
proc.startup.subscriber(proc, REQTailFile, nil)
if proc.configuration.StartSubCliCommandCont {
proc.startup.subscriber(proc, CliCommandCont, nil)
}
if proc.configuration.StartSubREQCliCommandCont {
proc.startup.subscriber(proc, REQCliCommandCont, nil)
}
proc.startup.subscriber(proc, REQPublicKey, nil)
proc.startup.subscriber(proc, PublicKey, nil)
}
// Stop all subscriber processes.
@ -379,7 +375,7 @@ func (s *startup) subscriber(p process, m Method, pf func(ctx context.Context, p
var sub Subject
switch {
case m == REQErrorLog:
case m == ErrorLog:
sub = newSubject(m, "errorCentral")
default:
sub = newSubject(m, string(p.node))

162
requests.go
View file

@ -32,7 +32,7 @@
// Requests used in sub processes should always start with the
// naming REQSUB. Since the method of a sub process are defined
// within the method handler of the owning reqest type we should
// use the methodREQSUB for these types. The methodREQSUB handler
// use the methodSUB for these types. The methodSUB handler
// does nothing.
//
// Check out the existing code below for more examples.
@ -56,13 +56,13 @@ type Method string
// when specifying what kind of Method to send or work with.
const (
// Initial parent method used to start other processes.
REQInitial Method = "REQInitial"
Initial Method = "initial"
// Get a list of all the running processes.
REQOpProcessList Method = "REQOpProcessList"
OpProcessList Method = "opProcessList"
// Start up a process.
REQOpProcessStart Method = "REQOpProcessStart"
OpProcessStart Method = "opProcessStart"
// Stop up a process.
REQOpProcessStop Method = "REQOpProcessStop"
OpProcessStop Method = "opProcessStop"
// 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
@ -71,94 +71,94 @@ const (
// 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"
CliCommand Method = "cliCommand"
// REQCliCommandCont 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 wait until the command is finished.
REQCliCommandCont Method = "REQCliCommandCont"
CliCommandCont Method = "cliCommandCont"
// 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"
Console Method = "console"
// 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"
FileAppend Method = "fileAppend"
// 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"
File Method = "file"
// Initiated by the user.
REQCopySrc Method = "REQCopySrc"
CopySrc Method = "copySrc"
// Initial request for file copying.
// Generated by the source to send initial information to the destination.
REQCopyDst Method = "REQCopyDst"
CopyDst Method = "copyDst"
// Read the source file to be copied to some node.
REQSUBCopySrc Method = "REQSUBCopySrc"
SUBCopySrc Method = "subCopySrc"
// Write the destination copied to some node.
REQSUBCopyDst Method = "REQSUBCopyDst"
SUBCopyDst Method = "subCopyDst"
// Send Hello I'm here message.
REQHello Method = "REQHello"
Hello Method = "hello"
// Error log methods to centralError node.
REQErrorLog Method = "REQErrorLog"
ErrorLog Method = "errorLog"
// Http Get
REQHttpGet Method = "REQHttpGet"
HttpGet Method = "httpGet"
// Http Get Scheduled
// The second element of the MethodArgs slice holds the timer defined in seconds.
REQHttpGetScheduled Method = "REQHttpGetScheduled"
HttpGetScheduled Method = "httpGetScheduled"
// Tail file
REQTailFile Method = "REQTailFile"
TailFile Method = "tailFile"
// REQNone is used when there should be no reply.
REQNone Method = "REQNone"
None Method = "none"
// REQTest is used only for testing to be able to grab the output
// of messages.
REQTest Method = "REQTest"
Test Method = "test"
// REQPublicKey will get the public ed25519 key from a node.
REQPublicKey Method = "REQPublicKey"
PublicKey Method = "publicKey"
// REQKeysRequestUpdate will get all the public keys from central if an update is available.
REQKeysRequestUpdate Method = "REQKeysRequestUpdate"
KeysRequestUpdate Method = "keysRequestUpdate"
// REQKeysDeliverUpdate will deliver the public from central to a node.
REQKeysDeliverUpdate Method = "REQKeysDeliverUpdate"
KeysDeliverUpdate Method = "keysDeliverUpdate"
// REQKeysAllow
REQKeysAllow Method = "REQKeysAllow"
KeysAllow Method = "keysAllow"
// REQKeysDelete
REQKeysDelete Method = "REQKeysDelete"
KeysDelete Method = "keysDelete"
// REQAclRequestUpdate will get all node acl's from central if an update is available.
REQAclRequestUpdate Method = "REQAclRequestUpdate"
AclRequestUpdate Method = "aclRequestUpdate"
// REQAclDeliverUpdate will deliver the acl from central to a node.
REQAclDeliverUpdate Method = "REQAclDeliverUpdate"
AclDeliverUpdate Method = "aclDeliverUpdate"
// REQAclAddCommand
REQAclAddCommand = "REQAclAddCommand"
AclAddCommand = "aclAddCommand"
// REQAclDeleteCommand
REQAclDeleteCommand = "REQAclDeleteCommand"
AclDeleteCommand = "aclDeleteCommand"
// REQAclDeleteSource
REQAclDeleteSource = "REQAclDeleteSource"
AclDeleteSource = "aclDeleteSource"
// REQGroupNodesAddNode
REQAclGroupNodesAddNode = "REQAclGroupNodesAddNode"
AclGroupNodesAddNode = "aclGroupNodesAddNode"
// REQAclGroupNodesDeleteNode
REQAclGroupNodesDeleteNode = "REQAclGroupNodesDeleteNode"
AclGroupNodesDeleteNode = "aclGroupNodesDeleteNode"
// REQAclGroupNodesDeleteGroup
REQAclGroupNodesDeleteGroup = "REQAclGroupNodesDeleteGroup"
AclGroupNodesDeleteGroup = "aclGroupNodesDeleteGroup"
// REQAclGroupCommandsAddCommand
REQAclGroupCommandsAddCommand = "REQAclGroupCommandsAddCommand"
AclGroupCommandsAddCommand = "aclGroupCommandsAddCommand"
// REQAclGroupCommandsDeleteCommand
REQAclGroupCommandsDeleteCommand = "REQAclGroupCommandsDeleteCommand"
AclGroupCommandsDeleteCommand = "aclGroupCommandsDeleteCommand"
// REQAclGroupCommandsDeleteGroup
REQAclGroupCommandsDeleteGroup = "REQAclGroupCommandsDeleteGroup"
AclGroupCommandsDeleteGroup = "aclGroupCommandsDeleteGroup"
// REQAclExport
REQAclExport = "REQAclExport"
AclExport = "aclExport"
// REQAclImport
REQAclImport = "REQAclImport"
AclImport = "aclImport"
)
type HandlerFunc func(proc process, message Message, node string) ([]byte, error)
@ -173,45 +173,45 @@ func (m Method) GetMethodsAvailable() MethodsAvailable {
ma := MethodsAvailable{
Methodhandlers: map[Method]HandlerFunc{
REQInitial: HandlerFunc(methodREQInitial),
REQOpProcessList: HandlerFunc(methodREQOpProcessList),
REQOpProcessStart: HandlerFunc(methodREQOpProcessStart),
REQOpProcessStop: HandlerFunc(methodREQOpProcessStop),
REQCliCommand: HandlerFunc(methodREQCliCommand),
REQCliCommandCont: HandlerFunc(methodREQCliCommandCont),
REQToConsole: HandlerFunc(methodREQToConsole),
REQToFileAppend: HandlerFunc(methodREQToFileAppend),
REQToFile: HandlerFunc(methodREQToFile),
REQCopySrc: HandlerFunc(methodREQCopySrc),
REQCopyDst: HandlerFunc(methodREQCopyDst),
REQSUBCopySrc: HandlerFunc(methodREQSUB),
REQSUBCopyDst: HandlerFunc(methodREQSUB),
REQHello: HandlerFunc(methodREQHello),
REQErrorLog: HandlerFunc(methodREQErrorLog),
REQHttpGet: HandlerFunc(methodREQHttpGet),
REQHttpGetScheduled: HandlerFunc(methodREQHttpGetScheduled),
REQTailFile: HandlerFunc(methodREQTailFile),
REQPublicKey: HandlerFunc(methodREQPublicKey),
REQKeysRequestUpdate: HandlerFunc(methodREQKeysRequestUpdate),
REQKeysDeliverUpdate: HandlerFunc(methodREQKeysDeliverUpdate),
REQKeysAllow: HandlerFunc(methodREQKeysAllow),
REQKeysDelete: HandlerFunc(methodREQKeysDelete),
Initial: HandlerFunc(methodInitial),
OpProcessList: HandlerFunc(methodOpProcessList),
OpProcessStart: HandlerFunc(methodOpProcessStart),
OpProcessStop: HandlerFunc(methodOpProcessStop),
CliCommand: HandlerFunc(methodCliCommand),
CliCommandCont: HandlerFunc(methodCliCommandCont),
Console: HandlerFunc(methodConsole),
FileAppend: HandlerFunc(methodFileAppend),
File: HandlerFunc(methodToFile),
CopySrc: HandlerFunc(methodCopySrc),
CopyDst: HandlerFunc(methodCopyDst),
SUBCopySrc: HandlerFunc(methodSUB),
SUBCopyDst: HandlerFunc(methodSUB),
Hello: HandlerFunc(methodHello),
ErrorLog: HandlerFunc(methodErrorLog),
HttpGet: HandlerFunc(methodHttpGet),
HttpGetScheduled: HandlerFunc(methodHttpGetScheduled),
TailFile: HandlerFunc(methodTailFile),
PublicKey: HandlerFunc(methodPublicKey),
KeysRequestUpdate: HandlerFunc(methodKeysRequestUpdate),
KeysDeliverUpdate: HandlerFunc(methodKeysDeliverUpdate),
KeysAllow: HandlerFunc(methodKeysAllow),
KeysDelete: HandlerFunc(methodKeysDelete),
REQAclRequestUpdate: HandlerFunc(methodREQAclRequestUpdate),
REQAclDeliverUpdate: HandlerFunc(methodREQAclDeliverUpdate),
AclRequestUpdate: HandlerFunc(methodAclRequestUpdate),
AclDeliverUpdate: HandlerFunc(methodAclDeliverUpdate),
REQAclAddCommand: HandlerFunc(methodREQAclAddCommand),
REQAclDeleteCommand: HandlerFunc(methodREQAclDeleteCommand),
REQAclDeleteSource: HandlerFunc(methodREQAclDeleteSource),
REQAclGroupNodesAddNode: HandlerFunc(methodREQAclGroupNodesAddNode),
REQAclGroupNodesDeleteNode: HandlerFunc(methodREQAclGroupNodesDeleteNode),
REQAclGroupNodesDeleteGroup: HandlerFunc(methodREQAclGroupNodesDeleteGroup),
REQAclGroupCommandsAddCommand: HandlerFunc(methodREQAclGroupCommandsAddCommand),
REQAclGroupCommandsDeleteCommand: HandlerFunc(methodREQAclGroupCommandsDeleteCommand),
REQAclGroupCommandsDeleteGroup: HandlerFunc(methodREQAclGroupCommandsDeleteGroup),
REQAclExport: HandlerFunc(methodREQAclExport),
REQAclImport: HandlerFunc(methodREQAclImport),
REQTest: HandlerFunc(methodREQTest),
AclAddCommand: HandlerFunc(methodAclAddCommand),
AclDeleteCommand: HandlerFunc(methodAclDeleteCommand),
AclDeleteSource: HandlerFunc(methodAclDeleteSource),
AclGroupNodesAddNode: HandlerFunc(methodAclGroupNodesAddNode),
AclGroupNodesDeleteNode: HandlerFunc(methodAclGroupNodesDeleteNode),
AclGroupNodesDeleteGroup: HandlerFunc(methodAclGroupNodesDeleteGroup),
AclGroupCommandsAddCommand: HandlerFunc(methodAclGroupCommandsAddCommand),
AclGroupCommandsDeleteCommand: HandlerFunc(methodAclGroupCommandsDeleteCommand),
AclGroupCommandsDeleteGroup: HandlerFunc(methodAclGroupCommandsDeleteGroup),
AclExport: HandlerFunc(methodAclExport),
AclImport: HandlerFunc(methodAclImport),
Test: HandlerFunc(methodTest),
},
}
@ -246,7 +246,7 @@ func getContextForMethodTimeout(ctx context.Context, message Message) (context.C
// ----
// Initial parent method used to start other processes.
func methodREQInitial(proc process, message Message, node string) ([]byte, error) {
func methodInitial(proc process, message Message, node string) ([]byte, error) {
// proc.procFuncCh <- message
ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
return ackMsg, nil
@ -258,7 +258,7 @@ func methodREQInitial(proc process, message Message, node string) ([]byte, error
// Methods used in sub processes are defined within the the requests
// they are spawned in, so this type is primarily for us to use the
// same logic with sub process requests as we do with normal requests.
func methodREQSUB(proc process, message Message, node string) ([]byte, error) {
func methodSUB(proc process, message Message, node string) ([]byte, error) {
// proc.procFuncCh <- message
ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
return ackMsg, nil
@ -277,7 +277,7 @@ type MethodsAvailable struct {
// will be returned.
func (ma MethodsAvailable) CheckIfExists(m Method) (HandlerFunc, bool) {
// First check if it is a sub process.
if strings.HasPrefix(string(m), "REQSUB") {
if strings.HasPrefix(string(m), "sub") {
// Strip of the uuid after the method name.
sp := strings.Split(string(m), ".")
m = Method(sp[0])
@ -310,14 +310,14 @@ func (ma MethodsAvailable) CheckIfExists(m Method) (HandlerFunc, bool) {
func newReplyMessage(proc process, message Message, outData []byte) {
// If REQNone is specified, we don't want to send a reply message
// so we silently just return without sending anything.
if message.ReplyMethod == "REQNone" || message.IsReply {
if message.ReplyMethod == None || message.IsReply {
return
}
// If no replyMethod is set we default to writing to writing to
// a log file.
if message.ReplyMethod == "" {
message.ReplyMethod = REQToFileAppend
message.ReplyMethod = FileAppend
}
// Make a copy of the message as it is right now to use

108
requests_acl.go
View file

@ -11,11 +11,11 @@ import (
// ----
// Handler to get all acl's from a central server.
func methodREQAclRequestUpdate(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- subscriber methodREQAclRequestUpdate received from: %v, hash data = %v", message.FromNode, message.Data)
func methodAclRequestUpdate(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- subscriber methodAclRequestUpdate received from: %v, hash data = %v", message.FromNode, message.Data)
proc.errorKernel.logDebug(er)
// fmt.Printf("\n --- subscriber methodREQAclRequestUpdate: the message brought to handler : %+v\n", message)
// fmt.Printf("\n --- subscriber methodAclRequestUpdate: the message brought to handler : %+v\n", message)
// Get a context with the timeout specified in message.MethodTimeout.
ctx, _ := getContextForMethodTimeout(proc.ctx, message)
@ -44,22 +44,22 @@ func methodREQAclRequestUpdate(proc process, message Message, node string) ([]by
proc.centralAuth.accessLists.schemaGenerated.mu.Lock()
defer proc.centralAuth.accessLists.schemaGenerated.mu.Unlock()
er := fmt.Errorf("info: subscriber methodREQAclRequestUpdate: got acl hash from NODE=%v, HASH data =%v", message.FromNode, message.Data)
er := fmt.Errorf("info: subscriber methodAclRequestUpdate: got acl hash from NODE=%v, HASH data =%v", message.FromNode, message.Data)
proc.errorKernel.logDebug(er)
// Check if the received hash is the same as the one currently active,
// If it is the same we exit the handler immediately.
hash32 := proc.centralAuth.accessLists.schemaGenerated.GeneratedACLsMap[message.FromNode].Hash
hash := hash32[:]
er = fmt.Errorf("info: subscriber methodREQAclRequestUpdate: the central acl hash=%v", hash32)
er = fmt.Errorf("info: subscriber methodAclRequestUpdate: the central acl hash=%v", hash32)
proc.errorKernel.logDebug(er)
if bytes.Equal(hash, message.Data) {
er := fmt.Errorf("info: subscriber methodREQAclRequestUpdate: NODE AND CENTRAL HAVE EQUAL ACL HASH, NOTHING TO DO, EXITING HANDLER")
er := fmt.Errorf("info: subscriber methodAclRequestUpdate: NODE AND CENTRAL HAVE EQUAL ACL HASH, NOTHING TO DO, EXITING HANDLER")
proc.errorKernel.logDebug(er)
return
}
er = fmt.Errorf("info: subscriber methodREQAclRequestUpdate: NODE AND CENTRAL HAD NOT EQUAL ACL, PREPARING TO SEND NEW VERSION OF Acl")
er = fmt.Errorf("info: subscriber methodAclRequestUpdate: NODE AND CENTRAL HAD NOT EQUAL ACL, PREPARING TO SEND NEW VERSION OF Acl")
proc.errorKernel.logDebug(er)
// Generate JSON for Message.Data
@ -76,7 +76,7 @@ func methodREQAclRequestUpdate(proc process, message Message, node string) ([]by
proc.errorKernel.errSend(proc, message, er, logWarning)
}
er = fmt.Errorf("----> subscriber methodREQAclRequestUpdate: SENDING ACL'S TO NODE=%v, serializedAndHash=%+v", message.FromNode, hdh)
er = fmt.Errorf("----> subscriber methodAclRequestUpdate: SENDING ACL'S TO NODE=%v, serializedAndHash=%+v", message.FromNode, hdh)
proc.errorKernel.logDebug(er)
newReplyMessage(proc, message, js)
@ -91,11 +91,11 @@ func methodREQAclRequestUpdate(proc process, message Message, node string) ([]by
// ----
// Handler to receive the acls from a central server.
func methodREQAclDeliverUpdate(proc process, message Message, node string) ([]byte, error) {
inf := fmt.Errorf("<--- subscriber methodREQAclDeliverUpdate received from: %v, containing: %v", message.FromNode, message.Data)
func methodAclDeliverUpdate(proc process, message Message, node string) ([]byte, error) {
inf := fmt.Errorf("<--- subscriber methodAclDeliverUpdate received from: %v, containing: %v", message.FromNode, message.Data)
proc.errorKernel.logDebug(inf)
// fmt.Printf("\n --- subscriber methodREQAclRequestUpdate: the message received on handler : %+v\n\n", message)
// fmt.Printf("\n --- subscriber methodAclRequestUpdate: the message received on handler : %+v\n\n", message)
// Get a context with the timeout specified in message.MethodTimeout.
ctx, _ := getContextForMethodTimeout(proc.ctx, message)
@ -168,8 +168,8 @@ func methodREQAclDeliverUpdate(proc process, message Message, node string) ([]by
// ---
func methodREQAclAddCommand(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodREQAclAddCommand received from: %v, containing: %v", message.FromNode, message.MethodArgs)
func methodAclAddCommand(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodAclAddCommand received from: %v, containing: %v", message.FromNode, message.MethodArgs)
proc.errorKernel.logDebug(er)
proc.processes.wg.Add(1)
@ -230,8 +230,8 @@ func methodREQAclAddCommand(proc process, message Message, node string) ([]byte,
// ---
func methodREQAclDeleteCommand(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodREQAclDeleteCommand received from: %v, containing: %v", message.FromNode, message.MethodArgs)
func methodAclDeleteCommand(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodAclDeleteCommand received from: %v, containing: %v", message.FromNode, message.MethodArgs)
proc.errorKernel.logDebug(er)
proc.processes.wg.Add(1)
@ -249,7 +249,7 @@ func methodREQAclDeleteCommand(proc process, message Message, node string) ([]by
switch {
case len(message.MethodArgs) < 3:
errCh <- fmt.Errorf("error: methodREQAclDeleteCommand: got <3 number methodArgs, want 3")
errCh <- fmt.Errorf("error: methodAclDeleteCommand: got <3 number methodArgs, want 3")
return
}
@ -275,7 +275,7 @@ func methodREQAclDeleteCommand(proc process, message Message, node string) ([]by
case <-ctx.Done():
cancel()
er := fmt.Errorf("error: methodREQAclDeleteCommand: method timed out: %v", message.MethodArgs)
er := fmt.Errorf("error: methodAclDeleteCommand: method timed out: %v", message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logInfo)
case out := <-outCh:
@ -292,8 +292,8 @@ func methodREQAclDeleteCommand(proc process, message Message, node string) ([]by
// ---
func methodREQAclDeleteSource(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodREQAclDeleteSource received from: %v, containing: %v", message.FromNode, message.MethodArgs)
func methodAclDeleteSource(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodAclDeleteSource received from: %v, containing: %v", message.FromNode, message.MethodArgs)
proc.errorKernel.logDebug(er)
proc.processes.wg.Add(1)
@ -311,7 +311,7 @@ func methodREQAclDeleteSource(proc process, message Message, node string) ([]byt
switch {
case len(message.MethodArgs) < 2:
errCh <- fmt.Errorf("error: methodREQAclDeleteSource: got <2 number methodArgs, want 2")
errCh <- fmt.Errorf("error: methodAclDeleteSource: got <2 number methodArgs, want 2")
return
}
@ -336,7 +336,7 @@ func methodREQAclDeleteSource(proc process, message Message, node string) ([]byt
case <-ctx.Done():
cancel()
er := fmt.Errorf("error: methodREQAclDeleteSource: method timed out: %v", message.MethodArgs)
er := fmt.Errorf("error: methodAclDeleteSource: method timed out: %v", message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logInfo)
case out := <-outCh:
@ -353,8 +353,8 @@ func methodREQAclDeleteSource(proc process, message Message, node string) ([]byt
// ---
func methodREQAclGroupNodesAddNode(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodREQAclGroupNodesAddNode received from: %v, containing: %v", message.FromNode, message.MethodArgs)
func methodAclGroupNodesAddNode(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodAclGroupNodesAddNode received from: %v, containing: %v", message.FromNode, message.MethodArgs)
proc.errorKernel.logDebug(er)
proc.processes.wg.Add(1)
@ -372,7 +372,7 @@ func methodREQAclGroupNodesAddNode(proc process, message Message, node string) (
switch {
case len(message.MethodArgs) < 2:
errCh <- fmt.Errorf("error: methodREQAclGroupNodesAddNode: got <2 number methodArgs, want 2")
errCh <- fmt.Errorf("error: methodAclGroupNodesAddNode: got <2 number methodArgs, want 2")
return
}
@ -397,7 +397,7 @@ func methodREQAclGroupNodesAddNode(proc process, message Message, node string) (
case <-ctx.Done():
cancel()
er := fmt.Errorf("error: methodREQAclGroupNodesAddNode: method timed out: %v", message.MethodArgs)
er := fmt.Errorf("error: methodAclGroupNodesAddNode: method timed out: %v", message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logInfo)
case out := <-outCh:
@ -414,8 +414,8 @@ func methodREQAclGroupNodesAddNode(proc process, message Message, node string) (
// ---
func methodREQAclGroupNodesDeleteNode(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodREQAclGroupNodesDeleteNode received from: %v, containing: %v", message.FromNode, message.MethodArgs)
func methodAclGroupNodesDeleteNode(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodAclGroupNodesDeleteNode received from: %v, containing: %v", message.FromNode, message.MethodArgs)
proc.errorKernel.logDebug(er)
proc.processes.wg.Add(1)
@ -433,7 +433,7 @@ func methodREQAclGroupNodesDeleteNode(proc process, message Message, node string
switch {
case len(message.MethodArgs) < 2:
errCh <- fmt.Errorf("error: methodREQAclGroupNodesDeleteNode: got <2 number methodArgs, want 2")
errCh <- fmt.Errorf("error: methodAclGroupNodesDeleteNode: got <2 number methodArgs, want 2")
return
}
@ -458,7 +458,7 @@ func methodREQAclGroupNodesDeleteNode(proc process, message Message, node string
case <-ctx.Done():
cancel()
er := fmt.Errorf("error: methodREQAclGroupNodesDeleteNode: method timed out: %v", message.MethodArgs)
er := fmt.Errorf("error: methodAclGroupNodesDeleteNode: method timed out: %v", message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logInfo)
case out := <-outCh:
@ -475,8 +475,8 @@ func methodREQAclGroupNodesDeleteNode(proc process, message Message, node string
// ---
func methodREQAclGroupNodesDeleteGroup(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodREQAclGroupNodesDeleteGroup received from: %v, containing: %v", message.FromNode, message.MethodArgs)
func methodAclGroupNodesDeleteGroup(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodAclGroupNodesDeleteGroup received from: %v, containing: %v", message.FromNode, message.MethodArgs)
proc.errorKernel.logDebug(er)
proc.processes.wg.Add(1)
@ -494,7 +494,7 @@ func methodREQAclGroupNodesDeleteGroup(proc process, message Message, node strin
switch {
case len(message.MethodArgs) < 1:
errCh <- fmt.Errorf("error: methodREQAclGroupNodesDeleteGroup: got <1 number methodArgs, want 1")
errCh <- fmt.Errorf("error: methodAclGroupNodesDeleteGroup: got <1 number methodArgs, want 1")
return
}
@ -518,7 +518,7 @@ func methodREQAclGroupNodesDeleteGroup(proc process, message Message, node strin
case <-ctx.Done():
cancel()
er := fmt.Errorf("error: methodREQAclGroupNodesDeleteGroup: method timed out: %v", message.MethodArgs)
er := fmt.Errorf("error: methodAclGroupNodesDeleteGroup: method timed out: %v", message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logInfo)
case out := <-outCh:
@ -535,8 +535,8 @@ func methodREQAclGroupNodesDeleteGroup(proc process, message Message, node strin
// ---
func methodREQAclGroupCommandsAddCommand(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodREQAclGroupCommandsAddCommand received from: %v, containing: %v", message.FromNode, message.MethodArgs)
func methodAclGroupCommandsAddCommand(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodAclGroupCommandsAddCommand received from: %v, containing: %v", message.FromNode, message.MethodArgs)
proc.errorKernel.logDebug(er)
proc.processes.wg.Add(1)
@ -554,7 +554,7 @@ func methodREQAclGroupCommandsAddCommand(proc process, message Message, node str
switch {
case len(message.MethodArgs) < 2:
errCh <- fmt.Errorf("error: methodREQAclGroupCommandsAddCommand: got <2 number methodArgs, want 1")
errCh <- fmt.Errorf("error: methodAclGroupCommandsAddCommand: got <2 number methodArgs, want 1")
return
}
@ -579,7 +579,7 @@ func methodREQAclGroupCommandsAddCommand(proc process, message Message, node str
case <-ctx.Done():
cancel()
er := fmt.Errorf("error: methodREQAclGroupCommandsAddCommand: method timed out: %v", message.MethodArgs)
er := fmt.Errorf("error: methodAclGroupCommandsAddCommand: method timed out: %v", message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logInfo)
case out := <-outCh:
@ -596,8 +596,8 @@ func methodREQAclGroupCommandsAddCommand(proc process, message Message, node str
// ---
func methodREQAclGroupCommandsDeleteCommand(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodREQAclGroupCommandsDeleteCommand received from: %v, containing: %v", message.FromNode, message.MethodArgs)
func methodAclGroupCommandsDeleteCommand(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodAclGroupCommandsDeleteCommand received from: %v, containing: %v", message.FromNode, message.MethodArgs)
proc.errorKernel.logDebug(er)
proc.processes.wg.Add(1)
@ -615,7 +615,7 @@ func methodREQAclGroupCommandsDeleteCommand(proc process, message Message, node
switch {
case len(message.MethodArgs) < 1:
errCh <- fmt.Errorf("error: methodREQAclGroupCommandsDeleteCommand: got <1 number methodArgs, want 1")
errCh <- fmt.Errorf("error: methodAclGroupCommandsDeleteCommand: got <1 number methodArgs, want 1")
return
}
@ -640,7 +640,7 @@ func methodREQAclGroupCommandsDeleteCommand(proc process, message Message, node
case <-ctx.Done():
cancel()
er := fmt.Errorf("error: methodREQAclGroupCommandsDeleteCommand: method timed out: %v", message.MethodArgs)
er := fmt.Errorf("error: methodAclGroupCommandsDeleteCommand: method timed out: %v", message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logInfo)
case out := <-outCh:
@ -657,8 +657,8 @@ func methodREQAclGroupCommandsDeleteCommand(proc process, message Message, node
// ---
func methodREQAclGroupCommandsDeleteGroup(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodREQAclGroupCommandsDeleteGroup received from: %v, containing: %v", message.FromNode, message.MethodArgs)
func methodAclGroupCommandsDeleteGroup(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodAclGroupCommandsDeleteGroup received from: %v, containing: %v", message.FromNode, message.MethodArgs)
proc.errorKernel.logDebug(er)
proc.processes.wg.Add(1)
@ -676,7 +676,7 @@ func methodREQAclGroupCommandsDeleteGroup(proc process, message Message, node st
switch {
case len(message.MethodArgs) < 1:
errCh <- fmt.Errorf("error: methodREQAclGroupCommandsDeleteGroup: got <1 number methodArgs, want 1")
errCh <- fmt.Errorf("error: methodAclGroupCommandsDeleteGroup: got <1 number methodArgs, want 1")
return
}
@ -700,7 +700,7 @@ func methodREQAclGroupCommandsDeleteGroup(proc process, message Message, node st
case <-ctx.Done():
cancel()
er := fmt.Errorf("error: methodREQAclGroupCommandsDeleteGroup: method timed out: %v", message.MethodArgs)
er := fmt.Errorf("error: methodAclGroupCommandsDeleteGroup: method timed out: %v", message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logInfo)
case out := <-outCh:
@ -717,8 +717,8 @@ func methodREQAclGroupCommandsDeleteGroup(proc process, message Message, node st
// ---
func methodREQAclExport(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodREQAclExport received from: %v, containing: %v", message.FromNode, message.MethodArgs)
func methodAclExport(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodAclExport received from: %v, containing: %v", message.FromNode, message.MethodArgs)
proc.errorKernel.logDebug(er)
proc.processes.wg.Add(1)
@ -736,7 +736,7 @@ func methodREQAclExport(proc process, message Message, node string) ([]byte, err
out, err := proc.centralAuth.exportACLs()
if err != nil {
errCh <- fmt.Errorf("error: methodREQAclExport failed: %v", err)
errCh <- fmt.Errorf("error: methodAclExport failed: %v", err)
return
}
@ -756,7 +756,7 @@ func methodREQAclExport(proc process, message Message, node string) ([]byte, err
case <-ctx.Done():
cancel()
er := fmt.Errorf("error: methodREQAclExport: method timed out: %v", message.MethodArgs)
er := fmt.Errorf("error: methodAclExport: method timed out: %v", message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logInfo)
case out := <-outCh:
@ -773,8 +773,8 @@ func methodREQAclExport(proc process, message Message, node string) ([]byte, err
// ---
func methodREQAclImport(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodREQAclImport received from: %v, containing: %v", message.FromNode, message.MethodArgs)
func methodAclImport(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- methodAclImport received from: %v, containing: %v", message.FromNode, message.MethodArgs)
proc.errorKernel.logDebug(er)
proc.processes.wg.Add(1)
@ -793,14 +793,14 @@ func methodREQAclImport(proc process, message Message, node string) ([]byte, err
switch {
case len(message.MethodArgs) < 1:
errCh <- fmt.Errorf("error: methodREQAclImport: got <1 number methodArgs, want 1")
errCh <- fmt.Errorf("error: methodAclImport: got <1 number methodArgs, want 1")
return
}
js := []byte(message.MethodArgs[0])
err := proc.centralAuth.importACLs(js)
if err != nil {
errCh <- fmt.Errorf("error: methodREQAclImport failed: %v", err)
errCh <- fmt.Errorf("error: methodAclImport failed: %v", err)
return
}
@ -820,7 +820,7 @@ func methodREQAclImport(proc process, message Message, node string) ([]byte, err
case <-ctx.Done():
cancel()
er := fmt.Errorf("error: methodREQAclImport: method timed out")
er := fmt.Errorf("error: methodAclImport: method timed out")
proc.errorKernel.errSend(proc, message, er, logInfo)
case out := <-outCh:

22
requests_cli.go
View file

@ -12,7 +12,7 @@ import (
// 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 methodREQCliCommand(proc process, message Message, node string) ([]byte, error) {
func methodCliCommand(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- CLICommandREQUEST received from: %v, containing: %v", message.FromNode, message.MethodArgs)
proc.errorKernel.logDebug(er)
@ -31,7 +31,7 @@ func methodREQCliCommand(proc process, message Message, node string) ([]byte, er
switch {
case len(message.MethodArgs) < 1:
er := fmt.Errorf("error: methodREQCliCommand: got <1 number methodArgs")
er := fmt.Errorf("error: methodCliCommand: got <1 number methodArgs")
proc.errorKernel.errSend(proc, message, er, logWarning)
newReplyMessage(proc, msgForErrors, []byte(er.Error()))
@ -83,7 +83,7 @@ func methodREQCliCommand(proc process, message Message, node string) ([]byte, er
cmd.WaitDelay = time.Second * 5
err := cmd.Run()
if err != nil {
er := fmt.Errorf("error: methodREQCliCommand: cmd.Run failed : %v, methodArgs: %v, error_output: %v", err, message.MethodArgs, stderr.String())
er := fmt.Errorf("error: methodCliCommand: cmd.Run failed : %v, methodArgs: %v, error_output: %v", err, message.MethodArgs, stderr.String())
proc.errorKernel.errSend(proc, message, er, logWarning)
newReplyMessage(proc, msgForErrors, []byte(er.Error()))
}
@ -98,7 +98,7 @@ func methodREQCliCommand(proc process, message Message, node string) ([]byte, er
select {
case <-ctx.Done():
cancel()
er := fmt.Errorf("error: methodREQCliCommand: method timed out: %v", message.MethodArgs)
er := fmt.Errorf("error: methodCliCommand: method timed out: %v", message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logWarning)
newReplyMessage(proc, msgForErrors, []byte(er.Error()))
case out := <-outCh:
@ -127,7 +127,7 @@ func methodREQCliCommand(proc process, message Message, node string) ([]byte, er
// 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 methodREQCliCommandCont(proc process, message Message, node string) ([]byte, error) {
func methodCliCommandCont(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- CLInCommandCont REQUEST received from: %v, containing: %v", message.FromNode, message.Data)
proc.errorKernel.logDebug(er)
@ -150,7 +150,7 @@ func methodREQCliCommandCont(proc process, message Message, node string) ([]byte
switch {
case len(message.MethodArgs) < 1:
er := fmt.Errorf("error: methodREQCliCommand: got <1 number methodArgs")
er := fmt.Errorf("error: methodCliCommand: got <1 number methodArgs")
proc.errorKernel.errSend(proc, message, er, logWarning)
newReplyMessage(proc, msgForErrors, []byte(er.Error()))
@ -179,21 +179,21 @@ func methodREQCliCommandCont(proc process, message Message, node string) ([]byte
// able to read the out put of the command.
outReader, err := cmd.StdoutPipe()
if err != nil {
er := fmt.Errorf("error: methodREQCliCommandCont: cmd.StdoutPipe failed : %v, methodArgs: %v", err, message.MethodArgs)
er := fmt.Errorf("error: methodCliCommandCont: cmd.StdoutPipe failed : %v, methodArgs: %v", err, message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logWarning)
newReplyMessage(proc, msgForErrors, []byte(er.Error()))
}
ErrorReader, err := cmd.StderrPipe()
if err != nil {
er := fmt.Errorf("error: methodREQCliCommandCont: cmd.StderrPipe failed : %v, methodArgs: %v", err, message.MethodArgs)
er := fmt.Errorf("error: methodCliCommandCont: cmd.StderrPipe failed : %v, methodArgs: %v", err, message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logWarning)
newReplyMessage(proc, msgForErrors, []byte(er.Error()))
}
cmd.WaitDelay = time.Second * 5
if err := cmd.Start(); err != nil {
er := fmt.Errorf("error: methodREQCliCommandCont: cmd.Start failed : %v, methodArgs: %v", err, message.MethodArgs)
er := fmt.Errorf("error: methodCliCommandCont: cmd.Start failed : %v, methodArgs: %v", err, message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logWarning)
newReplyMessage(proc, msgForErrors, []byte(er.Error()))
}
@ -224,7 +224,7 @@ func methodREQCliCommandCont(proc process, message Message, node string) ([]byte
cancel()
if err := cmd.Wait(); err != nil {
er := fmt.Errorf("info: methodREQCliCommandCont: method timeout reached, canceled: methodArgs: %v, %v", message.MethodArgs, err)
er := fmt.Errorf("info: methodCliCommandCont: method timeout reached, canceled: methodArgs: %v, %v", message.MethodArgs, err)
proc.errorKernel.errSend(proc, message, er, logWarning)
}
@ -235,7 +235,7 @@ func methodREQCliCommandCont(proc process, message Message, node string) ([]byte
select {
case <-ctx.Done():
cancel()
er := fmt.Errorf("info: methodREQCliCommandCont: method timeout reached, canceling: methodArgs: %v", message.MethodArgs)
er := fmt.Errorf("info: methodCliCommandCont: method timeout reached, canceling: methodArgs: %v", message.MethodArgs)
proc.errorKernel.infoSend(proc, message, er)
newReplyMessage(proc, msgForErrors, []byte(er.Error()))
return

42
requests_copy.go
View file

@ -32,7 +32,7 @@ type copyInitialData struct {
FolderPermission uint64
}
// methodREQCopySrc are handles the initial and first part of
// methodCopySrc are handles the initial and first part of
// the message flow for a copy to destination request.
// It's main role is to start up a sub process for the destination
// in which all the actual file copying is done.
@ -80,7 +80,7 @@ type copyInitialData struct {
// -----------------------------------------------------
// Handle writing to a file. Will truncate any existing data if the file did already
// exist.
func methodREQCopySrc(proc process, message Message, node string) ([]byte, error) {
func methodCopySrc(proc process, message Message, node string) ([]byte, error) {
// If the toNode field is not the same as nodeName of the receiving node
// we should forward the message to that specified toNode. This will allow
@ -127,7 +127,7 @@ func methodREQCopySrc(proc process, message Message, node string) ([]byte, error
// Verify and check the methodArgs
if len(message.MethodArgs) < 3 {
er := fmt.Errorf("error: methodREQCopySrc: got <3 number methodArgs: want srcfilePath,dstNode,dstFilePath")
er := fmt.Errorf("error: methodCopySrc: got <3 number methodArgs: want srcfilePath,dstNode,dstFilePath")
proc.errorKernel.errSend(proc, message, er, logWarning)
return
}
@ -137,7 +137,7 @@ func methodREQCopySrc(proc process, message Message, node string) ([]byte, error
var err error
splitChunkSize, err = strconv.Atoi(message.MethodArgs[3])
if err != nil {
er := fmt.Errorf("error: methodREQCopySrc: unble to convert splitChunkSize into int value: %v", err)
er := fmt.Errorf("error: methodCopySrc: unble to convert splitChunkSize into int value: %v", err)
proc.errorKernel.errSend(proc, message, er, logWarning)
}
}
@ -147,7 +147,7 @@ func methodREQCopySrc(proc process, message Message, node string) ([]byte, error
var err error
maxTotalCopyTime, err = strconv.Atoi(message.MethodArgs[4])
if err != nil {
er := fmt.Errorf("error: methodREQCopySrc: unble to convert maxTotalCopyTime into int value: %v", err)
er := fmt.Errorf("error: methodCopySrc: unble to convert maxTotalCopyTime into int value: %v", err)
proc.errorKernel.errSend(proc, message, er, logWarning)
}
}
@ -157,14 +157,14 @@ func methodREQCopySrc(proc process, message Message, node string) ([]byte, error
var err error
folderPermission, err = strconv.ParseUint(message.MethodArgs[5], 8, 32)
if err != nil {
er := fmt.Errorf("methodREQCopySrc: failed to parse uint, %v", err)
er := fmt.Errorf("methodCopySrc: failed to parse uint, %v", err)
proc.errorKernel.logError(er)
}
er := fmt.Errorf("info: FolderPermission defined in message for socket: %v, converted = %v", message.MethodArgs[5], folderPermission)
proc.errorKernel.logDebug(er)
if err != nil {
er := fmt.Errorf("error: methodREQCopySrc: unable to convert folderPermission into int value: %v", err)
er := fmt.Errorf("error: methodCopySrc: unable to convert folderPermission into int value: %v", err)
proc.errorKernel.errSend(proc, message, er, logWarning)
}
}
@ -183,7 +183,7 @@ func methodREQCopySrc(proc process, message Message, node string) ([]byte, error
// Create a subject for one copy request
uid := uuid.New()
subProcessName = fmt.Sprintf("REQSUBCopySrc.%v", uid.String())
subProcessName = fmt.Sprintf("%v.%v", SUBCopySrc, uid.String())
dstDir := filepath.Dir(DstFilePath)
dstFile := filepath.Base(DstFilePath)
@ -192,13 +192,13 @@ func methodREQCopySrc(proc process, message Message, node string) ([]byte, error
// Also choosing to create the naming for the dst method here so
// we can have all the information in the cia from the beginning
// at both ends.
dstSubProcessName := fmt.Sprintf("REQSUBCopyDst.%v", uid.String())
dstSubProcessName := fmt.Sprintf("%v.%v", SUBCopyDst, uid.String())
dstM := Method(dstSubProcessName)
// Get the file permissions
fileInfo, err := os.Stat(SrcFilePath)
if err != nil {
// errCh <- fmt.Errorf("error: methodREQCopySrc: failed to open file: %v, %v", SrcFilePath, err)
// errCh <- fmt.Errorf("error: methodCopySrc: failed to open file: %v, %v", SrcFilePath, err)
er := fmt.Errorf("error: copySrcSubProcFunc: failed to stat file: %v", err)
proc.errorKernel.logDebug(er)
return
@ -251,7 +251,7 @@ func methodREQCopySrc(proc process, message Message, node string) ([]byte, error
msg := message
msg.ToNode = Node(DstNode)
//msg.Method = REQToFile
msg.Method = REQCopyDst
msg.Method = CopyDst
msg.Data = cb
msg.ACKTimeout = message.ACKTimeout
msg.Retries = message.Retries
@ -289,11 +289,11 @@ func newSubProcess(ctx context.Context, server *server, subject Subject, process
// ----
// methodREQCopyDst are handles the initial and first part of
// methodCopyDst are handles the initial and first part of
// the message flow for a copy to destination request.
// It's main role is to start up a sub process for the destination
// in which all the actual file copying is done.
func methodREQCopyDst(proc process, message Message, node string) ([]byte, error) {
func methodCopyDst(proc process, message Message, node string) ([]byte, error) {
var subProcessName string
proc.processes.wg.Add(1)
@ -304,7 +304,7 @@ func methodREQCopyDst(proc process, message Message, node string) ([]byte, error
var cia copyInitialData
err := cbor.Unmarshal(message.Data, &cia)
if err != nil {
er := fmt.Errorf("error: methodREQCopyDst: failed to cbor Unmarshal data: %v, message=%v", err, message)
er := fmt.Errorf("error: methodCopyDst: failed to cbor Unmarshal data: %v, message=%v", err, message)
proc.errorKernel.errSend(proc, message, er, logWarning)
return
}
@ -341,7 +341,7 @@ func methodREQCopyDst(proc process, message Message, node string) ([]byte, error
proc.processes.active.mu.Unlock()
if ok {
er := fmt.Errorf("methodREQCopyDst: subprocesses already existed, will not start another subscriber for %v", pn)
er := fmt.Errorf("methodCopyDst: subprocesses already existed, will not start another subscriber for %v", pn)
proc.errorKernel.logDebug(er)
// HERE!!!
@ -549,7 +549,7 @@ func copySrcSubProcFunc(proc process, cia copyInitialData, cancel context.Cancel
ToNode: cia.DstNode,
FromNode: cia.SrcNode,
Method: cia.DstMethod,
ReplyMethod: REQNone,
ReplyMethod: None,
Data: csaSerialized,
IsSubPublishedMsg: true,
ACKTimeout: initialMessage.ACKTimeout,
@ -619,7 +619,7 @@ func copySrcSubProcFunc(proc process, cia copyInitialData, cancel context.Cancel
ToNode: cia.DstNode,
FromNode: cia.SrcNode,
Method: cia.DstMethod,
ReplyMethod: REQNone,
ReplyMethod: None,
Data: csaSerialized,
IsSubPublishedMsg: true,
ACKTimeout: initialMessage.ACKTimeout,
@ -683,7 +683,7 @@ func copyDstSubProcFunc(proc process, cia copyInitialData, message Message, canc
ToNode: cia.SrcNode,
FromNode: cia.DstNode,
Method: cia.SrcMethod,
ReplyMethod: REQNone,
ReplyMethod: None,
Data: csaSerialized,
IsSubPublishedMsg: true,
ACKTimeout: message.ACKTimeout,
@ -785,7 +785,7 @@ func copyDstSubProcFunc(proc process, cia copyInitialData, message Message, canc
ToNode: cia.SrcNode,
FromNode: cia.DstNode,
Method: cia.SrcMethod,
ReplyMethod: REQNone,
ReplyMethod: None,
Data: csaSer,
IsSubPublishedMsg: true,
ACKTimeout: message.ACKTimeout,
@ -818,7 +818,7 @@ func copyDstSubProcFunc(proc process, cia copyInitialData, message Message, canc
ToNode: cia.SrcNode,
FromNode: cia.DstNode,
Method: cia.SrcMethod,
ReplyMethod: REQNone,
ReplyMethod: None,
Data: csaSer,
IsSubPublishedMsg: true,
ACKTimeout: message.ACKTimeout,
@ -972,7 +972,7 @@ func copyDstSubProcFunc(proc process, cia copyInitialData, message Message, canc
ToNode: cia.SrcNode,
FromNode: cia.DstNode,
Method: cia.SrcMethod,
ReplyMethod: REQNone,
ReplyMethod: None,
Data: csaSerialized,
IsSubPublishedMsg: true,
ACKTimeout: message.ACKTimeout,

20
requests_file_handling.go
View file

@ -32,14 +32,14 @@ func reqWriteFileOrSocket(isAppend bool, proc process, message Message) error {
if fi != nil && fi.Mode().Type() == fs.ModeSocket {
socket, err := net.Dial("unix", file)
if err != nil {
er := fmt.Errorf("error: methodREQToFile/Append could to open socket file for writing: subject:%v, folderTree: %v, %v", proc.subject, folderTree, err)
er := fmt.Errorf("error: methodToFile/Append could to open socket file for writing: subject:%v, folderTree: %v, %v", proc.subject, folderTree, err)
return er
}
defer socket.Close()
_, err = socket.Write([]byte(message.Data))
if err != nil {
er := fmt.Errorf("error: methodREQToFile/Append could not write to socket: subject:%v, folderTree: %v, %v", proc.subject, folderTree, err)
er := fmt.Errorf("error: methodToFile/Append could not write to socket: subject:%v, folderTree: %v, %v", proc.subject, folderTree, err)
return er
}
@ -50,7 +50,7 @@ func reqWriteFileOrSocket(isAppend bool, proc process, message Message) error {
if _, err := os.Stat(folderTree); os.IsNotExist(err) {
err := os.MkdirAll(folderTree, 0770)
if err != nil {
er := fmt.Errorf("error: methodREQToFile/Append failed to create toFile directory tree: subject:%v, folderTree: %v, %v", proc.subject, folderTree, err)
er := fmt.Errorf("error: methodToFile/Append failed to create toFile directory tree: subject:%v, folderTree: %v, %v", proc.subject, folderTree, err)
return er
}
@ -70,7 +70,7 @@ func reqWriteFileOrSocket(isAppend bool, proc process, message Message) error {
// Open file and write data.
f, err := os.OpenFile(file, fileFlag, 0755)
if err != nil {
er := fmt.Errorf("error: methodREQToFile/Append: failed to open file, check that you've specified a value for fileName in the message: directory: %v, fileName: %v, %v", message.Directory, message.FileName, err)
er := fmt.Errorf("error: methodToFile/Append: failed to open file, check that you've specified a value for fileName in the message: directory: %v, fileName: %v, %v", message.Directory, message.FileName, err)
return er
}
@ -79,7 +79,7 @@ func reqWriteFileOrSocket(isAppend bool, proc process, message Message) error {
_, err = f.Write(message.Data)
f.Sync()
if err != nil {
er := fmt.Errorf("error: methodREQToFile/Append: failed to write to file: file: %v, %v", file, err)
er := fmt.Errorf("error: methodToFile/Append: failed to write to file: file: %v, %v", file, err)
return er
}
@ -88,7 +88,7 @@ func reqWriteFileOrSocket(isAppend bool, proc process, message Message) error {
}
// Handle appending data to file.
func methodREQToFileAppend(proc process, message Message, node string) ([]byte, error) {
func methodFileAppend(proc process, message Message, node string) ([]byte, error) {
err := reqWriteFileOrSocket(true, proc, message)
proc.errorKernel.errSend(proc, message, err, logWarning)
@ -100,7 +100,7 @@ func methodREQToFileAppend(proc process, message Message, node string) ([]byte,
// Handle writing to a file. Will truncate any existing data if the file did already
// exist.
func methodREQToFile(proc process, message Message, node string) ([]byte, error) {
func methodToFile(proc process, message Message, node string) ([]byte, error) {
err := reqWriteFileOrSocket(false, proc, message)
proc.errorKernel.errSend(proc, message, err, logWarning)
@ -108,12 +108,12 @@ func methodREQToFile(proc process, message Message, node string) ([]byte, error)
return ackMsg, nil
}
// --- methodREQTailFile
// --- methodTailFile
// 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 methodREQTailFile(proc process, message Message, node string) ([]byte, error) {
func methodTailFile(proc process, message Message, node string) ([]byte, error) {
inf := fmt.Errorf("<--- TailFile REQUEST received from: %v, containing: %v", message.FromNode, message.Data)
proc.errorKernel.logDebug(inf)
@ -123,7 +123,7 @@ func methodREQTailFile(proc process, message Message, node string) ([]byte, erro
switch {
case len(message.MethodArgs) < 1:
er := fmt.Errorf("error: methodREQTailFile: got <1 number methodArgs")
er := fmt.Errorf("error: methodTailFile: got <1 number methodArgs")
proc.errorKernel.errSend(proc, message, er, logWarning)
return

32
requests_http.go
View file

@ -10,7 +10,7 @@ import (
)
// handler to do a Http Get.
func methodREQHttpGet(proc process, message Message, node string) ([]byte, error) {
func methodHttpGet(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- REQHttpGet received from: %v, containing: %v", message.FromNode, message.Data)
proc.errorKernel.logDebug(er)
@ -23,7 +23,7 @@ func methodREQHttpGet(proc process, message Message, node string) ([]byte, error
switch {
case len(message.MethodArgs) < 1:
er := fmt.Errorf("error: methodREQHttpGet: got <1 number methodArgs")
er := fmt.Errorf("error: methodHttpGet: got <1 number methodArgs")
proc.errorKernel.errSend(proc, message, er, logWarning)
newReplyMessage(proc, msgForErrors, []byte(er.Error()))
@ -41,7 +41,7 @@ func methodREQHttpGet(proc process, message Message, node string) ([]byte, error
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.MethodArgs)
er := fmt.Errorf("error: methodHttpGet: NewRequest failed: %v, bailing out: %v", err, message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logWarning)
newReplyMessage(proc, msgForErrors, []byte(er.Error()))
cancel()
@ -56,7 +56,7 @@ func methodREQHttpGet(proc process, message Message, node string) ([]byte, error
resp, err := client.Do(req)
if err != nil {
er := fmt.Errorf("error: methodREQHttpGet: client.Do failed: %v, bailing out: %v", err, message.MethodArgs)
er := fmt.Errorf("error: methodHttpGet: client.Do failed: %v, bailing out: %v", err, message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logWarning)
newReplyMessage(proc, msgForErrors, []byte(er.Error()))
return
@ -65,7 +65,7 @@ func methodREQHttpGet(proc process, message Message, node string) ([]byte, error
if resp.StatusCode != 200 {
cancel()
er := fmt.Errorf("error: methodREQHttpGet: not 200, were %#v, bailing out: %v", resp.StatusCode, message)
er := fmt.Errorf("error: methodHttpGet: not 200, were %#v, bailing out: %v", resp.StatusCode, message)
proc.errorKernel.errSend(proc, message, er, logWarning)
newReplyMessage(proc, msgForErrors, []byte(er.Error()))
return
@ -73,7 +73,7 @@ func methodREQHttpGet(proc process, message Message, node string) ([]byte, error
body, err := io.ReadAll(resp.Body)
if err != nil {
er := fmt.Errorf("error: methodREQHttpGet: io.ReadAll failed : %v, methodArgs: %v", err, message.MethodArgs)
er := fmt.Errorf("error: methodHttpGet: io.ReadAll failed : %v, methodArgs: %v", err, message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logWarning)
newReplyMessage(proc, msgForErrors, []byte(er.Error()))
}
@ -90,7 +90,7 @@ func methodREQHttpGet(proc process, message Message, node string) ([]byte, error
select {
case <-ctx.Done():
cancel()
er := fmt.Errorf("error: methodREQHttpGet: method timed out: %v", message.MethodArgs)
er := fmt.Errorf("error: methodHttpGet: method timed out: %v", message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logWarning)
newReplyMessage(proc, msgForErrors, []byte(er.Error()))
case out := <-outCh:
@ -111,7 +111,7 @@ func methodREQHttpGet(proc process, message Message, node string) ([]byte, error
// handler to do a Http Get Scheduled.
// The second element of the MethodArgs slice holds the timer defined in seconds.
func methodREQHttpGetScheduled(proc process, message Message, node string) ([]byte, error) {
func methodHttpGetScheduled(proc process, message Message, node string) ([]byte, error) {
er := fmt.Errorf("<--- REQHttpGetScheduled received from: %v, containing: %v", message.FromNode, message.Data)
proc.errorKernel.logDebug(er)
@ -123,7 +123,7 @@ func methodREQHttpGetScheduled(proc process, message Message, node string) ([]by
switch {
case len(message.MethodArgs) < 3:
er := fmt.Errorf("error: methodREQHttpGet: got <3 number methodArgs. Want URL, Schedule Interval in seconds, and the total time in minutes the scheduler should run for")
er := fmt.Errorf("error: methodHttpGet: got <3 number methodArgs. Want URL, Schedule Interval in seconds, and the total time in minutes the scheduler should run for")
proc.errorKernel.errSend(proc, message, er, logWarning)
return
@ -133,14 +133,14 @@ func methodREQHttpGetScheduled(proc process, message Message, node string) ([]by
scheduleInterval, err := strconv.Atoi(message.MethodArgs[1])
if err != nil {
er := fmt.Errorf("error: methodREQHttpGetScheduled: schedule interval value is not a valid int number defined as a string value seconds: %v, bailing out: %v", err, message.MethodArgs)
er := fmt.Errorf("error: methodHttpGetScheduled: schedule interval value is not a valid int number defined as a string value seconds: %v, bailing out: %v", err, message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logWarning)
return
}
schedulerTotalTime, err := strconv.Atoi(message.MethodArgs[2])
if err != nil {
er := fmt.Errorf("error: methodREQHttpGetScheduled: scheduler total time value is not a valid int number defined as a string value minutes: %v, bailing out: %v", err, message.MethodArgs)
er := fmt.Errorf("error: methodHttpGetScheduled: scheduler total time value is not a valid int number defined as a string value minutes: %v, bailing out: %v", err, message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logWarning)
return
}
@ -174,7 +174,7 @@ func methodREQHttpGetScheduled(proc process, message Message, node string) ([]by
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
if err != nil {
er := fmt.Errorf("error: methodREQHttpGet: NewRequest failed: %v, error: %v", err, message.MethodArgs)
er := fmt.Errorf("error: methodHttpGet: NewRequest failed: %v, error: %v", err, message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logWarning)
cancel()
return
@ -187,7 +187,7 @@ func methodREQHttpGetScheduled(proc process, message Message, node string) ([]by
resp, err := client.Do(req)
if err != nil {
er := fmt.Errorf("error: methodREQHttpGet: client.Do failed: %v, error: %v", err, message.MethodArgs)
er := fmt.Errorf("error: methodHttpGet: client.Do failed: %v, error: %v", err, message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logWarning)
return
}
@ -195,14 +195,14 @@ func methodREQHttpGetScheduled(proc process, message Message, node string) ([]by
if resp.StatusCode != 200 {
cancel()
er := fmt.Errorf("error: methodREQHttpGet: not 200, were %#v, error: %v", resp.StatusCode, message)
er := fmt.Errorf("error: methodHttpGet: not 200, were %#v, error: %v", resp.StatusCode, message)
proc.errorKernel.errSend(proc, message, er, logWarning)
return
}
body, err := io.ReadAll(resp.Body)
if err != nil {
er := fmt.Errorf("error: methodREQHttpGet: io.ReadAll failed : %v, methodArgs: %v", err, message.MethodArgs)
er := fmt.Errorf("error: methodHttpGet: io.ReadAll failed : %v, methodArgs: %v", err, message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logWarning)
}
@ -233,7 +233,7 @@ func methodREQHttpGetScheduled(proc process, message Message, node string) ([]by
case <-ctxScheduler.Done():
// fmt.Printf(" * DEBUG: <-ctxScheduler.Done()\n")
cancel()
er := fmt.Errorf("error: methodREQHttpGet: schedule context timed out: %v", message.MethodArgs)
er := fmt.Errorf("error: methodHttpGet: schedule context timed out: %v", message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logWarning)
return
case out := <-outCh:

44
requests_keys.go
View file

@ -9,7 +9,7 @@ import (
// ---
// Handler to get the public ed25519 key from a node.
func methodREQPublicKey(proc process, message Message, node string) ([]byte, error) {
func methodPublicKey(proc process, message Message, node string) ([]byte, error) {
// Get a context with the timeout specified in message.MethodTimeout.
ctx, _ := getContextForMethodTimeout(proc.ctx, message)
@ -48,7 +48,7 @@ func methodREQPublicKey(proc process, message Message, node string) ([]byte, err
// ----
// Handler to get all the public ed25519 keys from a central server.
func methodREQKeysRequestUpdate(proc process, message Message, node string) ([]byte, error) {
func methodKeysRequestUpdate(proc process, message Message, node string) ([]byte, error) {
// Get a context with the timeout specified in message.MethodTimeout.
// TODO:
@ -85,30 +85,30 @@ func methodREQKeysRequestUpdate(proc process, message Message, node string) ([]b
proc.centralAuth.pki.nodesAcked.mu.Lock()
defer proc.centralAuth.pki.nodesAcked.mu.Unlock()
er := fmt.Errorf(" <---- methodREQKeysRequestUpdate: received hash from NODE=%v, HASH=%v", message.FromNode, message.Data)
er := fmt.Errorf(" <---- methodKeysRequestUpdate: received hash from NODE=%v, HASH=%v", message.FromNode, message.Data)
proc.errorKernel.logDebug(er)
// Check if the received hash is the same as the one currently active,
if bytes.Equal(proc.centralAuth.pki.nodesAcked.keysAndHash.Hash[:], message.Data) {
er := fmt.Errorf("info: methodREQKeysRequestUpdate: node %v and central have equal keys, nothing to do, exiting key update handler", message.FromNode)
er := fmt.Errorf("info: methodKeysRequestUpdate: node %v and central have equal keys, nothing to do, exiting key update handler", message.FromNode)
// proc.errorKernel.infoSend(proc, message, er)
proc.errorKernel.logDebug(er)
return
}
er = fmt.Errorf("info: methodREQKeysRequestUpdate: node %v and central had not equal keys, preparing to send new version of keys", message.FromNode)
er = fmt.Errorf("info: methodKeysRequestUpdate: node %v and central had not equal keys, preparing to send new version of keys", message.FromNode)
proc.errorKernel.logDebug(er)
er = fmt.Errorf("info: methodREQKeysRequestUpdate: marshalling new keys and hash to send: map=%v, hash=%v", proc.centralAuth.pki.nodesAcked.keysAndHash.Keys, proc.centralAuth.pki.nodesAcked.keysAndHash.Hash)
er = fmt.Errorf("info: methodKeysRequestUpdate: marshalling new keys and hash to send: map=%v, hash=%v", proc.centralAuth.pki.nodesAcked.keysAndHash.Keys, proc.centralAuth.pki.nodesAcked.keysAndHash.Hash)
proc.errorKernel.logDebug(er)
b, err := json.Marshal(proc.centralAuth.pki.nodesAcked.keysAndHash)
if err != nil {
er := fmt.Errorf("error: methodREQKeysRequestUpdate, failed to marshal keys map: %v", err)
er := fmt.Errorf("error: methodKeysRequestUpdate, failed to marshal keys map: %v", err)
proc.errorKernel.errSend(proc, message, er, logWarning)
}
er = fmt.Errorf("----> methodREQKeysRequestUpdate: SENDING KEYS TO NODE=%v", message.FromNode)
er = fmt.Errorf("----> methodKeysRequestUpdate: SENDING KEYS TO NODE=%v", message.FromNode)
proc.errorKernel.logDebug(er)
newReplyMessage(proc, message, b)
}()
@ -122,7 +122,7 @@ func methodREQKeysRequestUpdate(proc process, message Message, node string) ([]b
// ----
// Handler to receive the public keys from a central server.
func methodREQKeysDeliverUpdate(proc process, message Message, node string) ([]byte, error) {
func methodKeysDeliverUpdate(proc process, message Message, node string) ([]byte, error) {
// Get a context with the timeout specified in message.MethodTimeout.
// TODO:
@ -207,7 +207,7 @@ func methodREQKeysDeliverUpdate(proc process, message Message, node string) ([]b
// are recived on the central server they will be put into a temp key
// map, and we need to acknowledge them before they are moved into the
// main key map, and then allowed to be sent out to other nodes.
func methodREQKeysAllow(proc process, message Message, node string) ([]byte, error) {
func methodKeysAllow(proc process, message Message, node string) ([]byte, error) {
// Get a context with the timeout specified in message.MethodTimeout.
ctx, _ := getContextForMethodTimeout(proc.ctx, message)
@ -300,7 +300,7 @@ func pushKeys(proc process, message Message, nodes []Node) error {
b, err := json.Marshal(proc.centralAuth.pki.nodesAcked.keysAndHash)
if err != nil {
er := fmt.Errorf("error: methodREQKeysAllow, failed to marshal keys map: %v", err)
er := fmt.Errorf("error: methodKeysAllow, failed to marshal keys map: %v", err)
return er
}
@ -322,8 +322,8 @@ func pushKeys(proc process, message Message, nodes []Node) error {
proc.errorKernel.logDebug(er)
msg := Message{
ToNode: n,
Method: REQKeysDeliverUpdate,
ReplyMethod: REQNone,
Method: KeysDeliverUpdate,
ReplyMethod: None,
ACKTimeout: 0,
}
@ -336,7 +336,7 @@ func pushKeys(proc process, message Message, nodes []Node) error {
proc.toRingbufferCh <- []subjectAndMessage{sam}
er = fmt.Errorf("----> methodREQKeysAllow: SENDING KEYS TO NODE=%v", message.FromNode)
er = fmt.Errorf("----> methodKeysAllow: SENDING KEYS TO NODE=%v", message.FromNode)
proc.errorKernel.logDebug(er)
}
@ -344,7 +344,7 @@ func pushKeys(proc process, message Message, nodes []Node) error {
b, err := json.Marshal(proc.centralAuth.pki.nodesAcked.keysAndHash)
if err != nil {
er := fmt.Errorf("error: methodREQKeysAllow, failed to marshal keys map: %v", err)
er := fmt.Errorf("error: methodKeysAllow, failed to marshal keys map: %v", err)
proc.errorKernel.errSend(proc, message, er, logWarning)
}
@ -368,9 +368,9 @@ func pushKeys(proc process, message Message, nodes []Node) error {
proc.errorKernel.logDebug(er)
msg := Message{
ToNode: n,
Method: REQKeysDeliverUpdate,
Method: KeysDeliverUpdate,
Data: b,
ReplyMethod: REQNone,
ReplyMethod: None,
ACKTimeout: 0,
}
@ -383,7 +383,7 @@ func pushKeys(proc process, message Message, nodes []Node) error {
proc.toRingbufferCh <- []subjectAndMessage{sam}
er = fmt.Errorf("----> methodREQKeysAllow: sending keys update to node=%v", message.FromNode)
er = fmt.Errorf("----> methodKeysAllow: sending keys update to node=%v", message.FromNode)
proc.errorKernel.logDebug(er)
}
@ -391,8 +391,8 @@ func pushKeys(proc process, message Message, nodes []Node) error {
}
func methodREQKeysDelete(proc process, message Message, node string) ([]byte, error) {
inf := fmt.Errorf("<--- methodREQKeysDelete received from: %v, containing: %v", message.FromNode, message.MethodArgs)
func methodKeysDelete(proc process, message Message, node string) ([]byte, error) {
inf := fmt.Errorf("<--- methodKeysDelete received from: %v, containing: %v", message.FromNode, message.MethodArgs)
proc.errorKernel.logDebug(inf)
proc.processes.wg.Add(1)
@ -410,7 +410,7 @@ func methodREQKeysDelete(proc process, message Message, node string) ([]byte, er
switch {
case len(message.MethodArgs) < 1:
errCh <- fmt.Errorf("error: methodREQAclGroupNodesDeleteNode: got <1 number methodArgs, want >0")
errCh <- fmt.Errorf("error: methodAclGroupNodesDeleteNode: got <1 number methodArgs, want >0")
return
}
@ -460,7 +460,7 @@ func methodREQKeysDelete(proc process, message Message, node string) ([]byte, er
case <-ctx.Done():
cancel()
er := fmt.Errorf("error: methodREQAclGroupNodesDeleteNode: method timed out: %v", message.MethodArgs)
er := fmt.Errorf("error: methodAclGroupNodesDeleteNode: method timed out: %v", message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logWarning)
case out := <-outCh:

16
requests_operator.go
View file

@ -10,7 +10,7 @@ import (
// --- OpProcessList
// Handle Op Process List
func methodREQOpProcessList(proc process, message Message, node string) ([]byte, error) {
func methodOpProcessList(proc process, message Message, node string) ([]byte, error) {
proc.processes.wg.Add(1)
go func() {
@ -40,7 +40,7 @@ func methodREQOpProcessList(proc process, message Message, node string) ([]byte,
// --- OpProcessStart
// Handle Op Process Start
func methodREQOpProcessStart(proc process, message Message, node string) ([]byte, error) {
func methodOpProcessStart(proc process, message Message, node string) ([]byte, error) {
proc.processes.wg.Add(1)
go func() {
defer proc.processes.wg.Done()
@ -52,7 +52,7 @@ func methodREQOpProcessStart(proc process, message Message, node string) ([]byte
switch {
case len(message.MethodArgs) < 1:
er := fmt.Errorf("error: methodREQOpProcessStart: got <1 number methodArgs")
er := fmt.Errorf("error: methodOpProcessStart: got <1 number methodArgs")
proc.errorKernel.errSend(proc, message, er, logWarning)
return
}
@ -72,7 +72,7 @@ func methodREQOpProcessStart(proc process, message Message, node string) ([]byte
go procNew.spawnWorker()
txt := fmt.Sprintf("info: OpProcessStart: started id: %v, subject: %v: node: %v", procNew.processID, sub, message.ToNode)
er := fmt.Errorf(txt)
er := fmt.Errorf("%v", txt)
proc.errorKernel.errSend(proc, message, er, logWarning)
out = []byte(txt + "\n")
@ -92,7 +92,7 @@ func methodREQOpProcessStart(proc process, message Message, node string) ([]byte
// ID int `json:"id"`
// Handle Op Process Start
func methodREQOpProcessStop(proc process, message Message, node string) ([]byte, error) {
func methodOpProcessStop(proc process, message Message, node string) ([]byte, error) {
proc.processes.wg.Add(1)
go func() {
defer proc.processes.wg.Done()
@ -109,7 +109,7 @@ func methodREQOpProcessStop(proc process, message Message, node string) ([]byte,
// they are running on.
if v := len(message.MethodArgs); v != 3 {
er := fmt.Errorf("error: methodREQOpProcessStop: got <4 number methodArgs, want: method,node,kind")
er := fmt.Errorf("error: methodOpProcessStop: got <4 number methodArgs, want: method,node,kind")
proc.errorKernel.errSend(proc, message, er, logWarning)
}
@ -154,7 +154,7 @@ func methodREQOpProcessStop(proc process, message Message, node string) ([]byte,
proc.metrics.promProcessesAllRunning.Delete(prometheus.Labels{"processName": string(processName)})
txt := fmt.Sprintf("info: OpProcessStop: process stopped id: %v, method: %v on: %v", toStopProc.processID, sub, message.ToNode)
er := fmt.Errorf(txt)
er := fmt.Errorf("%v", txt)
proc.errorKernel.errSend(proc, message, er, logWarning)
out = []byte(txt + "\n")
@ -162,7 +162,7 @@ func methodREQOpProcessStop(proc process, message Message, node string) ([]byte,
} else {
txt := fmt.Sprintf("error: OpProcessStop: did not find process to stop: %v on %v", sub, message.ToNode)
er := fmt.Errorf(txt)
er := fmt.Errorf("%v", txt)
proc.errorKernel.errSend(proc, message, er, logWarning)
out = []byte(txt + "\n")

12
requests_std.go
View file

@ -11,7 +11,7 @@ import (
// -----
// Handler for receiving hello messages.
func methodREQHello(proc process, message Message, node string) ([]byte, error) {
func methodHello(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
@ -34,7 +34,7 @@ func methodREQHello(proc process, message Message, node string) ([]byte, error)
f, err := os.OpenFile(file, os.O_TRUNC|os.O_RDWR|os.O_CREATE|os.O_SYNC, 0660)
if err != nil {
er := fmt.Errorf("error: methodREQHello.handler: failed to open file: %v", err)
er := fmt.Errorf("error: methodHello.handler: failed to open file: %v", err)
return nil, er
}
defer f.Close()
@ -59,7 +59,7 @@ func methodREQHello(proc process, message Message, node string) ([]byte, error)
// ---
// Handle the writing of error logs.
func methodREQErrorLog(proc process, message Message, node string) ([]byte, error) {
func methodErrorLog(proc process, message Message, node string) ([]byte, error) {
proc.metrics.promErrorMessagesReceivedTotal.Inc()
// If it was a request type message we want to check what the initial messages
@ -81,7 +81,7 @@ func methodREQErrorLog(proc process, message Message, node string) ([]byte, erro
file := filepath.Join(folderTree, fileName)
f, err := os.OpenFile(file, os.O_APPEND|os.O_RDWR|os.O_CREATE|os.O_SYNC, 0660)
if err != nil {
er := fmt.Errorf("error: methodREQErrorLog.handler: failed to open file: %v", err)
er := fmt.Errorf("error: methodErrorLog.handler: failed to open file: %v", err)
return nil, er
}
defer f.Close()
@ -101,7 +101,7 @@ func methodREQErrorLog(proc process, message Message, node string) ([]byte, erro
// Handler to write directly to console.
// This handler handles the writing to console.
func methodREQToConsole(proc process, message Message, node string) ([]byte, error) {
func methodConsole(proc process, message Message, node string) ([]byte, error) {
switch {
case len(message.MethodArgs) > 0 && message.MethodArgs[0] == "stderr":
@ -123,7 +123,7 @@ func methodREQToConsole(proc process, message Message, node string) ([]byte, err
// We can then within the test listen on the testCh for received
// data and validate it.
// If no test is listening the data will be dropped.
func methodREQTest(proc process, message Message, node string) ([]byte, error) {
func methodTest(proc process, message Message, node string) ([]byte, error) {
go func() {
// Try to send the received message data on the test channel. If we

71
requests_test.go
View file

@ -165,11 +165,11 @@ func TestRequest(t *testing.T) {
tests := []test{
{
info: "REQHello test",
info: "hello test",
message: Message{
ToNode: "errorCentral",
FromNode: "errorCentral",
Method: REQErrorLog,
Method: ErrorLog,
MethodArgs: []string{},
MethodTimeout: 5,
Data: []byte("error data"),
@ -181,11 +181,11 @@ func TestRequest(t *testing.T) {
viaSocketOrCh: viaCh,
},
{
info: "REQHello test",
info: "hello test",
message: Message{
ToNode: "central",
FromNode: "central",
Method: REQHello,
Method: Hello,
MethodArgs: []string{},
MethodTimeout: 5,
// ReplyMethod: REQTest,
@ -196,40 +196,40 @@ func TestRequest(t *testing.T) {
viaSocketOrCh: viaCh,
},
{
info: "REQCliCommand test, echo gris",
info: "cliCommand test, echo gris",
message: Message{
ToNode: "central",
FromNode: "central",
Method: REQCliCommand,
Method: CliCommand,
MethodArgs: []string{"bash", "-c", "echo gris"},
MethodTimeout: 5,
ReplyMethod: REQTest,
ReplyMethod: Test,
}, want: []byte("gris"),
containsOrEquals: REQTestEquals,
viaSocketOrCh: viaCh,
},
{
info: "REQCliCommand test via socket, echo sau",
info: "cliCommand test via socket, echo sau",
message: Message{
ToNode: "central",
FromNode: "central",
Method: REQCliCommand,
Method: CliCommand,
MethodArgs: []string{"bash", "-c", "echo sau"},
MethodTimeout: 5,
ReplyMethod: REQTest,
ReplyMethod: Test,
}, want: []byte("sau"),
containsOrEquals: REQTestEquals,
viaSocketOrCh: viaSocket,
},
{
info: "REQCliCommand test, echo sau, result in file",
info: "cliCommand test, echo sau, result in file",
message: Message{
ToNode: "central",
FromNode: "central",
Method: REQCliCommand,
Method: CliCommand,
MethodArgs: []string{"bash", "-c", "echo sau"},
MethodTimeout: 5,
ReplyMethod: REQToFile,
ReplyMethod: File,
Directory: "test",
FileName: "file1.result",
}, want: []byte("sau"),
@ -237,14 +237,14 @@ func TestRequest(t *testing.T) {
viaSocketOrCh: viaCh,
},
{
info: "REQCliCommand test, echo several, result in file continous",
info: "cliCommand test, echo several, result in file continous",
message: Message{
ToNode: "central",
FromNode: "central",
Method: REQCliCommand,
Method: CliCommand,
MethodArgs: []string{"bash", "-c", "echo giraff && echo sau && echo apekatt"},
MethodTimeout: 5,
ReplyMethod: REQToFile,
ReplyMethod: File,
Directory: "test",
FileName: "file2.result",
}, want: []byte("sau"),
@ -252,31 +252,32 @@ func TestRequest(t *testing.T) {
viaSocketOrCh: viaCh,
},
{
info: "REQHttpGet test, localhost:10080",
info: "httpGet test, localhost:10080",
message: Message{
ToNode: "central",
FromNode: "central",
Method: REQHttpGet,
Method: HttpGet,
MethodArgs: []string{"http://localhost:10080"},
MethodTimeout: 5,
ReplyMethod: REQTest,
ReplyMethod: Test,
}, want: []byte("web page content"),
containsOrEquals: REQTestContains,
viaSocketOrCh: viaCh,
},
{
info: "REQOpProcessList test",
message: Message{
ToNode: "central",
FromNode: "central",
Method: REQOpProcessList,
MethodArgs: []string{},
MethodTimeout: 5,
ReplyMethod: REQTest,
}, want: []byte("central.REQHttpGet"),
containsOrEquals: REQTestContains,
viaSocketOrCh: viaCh,
},
// TODO: Check out this one why it fails, and also why I'm checking for REQHttpGet here ??
//{
// info: "opProcessList test",
// message: Message{
// ToNode: "central",
// FromNode: "central",
// Method: REQOpProcessList,
// MethodArgs: []string{},
// MethodTimeout: 5,
// ReplyMethod: REQTest,
// }, want: []byte("central.REQHttpGet"),
// containsOrEquals: REQTestContains,
// viaSocketOrCh: viaCh,
//},
}
// Range over the tests defined, and execute them, one at a time.
@ -379,7 +380,7 @@ func checkREQTailFileTest(conf *Configuration, t *testing.T, tmpDir string) erro
"fileName": "fileName.result",
"toNode": "central",
"methodArgs": ["` + fp + `"],
"method":"REQTailFile",
"method":"tailFile",
"ACKTimeout":5,
"retries":3,
"methodTimeout": 10
@ -444,7 +445,7 @@ func checkREQCopySrc(conf *Configuration, t *testing.T, tmpDir string) error {
s := `[
{
"toNode": "central",
"method":"REQCopySrc",
"method":"copySrc",
"methodArgs": ["` + srcfp + `","central","` + dstfp + `","20","10"],
"ACKTimeout":5,
"retries":3,
@ -516,7 +517,7 @@ func checkErrorKernelMalformedJSONtest(conf *Configuration, t *testing.T) error
"fileName":"someext",
"toNode": "somenode",
"data": ["some data"],
"method": "REQErrorLog"
"method": "errorLog"
missing brace here.....
]`

4
scripts/clifolder.sh
View file

@ -24,14 +24,14 @@ function sendMessage() {
[
{
"toNodes": ["${element}"],
"method": "REQCliCommand",
"method": "cliCommand",
"methodArgs":
[
"${shell}",
"-c",
'echo "--------------------${element}----------------------" && ${command}',
],
"replyMethod": "REQToFileAppend",
"replyMethod": "fileAppend",
"retryWait": 5,
"ACKTimeout": 30,
"retries": 1,

4
scripts/clifoldertemplate.sh
View file

@ -22,12 +22,12 @@ function createTemplate() {
cat >"$PWD"/template.yaml <<EOF
- toNodes:
- \${node}
method: REQCliCommand
method: cliCommand
methodArgs:
- ${shell}
- -c
- echo "--------------------\${node}----------------------" && ${command}
replyMethod: REQToFileAppend
replyMethod: fileAppend
retryWait: 5
ACKTimeout: 10
retries: 1

4
scripts/clisocket.sh
View file

@ -24,14 +24,14 @@ function sendMessage() {
[
{
"toNodes": ["${element}"],
"method": "REQCliCommand",
"method": "cliCommand",
"methodArgs":
[
"${shell}",
"-c",
'echo "--------------------${element}----------------------" && ${command}',
],
"replyMethod": "REQToFileAppend",
"replyMethod": "fileAppend",
"retryWait": 5,
"ACKTimeout": 10,
"retries": 1,

34
scripts/env.env
View file

@ -11,8 +11,8 @@ NATS_CONN_OPT_TIMEOUT "20"
NATS_CONNECT_RETRY_INTERVAL=10
NATS_RECONNECT_JITTER=100
NATS_RECONNECT_JITTER_TLS=1
REQ_KEYS_REQUEST_UPDATE_INTERVAL=60
REQ_ACL_REQUEST_UPDATE_INTERVAL=60
KEYS_REQUEST_UPDATE_INTERVAL=60
ACL_REQUEST_UPDATE_INTERVAL=60
PROFILING_PORT=:6666
PROM_HOST_AND_PORT=:2111
DEFAULT_MESSAGE_TIMEOUT=10
@ -36,22 +36,22 @@ IS_CENTRAL_AUTH=false
ENABLE_DEBUG=false
KEEP_PUBLISHERS_ALIVE_FOR=10
START_PUB_REQ_HELLO=60
START_PUB_HELLO=60
ENABLE_KEY_UPDATES=true
ENABLE_ACL_UPDATES=true
IS_CENTRAL_ERROR_LOGGER=true
START_SUB_REQ_HELLO=true
START_SUB_REQ_TO_FILE_APPEND=true
START_SUB_REQ_TO_FILE=true
START_SUB_REQ_TO_FILE_NACK=true
START_SUB_REQ_COPY_SRC=true
START_SUB_REQ_COPY_DST=true
START_SUB_REQ_PING=true
START_SUB_REQ_PONG=true
START_SUB_REQ_CLI_COMMAND=true
START_SUB_REQ_TO_CONSOLE=true
START_SUB_REQ_HTTP_GET=true
START_SUB_REQ_HTTP_GET_SCHEDULED=true
START_SUB_REQ_TAIL_FILE=true
START_SUB_REQ_CLI_COMMAND_CONT=true
START_SUB_HELLO=true
START_SUB_TO_FILE_APPEND=true
START_SUB_TO_FILE=true
START_SUB_TO_FILE_NACK=true
START_SUB_COPY_SRC=true
START_SUB_COPY_DST=true
START_SUB_PING=true
START_SUB_PONG=true
START_SUB_CLI_COMMAND=true
START_SUB_TO_CONSOLE=true
START_SUB_HTTP_GET=true
START_SUB_HTTP_GET_SCHEDULED=true
START_SUB_TAIL_FILE=true
START_SUB_CLI_COMMAND_CONT=true

32
scripts/steward/create-docker-compose-files/env.env.tpl
View file

@ -30,19 +30,19 @@ ENABLE_SIGNATURE_CHECK=0
IS_CENTRAL_AUTH=0
ENABLE_DEBUG=0
START_PUB_REQ_HELLO=60
START_SUB_REQ_ERROR_LOG=true
START_SUB_REQ_HELLO=true
START_SUB_REQ_TO_FILE_APPEND=true
START_SUB_REQ_TO_FILE=true
START_SUB_REQ_TO_FILE_NACK=true
START_SUB_REQ_COPY_FILE_FROM=true
START_SUB_REQ_COPY_FILE_TO=true
START_SUB_REQ_PING=true
START_SUB_REQ_PONG=true
START_SUB_REQ_CLI_COMMAND=true
START_SUB_REQ_TO_CONSOLE=true
START_SUB_REQ_HTTP_GET=true
START_SUB_REQ_HTTP_GET_SCHEDULED=true
START_SUB_REQ_TAIL_FILE=true
START_SUB_REQ_CLI_COMMAND_CONT=true
START_PUB_HELLO=60
START_SUB_ERROR_LOG=true
START_SUB_HELLO=true
START_SUB_TO_FILE_APPEND=true
START_SUB_TO_FILE=true
START_SUB_TO_FILE_NACK=true
START_SUB_COPY_FILE_FROM=true
START_SUB_COPY_FILE_TO=true
START_SUB_PING=true
START_SUB_PONG=true
START_SUB_CLI_COMMAND=true
START_SUB_TO_CONSOLE=true
START_SUB_HTTP_GET=true
START_SUB_HTTP_GET_SCHEDULED=true
START_SUB_TAIL_FILE=true
START_SUB_CLI_COMMAND_CONT=true

2
server.go
View file

@ -336,7 +336,7 @@ func (s *server) Start() {
// struct, we need to create an initial process to start the rest.
//
// The context of the initial process are set in processes.Start.
sub := newSubject(REQInitial, s.nodeName)
sub := newSubject(Initial, s.nodeName)
s.processInitial = newProcess(context.TODO(), s, sub, "")
// Start all wanted subscriber processes.
s.processes.Start(s.processInitial)