diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java index 632007f37..242d5244e 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java @@ -75,7 +75,8 @@ private enum State { this.controlExecutor = config.getControlExecutor(); this.committer = new ReadPartitionCommitter(traceID, session, partition, lastCommittedOffset); - this.decoder = new ReadPartitionDecoder(traceID, sessionDecoder, partition, committer, this::sendDataToReaders); + this.decoder = new ReadPartitionDecoder(traceID, sessionDecoder, partition, committer, + this::sendDataToReaders, metrics); this.queue = new ReadPartitionQueue(traceID, decoder, config.getMaxBatchSize(), lastCommittedOffset); logger.info("[{}] started for {}", traceID, partition); diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionDecoder.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionDecoder.java index b7234a42b..bdf60891d 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionDecoder.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionDecoder.java @@ -12,6 +12,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import tech.ydb.core.metrics.Meter; import tech.ydb.proto.topic.YdbTopic; import tech.ydb.topic.description.Codec; import tech.ydb.topic.description.CodecRegistry; @@ -31,6 +32,7 @@ class ReadPartitionDecoder { private final PartitionSession partition; private final MessageCommitter committer; private final MessageDecoder decoder; + private final ReaderMetrics metrics; private final Runnable readyHandler; private final AtomicLong allocatedTotal = new AtomicLong(0); @@ -39,8 +41,14 @@ class ReadPartitionDecoder { ReadPartitionDecoder(String traceId, MessageDecoder decoder, PartitionSession partition, MessageCommitter committer, Runnable readyHandler) { + this(traceId, decoder, partition, committer, readyHandler, new ReaderMetrics(Meter.NOOP, null, null)); + } + + ReadPartitionDecoder(String traceId, MessageDecoder decoder, PartitionSession partition, + MessageCommitter committer, Runnable readyHandler, ReaderMetrics metrics) { this.traceID = traceId; this.decoder = decoder; + this.metrics = metrics; this.partition = partition; this.committer = committer; this.readyHandler = readyHandler; @@ -181,6 +189,7 @@ public void decode(CodecRegistry registry) { return; } + long compressionStarted = 0; try { Codec codec = registry.getCodec(codecCode); if (codec == null) { @@ -190,6 +199,7 @@ public void decode(CodecRegistry registry) { } logger.trace("[{}] Started decoding batch", traceID); + compressionStarted = System.nanoTime(); try (InputStream encoded = origin.newInput(); InputStream decoded = codec.decode(encoded)) { data = ByteStreams.toByteArray(decoded); logger.trace("[{}] Finished decoding batch", traceID); @@ -201,6 +211,8 @@ public void decode(CodecRegistry registry) { problem = new IOException("Cannot decode message", ex); } } finally { + metrics.reportDecoded(compressionStarted == 0 ? 0 : System.nanoTime() - compressionStarted, + System.nanoTime() - getReceivedAt(), codecCode, partition.getPath(), problem); if (problem == null) { origin = null; } else { diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReaderMetrics.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReaderMetrics.java index 1ef203108..97f639eb4 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReaderMetrics.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReaderMetrics.java @@ -6,9 +6,11 @@ import tech.ydb.core.Status; import tech.ydb.core.StatusCode; import tech.ydb.core.metrics.Attr; +import tech.ydb.core.metrics.DoubleHistogram; import tech.ydb.core.metrics.LongCounter; import tech.ydb.core.metrics.Meter; import tech.ydb.core.metrics.MetricRegistration; +import tech.ydb.proto.topic.YdbTopic; /** * Topic reader metrics. @@ -22,6 +24,8 @@ final class ReaderMetrics { private final LongCounter commitQueued; private final LongCounter commitAcknowledged; private final LongCounter sessionErrors; + private final DoubleHistogram compressionDuration; + private final DoubleHistogram processingDuration; private final Attr[] commonAttributes; private final boolean enabled; private final Meter meter; @@ -48,6 +52,10 @@ final class ReaderMetrics { "The number of messages acknowledged by advances of the committed offset."); this.sessionErrors = meter.createCounter("ydb.topic.reader.session.errors", "{error}", "The number of reader stream session errors by retry decision."); + this.compressionDuration = meter.createHistogram("ydb.topic.reader.compression.duration", "s", + "The time spent decompressing a message, excluding queueing."); + this.processingDuration = meter.createHistogram("ydb.topic.reader.message.processing.duration", "s", + "The time from receipt until a message is ready for delivery."); this.commonAttributes = createCommonAttributes(consumer, readerName); } @@ -122,6 +130,27 @@ void reportCommitAcknowledged(long messages, String topic) { report(commitAcknowledged, messages, topic); } + void reportDecoded(long compressionNanos, long processingNanos, int codec, String topic, Throwable error) { + recordDuration(compressionDuration, compressionNanos, codec, topic, error); + recordDuration(processingDuration, processingNanos, codec, topic, error); + } + + private void recordDuration(DoubleHistogram histogram, long durationNanos, int codec, String topic, + Throwable error) { + if (enabled) { + Attr[] attributes = Arrays.copyOf(commonAttributes, commonAttributes.length + (error == null ? 3 : 4)); + attributes[commonAttributes.length] = Attr.of("topic", topic); + YdbTopic.Codec codecType = YdbTopic.Codec.forNumber(codec); + attributes[commonAttributes.length + 1] = Attr.of("codec", codecType == null ? Integer.toString(codec) + : codecType.name().substring("CODEC_".length()).toLowerCase(java.util.Locale.ROOT)); + attributes[commonAttributes.length + 2] = Attr.of("outcome", error == null ? "ok" : "error"); + if (error != null) { + attributes[commonAttributes.length + 3] = Attr.of("error.type", error.getClass().getName()); + } + histogram.record(durationNanos / 1_000_000_000d, attributes); + } + } + private void report(LongCounter counter, long messages, String topic) { if (enabled && messages > 0) { record(counter, messages, Attr.of("topic", topic)); diff --git a/topic/src/main/java/tech/ydb/topic/write/impl/EnqueuedMessage.java b/topic/src/main/java/tech/ydb/topic/write/impl/EnqueuedMessage.java index 309fa7a33..c7edfc95d 100644 --- a/topic/src/main/java/tech/ydb/topic/write/impl/EnqueuedMessage.java +++ b/topic/src/main/java/tech/ydb/topic/write/impl/EnqueuedMessage.java @@ -12,6 +12,7 @@ */ public class EnqueuedMessage { private final MessageMeta meta; + private final long acceptedTimestamp = System.nanoTime(); private final CompletableFuture ackFuture = new CompletableFuture<>(); private volatile ByteString data = null; @@ -29,6 +30,10 @@ public MessageMeta getMeta() { return meta; } + long getAcceptedTimestamp() { + return acceptedTimestamp; + } + public ByteString getData() { return data; } diff --git a/topic/src/main/java/tech/ydb/topic/write/impl/WriterMetrics.java b/topic/src/main/java/tech/ydb/topic/write/impl/WriterMetrics.java index 3fc70f6da..f123df019 100644 --- a/topic/src/main/java/tech/ydb/topic/write/impl/WriterMetrics.java +++ b/topic/src/main/java/tech/ydb/topic/write/impl/WriterMetrics.java @@ -6,9 +6,11 @@ import tech.ydb.core.Status; import tech.ydb.core.StatusCode; import tech.ydb.core.metrics.Attr; +import tech.ydb.core.metrics.DoubleHistogram; import tech.ydb.core.metrics.LongCounter; import tech.ydb.core.metrics.Meter; import tech.ydb.core.metrics.MetricRegistration; +import tech.ydb.proto.topic.YdbTopic; /** * Topic writer metrics. @@ -20,13 +22,17 @@ final class WriterMetrics { private final LongCounter sendingBytes; private final LongCounter writtenMessages; private final LongCounter sessionErrors; + private final DoubleHistogram compressionDuration; + private final DoubleHistogram processingDuration; private final Attr[] commonAttributes; + private final boolean enabled; private final Meter meter; private MetricRegistration bufferUsedGauge = MetricRegistration.NOOP; private MetricRegistration bufferLimitGauge = MetricRegistration.NOOP; WriterMetrics(Meter meter, String topic, String writerName) { this.meter = meter; + this.enabled = meter != Meter.NOOP; this.sendingMessages = meter.createCounter("ydb.topic.writer.sending.messages", MESSAGE_UNIT, "The number of messages accepted by the SDK for sending."); this.sendingBytes = meter.createCounter("ydb.topic.writer.sending.bytes", "By", @@ -35,6 +41,10 @@ final class WriterMetrics { "The number of messages confirmed written by the server, including already written messages."); this.sessionErrors = meter.createCounter("ydb.topic.writer.session.errors", "{error}", "The number of writer stream session errors by retry decision."); + this.compressionDuration = meter.createHistogram("ydb.topic.writer.compression.duration", "s", + "The time spent compressing a message, excluding queueing."); + this.processingDuration = meter.createHistogram("ydb.topic.writer.message.processing.duration", "s", + "The time from buffer acceptance until a message is ready for sending."); this.commonAttributes = writerName == null ? new Attr[0] : new Attr[]{Attr.of("topic", topic), Attr.of("writer.name", writerName)}; @@ -65,6 +75,28 @@ void reportWritten() { writtenMessages.add(1, commonAttributes); } + void reportCompression(long durationNanos, int codec, Throwable error) { + recordDuration(compressionDuration, durationNanos, codec, error); + } + + void reportProcessing(long durationNanos, int codec, Throwable error) { + recordDuration(processingDuration, durationNanos, codec, error); + } + + private void recordDuration(DoubleHistogram histogram, long durationNanos, int codec, Throwable error) { + if (enabled) { + Attr[] attributes = Arrays.copyOf(commonAttributes, commonAttributes.length + (error == null ? 2 : 3)); + YdbTopic.Codec codecType = YdbTopic.Codec.forNumber(codec); + attributes[commonAttributes.length] = Attr.of("codec", codecType == null ? Integer.toString(codec) + : codecType.name().substring("CODEC_".length()).toLowerCase(java.util.Locale.ROOT)); + attributes[commonAttributes.length + 1] = Attr.of("outcome", error == null ? "ok" : "error"); + if (error != null) { + attributes[commonAttributes.length + 2] = Attr.of("error.type", error.getClass().getName()); + } + histogram.record(durationNanos / 1_000_000_000d, attributes); + } + } + void reportSessionError(Status status, boolean retry) { StatusCode code = status.getCode(); Attr[] attributes = Arrays.copyOf(commonAttributes, commonAttributes.length + 3); diff --git a/topic/src/main/java/tech/ydb/topic/write/impl/WriterQueue.java b/topic/src/main/java/tech/ydb/topic/write/impl/WriterQueue.java index f08a91034..23eec8f60 100644 --- a/topic/src/main/java/tech/ydb/topic/write/impl/WriterQueue.java +++ b/topic/src/main/java/tech/ydb/topic/write/impl/WriterQueue.java @@ -239,7 +239,9 @@ private CompletableFuture accept(Message message, YdbTransaction tx, l if (codec.getId() == Codec.RAW) { // fast track without compression - msg.completeWithData(UnsafeByteOperations.unsafeWrap(message.getData()), reservedSizeBytes); + ByteString data = UnsafeByteOperations.unsafeWrap(message.getData()); + metrics.reportProcessing(System.nanoTime() - msg.getAcceptedTimestamp(), codec.getId(), null); + msg.completeWithData(data, reservedSizeBytes); readyNotify.run(); return msg.getAckFuture(); } @@ -247,6 +249,7 @@ private CompletableFuture accept(Message message, YdbTransaction tx, l try { compressionExecutor.execute(() -> encode(message.getData(), reservedSizeBytes, msg)); } catch (Throwable ex) { + metrics.reportProcessing(System.nanoTime() - msg.getAcceptedTimestamp(), codec.getId(), ex); logger.warn("[{}] Message wasn't sent because of processing error", debugId, ex); msg.completeWithProblem(ex); readyNotify.run(); @@ -262,9 +265,16 @@ private void encode(byte[] data, long reservedSize, EnqueuedMessage msg) { logger.trace("[{}] Started encoding message", debugId); try (ByteString.Output encoded = ByteString.newOutput()) { - try (OutputStream os = codec.encode(encoded)) { - os.write(data, 0, data.length); + long compressionStarted = System.nanoTime(); + try { + try (OutputStream os = codec.encode(encoded)) { + os.write(data, 0, data.length); + } + } catch (Throwable ex) { + metrics.reportCompression(System.nanoTime() - compressionStarted, codec.getId(), ex); + throw ex; } + metrics.reportCompression(System.nanoTime() - compressionStarted, codec.getId(), null); if (logger.isTraceEnabled()) { logger.trace("[{}] Message compressed from {} to {} bytes", debugId, data.length, encoded.size()); @@ -286,8 +296,10 @@ private void encode(byte[] data, long reservedSize, EnqueuedMessage msg) { } bufferSizeBytes = updatedBufferSizeBytes; + metrics.reportProcessing(System.nanoTime() - msg.getAcceptedTimestamp(), codec.getId(), null); msg.completeWithData(encodedData, bufferSizeBytes); } catch (Throwable ex) { + metrics.reportProcessing(System.nanoTime() - msg.getAcceptedTimestamp(), codec.getId(), ex); logger.warn("[{}] Message wasn't sent because of encoding error", debugId, ex); msg.completeWithProblem(ex); }