Skip to content

Commit e0a097e

Browse files
rekbyrekby
committed
* Added traces and logs for read messages from topic within transaction
1 parent eeaec0e commit e0a097e

File tree

7 files changed

+671
-51
lines changed

7 files changed

+671
-51
lines changed

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
* Added traces and logs for read messages from topic within transaction
12
* Removed support of `go1.20`
23
* Added PopMessages from topic within transaction
34

‎internal/topic/topicreaderinternal/stream_reader_impl.go‎

Lines changed: 112 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -191,7 +191,20 @@ func (r *topicStreamReaderImpl) PopBatchTx(
191191
ctx context.Context,
192192
tx tx.Transaction,
193193
opts ReadMessageBatchOptions,
194-
) (*topicreadercommon.PublicBatch, error) {
194+
) (_ *topicreadercommon.PublicBatch, resErr error) {
195+
traceCtx := ctx
196+
onDone := trace.TopicOnReaderStreamPopBatchTx(
197+
r.cfg.Trace,
198+
&traceCtx,
199+
r.readerID,
200+
r.readConnectionID,
201+
tx.SessionID(),
202+
tx.ID(),
203+
)
204+
ctx = traceCtx
205+
defer func() {
206+
onDone(resErr)
207+
}()
195208

196209
batch, err := r.ReadMessageBatch(ctx, opts)
197210
if err != nil {
@@ -210,49 +223,45 @@ func (r *topicStreamReaderImpl) commitWithTransaction(
210223
tx tx.Transaction,
211224
batch *topicreadercommon.PublicBatch,
212225
) error {
213-
commitRange := topicreadercommon.GetCommitRange(batch)
226+
req := r.createUpdateOffsetRequest(ctx, batch, tx)
227+
214228
updateOffesetInTransactionErr := retry.Retry(ctx, func(ctx context.Context) (err error) {
215-
err = r.topicClient.UpdateOffsetsInTransaction(ctx, &rawtopic.UpdateOffsetsInTransactionRequest{
216-
OperationParams: rawydb.NewRawOperationParamsFromProto(operation.Params(ctx, 0, 0, operation.ModeSync)),
217-
Tx: rawtopic.UpdateOffsetsInTransactionRequest_TransactionIdentity{
218-
ID: tx.ID(),
219-
Session: tx.SessionID(),
220-
},
221-
Topics: []rawtopic.UpdateOffsetsInTransactionRequest_TopicOffsets{
222-
{
223-
Path: batch.Topic(),
224-
Partitions: []rawtopic.UpdateOffsetsInTransactionRequest_PartitionOffsets{
225-
{
226-
PartitionID: batch.PartitionID(),
227-
PartitionOffsets: []rawtopiccommon.OffsetRange{
228-
{
229-
Start: commitRange.CommitOffsetStart,
230-
End: commitRange.CommitOffsetEnd,
231-
},
232-
},
233-
},
234-
},
235-
},
236-
},
237-
Consumer: r.cfg.Consumer,
238-
})
229+
traceCtx := ctx
230+
onDone := trace.TopicOnReaderUpdateOffsetsInTransaction(
231+
r.cfg.Trace,
232+
&traceCtx,
233+
r.readerID,
234+
r.readConnectionID,
235+
tx.SessionID(),
236+
tx.ID(),
237+
)
238+
defer func() {
239+
onDone(err)
240+
}()
241+
242+
ctx = traceCtx
243+
err = r.topicClient.UpdateOffsetsInTransaction(ctx, req)
239244

240245
return err
241246
})
242247
if updateOffesetInTransactionErr == nil {
243-
tx.OnCompleted(func(transactionResult error) {
244-
//nolint:godox
245-
// TODO: trace
246-
if transactionResult == nil {
247-
topicreadercommon.BatchGetPartitionSession(batch).SetCommittedOffsetForward(commitRange.CommitOffsetEnd)
248-
} else {
249-
_ = r.CloseWithError(xcontext.ValueOnly(ctx), xerrors.WithStackTrace(xerrors.RetryableError(
250-
fmt.Errorf("ydb: failed batch commit because transaction doesn't committed: %w", updateOffesetInTransactionErr),
251-
)))
252-
}
253-
})
248+
r.addOnTransactionCompletedHandler(ctx, tx, batch, updateOffesetInTransactionErr)
254249
} else {
255250
_ = retry.Retry(ctx, func(ctx context.Context) (err error) {
251+
traceCtx := ctx
252+
onDone := trace.TopicOnReaderTransactionRollback(
253+
r.cfg.Trace,
254+
&traceCtx,
255+
r.readerID,
256+
r.readConnectionID,
257+
tx.SessionID(),
258+
tx.ID(),
259+
)
260+
ctx = traceCtx
261+
defer func() {
262+
onDone(err)
263+
}()
264+
256265
return tx.Rollback(ctx)
257266
})
258267

@@ -266,6 +275,72 @@ func (r *topicStreamReaderImpl) commitWithTransaction(
266275
return nil
267276
}
268277

278+
func (r *topicStreamReaderImpl) addOnTransactionCompletedHandler(
279+
ctx context.Context,
280+
tx tx.Transaction,
281+
batch *topicreadercommon.PublicBatch,
282+
updateOffesetInTransactionErr error,
283+
) {
284+
commitRange := topicreadercommon.GetCommitRange(batch)
285+
tx.OnCompleted(func(transactionResult error) {
286+
traceCtx := ctx
287+
onDone := trace.TopicOnReaderTransactionCompleted(
288+
r.cfg.Trace,
289+
&traceCtx,
290+
r.readerID,
291+
r.readConnectionID,
292+
tx.SessionID(),
293+
tx.ID(),
294+
transactionResult,
295+
)
296+
defer onDone()
297+
298+
ctx = traceCtx
299+
//nolint:godox
300+
// TODO: trace
301+
if transactionResult == nil {
302+
topicreadercommon.BatchGetPartitionSession(batch).SetCommittedOffsetForward(commitRange.CommitOffsetEnd)
303+
} else {
304+
_ = r.CloseWithError(xcontext.ValueOnly(ctx), xerrors.WithStackTrace(xerrors.RetryableError(
305+
fmt.Errorf("ydb: failed batch commit because transaction doesn't committed: %w", updateOffesetInTransactionErr),
306+
)))
307+
}
308+
})
309+
}
310+
311+
func (r *topicStreamReaderImpl) createUpdateOffsetRequest(
312+
ctx context.Context,
313+
batch *topicreadercommon.PublicBatch,
314+
tx tx.Transaction,
315+
) *rawtopic.UpdateOffsetsInTransactionRequest {
316+
commitRange := topicreadercommon.GetCommitRange(batch)
317+
318+
return &rawtopic.UpdateOffsetsInTransactionRequest{
319+
OperationParams: rawydb.NewRawOperationParamsFromProto(operation.Params(ctx, 0, 0, operation.ModeSync)),
320+
Tx: rawtopic.UpdateOffsetsInTransactionRequest_TransactionIdentity{
321+
ID: tx.ID(),
322+
Session: tx.SessionID(),
323+
},
324+
Topics: []rawtopic.UpdateOffsetsInTransactionRequest_TopicOffsets{
325+
{
326+
Path: batch.Topic(),
327+
Partitions: []rawtopic.UpdateOffsetsInTransactionRequest_PartitionOffsets{
328+
{
329+
PartitionID: batch.PartitionID(),
330+
PartitionOffsets: []rawtopiccommon.OffsetRange{
331+
{
332+
Start: commitRange.CommitOffsetStart,
333+
End: commitRange.CommitOffsetEnd,
334+
},
335+
},
336+
},
337+
},
338+
},
339+
},
340+
Consumer: r.cfg.Consumer,
341+
}
342+
}
343+
269344
func (r *topicStreamReaderImpl) ReadMessageBatch(
270345
ctx context.Context,
271346
opts ReadMessageBatchOptions,

‎log/topic.go‎

Lines changed: 196 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -438,6 +438,202 @@ func internalTopic(l Logger, d trace.Detailer) (t trace.Topic) {
438438
)
439439
}
440440

441+
t.OnReaderPopBatchTx = func(
442+
startInfo trace.TopicReaderPopBatchTxStartInfo,
443+
) func(trace.TopicReaderPopBatchTxDoneInfo) {
444+
if d.Details()&trace.TopicReaderCustomerEvents == 0 {
445+
return nil
446+
}
447+
448+
start := time.Now()
449+
ctx := with(*startInfo.Context, TRACE, "ydb", "topic", "reader", "customer", "popbatchtx")
450+
l.Log(WithLevel(ctx, TRACE), "starting pop batch tx",
451+
Int64("reader_id", startInfo.ReaderID),
452+
String("transaction_session_id", startInfo.TransactionSessionID),
453+
String("transaction_id", startInfo.TransactionID),
454+
)
455+
456+
return func(doneInfo trace.TopicReaderPopBatchTxDoneInfo) {
457+
if doneInfo.Error == nil {
458+
l.Log(
459+
WithLevel(ctx, DEBUG), "pop batch done",
460+
Int64("reader_id", startInfo.ReaderID),
461+
String("transaction_session_id", startInfo.TransactionSessionID),
462+
String("transaction_id", startInfo.TransactionID),
463+
Int("messaged_count", doneInfo.MessagesCount),
464+
Int64("start_offset", doneInfo.StartOffset),
465+
Int64("end_offset", doneInfo.EndOffset),
466+
latencyField(start),
467+
versionField(),
468+
)
469+
} else {
470+
l.Log(
471+
WithLevel(ctx, WARN), "pop batch failed",
472+
Int64("reader_id", startInfo.ReaderID),
473+
String("transaction_session_id", startInfo.TransactionSessionID),
474+
String("transaction_id", startInfo.TransactionID),
475+
Error(doneInfo.Error),
476+
latencyField(start),
477+
versionField(),
478+
)
479+
}
480+
}
481+
}
482+
483+
t.OnReaderStreamPopBatchTx = func(
484+
startInfo trace.TopicReaderStreamPopBatchTxStartInfo,
485+
) func(
486+
trace.TopicReaderStreamPopBatchTxDoneInfo,
487+
) {
488+
if d.Details()&trace.TopicReaderTransactionEvents == 0 {
489+
return nil
490+
}
491+
492+
start := time.Now()
493+
ctx := with(*startInfo.Context, TRACE, "ydb", "topic", "reader", "transaction", "popbatchtx_on_stream")
494+
l.Log(WithLevel(ctx, TRACE), "starting pop batch tx",
495+
Int64("reader_id", startInfo.ReaderID),
496+
String("reader_connection_id", startInfo.ReaderConnectionID),
497+
String("transaction_session_id", startInfo.TransactionSessionID),
498+
String("transaction_id", startInfo.TransactionID),
499+
versionField(),
500+
)
501+
502+
return func(doneInfo trace.TopicReaderStreamPopBatchTxDoneInfo) {
503+
if doneInfo.Error == nil {
504+
l.Log(
505+
WithLevel(ctx, DEBUG), "pop batch on stream done",
506+
Int64("reader_id", startInfo.ReaderID),
507+
String("transaction_session_id", startInfo.TransactionSessionID),
508+
String("transaction_id", startInfo.TransactionID),
509+
latencyField(start),
510+
versionField(),
511+
)
512+
} else {
513+
l.Log(
514+
WithLevel(ctx, WARN), "pop batch on stream failed",
515+
Int64("reader_id", startInfo.ReaderID),
516+
String("transaction_session_id", startInfo.TransactionSessionID),
517+
String("transaction_id", startInfo.TransactionID),
518+
Error(doneInfo.Error),
519+
latencyField(start),
520+
versionField(),
521+
)
522+
}
523+
}
524+
}
525+
526+
t.OnReaderUpdateOffsetsInTransaction = func(
527+
startInfo trace.TopicReaderOnUpdateOffsetsInTransactionStartInfo,
528+
) func(
529+
trace.TopicReaderOnUpdateOffsetsInTransactionDoneInfo,
530+
) {
531+
if d.Details()&trace.TopicReaderTransactionEvents == 0 {
532+
return nil
533+
}
534+
535+
start := time.Now()
536+
ctx := with(*startInfo.Context, TRACE, "ydb", "topic", "reader", "transaction", "update_offsets")
537+
l.Log(WithLevel(ctx, TRACE), "starting update offsets in transaction",
538+
Int64("reader_id", startInfo.ReaderID),
539+
String("reader_connection_id", startInfo.ReaderConnectionID),
540+
String("transaction_session_id", startInfo.TransactionSessionID),
541+
String("transaction_id", startInfo.TransactionID),
542+
versionField(),
543+
)
544+
545+
return func(doneInfo trace.TopicReaderOnUpdateOffsetsInTransactionDoneInfo) {
546+
if doneInfo.Error == nil {
547+
l.Log(
548+
WithLevel(ctx, DEBUG), "pop batch on stream done",
549+
Int64("reader_id", startInfo.ReaderID),
550+
String("transaction_session_id", startInfo.TransactionSessionID),
551+
String("transaction_id", startInfo.TransactionID),
552+
latencyField(start),
553+
versionField(),
554+
)
555+
} else {
556+
l.Log(
557+
WithLevel(ctx, WARN), "pop batch on stream failed",
558+
Int64("reader_id", startInfo.ReaderID),
559+
String("transaction_session_id", startInfo.TransactionSessionID),
560+
String("transaction_id", startInfo.TransactionID),
561+
Error(doneInfo.Error),
562+
latencyField(start),
563+
versionField(),
564+
)
565+
}
566+
}
567+
}
568+
569+
t.OnReaderTransactionRollback = func(
570+
startInfo trace.TopicReaderTransactionRollbackStartInfo,
571+
) func(
572+
trace.TopicReaderTransactionRollbackDoneInfo,
573+
) {
574+
if d.Details()&trace.TopicReaderTransactionEvents == 0 {
575+
return nil
576+
}
577+
578+
start := time.Now()
579+
ctx := with(*startInfo.Context, TRACE, "ydb", "topic", "reader", "transaction", "update_offsets")
580+
l.Log(WithLevel(ctx, TRACE), "starting update offsets in transaction",
581+
Int64("reader_id", startInfo.ReaderID),
582+
String("reader_connection_id", startInfo.ReaderConnectionID),
583+
String("transaction_session_id", startInfo.TransactionSessionID),
584+
String("transaction_id", startInfo.TransactionID),
585+
versionField(),
586+
)
587+
588+
return func(doneInfo trace.TopicReaderTransactionRollbackDoneInfo) {
589+
if doneInfo.RollbackError == nil {
590+
l.Log(
591+
WithLevel(ctx, DEBUG), "pop batch on stream done",
592+
Int64("reader_id", startInfo.ReaderID),
593+
String("transaction_session_id", startInfo.TransactionSessionID),
594+
String("transaction_id", startInfo.TransactionID),
595+
latencyField(start),
596+
versionField(),
597+
)
598+
} else {
599+
l.Log(
600+
WithLevel(ctx, WARN), "pop batch on stream failed",
601+
Int64("reader_id", startInfo.ReaderID),
602+
String("transaction_session_id", startInfo.TransactionSessionID),
603+
String("transaction_id", startInfo.TransactionID),
604+
Error(doneInfo.RollbackError),
605+
latencyField(start),
606+
versionField(),
607+
)
608+
}
609+
}
610+
}
611+
612+
t.OnReaderTransactionCompleted = func(
613+
startInfo trace.TopicReaderTransactionCompletedStartInfo,
614+
) func(
615+
trace.TopicReaderTransactionCompletedDoneInfo,
616+
) {
617+
if d.Details()&trace.TopicReaderTransactionEvents == 0 {
618+
return nil
619+
}
620+
621+
// expected as very short in memory operation without errors, no need log start separately
622+
start := time.Now()
623+
624+
return func(doneInfo trace.TopicReaderTransactionCompletedDoneInfo) {
625+
ctx := with(*startInfo.Context, TRACE, "ydb", "topic", "reader", "transaction", "update_offsets")
626+
l.Log(WithLevel(ctx, TRACE), "starting update offsets in transaction",
627+
Int64("reader_id", startInfo.ReaderID),
628+
String("reader_connection_id", startInfo.ReaderConnectionID),
629+
String("transaction_session_id", startInfo.TransactionSessionID),
630+
String("transaction_id", startInfo.TransactionID),
631+
latencyField(start),
632+
versionField(),
633+
)
634+
}
635+
}
636+
441637
///
442638
/// Topic writer
443639
///

0 commit comments

Comments
 (0)