HDDS-10362. Add S3 Gateway metric for pending operations

hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/PendingOperationsFilter.java

@@ -0,0 +1,126 @@
+/*
+ * 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.
+ *
+ * <p>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.
+ *
+ * <p>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;
+
+    // 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.
+        final AtomicBoolean decremented = new AtomicBoolean();
+        responseContext.setEntityStream(new WrappedOutputStream(out) {
+          @Override
+          public void close() throws IOException {
+            // finally, so a failure while closing still releases the count.
+            try {
+              super.close();
+            } finally {
+              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;
+  }
+}

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;
@@ -399,6 +404,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);
@@ -789,6 +797,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();
   }

hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/TestPendingOperationsFilter.java

@@ -0,0 +1,194 @@
+/*
+ * 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.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;
+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.hasEntity()).thenReturn(true);
+    when(response.getEntityStream()).thenReturn(new ByteArrayOutputStream());
+    ArgumentCaptor<OutputStream> 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());
+  }
+
+  @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<OutputStream> 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();
+    // 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<String, Object> 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;
+  }
+}