add a configurable number of workers to prometheus outputs

docs/user_guide/outputs/prometheus_output.md

@@ -82,6 +82,8 @@ outputs:
     target-template:
     # list of processors to apply on the message before writing
     event-processors: 
+    # an integer, sets the number of worker handling messages to be converted into Prometheus metrics
+    num-workers: 1
     # Enables Consul service registration
     service-registration:
       # Consul server address, default to localhost:8500
@@ -458,6 +460,13 @@ When caching is enabled, the received gNMI updates are not processed and convert
 
 Once a scrape request is received from `Prometheus`, all the cached gNMI updates are retrieved from the cache, converted to [events](../event_processors/intro.md#the-event-format), the configured processors, if any, are then applied to the whole list of events. Finally, The resulting event are converted into metrics and written back to `Prometheus` within the scrape response.
 
+## Prometheus Output Metrics
+
+When a Prometheus server (gNMI API) is enabled, `gnmic` prometheus output exposes 2 prometheus Gauges:
+
+* `number_of_prometheus_metrics_total`: Number of metrics stored by the prometheus output.
+* `number_of_prometheus_cached_metrics_total`: Number of metrics cached by the prometheus output.
+
 ## Examples
 
 ### **A simple Prometheus output**

docs/user_guide/outputs/prometheus_write_output.md

@@ -90,6 +90,8 @@ outputs:
     target-template:
     # list of processors to apply on the message before writing
     event-processors: 
+    # an integer, sets the number of worker handling messages to be converted into Prometheus metrics
+    num-workers: 1
 ```
 
 `gnmic` creates the prometheus metric name and its labels from the subscription name, the gnmic path and the value name.
@@ -137,3 +139,15 @@ For the previous example the labels would be:
 ```bash
 {interface_name="1/1/1",subinterface_index=0,source="$routerIP:Port",subscription_name="port-stats"}
 ```
+
+## Prometheus Write Metrics
+
+When a Prometheus server (gNMI API) is enabled, `gnmic` prometheus write output exposes 4 prometheus counters and 2 prometheus Gauges:
+
+* `number_of_prometheus_write_msgs_sent_success_total`: Number of msgs successfully sent by gnmic prometheus_write output.
+* `number_of_prometheus_write_msgs_sent_fail_total`: Number of failed msgs sent by gnmic prometheus_write output.
+* `msg_send_duration_ns`: gnmic prometheus_write output send duration in ns.
+
+* `number_of_prometheus_write_metadata_msgs_sent_success_total`: Number of metadata msgs successfully sent by gnmic prometheus_write output.
+* `number_of_prometheus_write_metadata_msgs_sent_fail_total`: Number of failed metadata msgs sent by gnmic prometheus_write output.
+* `metadata_msg_send_duration_ns`: gnmic prometheus_write output metadata send duration in ns.

outputs/prometheus_output/prometheus_output/prometheus_cache.go

@@ -24,8 +24,11 @@ func (p *prometheusOutput) collectFromCache(ch chan<- prometheus.Metric) {
 		p.logger.Printf("failed to read from cache: %v", err)
 		return
 	}
+	numNotifications := len(notifications)
+	prometheusNumberOfCachedMetrics.Set(float64(numNotifications))
+
 	p.targetsMeta.DeleteExpired()
-	events := make([]*formatters.EventMsg, 0, len(notifications))
+	events := make([]*formatters.EventMsg, 0, numNotifications)
 	for subName, notifs := range notifications {
 		// build events without processors
 		for _, notif := range notifs {
@@ -45,17 +48,17 @@ func (p *prometheusOutput) collectFromCache(ch chan<- prometheus.Metric) {
 		}
 	}
 
-	if p.Cfg.CacheConfig.Debug {
+	if p.cfg.CacheConfig.Debug {
 		p.logger.Printf("got %d events from cache pre processors", len(events))
 	}
 	for _, proc := range p.evps {
 		events = proc.Apply(events...)
 	}
-	if p.Cfg.CacheConfig.Debug {
+	if p.cfg.CacheConfig.Debug {
 		p.logger.Printf("got %d events from cache post processors", len(events))
 	}
 
-	ctx, cancel := context.WithTimeout(context.Background(), p.Cfg.Timeout)
+	ctx, cancel := context.WithTimeout(context.Background(), p.cfg.Timeout)
 	defer cancel()
 	now := time.Now()
 	for _, ev := range events {

outputs/prometheus_output/prometheus_output/prometheus_metrics.go

@@ -0,0 +1,48 @@
+// © 2023 Nokia.
+//
+// This code is a Contribution to the gNMIc project (“Work”) made under the Google Software Grant and Corporate Contributor License Agreement (“CLA”) and governed by the Apache License 2.0.
+// No other rights or licenses in or to any of Nokia’s intellectual property are granted for any other purpose.
+// This code is provided on an “as is” basis without any warranties of any kind.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package prometheus_output
+
+import "github.com/prometheus/client_golang/prometheus"
+
+const (
+	namespace = "gnmic"
+	subsystem = "prometheus_output"
+)
+
+var prometheusNumberOfMetrics = prometheus.NewGauge(
+	prometheus.GaugeOpts{
+		Namespace: namespace,
+		Subsystem: subsystem,
+		Name:      "number_of_prometheus_metrics_total",
+		Help:      "Number of metrics stored by the prometheus output",
+	})
+
+var prometheusNumberOfCachedMetrics = prometheus.NewGauge(
+	prometheus.GaugeOpts{
+		Namespace: namespace,
+		Subsystem: subsystem,
+		Name:      "number_of_prometheus_cached_metrics_total",
+		Help:      "Number of metrics cached by the prometheus output",
+	})
+
+func (p *prometheusOutput) initMetrics() {
+	if p.cfg.CacheConfig == nil {
+		prometheusNumberOfMetrics.Set(0)
+		return
+	}
+	prometheusNumberOfCachedMetrics.Set(0)
+}
+
+func (p *prometheusOutput) registerMetrics(reg *prometheus.Registry) error {
+	p.initMetrics()
+	if p.cfg.CacheConfig == nil {
+		return reg.Register(prometheusNumberOfMetrics)
+	}
+	return reg.Register(prometheusNumberOfCachedMetrics)
+}