Merge pull request #828 from Alegr37/issue-826

issue-826: write to log offset ranges for OnReaderSendCommitMessage

Author: asmyasnikov

CHANGELOG.md

@@ -1,3 +1,4 @@
+* Improved OnReaderSendCommitMessage logs
 * Added `table.Session.CopyTables` method
 * Added `x-ydb-trace-id` header into grpc calls
 * Improved topic reader logs

internal/topic/topicreaderinternal/commit_range.go

@@ -4,6 +4,7 @@ import (
 	"sort"
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawtopic/rawtopicreader"
+	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
 )
 
 // PublicCommitRangeGetter return data piece for commit messages range
@@ -23,20 +24,17 @@ func (r *CommitRanges) len() int {
 	return len(r.ranges)
 }
 
-// PartitionIDs implements trace.TopicReaderStreamSendCommitMessageStartMessageInfo
-func (r *CommitRanges) PartitionIDs() []int64 {
-	res := make([]int64, len(r.ranges))
+// GetCommitsInfo implements trace.TopicReaderStreamSendCommitMessageStartMessageInfo
+func (r *CommitRanges) GetCommitsInfo() []trace.TopicReaderStreamCommitInfo {
+	res := make([]trace.TopicReaderStreamCommitInfo, len(r.ranges))
 	for i := range res {
-		res[i] = r.ranges[i].partitionSession.PartitionID
-	}
-	return res
-}
-
-// PartitionSessionIDs implements trace.TopicReaderStreamSendCommitMessageStartMessageInfo
-func (r *CommitRanges) PartitionSessionIDs() []int64 {
-	res := make([]int64, len(r.ranges))
-	for i := range res {
-		res[i] = r.ranges[i].partitionSession.partitionSessionID.ToInt64()
+		res[i] = trace.TopicReaderStreamCommitInfo{
+			Topic:              r.ranges[i].partitionSession.Topic,
+			PartitionID:        r.ranges[i].partitionSession.PartitionID,
+			PartitionSessionID: r.ranges[i].partitionSession.partitionSessionID.ToInt64(),
+			StartOffset:        r.ranges[i].commitOffsetStart.ToInt64(),
+			EndOffset:          r.ranges[i].commitOffsetEnd.ToInt64(),
+		}
 	}
 	return res
 }

internal/topic/topicreaderinternal/committer.go

@@ -144,7 +144,10 @@ func (c *committer) pushCommitsLoop(ctx context.Context) {
 
 		commits.optimize()
 
-		onDone := trace.TopicOnReaderSendCommitMessage(c.tracer, &commits)
+		onDone := trace.TopicOnReaderSendCommitMessage(
+			c.tracer,
+			&commits,
+		)
 		err := sendCommitMessage(c.send, commits)
 		onDone(err)
 

log/topic.go

@@ -162,25 +162,37 @@ func internalTopic(l *wrapper, d trace.Detailer) (t trace.Topic) { //nolint:gocy
 		}
 		ctx := with(context.Background(), TRACE, "ydb", "topic", "reader", "send", "commit", "message")
 		start := time.Now()
-		l.Log(ctx, "start",
-			Any("partitions_id", info.CommitsInfo.PartitionIDs()),
-			Any("partitions_session_id", info.CommitsInfo.PartitionSessionIDs()),
-		)
+
+		commitInfo := info.CommitsInfo.GetCommitsInfo()
+		for i := range commitInfo {
+			l.Log(ctx, "start",
+				String("topic", commitInfo[i].Topic),
+				Int64("partitions_id", commitInfo[i].PartitionID),
+				Int64("partitions_session_id", commitInfo[i].PartitionSessionID),
+				Int64("commit_start_offset", commitInfo[i].StartOffset),
+				Int64("commit_end_offset", commitInfo[i].EndOffset),
+			)
+		}
 		return func(doneInfo trace.TopicReaderSendCommitMessageDoneInfo) {
-			fields := []Field{
-				Any("partitions_id", info.CommitsInfo.PartitionIDs()),
-				Any("partitions_session_id", info.CommitsInfo.PartitionSessionIDs()),
-				latencyField(start),
-			}
-			if doneInfo.Error == nil {
-				l.Log(ctx, "done", fields...)
-			} else {
-				l.Log(WithLevel(ctx, WARN), "commit message sent",
-					append(fields,
-						Error(doneInfo.Error),
-						versionField(),
-					)...,
-				)
+			for i := range commitInfo {
+				fields := []Field{
+					String("topic", commitInfo[i].Topic),
+					Int64("partitions_id", commitInfo[i].PartitionID),
+					Int64("partitions_session_id", commitInfo[i].PartitionSessionID),
+					Int64("commit_start_offset", commitInfo[i].StartOffset),
+					Int64("commit_end_offset", commitInfo[i].EndOffset),
+					latencyField(start),
+				}
+				if doneInfo.Error == nil {
+					l.Log(ctx, "done", fields...)
+				} else {
+					l.Log(WithLevel(ctx, WARN), "commit message sent",
+						append(fields,
+							Error(doneInfo.Error),
+							versionField(),
+						)...,
+					)
+				}
 			}
 		}
 	}

trace/topic.go

@@ -134,19 +134,31 @@ type (
 	// Notice: This API is EXPERIMENTAL and may be changed or removed in a
 	// later release.
 	TopicReaderSendCommitMessageStartInfo struct {
-		// ReaderConnectionID string unimplemented yet - need some internal changes
 		CommitsInfo TopicReaderStreamSendCommitMessageStartMessageInfo
 	}
 
+	// TopicReaderStreamCommitInfo
+	//
+	// Experimental
+	//
+	// Notice: This API is EXPERIMENTAL and may be changed or removed in a
+	// later release.
+	TopicReaderStreamCommitInfo struct {
+		Topic              string
+		PartitionID        int64
+		PartitionSessionID int64
+		StartOffset        int64
+		EndOffset          int64
+	}
+
 	// TopicReaderStreamSendCommitMessageStartMessageInfo
 	//
 	// Experimental
 	//
 	// Notice: This API is EXPERIMENTAL and may be changed or removed in a
 	// later release.
 	TopicReaderStreamSendCommitMessageStartMessageInfo interface {
-		PartitionIDs() []int64
-		PartitionSessionIDs() []int64
+		GetCommitsInfo() []TopicReaderStreamCommitInfo
 	}
 
 	// TopicReaderSendCommitMessageDoneInfo