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/SentMessage.java b/topic/src/main/java/tech/ydb/topic/write/impl/SentMessage.java index 53b840062..c96bba406 100644 --- a/topic/src/main/java/tech/ydb/topic/write/impl/SentMessage.java +++ b/topic/src/main/java/tech/ydb/topic/write/impl/SentMessage.java @@ -17,6 +17,8 @@ public class SentMessage implements WriterQueue.EncodedMsg { private final long seqNo; private final long bufferSize; + private final long sendTimestamp = System.nanoTime(); + private final long acceptedTimestamp; private final YdbTransaction tx; private final YdbTopic.StreamWriteMessage.WriteRequest.MessageData pb; private final CompletableFuture ackFuture; @@ -24,6 +26,7 @@ public class SentMessage implements WriterQueue.EncodedMsg { public SentMessage(EnqueuedMessage msg, long seqNo) { this.seqNo = seqNo; this.bufferSize = msg.getBufferSize(); + this.acceptedTimestamp = msg.getAcceptedTimestamp(); this.tx = msg.getMeta().getTransaction(); this.pb = buildPb(seqNo, msg); this.ackFuture = msg.getAckFuture(); @@ -42,6 +45,14 @@ public long getBufferSize() { return bufferSize; } + long getSendTimestamp() { + return sendTimestamp; + } + + long getAcceptedTimestamp() { + return acceptedTimestamp; + } + @Override public void confirm(WriteAck ack) { this.ackFuture.complete(ack); 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..29cfbc8a1 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,6 +6,7 @@ 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; @@ -20,13 +21,18 @@ final class WriterMetrics { private final LongCounter sendingBytes; private final LongCounter writtenMessages; private final LongCounter sessionErrors; + private final DoubleHistogram messageAckDuration; + private final DoubleHistogram bufferWaitDuration; private final Attr[] commonAttributes; private final Meter meter; + private final boolean enabled; private MetricRegistration bufferUsedGauge = MetricRegistration.NOOP; private MetricRegistration bufferLimitGauge = MetricRegistration.NOOP; + private MetricRegistration oldestAgeGauge = 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,25 +41,49 @@ 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.messageAckDuration = meter.createHistogram("ydb.topic.writer.message.ack.duration", "s", + "Time from first sending a message to the server to its acknowledgement."); + this.bufferWaitDuration = meter.createHistogram("ydb.topic.writer.buffer.wait.duration", "s", + "Time waiting for buffer capacity before a message is accepted."); this.commonAttributes = writerName == null ? new Attr[0] : new Attr[]{Attr.of("topic", topic), Attr.of("writer.name", writerName)}; } - void register(LongSupplier bufferUsed, LongSupplier bufferLimit) { + void register(LongSupplier bufferUsed, LongSupplier bufferLimit, LongSupplier oldestTimestamp) { bufferUsedGauge = meter.registerLongGauge("ydb.topic.writer.buffer.used.bytes", "By", "The occupied budget of the writer buffer limiter.", m -> m.record(bufferUsed.getAsLong(), commonAttributes)); bufferLimitGauge = meter.registerLongGauge("ydb.topic.writer.buffer.limit.bytes", "By", "The configured limit of the writer buffer limiter.", m -> m.record(bufferLimit.getAsLong(), commonAttributes)); + oldestAgeGauge = meter.registerDoubleGauge("ydb.topic.writer.sending.oldest_age", "s", + "The age of the oldest message in the writer's in-flight buffer.", m -> { + long timestamp = oldestTimestamp.getAsLong(); + m.record(timestamp == 0 ? 0 : (System.nanoTime() - timestamp) / 1_000_000_000d, + commonAttributes); + }); } void unregister() { bufferUsedGauge.close(); bufferLimitGauge.close(); + oldestAgeGauge.close(); bufferUsedGauge = MetricRegistration.NOOP; bufferLimitGauge = MetricRegistration.NOOP; + oldestAgeGauge = MetricRegistration.NOOP; + } + + void reportMessageAckDuration(long timestamp) { + if (enabled) { + messageAckDuration.record((System.nanoTime() - timestamp) / 1_000_000_000d, commonAttributes); + } + } + + void reportBufferWaitDuration(long timestamp) { + if (enabled) { + bufferWaitDuration.record((System.nanoTime() - timestamp) / 1_000_000_000d, commonAttributes); + } } void reportSending(long bytes) { 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..e0c918e5c 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 @@ -80,7 +80,17 @@ public WriterQueue(String debugId, WriterSettings settings, WriterMetrics metric this.compressionExecutor = compressionExecutor; this.readyNotify = readyNotify; this.metrics = metrics; - this.metrics.register(buffer::getUsedSize, buffer::getMaxSize); + this.metrics.register(buffer::getUsedSize, buffer::getMaxSize, this::getOlderAcceptedTimestamp); + } + + private long getOlderAcceptedTimestamp() { + for (EncodedMsg msg : sent) { + SentMessage sentMsg = msg.getSentMessage(); + if (sentMsg != null) { + return sentMsg.getAcceptedTimestamp(); + } + } + return 0; } CompletableFuture flush() { @@ -159,6 +169,7 @@ void confirmAck(WriteAck ack) { sentIt.remove(); buffer.releaseMessage(sentMsg.getBufferSize()); metrics.reportWritten(); + metrics.reportMessageAckDuration(sentMsg.getSendTimestamp()); msg.confirm(ack); } } @@ -206,7 +217,9 @@ List updateSeqNo(long newSeqNo) { CompletableFuture enqueue(Message message, YdbTransaction tx) throws QueueOverflowException, InterruptedException { long reservedSizeBytes = reservationSizeBytes(message.getData().length); + long waitStarted = System.nanoTime(); buffer.acquire(reservedSizeBytes); + metrics.reportBufferWaitDuration(waitStarted); return accept(message, tx, reservedSizeBytes); } @@ -219,7 +232,9 @@ CompletableFuture tryEnqueue(Message message, YdbTransaction tx) throw CompletableFuture tryEnqueue(Message message, YdbTransaction tx, long timeout, TimeUnit unit) throws QueueOverflowException, InterruptedException, TimeoutException { long reservedSizeBytes = reservationSizeBytes(message.getData().length); + long waitStarted = System.nanoTime(); buffer.tryAcquire(reservedSizeBytes, timeout, unit); + metrics.reportBufferWaitDuration(waitStarted); return accept(message, tx, reservedSizeBytes); }