Initialize Prometheus Counters to 0 when service(s) start (#3906)

Signed-off-by: Aaron Mell <aaronamell@gmail.com>

Author: aaronmell

USERS.md

@@ -35,6 +35,7 @@ Organizations below are **officially** using Argo Events. Please send a PR with
 1. [FikaWorks](https://fika.works/)
 1. [Kraken Technologies Ltd.](https://kraken.tech/)
 1. [Loam](https://www.getloam.com/)
+1. [Lumin Digital](https://lumindigital.com)
 1. [MariaDB](https://mariadb.com/)
 1. [Mobimeo GmbH](https://mobimeo.com/en/home/)
 1. [OneCause](https://www.onecause.com/)
@@ -63,4 +64,4 @@ Organizations below are **officially** using Argo Events. Please send a PR with
 1. [Zillow Group](https://www.zillow.com)
 1. [AlpacaTech](https://www.alpaca-tech.ai)
 1. [SBAB Bank AB](https://www.sbab.se)
-1. [Terradue Srl](https://www.terradue.com)
+1. [Terradue Srl](https://www.terradue.com)

pkg/eventsources/eventing.go

@@ -478,6 +478,7 @@ func (e *EventSourceAdaptor) run(ctx context.Context, servers map[aev1.EventSour
 		logger.Errorw("failed to connect to eventbus", zap.Error(err))
 		return err
 	}
+
 	defer e.eventBusConn.Close()
 
 	ctx, cancel := context.WithCancel(ctx)
@@ -527,6 +528,9 @@ func (e *EventSourceAdaptor) run(ctx context.Context, servers map[aev1.EventSour
 				// Continue starting other event services instead of failing all of them
 				continue
 			}
+
+			e.metrics.InitEventMetrics(server.GetEventSourceName(), server.GetEventName())
+
 			wg.Add(1)
 			go func(s EventingServer) {
 				defer wg.Done()

pkg/metrics/metrics.go

@@ -152,6 +152,19 @@ func (m *Metrics) Describe(ch chan<- *prometheus.Desc) {
 	m.actionDuration.Describe(ch)
 }
 
+func (m *Metrics) InitSensorMetrics(sensorName string, triggerName string) {
+	m.actionTriggered.WithLabelValues(sensorName, triggerName).Add(0)
+	m.actionFailed.WithLabelValues(sensorName, triggerName).Add(0)
+	m.actionRetriesFailed.WithLabelValues(sensorName, triggerName).Add(0)
+}
+
+func (m *Metrics) InitEventMetrics(eventSourceName string, eventName string) {
+	m.runningEventServices.WithLabelValues(eventSourceName).Set(0)
+	m.eventsSent.WithLabelValues(eventSourceName, eventName).Add(0)
+	m.eventsSentFailed.WithLabelValues(eventSourceName, eventName).Add(0)
+	m.eventsProcessingFailed.WithLabelValues(eventSourceName, eventName).Add(0)
+}
+
 func (m *Metrics) IncRunningServices(eventSourceName string) {
 	m.runningEventServices.WithLabelValues(eventSourceName).Inc()
 }

pkg/sensors/listener.go

@@ -124,6 +124,8 @@ func (sensorCtx *SensorContext) listenEvents(ctx context.Context) error {
 
 	wg := &sync.WaitGroup{}
 	for _, t := range sensor.Spec.Triggers {
+		sensorCtx.metrics.InitSensorMetrics(sensorCtx.sensor.Name, t.Template.Name)
+
 		initRateLimiter(t)
 		wg.Add(1)
 		go func(trigger v1alpha1.Trigger) {