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

241 lines
8.5 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"
"time"
"github.com/prometheus/client_golang/prometheus"
2021-04-12 08:51:26 +00:00
"github.com/prometheus/client_golang/prometheus/promauto"
2021-02-24 09:58:02 +00:00
)
func (p process) ProcessesStart() {
// --- Subscriber services that can be started via flags
// Allways start an REQOpCommand subscriber
{
fmt.Printf("Starting REQOpCommand subscriber: %#v\n", p.node)
sub := newSubject(REQOpCommand, string(p.node))
proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, []node{node(p.configuration.CentralNodeName)}, nil)
go proc.spawnWorker(p.processes, p.natsConn)
}
2021-02-24 09:58:02 +00:00
// Start a subscriber for textLogging messages
if p.configuration.StartSubREQTextToLogFile.OK {
p.startup.subREQTextToLogFile(p)
2021-02-24 09:58:02 +00:00
}
2021-04-06 17:42:03 +00:00
// Start a subscriber for text to file messages
if p.configuration.StartSubREQTextToFile.OK {
p.startup.subREQTextToFile(p)
2021-04-06 17:42:03 +00:00
}
2021-04-05 06:37:24 +00:00
// Start a subscriber for Hello messages
if p.configuration.StartSubREQHello.OK {
p.startup.subREQHello(p)
2021-02-24 09:58:02 +00:00
}
if p.configuration.StartSubREQErrorLog.OK {
2021-04-06 05:56:49 +00:00
// Start a subscriber for REQErrorLog messages
p.startup.subREQErrorLog(p)
}
2021-04-06 04:08:26 +00:00
// Start a subscriber for Ping Request messages
if p.configuration.StartSubREQPing.OK {
p.startup.subREQPing(p)
}
2021-04-06 04:08:26 +00:00
// Start a subscriber for REQPong messages
if p.configuration.StartSubREQPong.OK {
p.startup.subREQPong(p)
}
2021-04-04 09:19:17 +00:00
// Start a subscriber for REQCliCommand messages
if p.configuration.StartSubREQCliCommand.OK {
p.startup.subREQCliCommand(p)
}
// Start a subscriber for Not In Order Cli Command Request messages
if p.configuration.StartSubREQnCliCommand.OK {
p.startup.subREQnCliCommand(p)
}
// Start a subscriber for CLICommandReply messages
if p.configuration.StartSubREQTextToConsole.OK {
p.startup.subREQTextToConsole(p)
}
if p.configuration.StartPubREQHello != 0 {
p.startup.pubREQHello(p)
}
2021-04-06 17:42:03 +00:00
// Start a subscriber for Http Get Requests
if p.configuration.StartSubREQHttpGet.OK {
p.startup.subREQHttpGet(p)
2021-04-06 17:42:03 +00:00
}
2021-02-24 09:58:02 +00:00
}
// ---------------------------------------------------------------------------------------
type startup struct{}
func (s startup) subREQHttpGet(p process) {
fmt.Printf("Starting Http Get subscriber: %#v\n", p.node)
sub := newSubject(REQHttpGet, string(p.node))
proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQHttpGet.Values, nil)
// fmt.Printf("*** %#v\n", proc)
go proc.spawnWorker(p.processes, p.natsConn)
}
func (s startup) pubREQHello(p process) {
fmt.Printf("Starting Hello Publisher: %#v\n", p.node)
sub := newSubject(REQHello, p.configuration.CentralNodeName)
proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindPublisher, []node{}, nil)
// Define the procFunc to be used for the process.
proc.procFunc = procFunc(
func(ctx context.Context) error {
ticker := time.NewTicker(time.Second * time.Duration(p.configuration.StartPubREQHello))
for {
// fmt.Printf("--- DEBUG : procFunc call:kind=%v, Subject=%v, toNode=%v\n", proc.processKind, proc.subject, proc.subject.ToNode)
d := fmt.Sprintf("Hello from %v\n", p.node)
m := Message{
ToNode: "central",
FromNode: node(p.node),
Data: []string{d},
Method: REQHello,
}
sam, err := newSAM(m)
if err != nil {
// In theory the system should drop the message before it reaches here.
log.Printf("error: ProcessesStart: %v\n", err)
}
proc.toRingbufferCh <- []subjectAndMessage{sam}
select {
case <-ticker.C:
case <-ctx.Done():
fmt.Printf(" ** DEBUG: got <- ctx.Done\n")
er := fmt.Errorf("info: stopped handleFunc for: %v", proc.subject.name())
sendErrorLogMessage(proc.toRingbufferCh, proc.node, er)
return nil
}
}
})
go proc.spawnWorker(p.processes, p.natsConn)
}
func (s startup) subREQTextToConsole(p process) {
fmt.Printf("Starting Text To Console subscriber: %#v\n", p.node)
sub := newSubject(REQTextToConsole, string(p.node))
proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQTextToConsole.Values, nil)
go proc.spawnWorker(p.processes, p.natsConn)
}
func (s startup) subREQnCliCommand(p process) {
fmt.Printf("Starting CLICommand Not Sequential Request subscriber: %#v\n", p.node)
sub := newSubject(REQnCliCommand, string(p.node))
proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQnCliCommand.Values, nil)
go proc.spawnWorker(p.processes, p.natsConn)
}
func (s startup) subREQCliCommand(p process) {
fmt.Printf("Starting CLICommand Request subscriber: %#v\n", p.node)
sub := newSubject(REQCliCommand, string(p.node))
proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQCliCommand.Values, nil)
go proc.spawnWorker(p.processes, p.natsConn)
}
func (s startup) subREQPong(p process) {
fmt.Printf("Starting Pong subscriber: %#v\n", p.node)
sub := newSubject(REQPong, string(p.node))
proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQPong.Values, nil)
go proc.spawnWorker(p.processes, p.natsConn)
}
func (s startup) subREQPing(p process) {
fmt.Printf("Starting Ping Request subscriber: %#v\n", p.node)
sub := newSubject(REQPing, string(p.node))
proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQPing.Values, nil)
go proc.spawnWorker(p.processes, p.natsConn)
}
func (s startup) subREQErrorLog(p process) {
fmt.Printf("Starting REQErrorLog subscriber: %#v\n", p.node)
sub := newSubject(REQErrorLog, "errorCentral")
proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQErrorLog.Values, nil)
go proc.spawnWorker(p.processes, p.natsConn)
}
func (s startup) subREQHello(p process) {
fmt.Printf("Starting Hello subscriber: %#v\n", p.node)
sub := newSubject(REQHello, string(p.node))
proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQHello.Values, nil)
proc.procFuncCh = make(chan Message)
// 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.
proc.procFunc = func(ctx context.Context) error {
sayHelloNodes := make(map[node]struct{})
2021-04-12 08:51:26 +00:00
promHelloNodes := promauto.NewGauge(prometheus.GaugeOpts{
2021-04-12 14:49:16 +00:00
Name: "hello_nodes_total",
2021-04-12 08:51:26 +00:00
Help: "The current number of total nodes who have said hello",
})
promHelloNodesNameVec := promauto.NewGaugeVec(prometheus.GaugeOpts{
2021-04-12 14:49:16 +00:00
Name: "hello_node_last_hello",
Help: "Name of the nodes who have said hello",
}, []string{"nodeName"},
)
for {
// Receive a copy of the message sent from the method handler.
var m Message
select {
case m = <-proc.procFuncCh:
case <-ctx.Done():
er := fmt.Errorf("info: stopped handleFunc for: %v", proc.subject.name())
sendErrorLogMessage(proc.toRingbufferCh, proc.node, er)
return nil
}
// Add an entry for the node in the map
sayHelloNodes[m.FromNode] = struct{}{}
// update the prometheus metrics
2021-04-12 08:51:26 +00:00
promHelloNodes.Set(float64(len(sayHelloNodes)))
promHelloNodesNameVec.With(prometheus.Labels{"nodeName": string(m.FromNode)}).SetToCurrentTime()
}
}
go proc.spawnWorker(p.processes, p.natsConn)
}
func (s startup) subREQTextToFile(p process) {
fmt.Printf("Starting text to file subscriber: %#v\n", p.node)
sub := newSubject(REQTextToFile, string(p.node))
proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQTextToFile.Values, nil)
// fmt.Printf("*** %#v\n", proc)
go proc.spawnWorker(p.processes, p.natsConn)
}
func (s startup) subREQTextToLogFile(p process) {
fmt.Printf("Starting text logging subscriber: %#v\n", p.node)
sub := newSubject(REQTextToLogFile, string(p.node))
proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQTextToLogFile.Values, nil)
// fmt.Printf("*** %#v\n", proc)
go proc.spawnWorker(p.processes, p.natsConn)
}