Add Topic Writer metrics

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<WriteAck> 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;
     }

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

@@ -17,13 +17,16 @@
 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<WriteAck> ackFuture;
 
     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);

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) {

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<Void> 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<SentMessage> updateSeqNo(long newSeqNo) {
     CompletableFuture<WriteAck> 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<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);
+        long waitStarted = System.nanoTime();
         buffer.tryAcquire(reservedSizeBytes, timeout, unit);
+        metrics.reportBufferWaitDuration(waitStarted);
         return accept(message, tx, reservedSizeBytes);
     }