Add POSIX Antispam Metrics (#913)

* Antispam POSIX metrics

* dynamically generate batchEntriesBuckets

* comments about microseconds

* number of nrw before success

* fix names

* histogram

* fixes

* count sucesses rather than no_rewrite

* indent

* 1 based

Author: phbnf

storage/posix/antispam/badger.go

@@ -33,6 +33,7 @@ import (
 	"github.com/transparency-dev/tessera"
 	"github.com/transparency-dev/tessera/client"
 	"github.com/transparency-dev/tessera/internal/otel"
+	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/trace"
 )
 
@@ -96,6 +97,8 @@ func NewAntispam(ctx context.Context, badgerPath string, opts AntispamOpts) (*An
 	}
 
 	go func() {
+		// runsInTick tracks the number of GC runs per tick.
+		var runsInTick int64
 		ticker := time.NewTicker(1 * time.Minute)
 		defer ticker.Stop()
 		for {
@@ -105,8 +108,24 @@ func NewAntispam(ctx context.Context, badgerPath string, opts AntispamOpts) (*An
 			case <-ticker.C:
 			}
 
+			runsInTick = 0
 		again:
+			start := time.Now()
+			runsInTick++
 			err := db.RunValueLogGC(0.7)
+			status := "success"
+			if err != nil {
+				// We're done, export the number of runs we did.
+				gcRunsPerTick.Record(ctx, runsInTick)
+				if errors.Is(err, badger.ErrNoRewrite) {
+					status = "no_rewrite"
+				} else {
+					status = "failure"
+				}
+			}
+			attr := metric.WithAttributes(gcStatusKey.String(status))
+			gcCounter.Add(ctx, 1, attr)
+			gcDuration.Record(ctx, float64(time.Since(start).Milliseconds()), attr)
 			if err == nil {
 				goto again
 			}
@@ -126,32 +145,32 @@ type AntispamStorage struct {
 	// currently integrated tree size.
 	// When pushBack is true, the decorator will start returning ErrPushback to all calls.
 	pushBack atomic.Bool
-
-	numLookups atomic.Uint64
-	numWrites  atomic.Uint64
-	numHits    atomic.Uint64
 }
 
 // index returns the index (if any) previously associated with the provided hash
 func (d *AntispamStorage) index(ctx context.Context, h []byte) (*uint64, error) {
 	return otel.Trace(ctx, "tessera.antispam.badger.index", tracer, func(ctx context.Context, span trace.Span) (*uint64, error) {
-		d.numLookups.Add(1)
+		start := time.Now()
 		var idx *uint64
+		var hit bool
 		err := d.db.View(func(txn *badger.Txn) error {
 			item, err := txn.Get(h)
 			if err == badger.ErrKeyNotFound {
 				span.AddEvent("tessera.miss")
 				return nil
 			}
 			span.AddEvent("tessera.hit")
-			d.numHits.Add(1)
+			hit = true
 
 			return item.Value(func(v []byte) error {
 				i := binary.BigEndian.Uint64(v)
 				idx = &i
 				return nil
 			})
 		})
+		// Microseconds / 1000.0 and not milliseconds to record sub-millisecond durations.
+		lookupDuration.Record(ctx, float64(time.Since(start).Microseconds())/1000.0, metric.WithAttributes(hitKey.Bool(hit)))
+		lookupCounter.Add(ctx, 1, metric.WithAttributes(hitKey.Bool(hit)))
 		return idx, err
 	})
 }
@@ -240,7 +259,8 @@ func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) {
 		for moreWork := true; moreWork; {
 
 			err := f.as.db.Update(func(txn *badger.Txn) error {
-				return otel.TraceErr(ctx, "tessera.antispam.badger.FollowTxn", tracer, func(ctx context.Context, span trace.Span) error {
+				return otel.TraceErr(ctx, "tessera.antispam.badger.follow_txn", tracer, func(ctx context.Context, span trace.Span) error {
+					batchStart := time.Now()
 					ctx, cancel := context.WithTimeout(ctx, defaultBatchTimeout)
 					defer cancel()
 
@@ -352,7 +372,6 @@ func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) {
 					}
 
 					numAdded := uint64(len(curEntries))
-					f.as.numWrites.Add(numAdded)
 
 					// and update the follower state
 					b := make([]byte, 8)
@@ -361,6 +380,10 @@ func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) {
 						return fmt.Errorf("failed to update follower state: %v", err)
 					}
 
+		// Microseconds / 1000.0 and not milliseconds to record sub-millisecond durations.
+					followTxnDuration.Record(ctx, float64(time.Since(batchStart).Microseconds())/1000.0)
+					followTxnEntriesCounter.Record(ctx, int64(numAdded))
+
 					return ctx.Err()
 				})
 			})

storage/posix/antispam/otel.go

@@ -15,17 +15,128 @@
 package badger
 
 import (
+	"context"
+	"log/slog"
+	"os"
+
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
 )
 
 const name = "github.com/transparency-dev/tessera/storage/posix/antispam"
 
 var (
 	tracer = otel.Tracer(name)
+	meter  = otel.Meter(name)
 )
 
 var (
 	followFromKey = attribute.Key("tessera.followFrom")
 	pushbackKey   = attribute.Key("tessera.pushback")
+	gcStatusKey   = attribute.Key("status")
+	hitKey        = attribute.Key("hit")
+)
+
+var (
+	gcCounter                 metric.Int64Counter
+	gcRunsPerTick             metric.Int64Histogram
+	gcDuration                metric.Float64Histogram
+	lookupCounter             metric.Int64Counter
+	lookupDuration            metric.Float64Histogram
+	followTxnEntriesCounter   metric.Int64Histogram
+	followTxnDuration         metric.Float64Histogram
+
+	// Histogram buckets for operations (latency in ms)
+	histogramBuckets = []float64{0.01, 0.02, 0.05, 0.1, 0.2, 0.5, 1, 2, 5, 10, 20, 50, 100, 200, 300, 400, 500, 600, 700, 800, 900, 1000, 1200, 1400, 1600, 1800, 2000, 2500, 3000, 4000, 5000, 6000, 8000, 10000}
+
+	// GC buckets for longer running operations.
+	gcDurationBuckets = []float64{10, 50, 100, 200, 500, 1000, 2000, 5000, 10000, 20000, 30000, 60000}
 )
+
+func generateBatchBuckets(max int) []float64 {
+	bases := []float64{1, 2, 5}
+	buckets := []float64{}
+	for multiplier := 1.0; ; multiplier *= 10 {
+		for _, b := range bases {
+			v := b * multiplier
+			if v >= float64(max) {
+				buckets = append(buckets, float64(max))
+				return buckets
+			}
+			buckets = append(buckets, v)
+		}
+	}
+}
+
+func init() {
+	var err error
+
+	gcCounter, err = meter.Int64Counter(
+		"tessera.antispam.badger.gc.count",
+		metric.WithDescription("Number of BadgerDB Garbage Collection runs"),
+		metric.WithUnit("{run}"))
+	if err != nil {
+		slog.ErrorContext(context.Background(), "Failed to create tessera.antispam.badger.gc.count metric", slog.Any("error", err))
+		os.Exit(1)
+	}
+
+	gcRunsPerTick, err = meter.Int64Histogram(
+		"tessera.antispam.badger.gc.runs_per_tick",
+		metric.WithDescription("Number of GC runs per tick"),
+		metric.WithUnit("{run}"),
+		metric.WithExplicitBucketBoundaries(generateBatchBuckets(500)...))
+	if err != nil {
+		slog.ErrorContext(context.Background(), "Failed to create tessera.antispam.badger.gc.runs_per_tick metric", slog.Any("error", err))
+		os.Exit(1)
+	}
+
+	gcDuration, err = meter.Float64Histogram(
+		"tessera.antispam.badger.gc.duration",
+		metric.WithDescription("Duration of BadgerDB Garbage Collection runs"),
+		metric.WithUnit("ms"),
+		metric.WithExplicitBucketBoundaries(gcDurationBuckets...))
+	if err != nil {
+		slog.ErrorContext(context.Background(), "Failed to create tessera.antispam.badger.gc.duration metric", slog.Any("error", err))
+		os.Exit(1)
+	}
+
+	lookupCounter, err = meter.Int64Counter(
+		"tessera.antispam.badger.index.count",
+		metric.WithDescription("Number of BadgerDB identity lookups"),
+		metric.WithUnit("{lookup}"))
+	if err != nil {
+		slog.ErrorContext(context.Background(), "Failed to create tessera.antispam.badger.index.count metric", slog.Any("error", err))
+		os.Exit(1)
+	}
+
+	lookupDuration, err = meter.Float64Histogram(
+		"tessera.antispam.badger.index.duration",
+		metric.WithDescription("Duration of BadgerDB identity lookups"),
+		metric.WithUnit("ms"),
+		metric.WithExplicitBucketBoundaries(histogramBuckets...))
+	if err != nil {
+		slog.ErrorContext(context.Background(), "Failed to create tessera.antispam.badger.index.duration metric", slog.Any("error", err))
+		os.Exit(1)
+	}
+
+	followTxnEntriesCounter, err = meter.Int64Histogram(
+		"tessera.antispam.badger.follow_txn.entries.count",
+		metric.WithDescription("Number of entries processed by BadgerDB antispam Follow transactions"),
+		metric.WithUnit("{entry}"),
+		metric.WithExplicitBucketBoundaries(generateBatchBuckets(int(DefaultMaxBatchSize))...))
+	if err != nil {
+		slog.ErrorContext(context.Background(), "Failed to create tessera.antispam.badger.follow_txn.entries.count metric", slog.Any("error", err))
+		os.Exit(1)
+	}
+
+	followTxnDuration, err = meter.Float64Histogram(
+		"tessera.antispam.badger.FollowTxn.duration",
+		metric.WithDescription("Duration of BadgerDB Follow transactions"),
+		metric.WithUnit("ms"),
+		metric.WithExplicitBucketBoundaries(histogramBuckets...))
+	if err != nil {
+		slog.ErrorContext(context.Background(), "Failed to create tessera.antispam.badger.FollowTxn.duration metric", slog.Any("error", err))
+		os.Exit(1)
+	}
+}