Add Topic Writer metrics

topic/src/main/java/tech/ydb/topic/write/impl/BufferManager.java

@@ -60,11 +60,21 @@ public void close(Status status) {
     }
 
     public void acquire(long messageSize) throws InterruptedException, QueueOverflowException {
+        acquire(messageSize, null);
+    }
+
+    void acquire(long messageSize, WriterMetrics metrics) throws InterruptedException, QueueOverflowException {
         if (closed != null) {
             throw new IllegalStateException("Writer was closed with status " + closed);
         }
 
-        countAvailable.acquire();
+        long waitStarted = 0;
+        if (!countAvailable.tryAcquire(0, TimeUnit.NANOSECONDS)) {
+            if (metrics != null) {
+                waitStarted = metrics.reportBufferWaitStart();
+            }
+            countAvailable.acquire();
+        }
 
         if (closed != null) {
             countAvailable.release();
@@ -74,7 +84,12 @@ public void acquire(long messageSize) throws InterruptedException, QueueOverflow
         int messageBlocks = calculateBlocksCount(messageSize, blockBitsCount);
 
         try {
-            blocksAvailable.acquire(messageBlocks);
+            if (!blocksAvailable.tryAcquire(messageBlocks, 0, TimeUnit.NANOSECONDS)) {
+                if (waitStarted == 0 && metrics != null) {
+                    waitStarted = metrics.reportBufferWaitStart();
+                }
+                blocksAvailable.acquire(messageBlocks);
+            }
         } catch (InterruptedException ex) {
             countAvailable.release();
             throw ex;
@@ -85,6 +100,9 @@ public void acquire(long messageSize) throws InterruptedException, QueueOverflow
             countAvailable.release();
             throw new IllegalStateException("Writer was closed with status " + closed);
         }
+        if (metrics != null) {
+            metrics.reportBufferWaitDuration(waitStarted);
+        }
     }
 
     public void tryAcquire(long messageSize) throws QueueOverflowException {
@@ -125,12 +143,25 @@ public void tryAcquire(long messageSize) throws QueueOverflowException {
 
     public void tryAcquire(long messageSize, long timeout, TimeUnit unit) throws InterruptedException,
             QueueOverflowException, TimeoutException {
+        tryAcquire(messageSize, timeout, unit, null);
+    }
+
+    void tryAcquire(long messageSize, long timeout, TimeUnit unit, WriterMetrics metrics) throws InterruptedException,
+            QueueOverflowException, TimeoutException {
         if (closed != null) {
             throw new IllegalStateException("Writer was closed with status " + closed);
         }
 
         long expireAt = System.nanoTime() + unit.toNanos(timeout);
-        if (!countAvailable.tryAcquire(timeout, unit)) {
+        long waitStarted = 0;
+        boolean countAcquired = countAvailable.tryAcquire(0, TimeUnit.NANOSECONDS);
+        if (!countAcquired) {
+            if (metrics != null) {
+                waitStarted = metrics.reportBufferWaitStart();
+            }
+            countAcquired = countAvailable.tryAcquire(timeout, unit);
+        }
+        if (!countAcquired) {
             String errorMsg = "[" + debugId + "] Rejecting a message due to reaching message queue in-flight limit of "
                     + maxCount;
             logger.warn(errorMsg);
@@ -147,7 +178,14 @@ public void tryAcquire(long messageSize, long timeout, TimeUnit unit) throws Int
         try {
             // negative timeout is allowed for tryAcquire
             long timeout2 = expireAt - System.nanoTime();
-            if (!blocksAvailable.tryAcquire(messageBlocks, timeout2, TimeUnit.NANOSECONDS)) {
+            boolean blocksAcquired = blocksAvailable.tryAcquire(messageBlocks, 0, TimeUnit.NANOSECONDS);
+            if (!blocksAcquired) {
+                if (waitStarted == 0 && metrics != null) {
+                    waitStarted = metrics.reportBufferWaitStart();
+                }
+                blocksAcquired = blocksAvailable.tryAcquire(messageBlocks, timeout2, TimeUnit.NANOSECONDS);
+            }
+            if (!blocksAcquired) {
                 countAvailable.release();
                 int count = maxCount - countAvailable.availablePermits();
                 long size = ((long) blocksAvailable.availablePermits()) << blockBitsCount;
@@ -167,6 +205,9 @@ public void tryAcquire(long messageSize, long timeout, TimeUnit unit) throws Int
             countAvailable.release();
             throw new IllegalStateException("Writer was closed with status " + closed);
         }
+        if (metrics != null) {
+            metrics.reportBufferWaitDuration(waitStarted);
+        }
     }
 
     public void releaseMessage(long messageSize) {

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;
     private final CompletableFuture<WriteAck> ackFuture = new CompletableFuture<>();
 
     private volatile ByteString data = null;
@@ -20,7 +21,12 @@ public class EnqueuedMessage {
     private volatile boolean isReady = false;
 
     public EnqueuedMessage(MessageMeta meta, long bufferSize) {
+        this(meta, bufferSize, 0);
+    }
+
+    EnqueuedMessage(MessageMeta meta, long bufferSize, long acceptedTimestamp) {
         this.meta = meta;
+        this.acceptedTimestamp = acceptedTimestamp;
         this.data = null;
         this.bufferSize = bufferSize;
     }
@@ -29,6 +35,10 @@ public MessageMeta getMeta() {
         return meta;
     }
 
+    long getAcceptedTimestamp() {
+        return acceptedTimestamp;
+    }
+
     public ByteString getData() {
         return data;
     }

topic/src/main/java/tech/ydb/topic/write/impl/SentMessage.java

@@ -17,13 +17,21 @@
 public class SentMessage implements WriterQueue.EncodedMsg {
     private final long seqNo;
     private final long bufferSize;
+    private final long sendTimestamp;
+    private final long acceptedTimestamp;
     private final YdbTransaction tx;
     private final YdbTopic.StreamWriteMessage.WriteRequest.MessageData pb;
     private final CompletableFuture<WriteAck> ackFuture;
 
     public SentMessage(EnqueuedMessage msg, long seqNo) {
+        this(msg, seqNo, 0);
+    }
+
+    SentMessage(EnqueuedMessage msg, long seqNo, long sendTimestamp) {
         this.seqNo = seqNo;
         this.bufferSize = msg.getBufferSize();
+        this.sendTimestamp = sendTimestamp;
+        this.acceptedTimestamp = msg.getAcceptedTimestamp();
         this.tx = msg.getMeta().getTransaction();
         this.pb = buildPb(seqNo, msg);
         this.ackFuture = msg.getAckFuture();
@@ -42,6 +50,14 @@ public long getBufferSize() {
         return bufferSize;
     }
 
+    long getSendTimestamp() {
+        return sendTimestamp;
+    }
+
+    long getAcceptedTimestamp() {
+        return acceptedTimestamp;
+    }
+
     @Override
     public void confirm(WriteAck ack) {
         this.ackFuture.complete(ack);

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,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.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)};
@@ -49,11 +59,42 @@ void register(LongSupplier bufferUsed, LongSupplier bufferLimit) {
                 m -> m.record(bufferLimit.getAsLong(), commonAttributes));
     }
 
+    void register(LongSupplier oldestTimestamp) {
+        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;
+    }
+
+    long reportMessageSendStart() {
+        return enabled ? System.nanoTime() : 0;
+    }
+
+    void reportMessageAckDuration(long timestamp) {
+        if (timestamp != 0) {
+            messageAckDuration.record((System.nanoTime() - timestamp) / 1_000_000_000d, commonAttributes);
+        }
+    }
+
+    long reportBufferWaitStart() {
+        return enabled ? System.nanoTime() : 0;
+    }
+
+    void reportBufferWaitDuration(long timestamp) {
+        if (timestamp != 0) {
+            bufferWaitDuration.record((System.nanoTime() - timestamp) / 1_000_000_000d, commonAttributes);
+        }
     }
 
     void reportSending(long bytes) {

topic/src/main/java/tech/ydb/topic/write/impl/WriterQueue.java

@@ -81,6 +81,8 @@ public WriterQueue(String debugId, WriterSettings settings, WriterMetrics metric
         this.readyNotify = readyNotify;
         this.metrics = metrics;
         this.metrics.register(buffer::getUsedSize, buffer::getMaxSize);
+        this.metrics.register(() -> sent.stream().map(EncodedMsg::getSentMessage)
+                .filter(msg -> msg != null).mapToLong(SentMessage::getAcceptedTimestamp).findFirst().orElse(0));
     }
 
     CompletableFuture<Void> flush() {
@@ -121,7 +123,7 @@ SentMessage nextMessageToSend() {
             }
 
             lastSeqNo = actualSeqNo;
-            SentMessage sentMsg = new SentMessage(next, actualSeqNo);
+            SentMessage sentMsg = new SentMessage(next, actualSeqNo, metrics.reportMessageSendStart());
             logger.trace("[{}] prepare sent message with seqNo {}", debugId, actualSeqNo);
             sent.offer(sentMsg);
             return sentMsg;
@@ -159,6 +161,7 @@ void confirmAck(WriteAck ack) {
             sentIt.remove();
             buffer.releaseMessage(sentMsg.getBufferSize());
             metrics.reportWritten();
+            metrics.reportMessageAckDuration(sentMsg.getSendTimestamp());
             msg.confirm(ack);
         }
     }
@@ -206,7 +209,7 @@ List<SentMessage> updateSeqNo(long newSeqNo) {
     CompletableFuture<WriteAck> enqueue(Message message, YdbTransaction tx) throws QueueOverflowException,
             InterruptedException {
         long reservedSizeBytes = reservationSizeBytes(message.getData().length);
-        buffer.acquire(reservedSizeBytes);
+        buffer.acquire(reservedSizeBytes, metrics);
         return accept(message, tx, reservedSizeBytes);
     }
 
@@ -219,7 +222,7 @@ CompletableFuture<WriteAck> tryEnqueue(Message message, YdbTransaction tx) throw
     CompletableFuture<WriteAck> tryEnqueue(Message message, YdbTransaction tx, long timeout, TimeUnit unit)
             throws QueueOverflowException, InterruptedException, TimeoutException {
         long reservedSizeBytes = reservationSizeBytes(message.getData().length);
-        buffer.tryAcquire(reservedSizeBytes, timeout, unit);
+        buffer.tryAcquire(reservedSizeBytes, timeout, unit, metrics);
         return accept(message, tx, reservedSizeBytes);
     }
 
@@ -232,7 +235,8 @@ private long reservationSizeBytes(int inputSizeBytes) {
     }
 
     private CompletableFuture<WriteAck> accept(Message message, YdbTransaction tx, long reservedSizeBytes) {
-        EnqueuedMessage msg = new EnqueuedMessage(new MessageMeta(message, tx), reservedSizeBytes);
+        EnqueuedMessage msg = new EnqueuedMessage(new MessageMeta(message, tx), reservedSizeBytes,
+                metrics.reportMessageSendStart());
         lastAcceptedAckFuture = msg.getAckFuture();
         queue.add(msg);
         metrics.reportSending(message.getData().length);