[#5575] feat(metrics):add Hive Catalog Metrics for HMS connection pool

catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveCatalogOperations.java

@@ -51,6 +51,7 @@
 import java.util.stream.Collectors;
 import org.apache.commons.lang3.ArrayUtils;
 import org.apache.gravitino.Catalog;
+import org.apache.gravitino.GravitinoEnv;
 import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.Namespace;
 import org.apache.gravitino.SchemaChange;
@@ -72,6 +73,9 @@
 import org.apache.gravitino.hive.HiveSchema;
 import org.apache.gravitino.hive.HiveTable;
 import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.metrics.MetricsSystem;
+import org.apache.gravitino.metrics.source.HiveCatalogMetricsSource;
+import org.apache.gravitino.metrics.source.MetricsProvider;
 import org.apache.gravitino.rel.Column;
 import org.apache.gravitino.rel.Representation;
 import org.apache.gravitino.rel.Table;
@@ -111,6 +115,8 @@ public class HiveCatalogOperations
   private HiveViewCatalogOperations viewCatalogOperations;
 
   private boolean listAllTables = true;
+
+  private HiveCatalogMetricsSource catalogMetricsSource;
   // The maximum number of tables that can be returned by the listTableNamesByFilter function.
   // The default value is -1, which means that all tables are returned.
   private static final short MAX_TABLES = -1;
@@ -145,6 +151,30 @@ public void initialize(
     String catalogKey =
         String.format("hive-%s", info == null || info.id() == null ? "0" : info.id());
     this.clientPool = new CachedClientPool(catalogKey, prop, conf);
+    MetricsSystem metricsSystem = GravitinoEnv.getInstance().metricsSystem();
+    // Metrics System could be null in UT.
+    if (metricsSystem != null) {
+      this.catalogMetricsSource =
+          new HiveCatalogMetricsSource(info.namespace().toString(), info.name());
+      catalogMetricsSource.registerClientPoolMetrics(
+          new MetricsProvider() {
+            @Override
+            public int currentSize() {
+              return clientPool.totalCurrentSize();
+            }
+
+            @Override
+            public int idleCount() {
+              return clientPool.totalIdleCount();
+            }
+
+            @Override
+            public int poolSize() {
+              return clientPool.totalPoolSize();
+            }
+          });
+      metricsSystem.register(catalogMetricsSource);
+    }
     this.listAllTables = enableListAllTables(conf);
 
     // Initialize the HMS catalog name from catalog properties (default to DEFAULT_HMS_CATALOG)
@@ -193,6 +223,11 @@ boolean enableListAllTables(Map<String, String> conf) {
   /** Closes the Hive catalog and releases the associated client pool. */
   @Override
   public void close() {
+    // Metrics System could be null in UT.
+    MetricsSystem metricsSystem = GravitinoEnv.getInstance().metricsSystem();
+    if (metricsSystem != null) {
+      metricsSystem.unregister(catalogMetricsSource);
+    }
     if (clientPool != null) {
       clientPool.close();
       clientPool = null;

catalogs/hive-metastore-common/src/main/java/org/apache/gravitino/hive/CachedClientPool.java

@@ -148,6 +148,27 @@ private static ThreadFactory newDaemonThreadFactory() {
         .build();
   }
 
+  /**
+   * Returns the total number of connections currently created across all per-user pools. This
+   * provides an aggregated view of HMS connections for monitoring purposes.
+   */
+  public int totalCurrentSize() {
+    return clientPoolCache.asMap().values().stream().mapToInt(HiveClientPool::currentSize).sum();
+  }
+
+  /** Returns the total number of idle (available) connections across all per-user pools. */
+  public int totalIdleCount() {
+    return clientPoolCache.asMap().values().stream().mapToInt(HiveClientPool::idleCount).sum();
+  }
+
+  /**
+   * Returns the maximum configured pool size. Since all per-user pools share the same
+   * configuration, this returns the poolSize of any single pool.
+   */
+  public int totalPoolSize() {
+    return clientPoolCache.asMap().values().stream().mapToInt(HiveClientPool::poolSize).sum();
+  }
+
   public void close() {
     // Close all the HiveClientPool instances in the cache first and then shutdown the scheduler. As
     // Removal listener in Cache will be invoked by the scheduler asynchronously, we need to close

core/src/main/java/org/apache/gravitino/metrics/source/HiveCatalogMetricsSource.java

@@ -0,0 +1,45 @@
+/*
+ * 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.gravitino.metrics.source;
+
+import com.codahale.metrics.Gauge;
+import org.apache.gravitino.metrics.MetricNames;
+
+public class HiveCatalogMetricsSource extends CatalogMetricsSource {
+
+  public HiveCatalogMetricsSource(String metalakeName, String catalogName) {
+    super("hive", metalakeName, catalogName);
+  }
+  /**
+   * Registers connection pool metrics from the given provider. The provider should aggregate values
+   * across all underlying pools (e.g., all per-user pools in a cached client pool).
+   *
+   * @param metricsProvider the provider that supplies aggregated pool metrics
+   */
+  public void registerClientPoolMetrics(MetricsProvider metricsProvider) {
+    registerGauge(
+        MetricNames.DATASOURCE_ACTIVE_CONNECTIONS,
+        (Gauge<Integer>) () -> metricsProvider.currentSize() - metricsProvider.idleCount());
+    registerGauge(
+        MetricNames.DATASOURCE_IDLE_CONNECTIONS, (Gauge<Integer>) metricsProvider::idleCount);
+    registerGauge(
+        MetricNames.DATASOURCE_MAX_CONNECTIONS, (Gauge<Integer>) metricsProvider::poolSize);
+  }
+}

core/src/main/java/org/apache/gravitino/metrics/source/MetricsProvider.java

@@ -0,0 +1,36 @@
+/*
+ * 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.gravitino.metrics.source;
+
+/**
+ * Provides connection pool metrics values. Implementations should aggregate across all underlying
+ * pools (e.g., all per-user pools in a cached client pool).
+ */
+public interface MetricsProvider {
+
+  /** Returns the total number of connections currently created across all pools. */
+  int currentSize();
+
+  /** Returns the number of idle (available) connections across all pools. */
+  int idleCount();
+
+  /** Returns the maximum configured pool size per underlying pool. */
+  int poolSize();
+}

core/src/main/java/org/apache/gravitino/utils/ClientPoolImpl.java

@@ -165,6 +165,16 @@ public int poolSize() {
     return poolSize;
   }
 
+  public int currentSize() {
+    return currentSize;
+  }
+
+  public int idleCount() {
+    synchronized (this) {
+      return clients.size();
+    }
+  }
+
   public boolean isClosed() {
     return closed;
   }

core/src/test/java/org/apache/gravitino/metrics/TestGravitinoSampleBuilder.java

@@ -213,4 +213,56 @@ public void testCreateSampleWithAdditionalLabelsAndHyphens() {
         sample.labelValues);
     Assertions.assertEquals(value, sample.value);
   }
+
+  @Test
+  public void testCreateSampleWithHiveCatalogDatasourceMetrics() {
+    String dropwizardName =
+        "gravitino-catalog.hive.metalake1.catalog1." + MetricNames.DATASOURCE_ACTIVE_CONNECTIONS;
+    double value = 2.0;
+
+    Collector.MetricFamilySamples.Sample sample =
+        sampleBuilder.createSample(
+            dropwizardName, "", Collections.emptyList(), Collections.emptyList(), value);
+
+    Assertions.assertEquals(
+        "gravitino_catalog_"
+            + Collector.sanitizeMetricName(MetricNames.DATASOURCE_ACTIVE_CONNECTIONS),
+        sample.name);
+    Assertions.assertEquals(ImmutableList.of("provider", "metalake", "catalog"), sample.labelNames);
+    Assertions.assertEquals(ImmutableList.of("hive", "metalake1", "catalog1"), sample.labelValues);
+    Assertions.assertEquals(value, sample.value);
+
+    // Verify idle connections
+    dropwizardName =
+        "gravitino-catalog.hive.metalake1.catalog1." + MetricNames.DATASOURCE_IDLE_CONNECTIONS;
+    value = 3.0;
+
+    sample =
+        sampleBuilder.createSample(
+            dropwizardName, "", Collections.emptyList(), Collections.emptyList(), value);
+
+    Assertions.assertEquals(
+        "gravitino_catalog_"
+            + Collector.sanitizeMetricName(MetricNames.DATASOURCE_IDLE_CONNECTIONS),
+        sample.name);
+    Assertions.assertEquals(ImmutableList.of("provider", "metalake", "catalog"), sample.labelNames);
+    Assertions.assertEquals(ImmutableList.of("hive", "metalake1", "catalog1"), sample.labelValues);
+    Assertions.assertEquals(value, sample.value);
+
+    // Verify max connections
+    dropwizardName =
+        "gravitino-catalog.hive.metalake1.catalog1." + MetricNames.DATASOURCE_MAX_CONNECTIONS;
+    value = 10.0;
+
+    sample =
+        sampleBuilder.createSample(
+            dropwizardName, "", Collections.emptyList(), Collections.emptyList(), value);
+
+    Assertions.assertEquals(
+        "gravitino_catalog_" + Collector.sanitizeMetricName(MetricNames.DATASOURCE_MAX_CONNECTIONS),
+        sample.name);
+    Assertions.assertEquals(ImmutableList.of("provider", "metalake", "catalog"), sample.labelNames);
+    Assertions.assertEquals(ImmutableList.of("hive", "metalake1", "catalog1"), sample.labelValues);
+    Assertions.assertEquals(value, sample.value);
+  }
 }