Add export metrics

this is useful to figure out how much data is being exported and if the log
pipeline is behind in processing the incoming events.

Signed-off-by: Chance Zibolski <[email protected]>

cmd/tetragon/main.go

@@ -713,7 +713,9 @@ func startExporter(ctx context.Context, server *server.Server) error {
 		}()
 	}
 
-	encoder := encoder.NewProtojsonEncoder(writer)
+	// Track how many bytes are written to the event export location
+	encoderWriter := exporter.NewExportedBytesTotalWriter(writer)
+	encoder := encoder.NewProtojsonEncoder(encoderWriter)
 	var rateLimiter *ratelimit.RateLimiter
 	if option.Config.ExportRateLimit >= 0 {
 		rateLimiter = ratelimit.NewRateLimiter(ctx, 1*time.Minute, option.Config.ExportRateLimit, encoder)

pkg/exporter/exporter.go

@@ -58,9 +58,12 @@ func (e *Exporter) Send(event *tetragon.GetEventsResponse) error {
 		e.rateLimiter.Drop()
 		return nil
 	}
+
 	if err := e.encoder.Encode(event); err != nil {
 		logger.GetLogger().WithError(err).Warning("Failed to JSON encode")
 	}
+	eventsExportedTotal.Inc()
+	eventsExportTimestamp.Set(float64(event.GetTime().GetSeconds()))
 	return nil
 }
 

pkg/exporter/metrics.go

@@ -0,0 +1,56 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright Authors of Cilium
+
+package exporter
+
+import (
+	"io"
+
+	"github.com/cilium/tetragon/pkg/metrics/consts"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var (
+	eventsExportedTotal = prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace: consts.MetricsNamespace,
+		Name:      "events_exported_total",
+		Help:      "Total number of events exported",
+	})
+
+	eventsExportedBytesTotal = prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace: consts.MetricsNamespace,
+		Name:      "events_exported_bytes_total",
+		Help:      "Number of bytes exported for events",
+	})
+
+	eventsExportTimestamp = prometheus.NewGauge(prometheus.GaugeOpts{
+		Namespace: consts.MetricsNamespace,
+		Name:      "events_last_exported_timestamp",
+		Help:      "Timestamp of the most recent event to be exported",
+	})
+)
+
+func InitMetrics(registry *prometheus.Registry) {
+	registry.MustRegister(eventsExportedTotal)
+	registry.MustRegister(eventsExportedBytesTotal)
+	registry.MustRegister(eventsExportTimestamp)
+}
+
+func newExportedBytesCounterWriter(w io.Writer, c prometheus.Counter) io.Writer {
+	return byteCounterWriter{Writer: w, bytesWritten: c}
+}
+
+type byteCounterWriter struct {
+	io.Writer
+	bytesWritten prometheus.Counter
+}
+
+func (w byteCounterWriter) Write(p []byte) (int, error) {
+	n, err := w.Writer.Write(p)
+	w.bytesWritten.Add(float64(n))
+	return n, err
+}
+
+func NewExportedBytesTotalWriter(w io.Writer) io.Writer {
+	return newExportedBytesCounterWriter(w, eventsExportedBytesTotal)
+}

pkg/metrics/metricsconfig/initmetrics.go

@@ -5,6 +5,7 @@ package metricsconfig
 
 import (
 	"github.com/cilium/tetragon/pkg/eventcache"
+	"github.com/cilium/tetragon/pkg/exporter"
 	"github.com/cilium/tetragon/pkg/grpc/tracing"
 	"github.com/cilium/tetragon/pkg/metrics/errormetrics"
 	"github.com/cilium/tetragon/pkg/metrics/eventcachemetrics"
@@ -43,6 +44,7 @@ func initHealthMetrics(registry *prometheus.Registry) {
 	observer.InitMetrics(registry)
 	tracing.InitMetrics(registry)
 	ratelimitmetrics.InitMetrics(registry)
+	exporter.InitMetrics(registry)
 
 	// register common third-party collectors
 	registry.MustRegister(grpcmetrics.NewServerMetrics())