From b91167202b603d39ab487ea71e0e0a7586a5847f Mon Sep 17 00:00:00 2001 From: rich7420 Date: Mon, 10 Aug 2026 23:12:59 +0800 Subject: [PATCH 1/2] HDDS-10362. Add S3 Gateway metric for pending operations --- .../ozone/s3/PendingOperationsFilter.java | 119 +++++++++++++++ .../ozone/s3/metrics/S3GatewayMetrics.java | 20 +++ .../ozone/s3/TestPendingOperationsFilter.java | 137 ++++++++++++++++++ 3 files changed, 276 insertions(+) create mode 100644 hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/PendingOperationsFilter.java create mode 100644 hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/TestPendingOperationsFilter.java diff --git a/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/PendingOperationsFilter.java b/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/PendingOperationsFilter.java new file mode 100644 index 000000000000..f33c89101e06 --- /dev/null +++ b/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/PendingOperationsFilter.java @@ -0,0 +1,119 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.ozone.s3; + +import com.google.common.annotations.VisibleForTesting; +import java.io.IOException; +import java.io.OutputStream; +import java.util.concurrent.atomic.AtomicBoolean; +import javax.ws.rs.container.ContainerRequestContext; +import javax.ws.rs.container.ContainerRequestFilter; +import javax.ws.rs.container.ContainerResponseContext; +import javax.ws.rs.container.ContainerResponseFilter; +import javax.ws.rs.container.ResourceInfo; +import javax.ws.rs.core.Context; +import javax.ws.rs.ext.Provider; +import org.apache.hadoop.ozone.client.io.WrappedOutputStream; +import org.apache.hadoop.ozone.s3.metrics.S3GatewayMetrics; + +/** + * Tracks the number of S3 operations being processed but not yet fully replied + * to (in-flight), exposed as the {@code pendingOperations} gauge in + * {@link S3GatewayMetrics}. Modeled after the xceiver client {@code pendingOps} + * metric (HDDS-10362): increment when the operation starts, decrement when its + * response is complete. + * + *

The filter is post-matching, so it counts requests that reached an + * endpoint; requests rejected earlier (for example by the {@code @PreMatching} + * {@link AuthorizationFilter}) are not counted, since they are not operations + * the gateway performs. The request side increments and remembers the metrics + * instance on the request; the response side decrements only that same + * instance, so a request that never incremented can never drive the gauge + * negative. + * + *

For a streaming {@code GetObject}, the JAX-RS response filter runs before + * the body is written, so the decrement is deferred to the point where the + * response stream is closed (mirroring {@link TracingFilter}); the operation + * stays "pending" until the object body has finished streaming to the client. + */ +@Provider +public class PendingOperationsFilter implements ContainerRequestFilter, + ContainerResponseFilter { + + private static final String COUNTED_METRICS = "PENDING_OPERATION_METRICS"; + private static final String HTTP_GET_METHOD = "GET"; + private static final String OBJECT_ENDPOINT_CLASS_NAME = "ObjectEndpoint"; + private static final String OBJECT_GET_METHOD_NAME = "get"; + + @Context + private ResourceInfo resourceInfo; + + @Override + public void filter(ContainerRequestContext requestContext) { + S3GatewayMetrics metrics = S3GatewayMetrics.getMetrics(); + if (metrics != null) { + metrics.incrPendingOperations(); + requestContext.setProperty(COUNTED_METRICS, metrics); + } + } + + @Override + public void filter(ContainerRequestContext requestContext, + ContainerResponseContext responseContext) { + Object counted = requestContext.getProperty(COUNTED_METRICS); + if (!(counted instanceof S3GatewayMetrics)) { + return; + } + requestContext.removeProperty(COUNTED_METRICS); + final S3GatewayMetrics metrics = (S3GatewayMetrics) counted; + + if (isStreamingGetObject(requestContext)) { + OutputStream out = responseContext.getEntityStream(); + if (out != null) { + // Decrement only once the body has been fully streamed to the client. + final AtomicBoolean decremented = new AtomicBoolean(); + responseContext.setEntityStream(new WrappedOutputStream(out) { + @Override + public void close() throws IOException { + super.close(); + if (decremented.compareAndSet(false, true)) { + metrics.decrPendingOperations(); + } + } + }); + return; + } + } + metrics.decrPendingOperations(); + } + + private boolean isStreamingGetObject(ContainerRequestContext requestContext) { + if (!HTTP_GET_METHOD.equalsIgnoreCase(requestContext.getMethod())) { + return false; + } + String cls = resourceInfo.getResourceClass().getSimpleName(); + String method = resourceInfo.getResourceMethod().getName(); + return OBJECT_ENDPOINT_CLASS_NAME.equals(cls) + && OBJECT_GET_METHOD_NAME.equals(method); + } + + @VisibleForTesting + void setResourceInfo(ResourceInfo info) { + this.resourceInfo = info; + } +} diff --git a/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/metrics/S3GatewayMetrics.java b/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/metrics/S3GatewayMetrics.java index 160102e72905..61769a923b68 100644 --- a/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/metrics/S3GatewayMetrics.java +++ b/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/metrics/S3GatewayMetrics.java @@ -31,6 +31,7 @@ import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem; import org.apache.hadoop.metrics2.lib.MetricsRegistry; import org.apache.hadoop.metrics2.lib.MutableCounterLong; +import org.apache.hadoop.metrics2.lib.MutableGaugeLong; import org.apache.hadoop.ozone.OzoneConsts; import org.apache.hadoop.ozone.s3.S3GatewayConfigKeys; import org.apache.hadoop.ozone.util.PerformanceMetrics; @@ -71,6 +72,10 @@ public final class S3GatewayMetrics implements Closeable, MetricsSource { private @Metric MutableCounterLong listS3BucketsSuccess; private @Metric MutableCounterLong listS3BucketsFailure; + // Gateway-wide: S3 operations being processed, held until the response is + // fully sent (in-flight). A gauge, so it can rise and fall. + private @Metric MutableGaugeLong pendingOperations; + // ObjectEndpoint private @Metric MutableCounterLong createMultipartKeySuccess; private @Metric MutableCounterLong createMultipartKeyFailure; @@ -391,6 +396,9 @@ public void getMetrics(MetricsCollector collector, boolean all) { listS3BucketsFailure.snapshot(recordBuilder, true); listS3BucketsFailureLatencyNs.snapshot(recordBuilder, true); + // Gateway-wide + pendingOperations.snapshot(recordBuilder, true); + // ObjectEndpoint createMultipartKeySuccess.snapshot(recordBuilder, true); createMultipartKeySuccessLatencyNs.snapshot(recordBuilder, true); @@ -767,6 +775,18 @@ public long getHeadBucketSuccess() { return headBucketSuccess.value(); } + public void incrPendingOperations() { + pendingOperations.incr(); + } + + public void decrPendingOperations() { + pendingOperations.decr(); + } + + public long getPendingOperations() { + return pendingOperations.value(); + } + public long getHeadKeySuccess() { return headKeySuccess.value(); } diff --git a/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/TestPendingOperationsFilter.java b/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/TestPendingOperationsFilter.java new file mode 100644 index 000000000000..1685fa05ebc0 --- /dev/null +++ b/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/TestPendingOperationsFilter.java @@ -0,0 +1,137 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.ozone.s3; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.ByteArrayOutputStream; +import java.io.OutputStream; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import javax.ws.rs.container.ContainerRequestContext; +import javax.ws.rs.container.ContainerResponseContext; +import javax.ws.rs.container.ResourceInfo; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.ozone.s3.endpoint.ObjectEndpoint; +import org.apache.hadoop.ozone.s3.metrics.S3GatewayMetrics; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +/** + * Tests {@link PendingOperationsFilter}: the in-flight gauge rises on the + * request side and returns on the response side; a response without a matching + * request never drives it negative; and a streaming {@code GetObject} stays + * pending until its response body is fully written. + */ +public class TestPendingOperationsFilter { + + private final PendingOperationsFilter filter = new PendingOperationsFilter(); + + @Test + public void nonStreamingOperationIncrementsThenDecrements() { + S3GatewayMetrics metrics = S3GatewayMetrics.create(new OzoneConfiguration()); + ContainerRequestContext request = mockRequestContext("PUT"); + + long before = metrics.getPendingOperations(); + + filter.filter(request); + assertEquals(1L, metrics.getPendingOperations() - before); + + filter.filter(request, mock(ContainerResponseContext.class)); + assertEquals(before, metrics.getPendingOperations()); + } + + @Test + public void responseWithoutRequestDoesNotDecrement() { + S3GatewayMetrics metrics = S3GatewayMetrics.create(new OzoneConfiguration()); + ContainerRequestContext request = mockRequestContext("PUT"); + + long before = metrics.getPendingOperations(); + + // No request-side filter ran, so the response side must be a no-op. + filter.filter(request, mock(ContainerResponseContext.class)); + assertEquals(before, metrics.getPendingOperations()); + } + + @Test + public void streamingGetObjectStaysPendingUntilStreamClose() throws Exception { + S3GatewayMetrics metrics = S3GatewayMetrics.create(new OzoneConfiguration()); + filter.setResourceInfo(objectEndpointGetResourceInfo()); + ContainerRequestContext request = mockRequestContext("GET"); + + long before = metrics.getPendingOperations(); + + filter.filter(request); + assertEquals(1L, metrics.getPendingOperations() - before); + + ContainerResponseContext response = mock(ContainerResponseContext.class); + when(response.getEntityStream()).thenReturn(new ByteArrayOutputStream()); + ArgumentCaptor wrapped = + ArgumentCaptor.forClass(OutputStream.class); + + filter.filter(request, response); + + // The response filter ran but the body is not written yet, so the + // operation is still pending; the entity stream was wrapped instead. + assertEquals(1L, metrics.getPendingOperations() - before); + verify(response).setEntityStream(wrapped.capture()); + + // Closing the wrapped stream (body finished streaming) decrements once. + wrapped.getValue().close(); + assertEquals(before, metrics.getPendingOperations()); + + // A second close must not drive the gauge negative. + wrapped.getValue().close(); + assertEquals(before, metrics.getPendingOperations()); + } + + private static ResourceInfo objectEndpointGetResourceInfo() throws Exception { + ResourceInfo info = mock(ResourceInfo.class); + doReturn(ObjectEndpoint.class).when(info).getResourceClass(); + // Any Method whose name is "get" satisfies the streaming check. + when(info.getResourceMethod()).thenReturn(List.class.getMethod("get", int.class)); + return info; + } + + /** + * A mock request context whose property map behaves like a real one, so the + * request/response filter pair can hand the counted metrics instance across. + */ + private static ContainerRequestContext mockRequestContext(String method) { + Map properties = new HashMap<>(); + ContainerRequestContext context = mock(ContainerRequestContext.class); + when(context.getMethod()).thenReturn(method); + doAnswer(inv -> properties.put(inv.getArgument(0), inv.getArgument(1))) + .when(context).setProperty(anyString(), any()); + when(context.getProperty(anyString())) + .thenAnswer(inv -> properties.get(inv.getArgument(0))); + doAnswer(inv -> { + properties.remove(inv.getArgument(0)); + return null; + }).when(context).removeProperty(anyString()); + return context; + } +} From 9e801dda1c3d3fa8c1e849ca691b128f000996ed Mon Sep 17 00:00:00 2001 From: rich7420 Date: Mon, 17 Aug 2026 21:44:24 +0800 Subject: [PATCH 2/2] HDDS-10362. Release pending count for body-less responses and on close failure --- .../ozone/s3/PendingOperationsFilter.java | 15 +++-- .../ozone/s3/TestPendingOperationsFilter.java | 57 +++++++++++++++++++ 2 files changed, 68 insertions(+), 4 deletions(-) diff --git a/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/PendingOperationsFilter.java b/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/PendingOperationsFilter.java index f33c89101e06..6d5d25520561 100644 --- a/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/PendingOperationsFilter.java +++ b/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/PendingOperationsFilter.java @@ -82,7 +82,10 @@ public void filter(ContainerRequestContext requestContext, requestContext.removeProperty(COUNTED_METRICS); final S3GatewayMetrics metrics = (S3GatewayMetrics) counted; - if (isStreamingGetObject(requestContext)) { + // Defer the decrement only when there is a body to stream. A body-less + // response (for example 304 Not Modified) has an entity stream that the + // container never closes, so decrement it immediately instead. + if (isStreamingGetObject(requestContext) && responseContext.hasEntity()) { OutputStream out = responseContext.getEntityStream(); if (out != null) { // Decrement only once the body has been fully streamed to the client. @@ -90,9 +93,13 @@ public void filter(ContainerRequestContext requestContext, responseContext.setEntityStream(new WrappedOutputStream(out) { @Override public void close() throws IOException { - super.close(); - if (decremented.compareAndSet(false, true)) { - metrics.decrPendingOperations(); + // finally, so a failure while closing still releases the count. + try { + super.close(); + } finally { + if (decremented.compareAndSet(false, true)) { + metrics.decrPendingOperations(); + } } } }); diff --git a/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/TestPendingOperationsFilter.java b/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/TestPendingOperationsFilter.java index 1685fa05ebc0..4115c172cfb5 100644 --- a/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/TestPendingOperationsFilter.java +++ b/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/TestPendingOperationsFilter.java @@ -18,15 +18,18 @@ package org.apache.hadoop.ozone.s3; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.io.ByteArrayOutputStream; +import java.io.IOException; import java.io.OutputStream; import java.util.HashMap; import java.util.List; @@ -88,6 +91,7 @@ public void streamingGetObjectStaysPendingUntilStreamClose() throws Exception { assertEquals(1L, metrics.getPendingOperations() - before); ContainerResponseContext response = mock(ContainerResponseContext.class); + when(response.hasEntity()).thenReturn(true); when(response.getEntityStream()).thenReturn(new ByteArrayOutputStream()); ArgumentCaptor wrapped = ArgumentCaptor.forClass(OutputStream.class); @@ -108,6 +112,59 @@ public void streamingGetObjectStaysPendingUntilStreamClose() throws Exception { assertEquals(before, metrics.getPendingOperations()); } + @Test + public void bodylessStreamingGetObjectDecrementsImmediately() throws Exception { + S3GatewayMetrics metrics = S3GatewayMetrics.create(new OzoneConfiguration()); + filter.setResourceInfo(objectEndpointGetResourceInfo()); + ContainerRequestContext request = mockRequestContext("GET"); + + long before = metrics.getPendingOperations(); + filter.filter(request); + + // A GetObject with no body (for example 304 Not Modified): the entity + // stream would never be closed, so the count must be released right away. + ContainerResponseContext response = mock(ContainerResponseContext.class); + when(response.hasEntity()).thenReturn(false); + + filter.filter(request, response); + + assertEquals(before, metrics.getPendingOperations()); + verify(response, never()).setEntityStream(any()); + } + + @Test + public void decrementsEvenWhenStreamCloseThrows() throws Exception { + S3GatewayMetrics metrics = S3GatewayMetrics.create(new OzoneConfiguration()); + filter.setResourceInfo(objectEndpointGetResourceInfo()); + ContainerRequestContext request = mockRequestContext("GET"); + + long before = metrics.getPendingOperations(); + filter.filter(request); + + ContainerResponseContext response = mock(ContainerResponseContext.class); + when(response.hasEntity()).thenReturn(true); + OutputStream failing = new OutputStream() { + @Override + public void write(int b) { + } + + @Override + public void close() throws IOException { + throw new IOException("close failed"); + } + }; + when(response.getEntityStream()).thenReturn(failing); + ArgumentCaptor wrapped = + ArgumentCaptor.forClass(OutputStream.class); + + filter.filter(request, response); + verify(response).setEntityStream(wrapped.capture()); + + // close() propagates its failure, but the count is still released. + assertThrows(IOException.class, () -> wrapped.getValue().close()); + assertEquals(before, metrics.getPendingOperations()); + } + private static ResourceInfo objectEndpointGetResourceInfo() throws Exception { ResourceInfo info = mock(ResourceInfo.class); doReturn(ObjectEndpoint.class).when(info).getResourceClass();