[Backport 5.2] telemetry-export: add teestore, queue store metrics (#57256)

Co-authored-by: Joe Chen <[email protected]>
Co-authored-by: Robert Lin <[email protected]>

Author: 3 people

internal/database/telemetry_export_store.go

@@ -2,12 +2,15 @@ package database
 
 import (
 	"context"
+	"strconv"
 	"time"
 
 	"go.opentelemetry.io/otel/attribute"
 	"google.golang.org/protobuf/proto"
 
 	"github.com/lib/pq"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/sourcegraph/log"
 
 	"github.com/sourcegraph/sourcegraph/internal/database/basestore"
@@ -23,6 +26,13 @@ import (
 // queued for export via (TelemetryEventsExportQueueStore).QueueForExport
 const FeatureFlagTelemetryExport = "telemetry-export"
 
+var counterQueuedEvents = promauto.NewCounterVec(prometheus.CounterOpts{
+	Namespace: "src",
+	Subsystem: "telemetry_export_store",
+	Name:      "queued_events",
+	Help:      "Events added to the telemetry export queue",
+}, []string{"failed"})
+
 type TelemetryEventsExportQueueStore interface {
 	basestore.ShareableStore
 
@@ -80,7 +90,8 @@ func (s *telemetryEventsExportQueueStore) QueueForExport(ctx context.Context, ev
 	if len(events) == 0 {
 		return nil
 	}
-	return batch.InsertValues(ctx,
+
+	err := batch.InsertValues(ctx,
 		s.Handle(),
 		"telemetry_events_export_queue",
 		batch.MaxNumPostgresParameters,
@@ -90,6 +101,10 @@ func (s *telemetryEventsExportQueueStore) QueueForExport(ctx context.Context, ev
 			"payload_pb",
 		},
 		insertChannel(logger, events))
+	counterQueuedEvents.
+		WithLabelValues(strconv.FormatBool(err != nil)).
+		Add(float64(len(events)))
+	return err
 }
 
 func insertChannel(logger log.Logger, events []*telemetrygatewayv1.Event) <-chan []any {

internal/telemetry/teestore/BUILD.bazel

@@ -16,6 +16,8 @@ go_library(
         "//internal/telemetrygateway/v1:telemetrygateway",
         "//lib/errors",
         "//lib/pointers",
+        "@com_github_prometheus_client_golang//prometheus",
+        "@com_github_prometheus_client_golang//prometheus/promauto",
         "@com_github_sourcegraph_conc//pool",
     ],
 )

internal/telemetry/teestore/teestore.go

@@ -7,6 +7,8 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/sourcegraph/conc/pool"
 
 	"github.com/sourcegraph/sourcegraph/internal/database"
@@ -17,6 +19,13 @@ import (
 	"github.com/sourcegraph/sourcegraph/lib/pointers"
 )
 
+var counterV1Events = promauto.NewCounterVec(prometheus.CounterOpts{
+	Namespace: "src",
+	Subsystem: "telemetry_teestore",
+	Name:      "v1_events",
+	Help:      "Events tee'd to the V1 event_logs table",
+}, []string{"failed"})
+
 // Store tees events into both the event_logs table and the new telemetry export
 // queue, translating the message into the existing event_logs format on a
 // best-effort basis.
@@ -40,7 +49,11 @@ func (s *Store) StoreEvents(ctx context.Context, events []*telemetrygatewayv1.Ev
 	})
 	if !shouldDisableV1(ctx) {
 		wg.Go(func() error {
-			if err := s.eventLogs.BulkInsert(ctx, toEventLogs(time.Now, events)); err != nil {
+			err := s.eventLogs.BulkInsert(ctx, toEventLogs(time.Now, events))
+			counterV1Events.
+				WithLabelValues(strconv.FormatBool(err != nil)).
+				Add(float64(len(events)))
+			if err != nil {
 				return errors.Wrap(err, "bulk inserting events logs")
 			}
 			return nil