Add Topic Reader metrics

topic/src/main/java/tech/ydb/topic/read/impl/BatchMeta.java

@@ -14,6 +14,7 @@ public class BatchMeta {
     private final Map<String, String> writeSessionMeta;
     private final int codec;
     private final Instant writtenAt;
+    private final long receivedAt = System.nanoTime();
 
     public BatchMeta(YdbTopic.StreamReadMessage.ReadResponse.Batch batch) {
         this.producerId = batch.getProducerId();
@@ -22,6 +23,10 @@ public BatchMeta(YdbTopic.StreamReadMessage.ReadResponse.Batch batch) {
         this.writtenAt = ProtobufUtils.protoToInstant(batch.getWrittenAt());
     }
 
+    long getReceivedAt() {
+        return receivedAt;
+    }
+
     public String getProducerId() {
         return producerId;
     }

topic/src/main/java/tech/ydb/topic/read/impl/MessageImpl.java

@@ -48,6 +48,10 @@ public MessageImpl(PartitionSession session, MessageCommitter committer, BatchMe
 
     public abstract boolean isReady();
 
+    long getReceivedAt() {
+        return batchMeta.getReceivedAt();
+    }
+
     @Override
     public long getOffset() {
         return offset;

topic/src/main/java/tech/ydb/topic/read/impl/ReadConfig.java

@@ -1,6 +1,7 @@
 package tech.ydb.topic.read.impl;
 
 import java.util.concurrent.Executor;
+import java.util.function.LongSupplier;
 
 import tech.ydb.topic.description.CodecRegistry;
 import tech.ydb.topic.settings.ReaderSettings;
@@ -18,9 +19,16 @@ public class ReadConfig {
     private final long maxMemoryUsageBytes;
     private final int maxBatchSize;
     private final ReaderMetrics metrics;
+    private final LongSupplier readyMessages;
+    private final LongSupplier readyOldestTimestamp;
 
     public ReadConfig(CodecRegistry codecRegistry, Executor control, Executor processor, Executor decompressor,
             ReaderSettings settings) {
+        this(codecRegistry, control, processor, decompressor, settings, () -> 0, () -> 0);
+    }
+
+    public ReadConfig(CodecRegistry codecRegistry, Executor control, Executor processor, Executor decompressor,
+            ReaderSettings settings, LongSupplier readyMessages, LongSupplier readyOldestTimestamp) {
         this.codecRegistry = codecRegistry;
         this.control = control;
         this.processor = processor;
@@ -29,6 +37,8 @@ public ReadConfig(CodecRegistry codecRegistry, Executor control, Executor proces
         this.maxMemoryUsageBytes = settings.getMaxMemoryUsageBytes();
         this.maxBatchSize = settings.getMaxBatchSize();
         this.metrics = new ReaderMetrics(settings.getMeter(), settings.getConsumerName(), settings.getReaderName());
+        this.readyMessages = readyMessages;
+        this.readyOldestTimestamp = readyOldestTimestamp;
     }
 
     public CodecRegistry getCodecRegistry() {
@@ -59,6 +69,14 @@ public int getMaxBatchSize() {
         return maxBatchSize;
     }
 
+    public long getReadyMessages() {
+        return readyMessages.getAsLong();
+    }
+
+    public long getReadyOldestTimestamp() {
+        return readyOldestTimestamp.getAsLong();
+    }
+
     ReaderMetrics getMetrics() {
         return metrics;
     }

topic/src/main/java/tech/ydb/topic/read/impl/ReadPartition.java

@@ -90,6 +90,14 @@ public PartitionSession getPartition() {
         return partition;
     }
 
+    public long getCommitOffsetLag() {
+        return committer.getCommitOffsetLag();
+    }
+
+    void updateCommitTarget(long offset) {
+        committer.updateCommitTarget(offset);
+    }
+
     @Override
     public void confirmRangeProcessed(OffsetsRange range) {
         decoder.releaseRange(range);
@@ -104,11 +112,20 @@ public void confirmPartitionEnded(List<TopicPartition> childs) {
         sendDataToReaders();
     }
 
-    public void confirmCommittedOffset(long committedOffset) {
-        committer.updateCommittedOffset(committedOffset);
+    long getOldestReceivedTimestamp() {
+        return queue.getOldestReceivedTimestamp();
+    }
+
+    long getLocalBufferMessages() {
+        return queue.getLocalBufferMessages();
+    }
+
+    public long confirmCommittedOffset(long committedOffset) {
+        long acknowledged = committer.updateCommittedOffset(committedOffset);
         committer.completePendingCommits();
         commitOffsetAck.set(new CommitOffsetAcknowledgementEventImpl(partition, committedOffset));
         sendDataToReaders();
+        return acknowledged;
     }
 
     public void unpause() {

topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionCommitter.java

@@ -32,24 +32,36 @@ class ReadPartitionCommitter implements MessageCommitter {
     private final ReentrantLock commitFuturesLock = new ReentrantLock();
 
     private final AtomicLong lastCommittedOffset;
+    private final AtomicLong commitTargetOffset;
 
     ReadPartitionCommitter(String traceID, ReadSession session, PartitionSession partition, long lastCommittedOffset) {
         this.traceID = traceID;
         this.session = session;
         this.partition = partition;
         this.lastCommittedOffset = new AtomicLong(lastCommittedOffset);
+        this.commitTargetOffset = new AtomicLong(lastCommittedOffset);
     }
 
     private RuntimeException partitionIsClosedException() {
         return new RuntimeException("" + partition + " is already stopped");
     }
 
-    public void updateCommittedOffset(long offset) {
-        long old = lastCommittedOffset.get();
-        if (old != lastCommittedOffset.accumulateAndGet(offset, Math::max)) {
+    public long getCommitOffsetLag() {
+        return Math.max(0, commitTargetOffset.get() - lastCommittedOffset.get());
+    }
+
+    void updateCommitTarget(long offset) {
+        commitTargetOffset.accumulateAndGet(offset, Math::max);
+    }
+
+    public long updateCommittedOffset(long offset) {
+        long old = lastCommittedOffset.getAndAccumulate(offset, Math::max);
+        if (offset > old) {
             logger.debug("[{}] Updated last committed offset: {}. Previous committed offset: {} "
                     + "(diff is {} message(s)).", traceID, offset, old, offset - old);
+            return offset - old;
         }
+        return 0;
     }
 
     public void completePendingCommits() {

topic/src/main/java/tech/ydb/topic/read/impl/ReadPartitionQueue.java

@@ -36,6 +36,15 @@ void updateLastReadOffset(long offset) {
         lastReadOffset = offset;
     }
 
+    long getLocalBufferMessages() {
+        return readingQueue.size();
+    }
+
+    long getOldestReceivedTimestamp() {
+        MessageImpl first = readingQueue.peek();
+        return first == null ? 0 : first.getReceivedAt();
+    }
+
     void addBatches(List<YdbTopic.StreamReadMessage.ReadResponse.Batch> batchList) {
         for (YdbTopic.StreamReadMessage.ReadResponse.Batch batch : batchList) {
             if (batch.getMessageDataCount() == 0) {

topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java

@@ -128,6 +128,10 @@ public boolean commitOffsets(PartitionSession session, List<OffsetsRange> ranges
                         .build())
                 .build();
 
+        for (OffsetsRange range : rangesToCommit) {
+            partition.updateCommitTarget(range.getEnd());
+            config.getMetrics().reportCommitQueued(range.getEnd() - range.getStart(), session.getPath());
+        }
         send(FromClient.newBuilder().setCommitOffsetRequest(req).build());
         return true;
     }
@@ -189,7 +193,30 @@ public void removeLocks(PartitionSession ps, List<TopicPartition> childs) {
 
     public void onInit(YdbTopic.StreamReadMessage.InitResponse response) {
         bufferManager.init(response.getSessionId());
-        config.getMetrics().register(partitions::size, bufferManager::getCreditBalanceBytes);
+        config.getMetrics().register(partitions::size, bufferManager::getCreditBalanceBytes,
+                this::getCommitOffsetLagMax, this::getLocalBufferMessages, this::getOldestReceivedTimestamp);
+    }
+
+    private long getCommitOffsetLagMax() {
+        return partitions.values().stream().mapToLong(ReadPartition::getCommitOffsetLag).max().orElse(0);
+    }
+
+    private long getLocalBufferMessages() {
+        return config.getReadyMessages() + partitions.values().stream()
+                .filter(ReadPartition::isActive).mapToLong(ReadPartition::getLocalBufferMessages).sum();
+    }
+
+    private long getOldestReceivedTimestamp() {
+        long oldestTimestamp = config.getReadyOldestTimestamp();
+        for (ReadPartition partition : partitions.values()) {
+            if (partition.isActive()) {
+                long timestamp = partition.getOldestReceivedTimestamp();
+                if (timestamp != 0 && (oldestTimestamp == 0 || timestamp < oldestTimestamp)) {
+                    oldestTimestamp = timestamp;
+                }
+            }
+        }
+        return oldestTimestamp;
     }
 
     public void onStartPartition(YdbTopic.StreamReadMessage.StartPartitionSessionRequest req) {
@@ -299,7 +326,8 @@ public void onCommitOffset(YdbTopic.StreamReadMessage.CommitOffsetResponse respo
                 continue;
             }
 
-            partition.confirmCommittedOffset(offset.getCommittedOffset());
+            long acknowledged = partition.confirmCommittedOffset(offset.getCommittedOffset());
+            config.getMetrics().reportCommitAcknowledged(acknowledged, partition.getPartition().getPath());
         }
     }
 

topic/src/main/java/tech/ydb/topic/read/impl/ReaderMetrics.java

@@ -19,12 +19,17 @@ final class ReaderMetrics {
     private final LongCounter deliveredMessages;
     private final LongCounter receivedMessages;
     private final LongCounter receivedBytes;
+    private final LongCounter commitQueued;
+    private final LongCounter commitAcknowledged;
     private final LongCounter sessionErrors;
     private final Attr[] commonAttributes;
     private final boolean enabled;
     private final Meter meter;
     private MetricRegistration partitionsGauge = MetricRegistration.NOOP;
     private MetricRegistration creditGauge = MetricRegistration.NOOP;
+    private MetricRegistration commitLagGauge = MetricRegistration.NOOP;
+    private MetricRegistration bufferMessagesGauge = MetricRegistration.NOOP;
+    private MetricRegistration ageGauge = MetricRegistration.NOOP;
 
     ReaderMetrics(Meter meter, String consumer, String readerName) {
         this.meter = meter;
@@ -37,26 +42,50 @@ final class ReaderMetrics {
                 "The number of messages accepted by the SDK for an active partition session.");
         this.receivedBytes = meter.createCounter("ydb.topic.reader.received.bytes", "By",
                 "The protocol bytes_size received in read responses.");
+        this.commitQueued = meter.createCounter("ydb.topic.reader.commit.queued", MESSAGE_UNIT,
+                "The number of messages in commit ranges accepted by the SDK.");
+        this.commitAcknowledged = meter.createCounter("ydb.topic.reader.commit.acknowledged", MESSAGE_UNIT,
+                "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.commonAttributes = createCommonAttributes(consumer, readerName);
     }
 
-    void register(LongSupplier partitionCount, LongSupplier bufferBudget) {
+    void register(LongSupplier partitionCount, LongSupplier bufferBudget, LongSupplier commitLag,
+            LongSupplier bufferMessages, LongSupplier oldestTimestamp) {
         partitionsGauge = meter.registerLongGauge(
                 "ydb.topic.reader.partition_session.count", "{session}",
                 "The number of partition sessions currently in the reader session processing lifecycle.",
                 m -> m.record(partitionCount.getAsLong(), commonAttributes));
         creditGauge = meter.registerLongGauge("ydb.topic.reader.credit_balance_bytes", "By",
                 "The protocol credit granted to the server and not yet consumed by read responses.",
                 m -> m.record(bufferBudget.getAsLong(), commonAttributes));
+        commitLagGauge = meter.registerLongGauge("ydb.topic.reader.commit_offset.lag.max", MESSAGE_UNIT,
+                "The maximum gap between requested and acknowledged commit offsets.",
+                m -> m.record(commitLag.getAsLong(), commonAttributes));
+        bufferMessagesGauge = meter.registerLongGauge("ydb.topic.reader.local_buffer.messages", MESSAGE_UNIT,
+                "The number of messages currently buffered by the reader.",
+                m -> m.record(bufferMessages.getAsLong(), commonAttributes));
+        ageGauge = meter.registerDoubleGauge("ydb.topic.reader.local_buffer.message_age.max", "s",
+                "The age of the oldest batch retained in the reader's local buffer.",
+                m -> {
+                    long timestamp = oldestTimestamp.getAsLong();
+                    m.record(timestamp == 0 ? 0 : (System.nanoTime() - timestamp) / 1_000_000_000d,
+                            commonAttributes);
+                });
     }
 
     void unregister() {
         partitionsGauge.close();
         creditGauge.close();
+        commitLagGauge.close();
+        bufferMessagesGauge.close();
+        ageGauge.close();
         partitionsGauge = MetricRegistration.NOOP;
         creditGauge = MetricRegistration.NOOP;
+        commitLagGauge = MetricRegistration.NOOP;
+        bufferMessagesGauge = MetricRegistration.NOOP;
+        ageGauge = MetricRegistration.NOOP;
     }
 
     void reportDelivered(long messages, String topic) {
@@ -85,6 +114,14 @@ void reportSessionError(Status status, boolean retry) {
         sessionErrors.add(1, attributes);
     }
 
+    void reportCommitQueued(long messages, String topic) {
+        report(commitQueued, messages, topic);
+    }
+
+    void reportCommitAcknowledged(long messages, String topic) {
+        report(commitAcknowledged, messages, topic);
+    }
+
     private void report(LongCounter counter, long messages, String topic) {
         if (enabled && messages > 0) {
             record(counter, messages, Attr.of("topic", topic));

topic/src/main/java/tech/ydb/topic/read/impl/SyncReaderImpl.java

@@ -65,7 +65,8 @@ public SyncReaderImpl(TopicRpc topicRpc, ReaderSettings settings, @Nonnull Codec
         this.debugId = DebugTools.createDebugId(settings.getLogPrefix());
         this.decompressor = new LazyExecutor("reader[" + debugId + "]-decoder", settings.getDecompressionExecutor());
 
-        this.config = new ReadConfig(codecRegistry, Runnable::run, Runnable::run, decompressor, settings);
+        this.config = new ReadConfig(codecRegistry, Runnable::run, Runnable::run, decompressor, settings,
+                queue::size, this::getOldestReceivedTimestamp);
         this.impl = new Impl(topicRpc, debugId, settings, config);
 
         String readerName = settings.getReaderName();
@@ -83,6 +84,11 @@ public String getSessionId() {
         return sessionId;
     }
 
+    private long getOldestReceivedTimestamp() {
+        return queue.stream().filter(MessageWrapper::isActive)
+                .mapToLong(message -> ((MessageImpl) message.msg).getReceivedAt()).min().orElse(0);
+    }
+
     @Override
     public void init() {
         impl.start();