Add Topic session error counters

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

@@ -67,6 +67,7 @@ protected CompletableFuture<Result<ReadSession>> createNewStream(String id) {
 
     @Override
     protected void onRetry(ReadSession stream, Status status) {
+        config.getMetrics().reportSessionError(status, true);
         logger.warn("[{}] stopped by status {}", debugId, status);
         currentSessionId = null;
         if (errorHandler != null) {
@@ -85,6 +86,7 @@ protected void onRetry(ReadSession stream, Status status) {
     protected void onClose(ReadSession stream, Status status) {
         currentSessionId = null;
         if (!status.isSuccess()) {
+            config.getMetrics().reportSessionError(status, false);
             logger.warn("[{}] closed by status {}", debugId, status);
         } else {
             logger.info("[{}] closed by status {}", debugId, status);

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

@@ -3,6 +3,8 @@
 import java.util.Arrays;
 import java.util.function.LongSupplier;
 
+import tech.ydb.core.Status;
+import tech.ydb.core.StatusCode;
 import tech.ydb.core.metrics.Attr;
 import tech.ydb.core.metrics.LongCounter;
 import tech.ydb.core.metrics.Meter;
@@ -17,6 +19,7 @@ final class ReaderMetrics {
     private final LongCounter deliveredMessages;
     private final LongCounter receivedMessages;
     private final LongCounter receivedBytes;
+    private final LongCounter sessionErrors;
     private final Attr[] commonAttributes;
     private final boolean enabled;
     private final Meter meter;
@@ -34,6 +37,8 @@ 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.sessionErrors = meter.createCounter("ydb.topic.reader.session.errors", "{error}",
+                "The number of reader stream session errors by retry decision.");
         this.commonAttributes = createCommonAttributes(consumer, readerName);
     }
 
@@ -68,6 +73,18 @@ void reportReceivedBytes(long bytes) {
         }
     }
 
+    void reportSessionError(Status status, boolean retry) {
+        StatusCode code = status.getCode();
+        Attr[] attributes = Arrays.copyOf(commonAttributes, commonAttributes.length + 3);
+        attributes[commonAttributes.length] = Attr.of("retry_decision", retry ? "retry" : "stop");
+        attributes[commonAttributes.length + 1] = Attr.of("status_code", code.name());
+        String errorType = code == StatusCode.CLIENT_INTERNAL_ERROR ? "client_error"
+                : code == StatusCode.SUCCESS || code == StatusCode.UNUSED_STATUS
+                        ? "session_closed" : code.isTransportError() ? "transport_error" : "ydb_error";
+        attributes[commonAttributes.length + 2] = Attr.of("error.type", errorType);
+        sessionErrors.add(1, attributes);
+    }
+
     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/write/impl/WriterImpl.java

@@ -39,6 +39,7 @@ public class WriterImpl {
 
     private final String debugId;
     private final WriterQueue writeQueue;
+    private final WriterMetrics metrics;
     private final WriteSession stream;
     private final Runnable sendTask = new SerialRunnable(new SendTask());
 
@@ -57,8 +58,9 @@ public WriterImpl(TopicRpc topicRpc, WriterSettings settings, Executor compressi
     public WriterImpl(TopicRpc topicRpc, WriteStreamFactory factory, WriterSettings settings,
             Executor compressionExecutor, @Nonnull CodecRegistry codecRegistry) {
         this.debugId = DebugTools.createDebugId(settings.getLogPrefix());
+        this.metrics = new WriterMetrics(settings.getMeter(), settings.getTopicPath(), settings.getWriterName());
         this.stream = new WriteSession(debugId, factory, settings, topicRpc.getScheduler(), new ListenerImpl());
-        this.writeQueue = new WriterQueue(debugId, settings, codecRegistry, compressionExecutor, sendTask);
+        this.writeQueue = new WriterQueue(debugId, settings, metrics, codecRegistry, compressionExecutor, sendTask);
         Observability.reportMetricsUsage(settings.getMeter());
 
         logger.info("Writer with id {} created for topic \"{}\" with producerId \"{}\" and messageGroupId \"{}\"",
@@ -185,6 +187,7 @@ public void onStart(long lastSeqNo, String sessionId) {
         @Override
         public void onStop(Status status) {
             isReady = false;
+            metrics.reportSessionError(status, true);
         }
 
         @Override
@@ -194,7 +197,9 @@ public void onAck(WriteAck ack) {
 
         @Override
         public void onClose(Status status) {
-            isClosed.set(true);
+            if (isClosed.compareAndSet(false, true)) {
+                metrics.reportSessionError(status, false);
+            }
             isReady = false;
             initFuture.completeExceptionally(new UnexpectedResultException("Cannot init write session", status));
             shutdownFuture.complete(status);

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

@@ -1,7 +1,10 @@
 package tech.ydb.topic.write.impl;
 
+import java.util.Arrays;
 import java.util.function.LongSupplier;
 
+import tech.ydb.core.Status;
+import tech.ydb.core.StatusCode;
 import tech.ydb.core.metrics.Attr;
 import tech.ydb.core.metrics.LongCounter;
 import tech.ydb.core.metrics.Meter;
@@ -16,6 +19,7 @@ final class WriterMetrics {
     private final LongCounter sendingMessages;
     private final LongCounter sendingBytes;
     private final LongCounter writtenMessages;
+    private final LongCounter sessionErrors;
     private final Attr[] commonAttributes;
     private final Meter meter;
     private MetricRegistration bufferUsedGauge = MetricRegistration.NOOP;
@@ -29,6 +33,8 @@ final class WriterMetrics {
                 "The uncompressed body size of messages accepted by the writer.");
         this.writtenMessages = meter.createCounter("ydb.topic.writer.written.messages", MESSAGE_UNIT,
                 "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.commonAttributes = writerName == null
                 ? new Attr[0]
                 : new Attr[]{Attr.of("topic", topic), Attr.of("writer.name", writerName)};
@@ -58,4 +64,15 @@ void reportSending(long bytes) {
     void reportWritten() {
         writtenMessages.add(1, commonAttributes);
     }
+
+    void reportSessionError(Status status, boolean retry) {
+        StatusCode code = status.getCode();
+        Attr[] attributes = Arrays.copyOf(commonAttributes, commonAttributes.length + 3);
+        attributes[commonAttributes.length] = Attr.of("retry_decision", retry ? "retry" : "stop");
+        attributes[commonAttributes.length + 1] = Attr.of("status_code", code.name());
+        String errorType = code == StatusCode.SUCCESS || code == StatusCode.UNUSED_STATUS
+                ? "session_closed" : code.isTransportError() ? "transport_error" : "ydb_error";
+        attributes[commonAttributes.length + 2] = Attr.of("error.type", errorType);
+        sessionErrors.add(1, attributes);
+    }
 }

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

@@ -13,13 +13,15 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.UnsafeByteOperations;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import tech.ydb.common.transaction.YdbTransaction;
 import tech.ydb.core.Status;
+import tech.ydb.core.metrics.Meter;
 import tech.ydb.topic.description.Codec;
 import tech.ydb.topic.description.CodecRegistry;
 import tech.ydb.topic.settings.WriterSettings;
@@ -59,7 +61,14 @@ interface EncodedMsg {
     // Future for flush method. Only the future is stored to not retain the message data after sending
     private volatile CompletableFuture<WriteAck> lastAcceptedAckFuture = null;
 
-    public WriterQueue(String debugId, WriterSettings settings, CodecRegistry codecRegistry,
+    @VisibleForTesting
+    WriterQueue(String debugId, WriterSettings settings, CodecRegistry codecRegistry,
+                Executor compressionExecutor, Runnable readyNotify) {
+        this(debugId, settings, new WriterMetrics(Meter.NOOP, settings.getTopicPath(), settings.getWriterName()),
+                codecRegistry, compressionExecutor, readyNotify);
+    }
+
+    public WriterQueue(String debugId, WriterSettings settings, WriterMetrics metrics, CodecRegistry codecRegistry,
                        Executor compressionExecutor, Runnable readyNotify) {
         this.debugId = debugId;
         this.buffer = new BufferManager(debugId, settings);
@@ -70,7 +79,7 @@ public WriterQueue(String debugId, WriterSettings settings, CodecRegistry codecR
         }
         this.compressionExecutor = compressionExecutor;
         this.readyNotify = readyNotify;
-        this.metrics = new WriterMetrics(settings.getMeter(), settings.getTopicPath(), settings.getWriterName());
+        this.metrics = metrics;
         this.metrics.register(buffer::getUsedSize, buffer::getMaxSize);
     }