feat(storagenode): migrate metrics to OpenTelemetry histogram wrappers

Replace existing metrics with OpenTelemetry histogram wrappers. Convert several
metric types from counters to histograms to capture measurement distributions.

Converted from counter to histogram:

- sn.append.preparation
- sn.sequencer.operation.duration
- sn.sequencer.fanout.duration
- sn.writer.operation.duration
- sn.committer.operation.duration
- sn.committer.logs
- sn.replicate.client.operation.duration
- sn.replicate.duration
- sn.replicate.fanout.duration

Migrated to OpenTelemetry histogram wrapper (already histogram):

- log_rpc.server.duration
- log_rpc.server.log_entry.size
- log_rpc.server.batch.size
- log_rpc.server.log_entries_per_batch

This improves observability by enabling distribution analysis for these metrics.

Author: ijsong

internal/storagenode/log_server.go

@@ -7,11 +7,15 @@ import (
 	"time"
 
 	pbtypes "github.com/gogo/protobuf/types"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+	semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
 	"go.uber.org/multierr"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
+	"github.com/kakao/varlog/internal/stats/opentelemetry"
 	snerrors "github.com/kakao/varlog/internal/storagenode/errors"
 	"github.com/kakao/varlog/internal/storagenode/logstream"
 	"github.com/kakao/varlog/internal/storagenode/telemetry"
@@ -118,6 +122,7 @@ func (ls *logServer) appendStreamRecvLoop(stream snpb.LogIO_AppendServer, cq cha
 			lsid = req.LogStreamID
 		}
 
+		appendTask.TopicID = tpid
 		appendTask.LogStreamID = lsid
 		appendTask.RPCStartTime = time.Now()
 
@@ -160,6 +165,7 @@ func (ls *logServer) appendStreamSendLoop(stream snpb.LogIO_AppendServer, cq <-c
 				return nil
 			}
 
+			tpid := appendTask.TopicID
 			lsid := appendTask.LogStreamID
 			res, err = appendTask.WaitForCompletion(ctx)
 			elapsed := time.Since(appendTask.RPCStartTime)
@@ -188,10 +194,18 @@ func (ls *logServer) appendStreamSendLoop(stream snpb.LogIO_AppendServer, cq <-c
 				}
 			}
 			if !lsid.Invalid() {
-				metrics, ok := ls.sn.metrics.GetLogStreamMetrics(lsid)
-				if ok {
-					metrics.LogRPCServerDuration.Record(ctx, telemetry.RPCKindAppend, code, elapsed.Microseconds())
-				}
+				ls.sn.metrics.LogRPCServerDuration.Record(ctx, lsid, elapsed.Microseconds(), func() []metric.RecordOption {
+					return []metric.RecordOption{
+						metric.WithAttributeSet(attribute.NewSet(
+							opentelemetry.TopicID(tpid),
+							opentelemetry.LogStreamID(lsid),
+							semconv.RPCSystemGRPC,
+							semconv.RPCService(telemetry.ServiceNames[telemetry.RPCKindAppend]),
+							semconv.RPCMethod(telemetry.MethodNames[telemetry.RPCKindAppend]),
+							semconv.RPCGRPCStatusCodeKey.Int64(int64(code)),
+						)),
+					}
+				})
 			}
 			if err != nil {
 				return err

internal/storagenode/logstream/append.go

@@ -5,10 +5,7 @@ import (
 	"sync"
 	"time"
 
-	"google.golang.org/grpc/codes"
-
 	snerrors "github.com/kakao/varlog/internal/storagenode/errors"
-	"github.com/kakao/varlog/internal/storagenode/telemetry"
 	"github.com/kakao/varlog/pkg/types"
 	"github.com/kakao/varlog/pkg/verrors"
 	"github.com/kakao/varlog/proto/snpb"
@@ -28,6 +25,7 @@ type AppendTask struct {
 	apc          appendContext
 	dataBatchLen int
 
+	TopicID      types.TopicID
 	LogStreamID  types.LogStreamID
 	RPCStartTime time.Time
 }
@@ -117,9 +115,6 @@ func (at *AppendTask) WaitForCompletion(ctx context.Context) ([]snpb.AppendResul
 func appendTaskDeferredFunc(at *AppendTask) {
 	at.lse.inflight.Add(-1)
 	at.lse.inflightAppend.Add(-1)
-	if at.lse.lsm != nil {
-		at.lse.lsm.AppendDuration.Add(time.Since(at.start).Microseconds())
-	}
 }
 
 type appendContext struct {
@@ -154,13 +149,9 @@ func (lse *Executor) AppendAsync(ctx context.Context, dataBatch [][]byte, append
 	var preparationDuration time.Duration
 	defer func() {
 		if lse.lsm != nil {
-			lse.lsm.AppendLogs.Add(int64(dataBatchLen))
-			lse.lsm.AppendBytes.Add(totalBytes)
-			lse.lsm.AppendOperations.Add(1)
-			lse.lsm.AppendPreparationMicro.Add(preparationDuration.Microseconds())
-
-			lse.lsm.LogRPCServerBatchSize.Record(context.Background(), telemetry.RPCKindAppend, codes.OK, totalBytes)
-			lse.lsm.LogRPCServerLogEntriesPerBatch.Record(context.Background(), telemetry.RPCKindAppend, codes.OK, int64(dataBatchLen))
+			lse.lsm.AppendPreparationDuration.Record(context.Background(), preparationDuration.Microseconds())
+			lse.lsm.LogRPCServerBatchSize.Record(context.Background(), totalBytes)
+			lse.lsm.LogRPCServerLogEntriesPerBatch.Record(context.Background(), int64(dataBatchLen))
 		}
 	}()
 
@@ -199,8 +190,7 @@ func (lse *Executor) prepareAppendContext(dataBatch [][]byte, apc *appendContext
 		logEntrySize := int64(len(dataBatch[i]))
 		totalBytes += logEntrySize
 		if lse.lsm != nil {
-			// TODO: Set the correct status code.
-			lse.lsm.LogRPCServerLogEntrySize.Record(context.Background(), telemetry.RPCKindAppend, codes.OK, logEntrySize)
+			lse.lsm.LogRPCServerLogEntrySize.Record(context.Background(), logEntrySize)
 		}
 	}
 	awg := newAppendWaitGroup(st.wwg, len(dataBatch))

internal/storagenode/logstream/backup_writer.go

@@ -74,7 +74,7 @@ func (bw *backupWriter) writeLoop(ctx context.Context) {
 	}
 }
 
-func (bw *backupWriter) writeLoopInternal(_ context.Context, bwt *backupWriteTask) {
+func (bw *backupWriter) writeLoopInternal(ctx context.Context, bwt *backupWriteTask) {
 	startTime := time.Now()
 	var err error
 	wb, oldLLSN, newLLSN := bwt.wb, bwt.oldLLSN, bwt.newLLSN
@@ -88,8 +88,7 @@ func (bw *backupWriter) writeLoopInternal(_ context.Context, bwt *backupWriteTas
 		if bw.lse.lsm == nil {
 			return
 		}
-		bw.lse.lsm.WriterOperationDuration.Add(time.Since(startTime).Microseconds())
-		bw.lse.lsm.WriterOperations.Add(1)
+		bw.lse.lsm.WriterOperationDuration.Record(ctx, time.Since(startTime).Microseconds())
 	}()
 
 	if uncommittedLLSNEnd := bw.lse.lsc.uncommittedLLSNEnd.Load(); uncommittedLLSNEnd != oldLLSN {

internal/storagenode/logstream/committer.go

@@ -253,9 +253,8 @@ func (cm *committer) commitInternal(cc storage.CommitContext) (err error) {
 		if cm.lse.lsm == nil {
 			return
 		}
-		cm.lse.lsm.CommitterOperationDuration.Add(int64(time.Since(startTime).Microseconds()))
-		cm.lse.lsm.CommitterOperations.Add(1)
-		cm.lse.lsm.CommitterLogs.Add(int64(numCommits))
+		cm.lse.lsm.CommitterOperationDuration.Record(context.Background(), int64(time.Since(startTime).Microseconds()))
+		cm.lse.lsm.CommitterLogs.Record(context.Background(), int64(numCommits))
 	}()
 
 	iter := cm.commitWaitQ.peekIterator()

internal/storagenode/logstream/executor.go

@@ -177,21 +177,19 @@ func (lse *Executor) Replicate(ctx context.Context, beginLLSN types.LLSN, dataLi
 		return errors.New("log stream: not backup")
 	}
 
-	var preparationDuration time.Duration
-	startTime := time.Now()
+	var startTime, prepEndTime time.Time
 	dataBytes := int64(0)
 	batchSize := len(dataList)
 	defer func() {
 		if lse.lsm == nil {
 			return
 		}
-		lse.lsm.ReplicateLogs.Add(int64(batchSize))
-		lse.lsm.ReplicateBytes.Add(dataBytes)
-		lse.lsm.ReplicateDuration.Add(time.Since(startTime).Microseconds())
-		lse.lsm.ReplicateOperations.Add(1)
-		lse.lsm.ReplicatePreparationMicro.Add(preparationDuration.Microseconds())
+		lse.lsm.ReplicateDuration.Record(ctx, time.Since(startTime).Microseconds())
+		lse.lsm.ReplicateFanoutDuration.Record(ctx, time.Since(prepEndTime).Microseconds())
 	}()
 
+	startTime = time.Now()
+
 	oldLLSN, newLLSN := beginLLSN, beginLLSN+types.LLSN(batchSize)
 	wb := lse.stg.NewWriteBatch()
 	for i := 0; i < batchSize; i++ {
@@ -201,7 +199,7 @@ func (lse *Executor) Replicate(ctx context.Context, beginLLSN types.LLSN, dataLi
 	cwt := newCommitWaitTask(nil, batchSize)
 	bwt := newBackupWriteTask(wb, oldLLSN, newLLSN)
 
-	preparationDuration = time.Since(startTime)
+	prepEndTime = time.Now()
 
 	if err := lse.bw.send(ctx, bwt); err != nil {
 		lse.logger.Error("could not send backup batch write task", zap.Error(err))
@@ -216,6 +214,7 @@ func (lse *Executor) Replicate(ctx context.Context, beginLLSN types.LLSN, dataLi
 		cwt.release()
 		return err
 	}
+
 	return nil
 }
 

internal/storagenode/logstream/replicate_client.go

@@ -127,7 +127,7 @@ func (rc *replicateClient) sendLoop(ctx context.Context) {
 }
 
 // sendLoopInternal sends a replicate task to the backup replica.
-func (rc *replicateClient) sendLoopInternal(_ context.Context, rt *replicateTask, req *snpb.ReplicateRequest) error {
+func (rc *replicateClient) sendLoopInternal(ctx context.Context, rt *replicateTask, req *snpb.ReplicateRequest) error {
 	startTime := time.Now()
 	req.Data = rt.dataList
 	req.BeginLLSN = rt.beginLLSN
@@ -136,8 +136,7 @@ func (rc *replicateClient) sendLoopInternal(_ context.Context, rt *replicateTask
 	inflight := rc.inflight.Add(-1)
 	if rc.lse.lsm != nil {
 		rc.lse.lsm.ReplicateClientInflightOperations.Store(inflight)
-		rc.lse.lsm.ReplicateClientOperationDuration.Add(time.Since(startTime).Microseconds())
-		rc.lse.lsm.ReplicateClientOperations.Add(1)
+		rc.lse.lsm.ReplicateClientOperationDuration.Record(ctx, time.Since(startTime).Microseconds())
 	}
 	return err
 }

internal/storagenode/logstream/sequencer.go

@@ -94,9 +94,8 @@ func (sq *sequencer) sequenceLoopInternal(ctx context.Context, st *sequenceTask)
 		if sq.lse.lsm == nil {
 			return
 		}
-		sq.lse.lsm.SequencerFanoutDuration.Add(time.Since(operationEndTime).Microseconds())
-		sq.lse.lsm.SequencerOperationDuration.Add(int64(operationEndTime.Sub(startTime).Microseconds()))
-		sq.lse.lsm.SequencerOperations.Add(1)
+		sq.lse.lsm.SequencerFanoutDuration.Record(ctx, time.Since(operationEndTime).Microseconds())
+		sq.lse.lsm.SequencerOperationDuration.Record(ctx, int64(operationEndTime.Sub(startTime).Microseconds()))
 		sq.lse.lsm.ReplicateClientInflightOperations.Store(inflight)
 	}()
 

internal/storagenode/logstream/writer.go

@@ -83,7 +83,7 @@ func (w *writer) writeLoop(ctx context.Context) {
 }
 
 // writeLoopInternal stores a batch of writes to the storage and modifies uncommittedLLSNEnd of the log stream which presents the next expected LLSN to be written.
-func (w *writer) writeLoopInternal(_ context.Context, st *sequenceTask) {
+func (w *writer) writeLoopInternal(ctx context.Context, st *sequenceTask) {
 	startTime := time.Now()
 	var err error
 	cnt := len(st.dataBatch)
@@ -98,8 +98,7 @@ func (w *writer) writeLoopInternal(_ context.Context, st *sequenceTask) {
 		if w.lse.lsm == nil {
 			return
 		}
-		w.lse.lsm.WriterOperationDuration.Add(time.Since(startTime).Microseconds())
-		w.lse.lsm.WriterOperations.Add(1)
+		w.lse.lsm.WriterOperationDuration.Record(ctx, time.Since(startTime).Microseconds())
 		w.lse.lsm.WriterInflightOperations.Store(inflight)
 	}()