From 7cd41ce214aa5c2b1e34ddd88916a5ce8b50de2a Mon Sep 17 00:00:00 2001 From: Jorge Esteban Quilcate Otoya Date: Mon, 20 Jul 2026 14:26:03 +0300 Subject: [PATCH] feat(inkless:storage): split S3 error metrics by operation The S3 MetricCollector recorded every SDK error (throttling, server, timeout, IO, other) into a single global sensor per error type, so a throttle could not be attributed to a specific operation. The ops question "are our deletes being throttled?" was unanswerable from metrics. Record each error into an additional operation-tagged sensor (operation=, e.g. DeleteObjects) alongside the existing untagged global sensor. The global metrics are unchanged, so existing dashboards/alerts keep working; the tagged variants appear under a new MBean (type=s3-client-metrics,operation=...). Tagged sensors are created lazily and stored in a ConcurrentHashMap to stay safe under concurrent SDK publish callbacks. Per-operation attribution is skipped when the operation name is ambiguous. Adds MetricCollectorTest which builds SDK MetricCollections directly to exercise publish() without WireMock, and regenerates metrics.rst. --- docs/inkless/metrics.rst | 18 +++ .../storage_backend/s3/MetricCollector.java | 65 ++++++++++ .../storage_backend/s3/MetricRegistry.java | 78 ++++++++++- .../s3/MetricCollectorTest.java | 121 ++++++++++++++++++ 4 files changed, 281 insertions(+), 1 deletion(-) create mode 100644 storage/inkless/src/test/java/io/aiven/inkless/storage_backend/s3/MetricCollectorTest.java diff --git a/docs/inkless/metrics.rst b/docs/inkless/metrics.rst index 5814d024725..d8e67ade86b 100644 --- a/docs/inkless/metrics.rst +++ b/docs/inkless/metrics.rst @@ -56,6 +56,24 @@ upload-part-time-avg Average time spent uploading a single upload-part-time-max Maximum time spent uploading a single part ========================================= ============================================================================= +io.aiven.inkless.storage:type=s3-client-metrics,operation="{operation}" +----------------------------------------------------------------------- + +================================ ========================================================== +Attribute name Description +================================ ========================================================== +configured-timeout-errors-rate Rate of configured timeout errors per S3 operation +configured-timeout-errors-total Total number of configured timeout errors per S3 operation +io-errors-rate Rate of IO errors per S3 operation +io-errors-total Total number of IO errors per S3 operation +other-errors-rate Rate of other errors per S3 operation +other-errors-total Total number of other errors per S3 operation +server-errors-rate Rate of server errors per S3 operation +server-errors-total Total number of server errors per S3 operation +throttling-errors-rate Rate of throttling errors per S3 operation +throttling-errors-total Total number of throttling errors per S3 operation +================================ ========================================================== + AzureBlobStorage metrics ================================== diff --git a/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/MetricCollector.java b/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/MetricCollector.java index fb118868488..964030c23cc 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/MetricCollector.java +++ b/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/MetricCollector.java @@ -34,6 +34,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import software.amazon.awssdk.core.metrics.CoreMetric; import software.amazon.awssdk.metrics.MetricCollection; @@ -52,6 +53,8 @@ import static io.aiven.inkless.storage_backend.s3.MetricRegistry.COMPLETE_MULTIPART_UPLOAD_TIME_AVG_METRIC_NAME; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.COMPLETE_MULTIPART_UPLOAD_TIME_MAX_METRIC_NAME; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.CONFIGURED_TIMEOUT_ERRORS; +import static io.aiven.inkless.storage_backend.s3.MetricRegistry.CONFIGURED_TIMEOUT_ERRORS_BY_OPERATION_RATE_METRIC_NAME; +import static io.aiven.inkless.storage_backend.s3.MetricRegistry.CONFIGURED_TIMEOUT_ERRORS_BY_OPERATION_TOTAL_METRIC_NAME; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.CONFIGURED_TIMEOUT_ERRORS_RATE_METRIC_NAME; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.CONFIGURED_TIMEOUT_ERRORS_TOTAL_METRIC_NAME; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.CREATE_MULTIPART_UPLOAD_REQUESTS; @@ -79,9 +82,14 @@ import static io.aiven.inkless.storage_backend.s3.MetricRegistry.GET_OBJECT_TIME_AVG_METRIC_NAME; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.GET_OBJECT_TIME_MAX_METRIC_NAME; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.IO_ERRORS; +import static io.aiven.inkless.storage_backend.s3.MetricRegistry.IO_ERRORS_BY_OPERATION_RATE_METRIC_NAME; +import static io.aiven.inkless.storage_backend.s3.MetricRegistry.IO_ERRORS_BY_OPERATION_TOTAL_METRIC_NAME; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.IO_ERRORS_RATE_METRIC_NAME; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.IO_ERRORS_TOTAL_METRIC_NAME; +import static io.aiven.inkless.storage_backend.s3.MetricRegistry.OPERATION_TAG; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.OTHER_ERRORS; +import static io.aiven.inkless.storage_backend.s3.MetricRegistry.OTHER_ERRORS_BY_OPERATION_RATE_METRIC_NAME; +import static io.aiven.inkless.storage_backend.s3.MetricRegistry.OTHER_ERRORS_BY_OPERATION_TOTAL_METRIC_NAME; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.OTHER_ERRORS_RATE_METRIC_NAME; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.OTHER_ERRORS_TOTAL_METRIC_NAME; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.PUT_OBJECT_REQUESTS; @@ -91,9 +99,13 @@ import static io.aiven.inkless.storage_backend.s3.MetricRegistry.PUT_OBJECT_TIME_AVG_METRIC_NAME; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.PUT_OBJECT_TIME_MAX_METRIC_NAME; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.SERVER_ERRORS; +import static io.aiven.inkless.storage_backend.s3.MetricRegistry.SERVER_ERRORS_BY_OPERATION_RATE_METRIC_NAME; +import static io.aiven.inkless.storage_backend.s3.MetricRegistry.SERVER_ERRORS_BY_OPERATION_TOTAL_METRIC_NAME; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.SERVER_ERRORS_RATE_METRIC_NAME; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.SERVER_ERRORS_TOTAL_METRIC_NAME; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.THROTTLING_ERRORS; +import static io.aiven.inkless.storage_backend.s3.MetricRegistry.THROTTLING_ERRORS_BY_OPERATION_RATE_METRIC_NAME; +import static io.aiven.inkless.storage_backend.s3.MetricRegistry.THROTTLING_ERRORS_BY_OPERATION_TOTAL_METRIC_NAME; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.THROTTLING_ERRORS_RATE_METRIC_NAME; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.THROTTLING_ERRORS_TOTAL_METRIC_NAME; import static io.aiven.inkless.storage_backend.s3.MetricRegistry.UPLOAD_PART_REQUESTS; @@ -117,6 +129,10 @@ public class MetricCollector implements MetricPublisher { private final Map requestMetrics = new HashMap<>(); private final Map latencyMetrics = new HashMap<>(); private final Map errorMetrics = new HashMap<>(); + // {rate, total} templates for the per-operation (operation-tagged) variant of each error type. + private final Map errorByOperationTemplates = new HashMap<>(); + // Lazily created operation-tagged error sensors, keyed by errorType + ":" + operationName. + private final Map errorByOperationMetrics = new ConcurrentHashMap<>(); public MetricCollector(Metrics metrics) { this.metrics = metrics; @@ -247,6 +263,46 @@ public MetricCollector(Metrics metrics) { OTHER_ERRORS_TOTAL_METRIC_NAME ); errorMetrics.put(OTHER.toString(), otherErrorsSensor); + + errorByOperationTemplates.put(THROTTLING.toString(), new MetricNameTemplate[]{ + THROTTLING_ERRORS_BY_OPERATION_RATE_METRIC_NAME, + THROTTLING_ERRORS_BY_OPERATION_TOTAL_METRIC_NAME + }); + errorByOperationTemplates.put(SERVER_ERROR.toString(), new MetricNameTemplate[]{ + SERVER_ERRORS_BY_OPERATION_RATE_METRIC_NAME, + SERVER_ERRORS_BY_OPERATION_TOTAL_METRIC_NAME + }); + errorByOperationTemplates.put(CONFIGURED_TIMEOUT.toString(), new MetricNameTemplate[]{ + CONFIGURED_TIMEOUT_ERRORS_BY_OPERATION_RATE_METRIC_NAME, + CONFIGURED_TIMEOUT_ERRORS_BY_OPERATION_TOTAL_METRIC_NAME + }); + errorByOperationTemplates.put(IO.toString(), new MetricNameTemplate[]{ + IO_ERRORS_BY_OPERATION_RATE_METRIC_NAME, + IO_ERRORS_BY_OPERATION_TOTAL_METRIC_NAME + }); + errorByOperationTemplates.put(OTHER.toString(), new MetricNameTemplate[]{ + OTHER_ERRORS_BY_OPERATION_RATE_METRIC_NAME, + OTHER_ERRORS_BY_OPERATION_TOTAL_METRIC_NAME + }); + } + + /** + * Returns the operation-tagged sensor for the given SDK error type and S3 operation, creating it + * on first use. The untagged global error sensors are recorded separately; this variant lets an + * error (e.g. a throttle) be attributed to a specific operation such as DeleteObjects. + */ + private Sensor errorSensorByOperation(final String errorType, final String operationName) { + final MetricNameTemplate[] templates = errorByOperationTemplates.get(errorType); + if (templates == null) { + return null; + } + return errorByOperationMetrics.computeIfAbsent(errorType + ":" + operationName, key -> { + final Map tags = Map.of(OPERATION_TAG, operationName); + final Sensor sensor = metrics.sensor(key); + sensor.add(metrics.metricInstance(templates[0], tags), new Rate()); + sensor.add(metrics.metricInstance(templates[1], tags), new CumulativeCount()); + return sensor; + }); } private Sensor createRequestsSensor( @@ -275,6 +331,7 @@ private Sensor createLatencySensor( public void publish(final MetricCollection metricCollection) { final List metricValues = metricCollection.metricValues(CoreMetric.OPERATION_NAME); // metrics are reported per request, so 1 value can be assumed. + final String operationName = metricValues.size() == 1 ? metricValues.get(0) : null; if (metricValues.size() == 1) { final var metricValue = metricValues.get(0); final var requests = requestMetrics.get(metricValue); @@ -310,6 +367,14 @@ public void publish(final MetricCollection metricCollection) { if (sensor != null) { sensor.record(); } + // Attribute the error to its operation too (e.g. is it deletes being throttled?). + // Skipped when the operation name is ambiguous. + if (operationName != null) { + final var byOperation = errorSensorByOperation(errorValue, operationName); + if (byOperation != null) { + byOperation.record(); + } + } } } diff --git a/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/MetricRegistry.java b/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/MetricRegistry.java index 39136809938..72eb2ccfef6 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/MetricRegistry.java +++ b/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/MetricRegistry.java @@ -26,6 +26,7 @@ @CoverageIgnore // tested on integration level public class MetricRegistry { static final String METRIC_GROUP = "s3-client-metrics"; + static final String OPERATION_TAG = "operation"; static final String GET_OBJECT_REQUESTS = "get-object-requests"; static final String GET_OBJECT_REQUESTS_RATE = GET_OBJECT_REQUESTS + "-rate"; static final String GET_OBJECT_REQUESTS_TOTAL = GET_OBJECT_REQUESTS + "-total"; @@ -330,6 +331,71 @@ public class MetricRegistry { TOTAL_DOC_PREFIX + OTHER_ERRORS_DOC ); + // Per-operation variants of the error metrics above, tagged with the S3 operation name + // (e.g. DeleteObjects, PutObject). These are recorded in addition to the untagged global + // error metrics so that throttling can be attributed to a specific operation. + static final String PER_OPERATION_DOC_SUFFIX = " per S3 operation"; + static final MetricNameTemplate THROTTLING_ERRORS_BY_OPERATION_RATE_METRIC_NAME = new MetricNameTemplate( + THROTTLING_ERRORS_RATE, + METRIC_GROUP, + RATE_DOC_PREFIX + THROTTLING_ERRORS_DOC + PER_OPERATION_DOC_SUFFIX, + OPERATION_TAG + ); + static final MetricNameTemplate THROTTLING_ERRORS_BY_OPERATION_TOTAL_METRIC_NAME = new MetricNameTemplate( + THROTTLING_ERRORS_TOTAL, + METRIC_GROUP, + TOTAL_DOC_PREFIX + THROTTLING_ERRORS_DOC + PER_OPERATION_DOC_SUFFIX, + OPERATION_TAG + ); + static final MetricNameTemplate SERVER_ERRORS_BY_OPERATION_RATE_METRIC_NAME = new MetricNameTemplate( + SERVER_ERRORS_RATE, + METRIC_GROUP, + RATE_DOC_PREFIX + SERVER_ERRORS_DOC + PER_OPERATION_DOC_SUFFIX, + OPERATION_TAG + ); + static final MetricNameTemplate SERVER_ERRORS_BY_OPERATION_TOTAL_METRIC_NAME = new MetricNameTemplate( + SERVER_ERRORS_TOTAL, + METRIC_GROUP, + TOTAL_DOC_PREFIX + SERVER_ERRORS_DOC + PER_OPERATION_DOC_SUFFIX, + OPERATION_TAG + ); + static final MetricNameTemplate CONFIGURED_TIMEOUT_ERRORS_BY_OPERATION_RATE_METRIC_NAME = new MetricNameTemplate( + CONFIGURED_TIMEOUT_ERRORS_RATE, + METRIC_GROUP, + RATE_DOC_PREFIX + CONFIGURED_TIMEOUT_ERRORS_DOC + PER_OPERATION_DOC_SUFFIX, + OPERATION_TAG + ); + static final MetricNameTemplate CONFIGURED_TIMEOUT_ERRORS_BY_OPERATION_TOTAL_METRIC_NAME = new MetricNameTemplate( + CONFIGURED_TIMEOUT_ERRORS_TOTAL, + METRIC_GROUP, + TOTAL_DOC_PREFIX + CONFIGURED_TIMEOUT_ERRORS_DOC + PER_OPERATION_DOC_SUFFIX, + OPERATION_TAG + ); + static final MetricNameTemplate IO_ERRORS_BY_OPERATION_RATE_METRIC_NAME = new MetricNameTemplate( + IO_ERRORS_RATE, + METRIC_GROUP, + RATE_DOC_PREFIX + IO_ERRORS_DOC + PER_OPERATION_DOC_SUFFIX, + OPERATION_TAG + ); + static final MetricNameTemplate IO_ERRORS_BY_OPERATION_TOTAL_METRIC_NAME = new MetricNameTemplate( + IO_ERRORS_TOTAL, + METRIC_GROUP, + TOTAL_DOC_PREFIX + IO_ERRORS_DOC + PER_OPERATION_DOC_SUFFIX, + OPERATION_TAG + ); + static final MetricNameTemplate OTHER_ERRORS_BY_OPERATION_RATE_METRIC_NAME = new MetricNameTemplate( + OTHER_ERRORS_RATE, + METRIC_GROUP, + RATE_DOC_PREFIX + OTHER_ERRORS_DOC + PER_OPERATION_DOC_SUFFIX, + OPERATION_TAG + ); + static final MetricNameTemplate OTHER_ERRORS_BY_OPERATION_TOTAL_METRIC_NAME = new MetricNameTemplate( + OTHER_ERRORS_TOTAL, + METRIC_GROUP, + TOTAL_DOC_PREFIX + OTHER_ERRORS_DOC + PER_OPERATION_DOC_SUFFIX, + OPERATION_TAG + ); + public static List all() { return List.of( GET_OBJECT_REQUESTS_RATE_METRIC_NAME, @@ -373,7 +439,17 @@ public static List all() { IO_ERRORS_RATE_METRIC_NAME, IO_ERRORS_TOTAL_METRIC_NAME, OTHER_ERRORS_RATE_METRIC_NAME, - OTHER_ERRORS_TOTAL_METRIC_NAME + OTHER_ERRORS_TOTAL_METRIC_NAME, + THROTTLING_ERRORS_BY_OPERATION_RATE_METRIC_NAME, + THROTTLING_ERRORS_BY_OPERATION_TOTAL_METRIC_NAME, + SERVER_ERRORS_BY_OPERATION_RATE_METRIC_NAME, + SERVER_ERRORS_BY_OPERATION_TOTAL_METRIC_NAME, + CONFIGURED_TIMEOUT_ERRORS_BY_OPERATION_RATE_METRIC_NAME, + CONFIGURED_TIMEOUT_ERRORS_BY_OPERATION_TOTAL_METRIC_NAME, + IO_ERRORS_BY_OPERATION_RATE_METRIC_NAME, + IO_ERRORS_BY_OPERATION_TOTAL_METRIC_NAME, + OTHER_ERRORS_BY_OPERATION_RATE_METRIC_NAME, + OTHER_ERRORS_BY_OPERATION_TOTAL_METRIC_NAME ); } } diff --git a/storage/inkless/src/test/java/io/aiven/inkless/storage_backend/s3/MetricCollectorTest.java b/storage/inkless/src/test/java/io/aiven/inkless/storage_backend/s3/MetricCollectorTest.java new file mode 100644 index 00000000000..cc66a0ea491 --- /dev/null +++ b/storage/inkless/src/test/java/io/aiven/inkless/storage_backend/s3/MetricCollectorTest.java @@ -0,0 +1,121 @@ +/* + * Inkless + * Copyright (C) 2024 - 2025 Aiven OY + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ +package io.aiven.inkless.storage_backend.s3; + +import org.apache.kafka.common.metrics.KafkaMetric; +import org.apache.kafka.common.metrics.Metrics; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.util.Map; + +import software.amazon.awssdk.core.internal.metrics.SdkErrorType; +import software.amazon.awssdk.core.metrics.CoreMetric; +import software.amazon.awssdk.metrics.MetricCollection; + +import static org.assertj.core.api.Assertions.assertThat; + +class MetricCollectorTest { + private static final String GROUP = "s3-client-metrics"; + + private final Metrics metrics = new Metrics(); + private final MetricCollector collector = new MetricCollector(metrics); + + @AfterEach + void tearDown() { + metrics.close(); + } + + @Test + void recordsRequestAndLatencyForOperation() { + collector.publish(collection("DeleteObjects", Duration.ofMillis(7), null)); + + assertThat(doubleValue("delete-objects-requests-total")).isEqualTo(1.0); + assertThat(doubleValue("delete-objects-time-max")).isEqualTo(7.0); + } + + @Test + void recordsErrorBothGloballyAndPerOperation() { + collector.publish(collection("DeleteObjects", Duration.ofMillis(1), SdkErrorType.THROTTLING)); + + // Untagged global metric preserves existing behavior. + assertThat(doubleValue("throttling-errors-total")).isEqualTo(1.0); + // New per-operation variant attributes the throttle to DeleteObjects. + assertThat(doubleValue("throttling-errors-total", Map.of("operation", "DeleteObjects"))) + .isEqualTo(1.0); + // A different operation is not credited. + assertThat(taggedMetric("throttling-errors-total", Map.of("operation", "PutObject"))) + .isNull(); + } + + @Test + void perOperationErrorsAreSeparatedByOperation() { + collector.publish(collection("DeleteObjects", Duration.ofMillis(1), SdkErrorType.THROTTLING)); + collector.publish(collection("GetObject", Duration.ofMillis(1), SdkErrorType.THROTTLING)); + + assertThat(doubleValue("throttling-errors-total")).isEqualTo(2.0); + assertThat(doubleValue("throttling-errors-total", Map.of("operation", "DeleteObjects"))) + .isEqualTo(1.0); + assertThat(doubleValue("throttling-errors-total", Map.of("operation", "GetObject"))) + .isEqualTo(1.0); + } + + @Test + void ambiguousOperationSkipsPerOperationButKeepsGlobal() { + // Two OPERATION_NAME values make the operation ambiguous; the global error is still recorded + // but no per-operation attribution is possible. + final software.amazon.awssdk.metrics.MetricCollector root = + software.amazon.awssdk.metrics.MetricCollector.create("ApiCall"); + root.reportMetric(CoreMetric.OPERATION_NAME, "DeleteObjects"); + root.reportMetric(CoreMetric.OPERATION_NAME, "GetObject"); + root.createChild("ApiCallAttempt").reportMetric(CoreMetric.ERROR_TYPE, SdkErrorType.THROTTLING.toString()); + + collector.publish(root.collect()); + + assertThat(doubleValue("throttling-errors-total")).isEqualTo(1.0); + assertThat(taggedMetric("throttling-errors-total", Map.of("operation", "DeleteObjects"))) + .isNull(); + } + + private static MetricCollection collection(final String operation, + final Duration duration, + final SdkErrorType errorType) { + final software.amazon.awssdk.metrics.MetricCollector root = + software.amazon.awssdk.metrics.MetricCollector.create("ApiCall"); + root.reportMetric(CoreMetric.OPERATION_NAME, operation); + root.reportMetric(CoreMetric.API_CALL_DURATION, duration); + if (errorType != null) { + root.createChild("ApiCallAttempt").reportMetric(CoreMetric.ERROR_TYPE, errorType.toString()); + } + return root.collect(); + } + + private double doubleValue(final String name) { + return (double) metrics.metric(metrics.metricName(name, GROUP)).metricValue(); + } + + private double doubleValue(final String name, final Map tags) { + return (double) taggedMetric(name, tags).metricValue(); + } + + private KafkaMetric taggedMetric(final String name, final Map tags) { + return metrics.metric(metrics.metricName(name, GROUP, "", tags)); + } +}