export assign/integrate latency

Author: AlCutter

storage/aws/aws.go

@@ -297,6 +297,9 @@ func (a *Appender) integrateEntriesJob(ctx context.Context) {
 		}
 
 		if err := otel.TraceErr(ctx, "tessera.storage.aws.integrateEntriesJob", tracer, func(ctx context.Context, span trace.Span) error {
+			start := time.Now()
+			defer opsHistogram.Record(ctx, time.Since(start).Milliseconds(), metric.WithAttributes(opNameKey.String("integrateEntries")))
+			
 			ctx, cancel := context.WithTimeout(ctx, defaultIntegrationTimeout)
 			defer cancel() // Note: ok because we're in a func passed to TraceErr here!
 
@@ -1054,7 +1057,11 @@ func (s *mySQLSequencer) initDB(ctx context.Context) error {
 // index assigned to the first entry in the batch.
 func (s *mySQLSequencer) assignEntries(ctx context.Context, entries []*tessera.Entry) error {
 	return otel.TraceErr(ctx, "tessera.storage.gcp.assignEntries", tracer, func(ctx context.Context, span trace.Span) error {
+		start := time.Now()
+		defer opsHistogram.Record(ctx, time.Since(start).Milliseconds(), metric.WithAttributes(opNameKey.String("assignEntries")))
+
 		span.SetAttributes(numEntriesKey.Int(len(entries)))
+
 		// First grab the treeSize in a non-locking read-only fashion (we don't want to block/collide with integration).
 		// We'll use this value to determine whether we need to apply back-pressure.
 		var treeSize uint64

storage/gcp/gcp.go

@@ -345,6 +345,9 @@ func (a *Appender) integrateEntriesJob(ctx context.Context) {
 		}
 
 		if err := otel.TraceErr(ctx, "tessera.storage.gcp.integrateEntriesJob", tracer, func(ctx context.Context, span trace.Span) error {
+		start := time.Now()
+		defer opsHistogram.Record(ctx, time.Since(start).Milliseconds(), metric.WithAttributes(opNameKey.String("integrateEntries")))
+		
 			ctx, cancel := context.WithTimeout(ctx, defaultIntegrationTimeout)
 			defer cancel() // Note: ok because we're in a func passed to TraceErr here!
 
@@ -833,6 +836,9 @@ func (s *spannerCoordinator) checkDataCompatibility(ctx context.Context) error {
 // index assigned to the first entry in the batch.
 func (s *spannerCoordinator) assignEntries(ctx context.Context, entries []*tessera.Entry) error {
 	return otel.TraceErr(ctx, "tessera.storage.gcp.assignEntries", tracer, func(ctx context.Context, span trace.Span) error {
+		start := time.Now()
+		defer opsHistogram.Record(ctx, time.Since(start).Milliseconds(), metric.WithAttributes(opNameKey.String("assignEntries")))
+
 		span.SetAttributes(numEntriesKey.Int(len(entries)))
 
 		span.AddEvent("Reading IntCoord:seq")

storage/gcp/otel.go

@@ -32,16 +32,16 @@ var (
 )
 
 var (
-	outcomeTypeKey = attribute.Key("tessera.publishCheckpoint.outcome")
-	treeSizeKey    = attribute.Key("tessera.treeSize")
-	fromSizeKey    = attribute.Key("tessera.fromSize")
-	numEntriesKey  = attribute.Key("tessera.numEntries")
-	objectPathKey  = attribute.Key("tessera.objectPath")
-	opNameKey      = attribute.Key("op_name")
+	outcomeTypeKey           = attribute.Key("tessera.publishCheckpoint.outcome")
+	treeSizeKey              = attribute.Key("tessera.treeSize")
+	fromSizeKey              = attribute.Key("tessera.fromSize")
+	numEntriesKey            = attribute.Key("tessera.numEntries")
+	objectPathKey            = attribute.Key("tessera.objectPath")
+	opNameKey                = attribute.Key("op_name")
 
-	publishCount           metric.Int64Counter
-	opsHistogram           metric.Int64Histogram
-	checkpointAgeHistogram metric.Int64Histogram
+	publishCount               metric.Int64Counter
+	opsHistogram               metric.Int64Histogram
+	checkpointAgeHistogram     metric.Int64Histogram
 
 	// Custom histogram buckets as we're interested in low-millis upto low-seconds.
 	histogramBuckets = []float64{0, 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}
@@ -78,4 +78,4 @@ func init() {
 		slog.ErrorContext(context.Background(), "Failed to create checkpoint publication counter metric", slog.Any("error", err))
 		os.Exit(1)
 	}
-}
+}

storage/posix/files.go

@@ -306,6 +306,8 @@ func (l *logResourceStorage) NextIndex(ctx context.Context) (uint64, error) {
 // than one-by-one.
 func (a *appender) sequenceBatch(ctx context.Context, entries []*tessera.Entry) error {
 	return otel.TraceErr(ctx, "tessera.storage.posix.assignEntries", tracer, func(ctx context.Context, span trace.Span) error {
+		start := time.Now()
+		defer posixOpsHistogram.Record(ctx, time.Since(start).Milliseconds(), metric.WithAttributes(opNameKey.String("assignIntegrateBatch")))
 		span.SetAttributes(numEntriesKey.Int(len(entries)))
 
 		// Double locking:
@@ -398,6 +400,7 @@ func (a *appender) sequenceBatch(ctx context.Context, entries []*tessera.Entry)
 		if err := a.s.writeTreeState(ctx, newSize, newRoot); err != nil {
 			return fmt.Errorf("failed to write new tree state: %v", err)
 		}
+
 		// Notify that we know for sure there's a new checkpoint, but don't block if there's already
 		// an outstanding notification in the channel.
 		select {