diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReaderImpl.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReaderImpl.java index e3548c093..e643a85e3 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReaderImpl.java +++ b/topic/src/main/java/tech/ydb/topic/read/impl/ReaderImpl.java @@ -67,6 +67,7 @@ protected CompletableFuture> 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); diff --git a/topic/src/main/java/tech/ydb/topic/read/impl/ReaderMetrics.java b/topic/src/main/java/tech/ydb/topic/read/impl/ReaderMetrics.java index d2f6efdd6..3ccd4d520 100644 --- a/topic/src/main/java/tech/ydb/topic/read/impl/ReaderMetrics.java +++ b/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)); diff --git a/topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java b/topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java index 72289751e..744e98a44 100644 --- a/topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java +++ b/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); 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 3cf097710..3fc70f6da 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 @@ -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); + } } 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 709725373..19e5866f8 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 @@ -13,6 +13,7 @@ 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; @@ -20,6 +21,7 @@ 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 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); }