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/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 {