Skip to content

feat(inkless:storage): split S3 error metrics by operation #716

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jeqo wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(inkless:storage): split S3 error metrics by operation #716

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions docs/inkless/metrics.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
==================================
Expand Down
65 changes: 65 additions & 0 deletions storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/MetricCollector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -117,6 +129,10 @@ public class MetricCollector implements MetricPublisher {
private final Map<String, Sensor> requestMetrics = new HashMap<>();
private final Map<String, Sensor> latencyMetrics = new HashMap<>();
private final Map<String, Sensor> errorMetrics = new HashMap<>();
// {rate, total} templates for the per-operation (operation-tagged) variant of each error type.
private final Map<String, MetricNameTemplate[]> errorByOperationTemplates = new HashMap<>();
// Lazily created operation-tagged error sensors, keyed by errorType + ":" + operationName.
private final Map<String, Sensor> errorByOperationMetrics = new ConcurrentHashMap<>();

public MetricCollector(Metrics metrics) {
this.metrics = metrics;
Expand Down Expand Up @@ -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<String, String> 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(
Expand Down Expand Up @@ -275,6 +331,7 @@ private Sensor createLatencySensor(
public void publish(final MetricCollection metricCollection) {
final List<String> 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);
Expand Down Expand Up @@ -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();
}
}
}
}

Expand Down
78 changes: 77 additions & 1 deletion storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/MetricRegistry.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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<MetricNameTemplate> all() {
return List.of(
GET_OBJECT_REQUESTS_RATE_METRIC_NAME,
Expand Down Expand Up @@ -373,7 +439,17 @@ public static List<MetricNameTemplate> 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
);
}
}
121 changes: 121 additions & 0 deletions storage/inkless/src/test/java/io/aiven/inkless/storage_backend/s3/MetricCollectorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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 <http://www.gnu.org/licenses/>.
*/
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";
Comment thread
jeqo marked this conversation as resolved.

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<String, String> tags) {
return (double) taggedMetric(name, tags).metricValue();
}

private KafkaMetric taggedMetric(final String name, final Map<String, String> tags) {
return metrics.metric(metrics.metricName(name, GROUP, "", tags));
}
}
Loading