feat(topic): log endpoint for topic writer session (#2029)

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: kprokopenko

CHANGELOG.md

@@ -1,3 +1,5 @@
+* Added `endpoint` (`node_id`, `address`, `location`) to topic writer init stream logs
+
 ## v3.127.0
 * Changed behaviour of `table.Client` and `query.Client` internal session pool. When `ydb.WithPreferredNodeID` is set and there is no idle session on preferred node, the pool closes most idle session to create a new one on the preferred node
 * Fixed a bug when sometimes in cases of context cancellation returner error was not `errors.Is(err, context.Canceled)`

internal/conn/grpc_client_stream.go

@@ -8,6 +8,7 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/metadata"
 
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/endpoint"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/meta"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/operation"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/stack"
@@ -37,6 +38,12 @@ func (s *grpcClientStream) Context() context.Context {
 	return s.stream.Context()
 }
 
+// Endpoint returns the endpoint of the connection used by this stream.
+// It implements the optional interface used by topic writer for session logging.
+func (s *grpcClientStream) Endpoint() endpoint.Endpoint {
+	return s.parentConn.Endpoint()
+}
+
 func (s *grpcClientStream) CloseSend() (err error) {
 	var (
 		ctx    = s.streamCtx

internal/grpcwrapper/rawtopic/client.go

@@ -7,6 +7,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/ydb-platform/ydb-go-genproto/Ydb_Topic_V1"
 
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/endpoint"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawtopic/rawtopicreader"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawtopic/rawtopicwriter"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawydb"
@@ -122,11 +123,17 @@ func (c *Client) StreamWrite(
 		)
 	}
 
+	var ep endpoint.Endpoint
+	if withEp, ok := protoResp.(interface{ Endpoint() endpoint.Endpoint }); ok {
+		ep = withEp.Endpoint()
+	}
+
 	return &rawtopicwriter.StreamWriter{
 		Stream:           protoResp,
 		Tracer:           tracer,
 		InternalStreamID: uuid.New().String(),
 		LogContext:       &ctxStreamLifeTime,
+		PeerEndpoint:     ep,
 	}, nil
 }
 

internal/grpcwrapper/rawtopic/rawtopicwriter/streamwriter.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Topic"
 
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/endpoint"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawtopic/rawtopiccommon"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawydb"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
@@ -36,8 +37,13 @@ type StreamWriter struct {
 	writtenMessagesCount int
 	sessionID            string
 	LogContext           *context.Context
+	// PeerEndpoint is the endpoint this stream is connected to (for logging).
+	PeerEndpoint endpoint.Endpoint
 }
 
+// Endpoint returns the endpoint this stream is connected to (for logging).
+func (w *StreamWriter) Endpoint() endpoint.Endpoint { return w.PeerEndpoint }
+
 //nolint:funlen
 func (w *StreamWriter) Recv() (ServerMessage, error) {
 	readCnt := atomic.AddInt32(&w.readCounter, 1)

internal/topic/topicwriterinternal/raw_topic_writer_stream_mock_test.go

@@ -610,13 +610,19 @@ func (w *WriterReconnector) waitFirstInitResponse(ctx context.Context) error {
 }
 
 func (w *WriterReconnector) createWriterStreamConfig(stream RawTopicWriterStream) SingleStreamWriterConfig {
+	var ep trace.EndpointInfo
+	if stream != nil {
+		ep = stream.Endpoint() // endpoint.Endpoint implements trace.EndpointInfo
+	}
+
 	cfg := newSingleStreamWriterConfig(
 		w.cfg.WritersCommonConfig,
 		stream,
 		&w.queue,
 		w.encodersMap,
 		w.needReceiveLastSeqNo(),
 		w.writerInstanceID,
+		ep,
 	)
 
 	return cfg

internal/topic/topicwriterinternal/writer_reconnector.go

@@ -555,6 +555,7 @@ func TestWriterImpl_Reconnect(t *testing.T) {
 		}
 
 		initRequest := testCreateInitRequest(w)
+		strm.EXPECT().Endpoint().Return(nil).AnyTimes()
 		strm.EXPECT().Send(&initRequest)
 		strm.EXPECT().Recv().Return(nil, testErr)
 		strm.EXPECT().CloseSend()
@@ -581,6 +582,7 @@ func TestWriterImpl_Reconnect(t *testing.T) {
 		isFirstConnection := true
 		newStream := func(name string) *MockRawTopicWriterStream {
 			strm := NewMockRawTopicWriterStream(mc)
+			strm.EXPECT().Endpoint().Return(nil).AnyTimes()
 			initReq := testCreateInitRequest(w)
 			if isFirstConnection {
 				isFirstConnection = false
@@ -1194,6 +1196,7 @@ func newTestEnv(t testing.TB, options *testEnvOptions) *testEnv {
 
 	res.writer = newWriterReconnectorStopped(NewWriterReconnectorConfig(writerOptions...))
 
+	res.stream.EXPECT().Endpoint().Return(nil).AnyTimes()
 	res.stream.EXPECT().Recv().DoAndReturn(res.receiveMessageHandler).AnyTimes()
 
 	req := testCreateInitRequest(res.writer)

internal/topic/topicwriterinternal/writer_reconnector_test.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/background"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/empty"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/endpoint"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawtopic/rawtopiccommon"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawtopic/rawtopicwriter"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xcontext"
@@ -26,6 +27,7 @@ type SingleStreamWriterConfig struct {
 	encodersMap           *MultiEncoder
 	getLastSeqNum         bool
 	reconnectorInstanceID string
+	endpoint              trace.EndpointInfo
 }
 
 func newSingleStreamWriterConfig(
@@ -35,6 +37,7 @@ func newSingleStreamWriterConfig(
 	encodersMap *MultiEncoder,
 	getLastSeqNum bool,
 	reconnectorID string,
+	endpoint trace.EndpointInfo,
 ) SingleStreamWriterConfig {
 	return SingleStreamWriterConfig{
 		WritersCommonConfig:   common,
@@ -43,6 +46,7 @@ func newSingleStreamWriterConfig(
 		encodersMap:           encodersMap,
 		getLastSeqNum:         getLastSeqNum,
 		reconnectorInstanceID: reconnectorID,
+		endpoint:              endpoint,
 	}
 }
 
@@ -125,16 +129,16 @@ func (w *SingleStreamWriter) start() {
 }
 
 func (w *SingleStreamWriter) initStream() (err error) {
+	logCtx := w.cfg.LogContext
+	traceOnDone := trace.TopicOnWriterInitStream(
+		w.cfg.Tracer,
+		&logCtx,
+		w.cfg.reconnectorInstanceID,
+		w.cfg.topic,
+		w.cfg.producerID,
+	)
 	defer func() {
-		logCtx := w.cfg.LogContext
-		traceOnDone := trace.TopicOnWriterInitStream(
-			w.cfg.Tracer,
-			&logCtx,
-			w.cfg.reconnectorInstanceID,
-			w.cfg.topic,
-			w.cfg.producerID,
-		)
-		traceOnDone(w.SessionID, err)
+		traceOnDone(w.SessionID, w.cfg.endpoint, err)
 	}()
 
 	req := w.createInitRequest()
@@ -322,6 +326,7 @@ type RawTopicWriterStream interface {
 	Recv() (rawtopicwriter.ServerMessage, error)
 	Send(mess rawtopicwriter.ClientMessage) error
 	CloseSend() error
+	Endpoint() endpoint.Endpoint
 }
 
 func sendMessagesToStream(

internal/topic/topicwriterinternal/writer_single_stream.go

@@ -702,23 +702,25 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
 		)
 
 		return func(doneInfo trace.TopicWriterInitStreamDoneInfo) {
-			if doneInfo.Error == nil {
-				l.Log(WithLevel(ctx, DEBUG), "topic writer init stream done",
-					kv.Error(doneInfo.Error),
-					kv.String("topic", info.Topic),
-					kv.String("producer_id", info.ProducerID),
-					kv.String("writer_instance_id", info.WriterInstanceID),
-					kv.Latency(start),
-					kv.String("session_id", doneInfo.SessionID),
+			fields := []Field{
+				kv.String("topic", info.Topic),
+				kv.String("producer_id", info.ProducerID),
+				kv.String("writer_instance_id", info.WriterInstanceID),
+				kv.Latency(start),
+				kv.String("session_id", doneInfo.SessionID),
+			}
+			if doneInfo.Endpoint != nil {
+				fields = append(fields,
+					kv.Int64("node_id", int64(doneInfo.Endpoint.NodeID())),
+					kv.String("address", doneInfo.Endpoint.Address()),
+					kv.String("location", doneInfo.Endpoint.Location()),
 				)
+			}
+			if doneInfo.Error == nil {
+				l.Log(WithLevel(ctx, DEBUG), "topic writer init stream done", fields...)
 			} else {
 				l.Log(WithLevel(ctx, WARN), "topic writer init stream failed",
-					kv.Error(doneInfo.Error),
-					kv.String("topic", info.Topic),
-					kv.String("producer_id", info.ProducerID),
-					kv.String("writer_instance_id", info.WriterInstanceID),
-					kv.Latency(start),
-					kv.String("session_id", doneInfo.SessionID),
+					append(append(fields, kv.Error(doneInfo.Error)), kv.Version())...,
 				)
 			}
 		}

log/topic.go

@@ -546,6 +546,7 @@ type (
 	// Internals: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#internals
 	TopicWriterInitStreamDoneInfo struct {
 		SessionID string
+		Endpoint  EndpointInfo
 		Error     error
 	}