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

added sendInfoLogMessage function

Browse source
This commit is contained in:
postmannen 2021-09-23 08:31:30 +02:00
parent de7a6c0dda
commit 8239c76bda
4 changed files with 25 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
metrics.go
View file

@ -55,6 +55,8 @@ type metrics struct {
promErrorMessagesReceivedTotal prometheus.Counter
// Metrics for sent error messages
promErrorMessagesSentTotal prometheus.Counter
// Metrics for sent info messages
promInfoMessagesSentTotal prometheus.Counter
// Metrics for the amount of messages currently in db.
promDBMessagesCurrent prometheus.Gauge
}
@ -157,6 +159,12 @@ func newMetrics(hostAndPort string) *metrics {
})
m.promRegistry.MustRegister(m.promErrorMessagesSentTotal)
m.promInfoMessagesSentTotal = prometheus.NewCounter(prometheus.CounterOpts{
Name: "steward_info_messages_sent_total",
Help: "Number of info messages sent total",
})
m.promRegistry.MustRegister(m.promInfoMessagesSentTotal)
m.promDBMessagesCurrent = prometheus.NewGauge(prometheus.GaugeOpts{
Name: "steward_db_messages_current",
Help: "The current value messages in database",

4
process.go
View file

@ -273,7 +273,7 @@ func (p process) messageDeliverNats(natsConn *nats.Conn, message Message) {
// We do not want to send errorLogs for REQErrorLog type since
// it will just cause an endless loop.
if message.Method != REQErrorLog {
sendErrorLogMessage(p.configuration, p.processes.metrics, p.toRingbufferCh, p.node, er)
sendInfoLogMessage(p.configuration, p.processes.metrics, p.toRingbufferCh, p.node, er)
}
log.Printf("%v\n", er)
@ -364,7 +364,7 @@ func (p process) subscriberHandler(natsConn *nats.Conn, thisNode string, msg *na
default:
er := fmt.Errorf("info: did not find that specific type of command or event: %#v", p.subject.CommandOrEvent)
sendErrorLogMessage(p.configuration, p.processes.metrics, p.toRingbufferCh, Node(thisNode), er)
sendInfoLogMessage(p.configuration, p.processes.metrics, p.toRingbufferCh, Node(thisNode), er)
}
}

8
requests.go
View file

@ -424,7 +424,7 @@ func (m methodREQOpCommand) handler(proc process, message Message, nodeName stri
go procNew.spawnWorker(proc.processes, proc.natsConn)
er := fmt.Errorf("info: startProc: started id: %v, subject: %v: node: %v", procNew.processID, sub, message.ToNode)
sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
sendInfoLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
case "stopProc":
// Set the interface type dst to &OpStart.
@ -485,7 +485,7 @@ func (m methodREQOpCommand) handler(proc process, message Message, nodeName stri
proc.processes.metrics.promProcessesAllRunning.Delete(prometheus.Labels{"processName": string(processName)})
er := fmt.Errorf("info: stopProc: stopped %v on %v", sub, message.ToNode)
sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
sendInfoLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
log.Printf("%v\n", er)
newReplyMessage(proc, message, []byte(er.Error()))
@ -1325,7 +1325,7 @@ func (m methodREQTailFile) handler(proc process, message Message, node string) (
// go routine.
// close(t.Lines)
er := fmt.Errorf("info: method timeout reached REQTailFile, canceling: %v", message.MethodArgs)
sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
sendInfoLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
return
case out := <-outCh:
@ -1444,7 +1444,7 @@ func (m methodREQCliCommandCont) handler(proc process, message Message, node str
case <-ctx.Done():
cancel()
er := fmt.Errorf("info: methodREQCliCommandCont: method timeout reached, canceling: methodArgs: %v", message.MethodArgs)
sendErrorLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
sendInfoLogMessage(proc.configuration, proc.processes.metrics, proc.toRingbufferCh, proc.node, er)
return
case out := <-outCh:
// Prepare and queue for sending a new message with the output

11
server.go
View file

@ -301,6 +301,17 @@ func sendErrorLogMessage(conf *Configuration, metrics *metrics, newMessagesCh ch
metrics.promErrorMessagesSentTotal.Inc()
}
// sendInfoMessage will put the error message directly on the channel that is
// read by the nats publishing functions.
func sendInfoLogMessage(conf *Configuration, metrics *metrics, newMessagesCh chan<- []subjectAndMessage, FromNode Node, theError error) {
// NB: Adding log statement here for more visuality during development.
log.Printf("%v\n", theError)
sam := createErrorMsgContent(conf, FromNode, theError)
newMessagesCh <- []subjectAndMessage{sam}
metrics.promInfoMessagesSentTotal.Inc()
}
// createErrorMsgContent will prepare a subject and message with the content
// of the error
func createErrorMsgContent(conf *Configuration, FromNode Node, theError error) subjectAndMessage {