HDDS-16238. Latency metrics for internal OM requests are not populated

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java

@@ -5912,6 +5912,10 @@ public OMExecutionFlow getOmExecutionFlow() {
     return omExecutionFlow;
   }
 
+  public ProtocolMessageMetrics<OzoneManagerProtocolProtos.Type> getOmClientProtocolMetrics() {
+    return omClientProtocolMetrics;
+  }
+
   /**
    * OM Startup mode.
    */

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/utils/OzoneManagerRatisUtils.java

@@ -109,6 +109,7 @@
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
 import org.apache.ratis.grpc.GrpcTlsConfig;
 import org.apache.ratis.protocol.ClientId;
+import org.apache.ratis.util.UncheckedAutoCloseable;
 import org.rocksdb.RocksDBException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -526,7 +527,12 @@ public static GrpcTlsConfig createServerTlsConfig(SecurityConfig conf,
 
   public static OzoneManagerProtocolProtos.OMResponse submitRequest(
       OzoneManager om, OMRequest omRequest, ClientId clientId, long callId) throws ServiceException {
-    return om.getOmRatisServer().submitRequest(omRequest, clientId, callId);
+    // Internally-submitted requests (e.g. PurgeKeys) bypass the RPC endpoint dispatcher, so measure
+    // them here to populate the same OmClientProtocol per-type metrics that client requests get.
+    try (UncheckedAutoCloseable ignored =
+        om.getOmClientProtocolMetrics().measure(omRequest.getCmdType())) {
+      return om.getOmRatisServer().submitRequest(omRequest, clientId, callId);
+    }
   }
 
   public static OzoneManagerProtocolProtos.OMResponse createErrorResponse(

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestTrashOzoneFileSystem.java

@@ -20,6 +20,9 @@
 import static org.apache.hadoop.fs.FileSystem.TRASH_PREFIX;
 import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_CONTAINER_REPORT_INTERVAL;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_LISTING_PAGE_SIZE_MAX;
+import static org.apache.hadoop.ozone.om.ratis.utils.ProtocolMessageMetricsTestUtils.getRequestCount;
+import static org.apache.hadoop.ozone.om.ratis.utils.ProtocolMessageMetricsTestUtils.getRequestTime;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
 import java.io.File;
@@ -30,6 +33,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.server.ServerUtils;
 import org.apache.hadoop.hdds.utils.db.DBConfigFromFile;
@@ -39,7 +43,9 @@
 import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
 import org.apache.hadoop.ozone.om.protocol.OzoneManagerProtocol;
 import org.apache.hadoop.ozone.om.request.OMRequestTestUtils;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
 import org.apache.hadoop.security.SecurityUtil;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
@@ -94,6 +100,82 @@ void testGetTrashRootsBeyondPageSize(BucketLayout bucketLayout,
     }
   }
 
+  /**
+   * The Trash emptier renames and deletes trash directories in FSO buckets through the
+   * TrashOzoneFileSystem, submitting internal {@code RenameKey} and {@code DeleteKey} requests.
+   * Both should be counted in the OmClientProtocol per-type metrics just like a client request.
+   */
+  @Test
+  void testRenameAndDeleteInFsoBucketIncrementMetrics(@TempDir File testDir) throws Exception {
+    OmTestManagers omTestManagers = newOmTestManagers(testDir);
+    try {
+      OzoneManager om = omTestManagers.getOzoneManager();
+      OzoneManagerProtocol writeClient = omTestManagers.getWriteClient();
+      final String volumeName = "vol-" + objectId.incrementAndGet();
+      final String bucketName = "bucket-" + objectId.incrementAndGet();
+      createVolumeAndBucket(omTestManagers, volumeName, bucketName,
+          BucketLayout.FILE_SYSTEM_OPTIMIZED, writeClient);
+      createDirectory(writeClient, volumeName, bucketName, TRASH_PREFIX + "/user1");
+
+      Path src = trashPath(volumeName, bucketName, "user1");
+      Path dst = trashPath(volumeName, bucketName, "user2");
+      try (FileSystem fs = SecurityUtil.doAsLoginUser(
+          (PrivilegedExceptionAction<FileSystem>) () -> new TrashOzoneFileSystem(om))) {
+        // This OM is freshly created, so no RenameKey call has been recorded yet.
+        assertEquals(0, getRequestCount(om.getOmClientProtocolMetrics(), Type.RenameKey));
+        assertEquals(0, getRequestTime(om.getOmClientProtocolMetrics(), Type.RenameKey));
+        fs.rename(src, dst);
+        assertThat(getRequestCount(om.getOmClientProtocolMetrics(), Type.RenameKey)).isGreaterThan(0L);
+        assertThat(getRequestTime(om.getOmClientProtocolMetrics(), Type.RenameKey)).isGreaterThan(0L);
+
+        // Likewise no DeleteKey call has been recorded yet.
+        assertEquals(0, getRequestCount(om.getOmClientProtocolMetrics(), Type.DeleteKey));
+        assertEquals(0, getRequestTime(om.getOmClientProtocolMetrics(), Type.DeleteKey));
+        fs.delete(dst, true);
+        assertThat(getRequestCount(om.getOmClientProtocolMetrics(), Type.DeleteKey)).isGreaterThan(0L);
+        assertThat(getRequestTime(om.getOmClientProtocolMetrics(), Type.DeleteKey)).isGreaterThan(0L);
+      }
+    } finally {
+      omTestManagers.stop();
+    }
+  }
+
+  /**
+   * The Trash emptier deletes trash contents in non-FSO buckets one key at a time through the
+   * TrashOzoneFileSystem, submitting an internal {@code DeleteKeys} request per key. These should
+   * be counted in the OmClientProtocol per-type metrics just like a client request.
+   */
+  @Test
+  void deleteInObjectStoreBucketIncrementsDeleteKeysMetric(@TempDir File testDir) throws Exception {
+    OmTestManagers omTestManagers = newOmTestManagers(testDir);
+    try {
+      OzoneManager om = omTestManagers.getOzoneManager();
+      OzoneManagerProtocol writeClient = omTestManagers.getWriteClient();
+      final String volumeName = "vol-" + objectId.incrementAndGet();
+      final String bucketName = "bucket-" + objectId.incrementAndGet();
+      createVolumeAndBucket(omTestManagers, volumeName, bucketName,
+          BucketLayout.OBJECT_STORE, writeClient);
+      createDirectory(writeClient, volumeName, bucketName, TRASH_PREFIX + "/user1");
+
+      Path trashRoot = new Path("/" + volumeName + "/" + bucketName + "/" + TRASH_PREFIX);
+      try (FileSystem fs = SecurityUtil.doAsLoginUser(
+          (PrivilegedExceptionAction<FileSystem>) () -> new TrashOzoneFileSystem(om))) {
+        // This OM is freshly created, so no DeleteKeys call has been recorded yet.
+        assertEquals(0, getRequestCount(om.getOmClientProtocolMetrics(), Type.DeleteKeys));
+        assertEquals(0, getRequestTime(om.getOmClientProtocolMetrics(), Type.DeleteKeys));
+        fs.delete(trashRoot, true);
+        assertThat(getRequestCount(om.getOmClientProtocolMetrics(), Type.DeleteKeys)).isGreaterThan(0L);
+        assertThat(getRequestTime(om.getOmClientProtocolMetrics(), Type.DeleteKeys)).isGreaterThan(0L);
+      }
+    } finally {
+      omTestManagers.stop();
+    }
+  }
+
+  private static Path trashPath(String volumeName, String bucketName, String userDir) {
+    return new Path("/" + volumeName + "/" + bucketName + "/" + TRASH_PREFIX + "/" + userDir);
+  }
+
   private void createVolumeAndBucket(OmTestManagers omTestManagers,
       String volumeName, String bucketName, BucketLayout bucketLayout,
       OzoneManagerProtocol writeClient) throws IOException {

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/utils/ProtocolMessageMetricsTestUtils.java

@@ -0,0 +1,73 @@
+/*
+ * 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.om.ratis.utils;
+
+import org.apache.hadoop.hdds.utils.ProtocolMessageMetrics;
+import org.apache.hadoop.metrics2.AbstractMetric;
+import org.apache.hadoop.metrics2.MetricsRecord;
+import org.apache.hadoop.metrics2.MetricsTag;
+import org.apache.hadoop.metrics2.impl.MetricsCollectorImpl;
+
+/**
+ * Test helpers for reading counters back out of a live {@link ProtocolMessageMetrics}.
+ */
+public final class ProtocolMessageMetricsTestUtils {
+
+  private ProtocolMessageMetricsTestUtils() {
+  }
+
+  /**
+   * Reads the {@code counter} value (number of calls) recorded for the given request type from the
+   * live {@link ProtocolMessageMetrics} source. Returns {@code 0} if the type has no recorded calls.
+   */
+  public static long getRequestCount(ProtocolMessageMetrics<?> metrics, Enum<?> type) {
+    return readMetric(metrics, type, "counter");
+  }
+
+  /**
+   * Reads the {@code time} value (summed call latency, in milliseconds) recorded for the given
+   * request type from the live {@link ProtocolMessageMetrics} source. Returns {@code 0} if the type
+   * has no recorded calls.
+   */
+  public static long getRequestTime(ProtocolMessageMetrics<?> metrics, Enum<?> type) {
+    return readMetric(metrics, type, "time");
+  }
+
+  private static long readMetric(ProtocolMessageMetrics<?> metrics, Enum<?> type, String metricName) {
+    MetricsCollectorImpl collector = new MetricsCollectorImpl();
+    metrics.getMetrics(collector, true);
+    for (MetricsRecord record : collector.getRecords()) {
+      boolean matchesType = false;
+      for (MetricsTag tag : record.tags()) {
+        if ("type".equals(tag.name()) && type.toString().equals(tag.value())) {
+          matchesType = true;
+          break;
+        }
+      }
+      if (!matchesType) {
+        continue;
+      }
+      for (AbstractMetric metric : record.metrics()) {
+        if (metricName.equals(metric.name())) {
+          return metric.value().longValue();
+        }
+      }
+    }
+    return 0;
+  }
+}

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/utils/TestOzoneManagerRatisUtils.java

@@ -0,0 +1,132 @@
+/*
+ * 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.om.ratis.utils;
+
+import static org.apache.hadoop.ozone.om.ratis.utils.ProtocolMessageMetricsTestUtils.getRequestCount;
+import static org.apache.hadoop.ozone.om.ratis.utils.ProtocolMessageMetricsTestUtils.getRequestTime;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.protobuf.ServiceException;
+import org.apache.hadoop.hdds.utils.ProtocolMessageMetrics;
+import org.apache.hadoop.ozone.om.OzoneManager;
+import org.apache.hadoop.ozone.om.ratis.OzoneManagerRatisServer;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
+import org.apache.ratis.protocol.ClientId;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for {@link OzoneManagerRatisUtils#submitRequest}, in particular that
+ * internally-submitted requests populate the OmClientProtocol per-type metrics.
+ */
+public class TestOzoneManagerRatisUtils {
+
+  private OzoneManager ozoneManager;
+  private OzoneManagerRatisServer ratisServer;
+  private ProtocolMessageMetrics<Type> metrics;
+
+  @BeforeEach
+  public void setup() {
+    ozoneManager = mock(OzoneManager.class);
+    ratisServer = mock(OzoneManagerRatisServer.class);
+    metrics = ProtocolMessageMetrics.create(
+        "OmClientProtocol", "Ozone Manager RPC endpoint", Type.class);
+    when(ozoneManager.getOmClientProtocolMetrics()).thenReturn(metrics);
+    when(ozoneManager.getOmRatisServer()).thenReturn(ratisServer);
+  }
+
+  @Test
+  public void testSubmitRequestRecordsMetricForRequestType() throws Exception {
+    OMRequest request = newRequest(Type.PurgeKeys);
+    OMResponse expected = newResponse(Type.PurgeKeys);
+    mockSubmitRequestWithDelay(expected);
+
+    OMResponse actual = OzoneManagerRatisUtils.submitRequest(
+        ozoneManager, request, ClientId.randomId(), 1L);
+
+    assertSame(expected, actual);
+    assertEquals(1, getRequestCount(metrics, Type.PurgeKeys));
+    assertThat(getRequestTime(metrics, Type.PurgeKeys)).isGreaterThan(0L);
+    // Only the submitted type should be counted.
+    assertEquals(0, getRequestCount(metrics, Type.RenameKey));
+    assertEquals(0, getRequestTime(metrics, Type.RenameKey));
+  }
+
+  @Test
+  public void testSubmitRequestIncrementsMetricPerCall() throws Exception {
+    OMRequest request = newRequest(Type.PurgeKeys);
+    mockSubmitRequestWithDelay(newResponse(Type.PurgeKeys));
+
+    OzoneManagerRatisUtils.submitRequest(ozoneManager, request, ClientId.randomId(), 1L);
+    OzoneManagerRatisUtils.submitRequest(ozoneManager, request, ClientId.randomId(), 2L);
+
+    assertEquals(2, getRequestCount(metrics, Type.PurgeKeys));
+    assertThat(getRequestTime(metrics, Type.PurgeKeys)).isGreaterThan(0L);
+  }
+
+  @Test
+  public void testSubmitRequestRecordsMetricOnFailure() throws Exception {
+    OMRequest request = newRequest(Type.PurgeKeys);
+    // Sleep briefly before failing so the measured latency is reliably greater than zero.
+    doAnswer(invocation -> {
+      Thread.sleep(2);
+      throw new ServiceException("submit failed");
+    }).when(ratisServer).submitRequest(any(OMRequest.class), any(ClientId.class), anyLong());
+
+    assertThrows(ServiceException.class, () -> OzoneManagerRatisUtils.submitRequest(
+        ozoneManager, request, ClientId.randomId(), 1L));
+
+    // The measurement wraps the submission, so the metric is recorded even when it fails.
+    assertEquals(1, getRequestCount(metrics, Type.PurgeKeys));
+    assertThat(getRequestTime(metrics, Type.PurgeKeys)).isGreaterThan(0L);
+  }
+
+  private void mockSubmitRequestWithDelay(OMResponse expectedResponse) throws ServiceException {
+    // Sleep briefly inside the submission so the measured latency is reliably greater than zero.
+    doAnswer(invocation -> {
+      Thread.sleep(2);
+      return expectedResponse;
+    }).when(ratisServer).submitRequest(any(OMRequest.class), any(ClientId.class), anyLong());
+  }
+
+  private static OMRequest newRequest(Type type) {
+    return OMRequest.newBuilder()
+        .setCmdType(type)
+        .setClientId(ClientId.randomId().toString())
+        .build();
+  }
+
+  private static OMResponse newResponse(Type type) {
+    return OMResponse.newBuilder()
+        .setCmdType(type)
+        .setStatus(Status.OK)
+        .setSuccess(true)
+        .build();
+  }
+}

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequestTests.java

@@ -53,6 +53,7 @@
 import org.apache.hadoop.hdds.scm.protocol.ScmBlockLocationProtocol;
 import org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationProtocol;
 import org.apache.hadoop.hdds.security.token.OzoneBlockTokenSecretManager;
+import org.apache.hadoop.hdds.utils.ProtocolMessageMetrics;
 import org.apache.hadoop.hdds.utils.TransactionInfo;
 import org.apache.hadoop.hdds.utils.db.BatchOperation;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
@@ -118,6 +119,7 @@ public class OMKeyRequestTests {
   protected StorageContainerLocationProtocol scmContainerLocationProtocol;
   protected OMPerformanceMetrics perfMetrics;
   protected DeletingServiceMetrics delMetrics;
+  protected ProtocolMessageMetrics<OzoneManagerProtocolProtos.Type> omClientProtocolMetrics;
 
   protected static final long CONTAINER_ID = 1000L;
   protected static final long LOCAL_ID = 100L;
@@ -140,6 +142,9 @@ public void setup() throws Exception {
     omMetrics = OMMetrics.create(ozoneConfiguration);
     perfMetrics = OMPerformanceMetrics.register();
     delMetrics = DeletingServiceMetrics.create();
+    omClientProtocolMetrics = ProtocolMessageMetrics.create(
+        "OmClientProtocol", "Ozone Manager RPC endpoint",
+        OzoneManagerProtocolProtos.Type.class);
     ozoneConfiguration.set(OMConfigKeys.OZONE_OM_DB_DIRS,
         folder.toAbsolutePath().toString());
     ozoneConfiguration.set(OzoneConfigKeys.OZONE_METADATA_DIRS,
@@ -151,6 +156,7 @@ public void setup() throws Exception {
     when(ozoneManager.getMetrics()).thenReturn(omMetrics);
     when(ozoneManager.getPerfMetrics()).thenReturn(perfMetrics);
     when(ozoneManager.getDeletionMetrics()).thenReturn(delMetrics);
+    when(ozoneManager.getOmClientProtocolMetrics()).thenReturn(omClientProtocolMetrics);
     when(ozoneManager.getMetadataManager()).thenReturn(omMetadataManager);
     when(ozoneManager.getConfiguration()).thenReturn(ozoneConfiguration);
     when(ozoneManager.getConfig()).thenReturn(ozoneConfiguration.getObject(OmConfig.class));