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

728 lines
23 KiB
Go
Raw Normal View History

2021-02-24 09:58:02 +00:00
package steward
import (
"context"
2021-02-24 09:58:02 +00:00
"fmt"
"log"
"sync"
"time"
"github.com/prometheus/client_golang/prometheus"
2021-02-24 09:58:02 +00:00
)
// processes holds all the information about running processes
type processes struct {
2021-08-11 10:23:37 +00:00
// The main context for subscriber processes.
ctx context.Context
// cancel func to send cancel signal to the subscriber processes context.
cancel context.CancelFunc
// The active spawned processes
2022-04-01 05:09:55 +00:00
// server
server *server
2021-10-08 10:07:10 +00:00
active procsMap
// mutex to lock the map
2021-10-08 20:39:46 +00:00
// mu sync.RWMutex
// The last processID created
lastProcessID int
2021-08-18 10:16:21 +00:00
// The instance global prometheus registry.
metrics *metrics
2022-02-02 07:54:36 +00:00
// Waitgroup to keep track of all the processes started.
2021-08-12 07:21:56 +00:00
wg sync.WaitGroup
2022-01-12 06:42:41 +00:00
// tui
tui *tui
// errorKernel
errorKernel *errorKernel
2022-02-02 07:54:36 +00:00
// configuration
configuration *Configuration
// Signatures
2022-04-21 11:21:36 +00:00
nodeAuth *nodeAuth
}
// newProcesses will prepare and return a *processes which
// is map containing all the currently running processes.
2022-04-01 05:09:55 +00:00
func newProcesses(ctx context.Context, server *server) *processes {
p := processes{
2022-04-01 05:09:55 +00:00
server: server,
2022-02-02 07:54:36 +00:00
active: *newProcsMap(),
2022-04-01 05:09:55 +00:00
tui: server.tui,
errorKernel: server.errorKernel,
configuration: server.configuration,
2022-04-21 11:21:36 +00:00
nodeAuth: server.nodeAuth,
2022-04-01 06:51:14 +00:00
metrics: server.metrics,
}
2021-08-16 11:01:12 +00:00
// Prepare the parent context for the subscribers.
2021-08-11 10:23:37 +00:00
ctx, cancel := context.WithCancel(ctx)
2021-10-08 11:42:19 +00:00
// // Start the processes map.
// go func() {
// p.active.run(ctx)
// }()
2021-10-08 10:07:10 +00:00
2021-08-11 10:23:37 +00:00
p.ctx = ctx
p.cancel = cancel
return &p
}
// ----------------------
2022-02-02 07:54:36 +00:00
// ----------------------
type procsMap struct {
2021-11-16 18:07:24 +00:00
procNames map[processName]process
2021-11-16 09:21:44 +00:00
mu sync.Mutex
}
func newProcsMap() *procsMap {
cM := procsMap{
2021-11-16 18:07:24 +00:00
procNames: make(map[processName]process),
}
return &cM
}
// ----------------------
// Start all the subscriber processes.
// Takes an initial process as it's input. All processes
// will be tied to this single process's context.
func (p *processes) Start(proc process) {
2021-08-11 10:23:37 +00:00
// Set the context for the initial process.
proc.ctx = p.ctx
// --- Subscriber services that can be started via flags
{
log.Printf("Starting REQOpProcessList subscriber: %#v\n", proc.node)
sub := newSubject(REQOpProcessList, string(proc.node))
2022-04-01 05:09:55 +00:00
proc := newProcess(proc.ctx, p.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
{
log.Printf("Starting REQOpProcessStart subscriber: %#v\n", proc.node)
sub := newSubject(REQOpProcessStart, string(proc.node))
2022-04-01 05:09:55 +00:00
proc := newProcess(proc.ctx, p.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
2021-09-20 09:53:17 +00:00
{
log.Printf("Starting REQOpProcessStop subscriber: %#v\n", proc.node)
sub := newSubject(REQOpProcessStop, string(proc.node))
2022-04-01 05:09:55 +00:00
proc := newProcess(proc.ctx, p.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
2021-09-20 09:53:17 +00:00
}
2022-05-22 04:36:02 +00:00
{
log.Printf("Starting REQTest subscriber: %#v\n", proc.node)
sub := newSubject(REQTest, string(proc.node))
proc := newProcess(proc.ctx, p.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
2021-09-08 16:56:23 +00:00
if proc.configuration.StartSubREQToFileAppend {
proc.startup.subREQToFileAppend(proc)
2021-02-24 09:58:02 +00:00
}
2021-09-08 16:56:23 +00:00
if proc.configuration.StartSubREQToFile {
proc.startup.subREQToFile(proc)
2021-04-06 17:42:03 +00:00
}
2022-03-04 14:02:43 +00:00
if proc.configuration.StartSubREQToFileNACK {
proc.startup.subREQToFileNACK(proc)
}
if proc.configuration.StartSubREQCopySrc {
proc.startup.subREQCopySrc(proc)
}
if proc.configuration.StartSubREQCopyDst {
proc.startup.subREQCopyDst(proc)
}
2021-09-08 16:56:23 +00:00
if proc.configuration.StartSubREQHello {
proc.startup.subREQHello(proc)
2021-02-24 09:58:02 +00:00
}
if proc.configuration.IsCentralErrorLogger {
proc.startup.subREQErrorLog(proc)
}
2021-09-08 16:56:23 +00:00
if proc.configuration.StartSubREQPing {
proc.startup.subREQPing(proc)
}
2021-09-08 16:56:23 +00:00
if proc.configuration.StartSubREQPong {
proc.startup.subREQPong(proc)
}
2021-09-08 16:56:23 +00:00
if proc.configuration.StartSubREQCliCommand {
proc.startup.subREQCliCommand(proc)
}
2021-09-08 16:56:23 +00:00
if proc.configuration.StartSubREQToConsole {
proc.startup.subREQToConsole(proc)
}
2022-01-13 07:34:22 +00:00
if proc.configuration.EnableTUI {
proc.startup.subREQTuiToConsole(proc)
}
if proc.configuration.StartPubREQHello != 0 {
proc.startup.pubREQHello(proc)
}
2021-04-06 17:42:03 +00:00
if proc.configuration.EnableKeyUpdates {
2022-05-24 05:21:48 +00:00
proc.startup.pubREQKeysRequestUpdate(proc)
proc.startup.subREQKeysDeliverUpdate(proc)
}
if proc.configuration.EnableAclUpdates {
proc.startup.pubREQAclRequestUpdate(proc)
2022-05-26 05:13:34 +00:00
proc.startup.subREQAclDeliverUpdate(proc)
2022-04-07 07:34:06 +00:00
}
if proc.configuration.IsCentralAuth {
2022-05-24 05:21:48 +00:00
proc.startup.subREQKeysRequestUpdate(proc)
proc.startup.subREQKeysAllow(proc)
2022-06-01 05:29:25 +00:00
proc.startup.subREQKeysDelete(proc)
proc.startup.subREQAclRequestUpdate(proc)
2022-05-18 12:43:35 +00:00
proc.startup.subREQAclAddCommand(proc)
proc.startup.subREQAclDeleteCommand(proc)
2022-05-19 19:35:14 +00:00
proc.startup.subREQAclDeleteSource(proc)
proc.startup.subREQAclGroupNodesAddNode(proc)
proc.startup.subREQAclGroupNodesDeleteNode(proc)
2022-05-20 03:18:26 +00:00
proc.startup.subREQAclGroupNodesDeleteGroup(proc)
proc.startup.subREQAclGroupCommandsAddCommand(proc)
proc.startup.subREQAclGroupCommandsDeleteCommand(proc)
proc.startup.subREQAclGroupCommandsDeleteGroup(proc)
2022-05-21 05:09:35 +00:00
proc.startup.subREQAclExport(proc)
2022-05-21 05:26:36 +00:00
proc.startup.subREQAclImport(proc)
2022-04-07 07:34:06 +00:00
}
// Moved this together with proc.configuration.StartPubREQKeysRequestUpdate since they belong together.
// if proc.configuration.StartSubREQKeysDeliverUpdate {
// proc.startup.subREQKeysDeliverUpdate(proc)
// }
2022-04-07 07:34:06 +00:00
2021-09-08 16:56:23 +00:00
if proc.configuration.StartSubREQHttpGet {
proc.startup.subREQHttpGet(proc)
2021-04-06 17:42:03 +00:00
}
2021-04-13 09:28:52 +00:00
2022-02-11 06:27:51 +00:00
if proc.configuration.StartSubREQHttpGetScheduled {
proc.startup.subREQHttpGetScheduled(proc)
}
2021-09-08 16:56:23 +00:00
if proc.configuration.StartSubREQTailFile {
proc.startup.subREQTailFile(proc)
2021-04-13 09:28:52 +00:00
}
if proc.configuration.StartSubREQCliCommandCont {
proc.startup.subREQCliCommandCont(proc)
}
2022-02-08 10:49:32 +00:00
proc.startup.subREQPublicKey(proc)
2021-02-24 09:58:02 +00:00
}
2021-08-11 10:23:37 +00:00
// Stop all subscriber processes.
func (p *processes) Stop() {
log.Printf("info: canceling all subscriber processes...\n")
2021-08-11 10:23:37 +00:00
p.cancel()
p.wg.Wait()
log.Printf("info: done canceling all subscriber processes.\n")
2021-08-11 10:23:37 +00:00
}
// ---------------------------------------------------------------------------------------
// Startup holds all the startup methods for subscribers.
2021-08-18 10:16:21 +00:00
type startup struct {
server *server
centralAuth *centralAuth
metrics *metrics
2021-08-18 10:16:21 +00:00
}
2022-04-01 05:09:55 +00:00
func newStartup(server *server) *startup {
2022-04-01 07:21:50 +00:00
s := startup{
server: server,
centralAuth: server.centralAuth,
metrics: server.metrics,
2022-04-01 07:21:50 +00:00
}
2021-08-18 10:16:21 +00:00
return &s
}
func (s startup) subREQHttpGet(p process) {
log.Printf("Starting Http Get subscriber: %#v\n", p.node)
sub := newSubject(REQHttpGet, string(p.node))
2022-04-01 05:09:55 +00:00
proc := newProcess(p.ctx, p.processes.server, sub, processKindSubscriber, nil)
2021-11-09 13:01:42 +00:00
2022-04-01 05:09:55 +00:00
go proc.spawnWorker()
}
2022-02-11 06:27:51 +00:00
func (s startup) subREQHttpGetScheduled(p process) {
log.Printf("Starting Http Get Scheduled subscriber: %#v\n", p.node)
sub := newSubject(REQHttpGetScheduled, string(p.node))
2022-04-01 05:09:55 +00:00
proc := newProcess(p.ctx, p.server, sub, processKindSubscriber, nil)
2022-02-11 06:27:51 +00:00
2022-04-01 05:09:55 +00:00
go proc.spawnWorker()
2022-02-11 06:27:51 +00:00
}
func (s startup) pubREQHello(p process) {
log.Printf("Starting Hello Publisher: %#v\n", p.node)
sub := newSubject(REQHello, p.configuration.CentralNodeName)
2022-04-01 05:09:55 +00:00
proc := newProcess(p.ctx, s.server, sub, processKindPublisher, nil)
// Define the procFunc to be used for the process.
2022-03-31 12:57:30 +00:00
proc.procFunc = func(ctx context.Context, procFuncCh chan Message) error {
ticker := time.NewTicker(time.Second * time.Duration(p.configuration.StartPubREQHello))
for {
// d := fmt.Sprintf("Hello from %v\n", p.node)
2022-04-19 16:30:00 +00:00
// Send the ed25519 public key used for signing as the payload of the message.
2022-04-21 11:21:36 +00:00
d := s.server.nodeAuth.SignPublicKey
2022-03-31 12:57:30 +00:00
m := Message{
FileName: "hello.log",
Directory: "hello-messages",
ToNode: Node(p.configuration.CentralNodeName),
FromNode: Node(p.node),
Data: []byte(d),
Method: REQHello,
ACKTimeout: 10,
Retries: 1,
}
2022-03-31 12:57:30 +00:00
sam, err := newSubjectAndMessage(m)
if err != nil {
// In theory the system should drop the message before it reaches here.
2022-04-01 06:43:14 +00:00
p.errorKernel.errSend(p, m, err)
2022-03-31 12:57:30 +00:00
log.Printf("error: ProcessesStart: %v\n", err)
}
proc.toRingbufferCh <- []subjectAndMessage{sam}
select {
case <-ticker.C:
case <-ctx.Done():
er := fmt.Errorf("info: stopped handleFunc for: publisher %v", proc.subject.name())
// sendErrorLogMessage(proc.toRingbufferCh, proc.node, er)
log.Printf("%v\n", er)
return nil
}
}
}
2022-04-01 05:09:55 +00:00
go proc.spawnWorker()
}
2022-05-24 05:21:48 +00:00
// pubREQKeysRequestUpdate defines the startup of a publisher that will send REQREQKeysRequestUpdate
2022-04-07 07:34:06 +00:00
// to central server and ask for publics keys, and to get them deliver back with a request
2022-05-24 05:21:48 +00:00
// of type pubREQKeysDeliverUpdate.
func (s startup) pubREQKeysRequestUpdate(p process) {
2022-04-07 07:34:06 +00:00
log.Printf("Starting PublicKeysGet Publisher: %#v\n", p.node)
2022-05-24 05:21:48 +00:00
sub := newSubject(REQKeysRequestUpdate, p.configuration.CentralNodeName)
2022-04-07 07:34:06 +00:00
proc := newProcess(p.ctx, s.server, sub, processKindPublisher, nil)
// Define the procFunc to be used for the process.
proc.procFunc = func(ctx context.Context, procFuncCh chan Message) error {
ticker := time.NewTicker(time.Second * time.Duration(p.configuration.REQKeysRequestUpdateInterval))
2022-04-07 07:34:06 +00:00
for {
2022-05-11 11:45:49 +00:00
// TODO: We could send with the hash of the currently stored keys,
// so we would know on the subscriber at central if it should send
// and update with new keys back.
2022-05-16 05:15:38 +00:00
proc.nodeAuth.publicKeys.mu.Lock()
fmt.Printf(" ----> publisher REQKeysRequestUpdate: sending our current hash: %v\n", []byte(proc.nodeAuth.publicKeys.keysAndHash.Hash[:]))
2022-05-16 05:15:38 +00:00
2022-04-07 07:34:06 +00:00
m := Message{
2022-05-16 05:15:38 +00:00
FileName: "publickeysget.log",
Directory: "publickeysget",
ToNode: Node(p.configuration.CentralNodeName),
FromNode: Node(p.node),
Data: []byte(proc.nodeAuth.publicKeys.keysAndHash.Hash[:]),
2022-05-24 05:21:48 +00:00
Method: REQKeysRequestUpdate,
ReplyMethod: REQKeysDeliverUpdate,
2022-04-07 07:34:06 +00:00
ACKTimeout: proc.configuration.DefaultMessageTimeout,
Retries: 1,
}
2022-05-16 05:15:38 +00:00
proc.nodeAuth.publicKeys.mu.Unlock()
2022-04-07 07:34:06 +00:00
sam, err := newSubjectAndMessage(m)
if err != nil {
// In theory the system should drop the message before it reaches here.
p.errorKernel.errSend(p, m, err)
log.Printf("error: ProcessesStart: %v\n", err)
}
proc.toRingbufferCh <- []subjectAndMessage{sam}
select {
case <-ticker.C:
case <-ctx.Done():
er := fmt.Errorf("info: stopped handleFunc for: publisher %v", proc.subject.name())
// sendErrorLogMessage(proc.toRingbufferCh, proc.node, er)
log.Printf("%v\n", er)
return nil
}
}
}
go proc.spawnWorker()
}
// pubREQAclRequestUpdate defines the startup of a publisher that will send REQREQKeysRequestUpdate
// to central server and ask for publics keys, and to get them deliver back with a request
// of type pubREQKeysDeliverUpdate.
func (s startup) pubREQAclRequestUpdate(p process) {
log.Printf("Starting REQAclRequestUpdate Publisher: %#v\n", p.node)
sub := newSubject(REQAclRequestUpdate, p.configuration.CentralNodeName)
proc := newProcess(p.ctx, s.server, sub, processKindPublisher, nil)
// Define the procFunc to be used for the process.
proc.procFunc = func(ctx context.Context, procFuncCh chan Message) error {
ticker := time.NewTicker(time.Second * time.Duration(p.configuration.REQAclRequestUpdateInterval))
for {
// TODO: We could send with the hash of the currently stored hash,
// so we would know on the subscriber at central if it should send
// and update with new keys back.
proc.nodeAuth.nodeAcl.mu.Lock()
fmt.Printf(" ----> publisher REQAclRequestUpdate: sending our current hash: %v\n", []byte(proc.nodeAuth.nodeAcl.aclAndHash.Hash[:]))
m := Message{
FileName: "aclRequestUpdate.log",
Directory: "aclRequestUpdate",
ToNode: Node(p.configuration.CentralNodeName),
FromNode: Node(p.node),
Data: []byte(proc.nodeAuth.nodeAcl.aclAndHash.Hash[:]),
Method: REQAclRequestUpdate,
ReplyMethod: REQAclDeliverUpdate,
ACKTimeout: proc.configuration.DefaultMessageTimeout,
Retries: 1,
}
proc.nodeAuth.nodeAcl.mu.Unlock()
sam, err := newSubjectAndMessage(m)
if err != nil {
// In theory the system should drop the message before it reaches here.
2022-04-07 07:34:06 +00:00
p.errorKernel.errSend(p, m, err)
log.Printf("error: ProcessesStart: %v\n", err)
}
proc.toRingbufferCh <- []subjectAndMessage{sam}
select {
case <-ticker.C:
case <-ctx.Done():
er := fmt.Errorf("info: stopped handleFunc for: publisher %v", proc.subject.name())
// sendErrorLogMessage(proc.toRingbufferCh, proc.node, er)
log.Printf("%v\n", er)
return nil
}
}
}
go proc.spawnWorker()
}
2022-05-24 05:21:48 +00:00
func (s startup) subREQKeysRequestUpdate(p process) {
log.Printf("Starting Public keys request update subscriber: %#v\n", p.node)
sub := newSubject(REQKeysRequestUpdate, string(p.node))
2022-04-07 07:34:06 +00:00
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
func (s startup) subREQKeysDeliverUpdate(p process) {
log.Printf("Starting Public keys to Node subscriber: %#v\n", p.node)
sub := newSubject(REQKeysDeliverUpdate, string(p.node))
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
2022-05-24 05:21:48 +00:00
func (s startup) subREQKeysAllow(p process) {
2022-04-20 16:33:52 +00:00
log.Printf("Starting Public keys allow subscriber: %#v\n", p.node)
2022-05-24 05:21:48 +00:00
sub := newSubject(REQKeysAllow, string(p.node))
2022-04-20 16:33:52 +00:00
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
2022-06-01 05:29:25 +00:00
func (s startup) subREQKeysDelete(p process) {
log.Printf("Starting Public keys delete subscriber: %#v\n", p.node)
sub := newSubject(REQKeysDelete, string(p.node))
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
func (s startup) subREQAclRequestUpdate(p process) {
log.Printf("Starting Acl Request update subscriber: %#v\n", p.node)
sub := newSubject(REQAclRequestUpdate, string(p.node))
2022-04-07 07:34:06 +00:00
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
2022-05-26 05:13:34 +00:00
func (s startup) subREQAclDeliverUpdate(p process) {
log.Printf("Starting Acl deliver update subscriber: %#v\n", p.node)
sub := newSubject(REQAclDeliverUpdate, string(p.node))
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
// HERE!
2022-05-18 12:43:35 +00:00
func (s startup) subREQAclAddCommand(p process) {
log.Printf("Starting Acl Add Command subscriber: %#v\n", p.node)
sub := newSubject(REQAclAddCommand, string(p.node))
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
func (s startup) subREQAclDeleteCommand(p process) {
log.Printf("Starting Acl Delete Command subscriber: %#v\n", p.node)
sub := newSubject(REQAclDeleteCommand, string(p.node))
2022-05-18 09:26:06 +00:00
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
2022-05-19 19:35:14 +00:00
func (s startup) subREQAclDeleteSource(p process) {
log.Printf("Starting Acl Delete Source subscriber: %#v\n", p.node)
sub := newSubject(REQAclDeleteSource, string(p.node))
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
func (s startup) subREQAclGroupNodesAddNode(p process) {
log.Printf("Starting Acl Add node to nodeGroup subscriber: %#v\n", p.node)
sub := newSubject(REQAclGroupNodesAddNode, string(p.node))
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
func (s startup) subREQAclGroupNodesDeleteNode(p process) {
log.Printf("Starting Acl Delete node from nodeGroup subscriber: %#v\n", p.node)
sub := newSubject(REQAclGroupNodesDeleteNode, string(p.node))
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
2022-05-20 03:18:26 +00:00
func (s startup) subREQAclGroupNodesDeleteGroup(p process) {
log.Printf("Starting Acl Delete nodeGroup subscriber: %#v\n", p.node)
sub := newSubject(REQAclGroupNodesDeleteGroup, string(p.node))
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
func (s startup) subREQAclGroupCommandsAddCommand(p process) {
log.Printf("Starting Acl add command to command group subscriber: %#v\n", p.node)
sub := newSubject(REQAclGroupCommandsAddCommand, string(p.node))
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
func (s startup) subREQAclGroupCommandsDeleteCommand(p process) {
log.Printf("Starting Acl delete command from command group subscriber: %#v\n", p.node)
sub := newSubject(REQAclGroupCommandsDeleteCommand, string(p.node))
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
func (s startup) subREQAclGroupCommandsDeleteGroup(p process) {
log.Printf("Starting Acl delete command group subscriber: %#v\n", p.node)
sub := newSubject(REQAclGroupCommandsDeleteGroup, string(p.node))
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
2022-05-21 05:09:35 +00:00
func (s startup) subREQAclExport(p process) {
log.Printf("Starting Acl export subscriber: %#v\n", p.node)
sub := newSubject(REQAclExport, string(p.node))
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
2022-05-21 05:26:36 +00:00
func (s startup) subREQAclImport(p process) {
log.Printf("Starting Acl import subscriber: %#v\n", p.node)
sub := newSubject(REQAclImport, string(p.node))
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
func (s startup) subREQToConsole(p process) {
log.Printf("Starting Text To Console subscriber: %#v\n", p.node)
sub := newSubject(REQToConsole, string(p.node))
2022-04-01 05:09:55 +00:00
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
2022-01-13 07:34:22 +00:00
}
func (s startup) subREQTuiToConsole(p process) {
log.Printf("Starting Tui To Console subscriber: %#v\n", p.node)
sub := newSubject(REQTuiToConsole, string(p.node))
2022-04-01 05:09:55 +00:00
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
func (s startup) subREQCliCommand(p process) {
log.Printf("Starting CLICommand Request subscriber: %#v\n", p.node)
sub := newSubject(REQCliCommand, string(p.node))
2022-04-01 05:09:55 +00:00
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
func (s startup) subREQPong(p process) {
log.Printf("Starting Pong subscriber: %#v\n", p.node)
sub := newSubject(REQPong, string(p.node))
2022-04-01 05:09:55 +00:00
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
func (s startup) subREQPing(p process) {
log.Printf("Starting Ping Request subscriber: %#v\n", p.node)
sub := newSubject(REQPing, string(p.node))
2022-04-01 05:09:55 +00:00
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
func (s startup) subREQErrorLog(p process) {
log.Printf("Starting REQErrorLog subscriber: %#v\n", p.node)
sub := newSubject(REQErrorLog, "errorCentral")
2022-04-01 05:09:55 +00:00
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
2022-01-01 09:13:35 +00:00
// 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,
// where the procFunc will receive messages from the handler when a message is
// received, the handler will deliver the message to the procFunc on the
// proc.procFuncCh, and we can then read that message from the procFuncCh in
// the procFunc running.
func (s startup) subREQHello(p process) {
log.Printf("Starting Hello subscriber: %#v\n", p.node)
sub := newSubject(REQHello, string(p.node))
2022-04-01 05:09:55 +00:00
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
// The reason for running the say hello subscriber as a procFunc is that
// a handler are not able to hold state, and we need to hold the state
// of the nodes we've received hello's from in the sayHelloNodes map,
// which is the information we pass along to generate metrics.
2022-03-31 12:57:30 +00:00
proc.procFunc = func(ctx context.Context, procFuncCh chan Message) error {
// sayHelloNodes := make(map[Node]struct{})
2022-03-31 12:57:30 +00:00
for {
// Receive a copy of the message sent from the method handler.
var m Message
select {
case m = <-procFuncCh:
case <-ctx.Done():
er := fmt.Errorf("info: stopped handleFunc for: subscriber %v", proc.subject.name())
// sendErrorLogMessage(proc.toRingbufferCh, proc.node, er)
log.Printf("%v\n", er)
return nil
2022-03-31 12:54:39 +00:00
}
2022-03-31 12:57:30 +00:00
s.centralAuth.addPublicKey(proc, m)
2022-03-31 12:57:30 +00:00
// update the prometheus metrics
2022-04-20 16:33:52 +00:00
2022-05-16 05:15:38 +00:00
s.server.centralAuth.pki.nodesAcked.mu.Lock()
mapLen := len(s.server.centralAuth.pki.nodesAcked.keysAndHash.Keys)
s.server.centralAuth.pki.nodesAcked.mu.Unlock()
2022-04-20 16:33:52 +00:00
s.metrics.promHelloNodesTotal.Set(float64(mapLen))
2022-04-01 07:21:50 +00:00
s.metrics.promHelloNodesContactLast.With(prometheus.Labels{"nodeName": string(m.FromNode)}).SetToCurrentTime()
2022-03-31 12:57:30 +00:00
}
}
2022-04-01 05:09:55 +00:00
go proc.spawnWorker()
}
2021-04-13 15:15:13 +00:00
func (s startup) subREQToFile(p process) {
log.Printf("Starting text to file subscriber: %#v\n", p.node)
2021-04-13 15:15:13 +00:00
sub := newSubject(REQToFile, string(p.node))
2022-04-01 05:09:55 +00:00
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
2021-11-09 13:01:42 +00:00
2022-04-01 05:09:55 +00:00
go proc.spawnWorker()
}
2022-03-04 14:02:43 +00:00
func (s startup) subREQToFileNACK(p process) {
log.Printf("Starting text to file subscriber: %#v\n", p.node)
sub := newSubject(REQToFileNACK, string(p.node))
2022-04-01 05:09:55 +00:00
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
2022-03-04 14:02:43 +00:00
2022-04-01 05:09:55 +00:00
go proc.spawnWorker()
2022-03-04 14:02:43 +00:00
}
func (s startup) subREQCopySrc(p process) {
log.Printf("Starting copy src subscriber: %#v\n", p.node)
sub := newSubject(REQCopySrc, string(p.node))
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
func (s startup) subREQCopyDst(p process) {
log.Printf("Starting copy dst subscriber: %#v\n", p.node)
sub := newSubject(REQCopyDst, string(p.node))
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
go proc.spawnWorker()
}
func (s startup) subREQToFileAppend(p process) {
log.Printf("Starting text logging subscriber: %#v\n", p.node)
sub := newSubject(REQToFileAppend, string(p.node))
2022-04-01 05:09:55 +00:00
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
2021-11-09 13:01:42 +00:00
2022-04-01 05:09:55 +00:00
go proc.spawnWorker()
}
2021-04-13 09:28:52 +00:00
func (s startup) subREQTailFile(p process) {
log.Printf("Starting tail log files subscriber: %#v\n", p.node)
2021-04-13 09:28:52 +00:00
sub := newSubject(REQTailFile, string(p.node))
2022-04-01 05:09:55 +00:00
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
2021-11-09 13:01:42 +00:00
2022-04-01 05:09:55 +00:00
go proc.spawnWorker()
2021-04-13 09:28:52 +00:00
}
func (s startup) subREQCliCommandCont(p process) {
log.Printf("Starting cli command with continous delivery: %#v\n", p.node)
sub := newSubject(REQCliCommandCont, string(p.node))
2022-04-01 05:09:55 +00:00
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
2021-11-10 10:21:38 +00:00
2022-04-01 05:09:55 +00:00
go proc.spawnWorker()
2021-11-10 10:21:38 +00:00
}
2022-02-08 10:49:32 +00:00
func (s startup) subREQPublicKey(p process) {
log.Printf("Starting get Public Key subscriber: %#v\n", p.node)
sub := newSubject(REQPublicKey, string(p.node))
2022-04-01 05:09:55 +00:00
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
2022-02-08 10:49:32 +00:00
2022-04-01 05:09:55 +00:00
go proc.spawnWorker()
2022-02-08 10:49:32 +00:00
}
2021-08-11 10:23:37 +00:00
// ---------------------------------------------------------------
// Print the content of the processes map.
func (p *processes) printProcessesMap() {
log.Printf("*** Output of processes map :\n")
2021-10-08 10:07:10 +00:00
2021-11-16 09:21:44 +00:00
{
p.active.mu.Lock()
2021-10-08 10:07:10 +00:00
2021-11-16 18:07:24 +00:00
for pName, proc := range p.active.procNames {
log.Printf("* proc - pub/sub: %v, procName in map: %v , id: %v, subject: %v\n", proc.processKind, pName, proc.processID, proc.subject.name())
2021-08-11 10:23:37 +00:00
}
2022-04-01 07:21:50 +00:00
p.metrics.promProcessesTotal.Set(float64(len(p.active.procNames)))
2021-11-16 09:21:44 +00:00
p.active.mu.Unlock()
}
2021-08-11 10:23:37 +00:00
}