From 28ff7f2abfab748128c2a5f3fcf554ad20ded9e0 Mon Sep 17 00:00:00 2001 From: yangyuxia <13853186257@139.com> Date: Tue, 21 Jul 2026 02:29:10 -0400 Subject: [PATCH] feat(metrics):add Hive Catalog Metrics for HMS connection pool --- .../catalog/hive/HiveCatalogOperations.java | 35 ++++ .../gravitino/hive/CachedClientPool.java | 21 +++ .../source/HiveCatalogMetricsSource.java | 45 +++++ .../metrics/source/MetricsProvider.java | 36 ++++ .../gravitino/utils/ClientPoolImpl.java | 10 ++ .../metrics/TestGravitinoSampleBuilder.java | 52 ++++++ .../source/TestHiveCatalogMetricsSource.java | 165 ++++++++++++++++++ 7 files changed, 364 insertions(+) create mode 100644 core/src/main/java/org/apache/gravitino/metrics/source/HiveCatalogMetricsSource.java create mode 100644 core/src/main/java/org/apache/gravitino/metrics/source/MetricsProvider.java create mode 100644 core/src/test/java/org/apache/gravitino/metrics/source/TestHiveCatalogMetricsSource.java diff --git a/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveCatalogOperations.java b/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveCatalogOperations.java index 99bcc1431d0..d3c762c2a32 100644 --- a/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveCatalogOperations.java +++ b/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 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; diff --git a/catalogs/hive-metastore-common/src/main/java/org/apache/gravitino/hive/CachedClientPool.java b/catalogs/hive-metastore-common/src/main/java/org/apache/gravitino/hive/CachedClientPool.java index 7d93a25de79..31eaf861959 100644 --- a/catalogs/hive-metastore-common/src/main/java/org/apache/gravitino/hive/CachedClientPool.java +++ b/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 diff --git a/core/src/main/java/org/apache/gravitino/metrics/source/HiveCatalogMetricsSource.java b/core/src/main/java/org/apache/gravitino/metrics/source/HiveCatalogMetricsSource.java new file mode 100644 index 00000000000..c38ad43e7a5 --- /dev/null +++ b/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) () -> metricsProvider.currentSize() - metricsProvider.idleCount()); + registerGauge( + MetricNames.DATASOURCE_IDLE_CONNECTIONS, (Gauge) metricsProvider::idleCount); + registerGauge( + MetricNames.DATASOURCE_MAX_CONNECTIONS, (Gauge) metricsProvider::poolSize); + } +} diff --git a/core/src/main/java/org/apache/gravitino/metrics/source/MetricsProvider.java b/core/src/main/java/org/apache/gravitino/metrics/source/MetricsProvider.java new file mode 100644 index 00000000000..35ecafcba76 --- /dev/null +++ b/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(); +} diff --git a/core/src/main/java/org/apache/gravitino/utils/ClientPoolImpl.java b/core/src/main/java/org/apache/gravitino/utils/ClientPoolImpl.java index ae2cd9910e2..80f5eed8f49 100644 --- a/core/src/main/java/org/apache/gravitino/utils/ClientPoolImpl.java +++ b/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; } diff --git a/core/src/test/java/org/apache/gravitino/metrics/TestGravitinoSampleBuilder.java b/core/src/test/java/org/apache/gravitino/metrics/TestGravitinoSampleBuilder.java index 3a6941fe178..0a2d7cb0263 100644 --- a/core/src/test/java/org/apache/gravitino/metrics/TestGravitinoSampleBuilder.java +++ b/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); + } } diff --git a/core/src/test/java/org/apache/gravitino/metrics/source/TestHiveCatalogMetricsSource.java b/core/src/test/java/org/apache/gravitino/metrics/source/TestHiveCatalogMetricsSource.java new file mode 100644 index 00000000000..cf4450bd7a1 --- /dev/null +++ b/core/src/test/java/org/apache/gravitino/metrics/source/TestHiveCatalogMetricsSource.java @@ -0,0 +1,165 @@ +/* + * 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 com.codahale.metrics.MetricRegistry; +import java.util.SortedMap; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.gravitino.metrics.MetricNames; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class TestHiveCatalogMetricsSource { + + @Test + void testMetricsSourceName() { + HiveCatalogMetricsSource source = new HiveCatalogMetricsSource("test_metalake", "test_catalog"); + Assertions.assertEquals( + "gravitino-catalog.hive.test_metalake.test_catalog", source.getMetricsSourceName()); + } + + @Test + void testRegisterClientPoolMetrics() { + HiveCatalogMetricsSource source = new HiveCatalogMetricsSource("test_metalake", "test_catalog"); + MetricRegistry registry = source.getMetricRegistry(); + + // Create a simple MetricsProvider with fixed values + MetricsProvider provider = + new MetricsProvider() { + @Override + public int currentSize() { + return 8; + } + + @Override + public int idleCount() { + return 3; + } + + @Override + public int poolSize() { + return 10; + } + }; + + source.registerClientPoolMetrics(provider); + + SortedMap gauges = registry.getGauges(); + Assertions.assertTrue(gauges.containsKey(MetricNames.DATASOURCE_ACTIVE_CONNECTIONS)); + Assertions.assertTrue(gauges.containsKey(MetricNames.DATASOURCE_IDLE_CONNECTIONS)); + Assertions.assertTrue(gauges.containsKey(MetricNames.DATASOURCE_MAX_CONNECTIONS)); + + // active = currentSize - idleCount = 8 - 3 = 5 + Assertions.assertEquals(5, gauges.get(MetricNames.DATASOURCE_ACTIVE_CONNECTIONS).getValue()); + // idle = 3 + Assertions.assertEquals(3, gauges.get(MetricNames.DATASOURCE_IDLE_CONNECTIONS).getValue()); + // max = 10 + Assertions.assertEquals(10, gauges.get(MetricNames.DATASOURCE_MAX_CONNECTIONS).getValue()); + } + + @Test + void testMetricsReflectDynamicProviderValues() { + HiveCatalogMetricsSource source = new HiveCatalogMetricsSource("test_metalake", "test_catalog"); + MetricRegistry registry = source.getMetricRegistry(); + + // Use AtomicInteger to simulate changing pool state + AtomicInteger currentSize = new AtomicInteger(5); + AtomicInteger idleCount = new AtomicInteger(2); + AtomicInteger poolSize = new AtomicInteger(10); + + MetricsProvider provider = + new MetricsProvider() { + @Override + public int currentSize() { + return currentSize.get(); + } + + @Override + public int idleCount() { + return idleCount.get(); + } + + @Override + public int poolSize() { + return poolSize.get(); + } + }; + + source.registerClientPoolMetrics(provider); + + SortedMap gauges = registry.getGauges(); + + // Initial state: active=3, idle=2, max=10 + Assertions.assertEquals(3, gauges.get(MetricNames.DATASOURCE_ACTIVE_CONNECTIONS).getValue()); + Assertions.assertEquals(2, gauges.get(MetricNames.DATASOURCE_IDLE_CONNECTIONS).getValue()); + Assertions.assertEquals(10, gauges.get(MetricNames.DATASOURCE_MAX_CONNECTIONS).getValue()); + + // Simulate more users connecting: currentSize increases, idle decreases + currentSize.set(9); + idleCount.set(1); + + // Gauges should reflect the new values dynamically + Assertions.assertEquals(8, gauges.get(MetricNames.DATASOURCE_ACTIVE_CONNECTIONS).getValue()); + Assertions.assertEquals(1, gauges.get(MetricNames.DATASOURCE_IDLE_CONNECTIONS).getValue()); + Assertions.assertEquals(10, gauges.get(MetricNames.DATASOURCE_MAX_CONNECTIONS).getValue()); + } + + @Test + void testAggregatedMetricsAcrossMultiplePools() { + // Simulate the scenario: 3 users, each with their own pool + // User A: currentSize=3, idleCount=1, poolSize=10 + // User B: currentSize=5, idleCount=2, poolSize=10 + // User C: currentSize=2, idleCount=0, poolSize=10 + // Aggregated: currentSize=10, idleCount=3, poolSize=30 + + HiveCatalogMetricsSource source = new HiveCatalogMetricsSource("test_metalake", "test_catalog"); + MetricRegistry registry = source.getMetricRegistry(); + + MetricsProvider provider = + new MetricsProvider() { + @Override + public int currentSize() { + return 3 + 5 + 2; // sum across 3 pools + } + + @Override + public int idleCount() { + return 1 + 2 + 0; // sum across 3 pools + } + + @Override + public int poolSize() { + return 10 + 10 + 10; // sum across 3 pools (each poolSize=10) + } + }; + + source.registerClientPoolMetrics(provider); + + SortedMap gauges = registry.getGauges(); + + // active = 10 - 3 = 7 + Assertions.assertEquals(7, gauges.get(MetricNames.DATASOURCE_ACTIVE_CONNECTIONS).getValue()); + // idle = 3 + Assertions.assertEquals(3, gauges.get(MetricNames.DATASOURCE_IDLE_CONNECTIONS).getValue()); + // max = 10 + 10 + 10 = 30 (sum across 3 pools) + Assertions.assertEquals(30, gauges.get(MetricNames.DATASOURCE_MAX_CONNECTIONS).getValue()); + } +}