11package log
22
33import (
4- "context"
54 "time"
65
76 "github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Topic"
@@ -25,7 +24,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
2524 if d .Details ()& trace .TopicReaderStreamLifeCycleEvents == 0 {
2625 return nil
2726 }
28- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "reader" , "reconnect" )
27+ ctx := with (* info . Context , TRACE , "ydb" , "topic" , "reader" , "reconnect" )
2928 start := time .Now ()
3029 l .Log (ctx , "topic reader reconnect starting..." )
3130
@@ -40,7 +39,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
4039 if d .Details ()& trace .TopicReaderStreamLifeCycleEvents == 0 {
4140 return
4241 }
43- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "reader" , "reconnect" , "request" )
42+ ctx := with (* info . Context , TRACE , "ydb" , "topic" , "reader" , "reconnect" , "request" )
4443 l .Log (ctx , "topic reader reconnect request" ,
4544 kv .NamedError ("reason" , info .Reason ),
4645 kv .Bool ("was_sent" , info .WasSent ),
@@ -52,7 +51,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
5251 if d .Details ()& trace .TopicReaderPartitionEvents == 0 {
5352 return nil
5453 }
55- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "reader" , "partition" , "read" , "start" , "response" )
54+ ctx := with (* info . PartitionContext , TRACE , "ydb" , "topic" , "reader" , "partition" , "read" , "start" , "response" )
5655 start := time .Now ()
5756 l .Log (ctx , "topic reader start partition read response starting..." ,
5857 kv .String ("topic" , info .Topic ),
@@ -97,7 +96,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
9796 if d .Details ()& trace .TopicReaderPartitionEvents == 0 {
9897 return nil
9998 }
100- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "reader" , "partition" , "read" , "stop" , "response" )
99+ ctx := with (info . PartitionContext , TRACE , "ydb" , "topic" , "reader" , "partition" , "read" , "stop" , "response" )
101100 start := time .Now ()
102101 l .Log (ctx , "topic reader stop partition read response starting..." ,
103102 kv .String ("reader_connection_id" , info .ReaderConnectionID ),
@@ -133,7 +132,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
133132 if d .Details ()& trace .TopicReaderStreamEvents == 0 {
134133 return nil
135134 }
136- ctx := with (* info .RequestContext , TRACE , "ydb" , "topic" , "reader" , "commit" )
135+ ctx := with (* info .Context , TRACE , "ydb" , "topic" , "reader" , "commit" )
137136 start := time .Now ()
138137 l .Log (ctx , "topic reader commit starting..." ,
139138 kv .String ("topic" , info .Topic ),
@@ -170,7 +169,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
170169 if d .Details ()& trace .TopicReaderStreamEvents == 0 {
171170 return nil
172171 }
173- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "reader" , "send" , "commit" , "message" )
172+ ctx := with (* info . Context , TRACE , "ydb" , "topic" , "reader" , "send" , "commit" , "message" )
174173 start := time .Now ()
175174
176175 commitInfo := info .CommitsInfo .GetCommitsInfo ()
@@ -211,7 +210,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
211210 if d .Details ()& trace .TopicReaderStreamEvents == 0 {
212211 return
213212 }
214- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "reader" , "committed" , "notify" )
213+ ctx := with (* info . Context , TRACE , "ydb" , "topic" , "reader" , "committed" , "notify" )
215214 l .Log (ctx , "topic reader received commit ack" ,
216215 kv .String ("reader_connection_id" , info .ReaderConnectionID ),
217216 kv .String ("topic" , info .Topic ),
@@ -224,7 +223,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
224223 if d .Details ()& trace .TopicReaderStreamEvents == 0 {
225224 return nil
226225 }
227- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "reader" , "close" )
226+ ctx := with (* info . Context , TRACE , "ydb" , "topic" , "reader" , "close" )
228227 start := time .Now ()
229228 l .Log (ctx , "topic reader close starting..." ,
230229 kv .String ("reader_connection_id" , info .ReaderConnectionID ),
@@ -253,7 +252,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
253252 if d .Details ()& trace .TopicReaderStreamEvents == 0 {
254253 return nil
255254 }
256- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "reader" , "init" )
255+ ctx := with (* info . Context , TRACE , "ydb" , "topic" , "reader" , "init" )
257256 start := time .Now ()
258257 l .Log (ctx , "topic reader init starting..." ,
259258 kv .String ("pre_init_reader_connection_id" , info .PreInitReaderConnectionID ),
@@ -284,7 +283,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
284283 if d .Details ()& trace .TopicReaderStreamEvents == 0 {
285284 return
286285 }
287- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "reader" , "error" )
286+ ctx := with (* info . Context , TRACE , "ydb" , "topic" , "reader" , "error" )
288287 l .Log (WithLevel (ctx , INFO ), "topic reader has grpc stream error" ,
289288 kv .Error (info .Error ),
290289 kv .String ("reader_connection_id" , info .ReaderConnectionID ),
@@ -299,7 +298,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
299298 if d .Details ()& trace .TopicReaderStreamEvents == 0 {
300299 return nil
301300 }
302- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "reader" , "update" , "token" )
301+ ctx := with (* info . Context , TRACE , "ydb" , "topic" , "reader" , "update" , "token" )
303302 start := time .Now ()
304303 l .Log (ctx , "topic reader token update starting..." ,
305304 kv .String ("reader_connection_id" , info .ReaderConnectionID ),
@@ -347,7 +346,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
347346 if d .Details ()& trace .TopicReaderMessageEvents == 0 {
348347 return
349348 }
350- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "reader" , "sent" , "data" , "request" )
349+ ctx := with (* info . Context , TRACE , "ydb" , "topic" , "reader" , "sent" , "data" , "request" )
351350 l .Log (ctx , "topic reader sent data request" ,
352351 kv .String ("reader_connection_id" , info .ReaderConnectionID ),
353352 kv .Int ("request_bytes" , info .RequestBytes ),
@@ -360,7 +359,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
360359 if d .Details ()& trace .TopicReaderMessageEvents == 0 {
361360 return nil
362361 }
363- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "reader" , "receive" , "data" , "response" )
362+ ctx := with (* info . Context , TRACE , "ydb" , "topic" , "reader" , "receive" , "data" , "response" )
364363 start := time .Now ()
365364 partitionsCount , batchesCount , messagesCount := info .DataResponse .GetPartitionBatchMessagesCounts ()
366365 l .Log (ctx , "topic reader data response received, process starting..." ,
@@ -404,7 +403,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
404403 if d .Details ()& trace .TopicReaderMessageEvents == 0 {
405404 return nil
406405 }
407- ctx := with (* info .RequestContext , TRACE , "ydb" , "topic" , "reader" , "read" , "messages" )
406+ ctx := with (* info .Context , TRACE , "ydb" , "topic" , "reader" , "read" , "messages" )
408407 start := time .Now ()
409408 l .Log (ctx , "topic read messages, waiting..." ,
410409 kv .Int ("min_count" , info .MinCount ),
@@ -436,7 +435,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
436435 if d .Details ()& trace .TopicReaderMessageEvents == 0 {
437436 return
438437 }
439- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "reader" , "unknown" , "grpc" , "message" )
438+ ctx := with (* info . Context , TRACE , "ydb" , "topic" , "reader" , "unknown" , "grpc" , "message" )
440439 l .Log (WithLevel (ctx , INFO ), "topic reader received unknown grpc message" ,
441440 kv .Error (info .Error ),
442441 kv .String ("reader_connection_id" , info .ReaderConnectionID ),
@@ -648,7 +647,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
648647 if d .Details ()& trace .TopicWriterStreamLifeCycleEvents == 0 {
649648 return nil
650649 }
651- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "writer" , "reconnect" )
650+ ctx := with (* info . Context , TRACE , "ydb" , "topic" , "writer" , "reconnect" )
652651 start := time .Now ()
653652 l .Log (ctx , "connect to topic writer stream starting..." ,
654653 kv .String ("topic" , info .Topic ),
@@ -694,7 +693,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
694693 if d .Details ()& trace .TopicWriterStreamLifeCycleEvents == 0 {
695694 return nil
696695 }
697- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "writer" , "stream" , "init" )
696+ ctx := with (* info . Context , TRACE , "ydb" , "topic" , "writer" , "stream" , "init" )
698697 start := time .Now ()
699698 l .Log (ctx , "topic writer init stream starting..." ,
700699 kv .String ("topic" , info .Topic ),
@@ -734,7 +733,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
734733 start := time .Now ()
735734
736735 return func (doneInfo trace.TopicOnWriterBeforeCommitTransactionDoneInfo ) {
737- ctx := with (* info .Ctx , TRACE , "ydb" , "topic" , "writer" , "beforecommit" )
736+ ctx := with (* info .Context , TRACE , "ydb" , "topic" , "writer" , "beforecommit" )
738737 l .Log (ctx , "topic writer wait of flush messages before commit transaction" ,
739738 kv .String ("kqp_session_id" , info .KqpSessionID ),
740739 kv .String ("topic_session_id_start" , info .TopicSessionID ),
@@ -750,7 +749,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
750749 start := time .Now ()
751750
752751 return func (doneInfo trace.TopicOnWriterAfterFinishTransactionDoneInfo ) {
753- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "writer" , "beforecommit" )
752+ ctx := with (* info . Context , TRACE , "ydb" , "topic" , "writer" , "beforecommit" )
754753 l .Log (ctx , "topic writer close writer after transaction finished" ,
755754 kv .String ("kqp_session_id" , info .SessionID ),
756755 kv .String ("tx_id" , info .TransactionID ),
@@ -762,7 +761,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
762761 if d .Details ()& trace .TopicWriterStreamLifeCycleEvents == 0 {
763762 return nil
764763 }
765- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "writer" , "close" )
764+ ctx := with (* info . Context , TRACE , "ydb" , "topic" , "writer" , "close" )
766765 start := time .Now ()
767766 l .Log (ctx , "topic writer close starting..." ,
768767 kv .String ("writer_instance_id" , info .WriterInstanceID ),
@@ -793,7 +792,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
793792 if d .Details ()& trace .TopicWriterStreamEvents == 0 {
794793 return nil
795794 }
796- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "writer" , "compress" , "messages" )
795+ ctx := with (* info . Context , TRACE , "ydb" , "topic" , "writer" , "compress" , "messages" )
797796 start := time .Now ()
798797 l .Log (ctx , "topic writer compress messages starting..." ,
799798 kv .String ("writer_instance_id" , info .WriterInstanceID ),
@@ -836,7 +835,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
836835 if d .Details ()& trace .TopicWriterStreamEvents == 0 {
837836 return nil
838837 }
839- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "writer" , "send" , "messages" )
838+ ctx := with (* info . Context , TRACE , "ydb" , "topic" , "writer" , "send" , "messages" )
840839 start := time .Now ()
841840 l .Log (ctx , "topic writer send messages starting..." ,
842841 kv .String ("writer_instance_id" , info .WriterInstanceID ),
@@ -874,7 +873,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
874873 return
875874 }
876875 acks := info .Acks .GetAcks ()
877- ctx := with (context . Background () , DEBUG , "ydb" , "topic" , "writer" , "receive" , "result" )
876+ ctx := with (* info . Context , DEBUG , "ydb" , "topic" , "writer" , "receive" , "result" )
878877 l .Log (ctx , "topic writer received result from server" ,
879878 kv .String ("writer_instance_id" , info .WriterInstanceID ),
880879 kv .String ("session_id" , info .SessionID ),
@@ -895,7 +894,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
895894 return
896895 }
897896
898- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "writer" , "grpc" )
897+ ctx := with (* info . Context , TRACE , "ydb" , "topic" , "writer" , "grpc" )
899898 l .Log (
900899 ctx , "topic writer sent grpc message (message body and metadata are removed)" ,
901900 kv .String ("topic_stream_internal_id" , info .TopicStreamInternalID ),
@@ -911,7 +910,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
911910 return
912911 }
913912
914- ctx := with (context . Background () , TRACE , "ydb" , "topic" , "writer" , "grpc" )
913+ ctx := with (* info . Context , TRACE , "ydb" , "topic" , "writer" , "grpc" )
915914 l .Log (
916915 ctx , "topic writer received grpc message (message body and metadata are removed)" ,
917916 kv .String ("topic_stream_internal_id" , info .TopicStreamInternalID ),
@@ -926,7 +925,7 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
926925 if d .Details ()& trace .TopicWriterStreamEvents == 0 {
927926 return
928927 }
929- ctx := with (context . Background () , DEBUG , "ydb" , "topic" , "writer" , "read" , "unknown" , "grpc" , "message" )
928+ ctx := with (* info . Context , DEBUG , "ydb" , "topic" , "writer" , "read" , "unknown" , "grpc" , "message" )
930929 l .Log (ctx , "topic writer receive unknown grpc message from server" ,
931930 kv .Error (info .Error ),
932931 kv .String ("writer_instance_id" , info .WriterInstanceID ),
0 commit comments