diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java index 92477e3cc3205c..b8a6f833d7aed5 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java @@ -82,6 +82,9 @@ public class CloudTabletRebalancer extends MasterDaemon { private static final Logger LOG = LogManager.getLogger(CloudTabletRebalancer.class); + private final CloudTabletRebalancerMetrics rebalancerMetrics; + private long currentRoundTabletScanCount; + private volatile ConcurrentHashMap> beToTabletsGlobal = new ConcurrentHashMap>(); @@ -245,8 +248,14 @@ private boolean isComputeGroupBalanceChanged(String clusterId) { } public CloudTabletRebalancer(CloudSystemInfoService cloudSystemInfoService) { + this(cloudSystemInfoService, CloudTabletRebalancerMetrics.create()); + } + + CloudTabletRebalancer(CloudSystemInfoService cloudSystemInfoService, + CloudTabletRebalancerMetrics rebalancerMetrics) { super("cloud tablet rebalancer", Config.cloud_tablet_rebalancer_interval_second * 1000); this.cloudSystemInfoService = cloudSystemInfoService; + this.rebalancerMetrics = rebalancerMetrics; } private void initializeWarmupExecutorsIfNeeded() { @@ -505,45 +514,51 @@ protected void runAfterCatalogReady() { } LOG.info("cloud tablet rebalance begin"); - long start = System.currentTimeMillis(); - refreshActiveTabletIdsIfNeeded(); - globalBalanceTypeEnum = BalanceTypeEnum.getCloudWarmUpForRebalanceTypeEnum(); + CloudTabletRebalancerMetrics.Round metricRound = rebalancerMetrics.startRound(); + currentRoundTabletScanCount = 0L; + try { + long start = System.currentTimeMillis(); + refreshActiveTabletIdsIfNeeded(); + globalBalanceTypeEnum = BalanceTypeEnum.getCloudWarmUpForRebalanceTypeEnum(); - buildClusterToBackendMap(); - if (!completeRouteInfo()) { - return; - } + buildClusterToBackendMap(); + if (!completeRouteInfo()) { + return; + } - statRouteInfo(); - boolean migrated = migrateTabletsForSmoothUpgrade(); - if (migrated) { statRouteInfo(); - } + boolean migrated = migrateTabletsForSmoothUpgrade(); + if (migrated) { + statRouteInfo(); + } - indexBalanced = true; - tableBalanced = true; + indexBalanced = true; + tableBalanced = true; - performBalancing(); + performBalancing(); - checkDecommissionState(clusterToBes); - inited = true; - long sleepSeconds = Config.cloud_tablet_rebalancer_interval_second; - if (sleepSeconds < 0L) { - LOG.warn("cloud tablet rebalance interval second is negative, change it to default 1s"); - sleepSeconds = 1L; - } - long balanceEnd = System.currentTimeMillis(); - if (DebugPointUtil.isEnable("CloudTabletRebalancer.balanceEnd.tooLong")) { - LOG.info("debug pointCloudTabletRebalancer.balanceEnd.tooLong"); - // slower the balance end time to trigger next balance immediately - balanceEnd += (Config.cloud_tablet_rebalancer_interval_second + 10L) * 1000L; - } - if (balanceEnd - start > Config.cloud_tablet_rebalancer_interval_second * 1000L) { - sleepSeconds = 1L; + checkDecommissionState(clusterToBes); + inited = true; + long sleepSeconds = Config.cloud_tablet_rebalancer_interval_second; + if (sleepSeconds < 0L) { + LOG.warn("cloud tablet rebalance interval second is negative, change it to default 1s"); + sleepSeconds = 1L; + } + long balanceEnd = System.currentTimeMillis(); + if (DebugPointUtil.isEnable("CloudTabletRebalancer.balanceEnd.tooLong")) { + LOG.info("debug pointCloudTabletRebalancer.balanceEnd.tooLong"); + // slower the balance end time to trigger next balance immediately + balanceEnd += (Config.cloud_tablet_rebalancer_interval_second + 10L) * 1000L; + } + if (balanceEnd - start > Config.cloud_tablet_rebalancer_interval_second * 1000L) { + sleepSeconds = 1L; + } + setInterval(sleepSeconds * 1000L); + LOG.info("finished to rebalancer. cost: {} ms, rebalancer sche interval {} s", + (System.currentTimeMillis() - start), sleepSeconds); + } finally { + rebalancerMetrics.finishRound(metricRound, currentRoundTabletScanCount); } - setInterval(sleepSeconds * 1000L); - LOG.info("finished to rebalancer. cost: {} ms, rebalancer sche interval {} s", - (System.currentTimeMillis() - start), sleepSeconds); } private void buildClusterToBackendMap() { @@ -1277,6 +1292,7 @@ public void loopCloudReplica(Operator operator) { for (MaterializedIndex index : partition.getMaterializedIndices(IndexExtState.VISIBLE, true)) { for (Map.Entry> entry : clusterToBes.entrySet()) { String cluster = entry.getKey(); + currentRoundTabletScanCount += index.getTablets().size(); operator.op(db, table, partition, index, cluster); } } // end for indices diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancerMetrics.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancerMetrics.java new file mode 100644 index 00000000000000..84250a8465f43f --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancerMetrics.java @@ -0,0 +1,78 @@ +// 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.doris.cloud.catalog; + +import org.apache.doris.metric.MetricRepo; + +import java.lang.management.ManagementFactory; +import java.util.concurrent.TimeUnit; +import java.util.function.LongSupplier; + +final class CloudTabletRebalancerMetrics { + private static final long ALLOCATED_BYTES_UNAVAILABLE = -1L; + + private final LongSupplier nanoTimeSupplier; + private final LongSupplier allocatedBytesSupplier; + + CloudTabletRebalancerMetrics(LongSupplier nanoTimeSupplier, LongSupplier allocatedBytesSupplier) { + this.nanoTimeSupplier = nanoTimeSupplier; + this.allocatedBytesSupplier = allocatedBytesSupplier; + } + + static CloudTabletRebalancerMetrics create() { + com.sun.management.ThreadMXBean threadMxBean = + ManagementFactory.getPlatformMXBean(com.sun.management.ThreadMXBean.class); + return new CloudTabletRebalancerMetrics(System::nanoTime, createAllocatedBytesSupplier(threadMxBean)); + } + + Round startRound() { + return new Round(nanoTimeSupplier.getAsLong(), allocatedBytesSupplier.getAsLong()); + } + + void finishRound(Round round, long tabletScanCount) { + long durationMs = TimeUnit.NANOSECONDS.toMillis(nanoTimeSupplier.getAsLong() - round.startNanos); + long currentAllocatedBytes = allocatedBytesSupplier.getAsLong(); + long allocatedBytes = round.startAllocatedBytes < 0L || currentAllocatedBytes < 0L + ? ALLOCATED_BYTES_UNAVAILABLE : currentAllocatedBytes - round.startAllocatedBytes; + MetricRepo.updateCloudTabletRebalancerMetrics(durationMs, allocatedBytes, tabletScanCount); + } + + static LongSupplier createAllocatedBytesSupplier(com.sun.management.ThreadMXBean threadMxBean) { + if (threadMxBean == null || !threadMxBean.isThreadAllocatedMemorySupported()) { + return () -> ALLOCATED_BYTES_UNAVAILABLE; + } + if (!threadMxBean.isThreadAllocatedMemoryEnabled()) { + try { + threadMxBean.setThreadAllocatedMemoryEnabled(true); + } catch (SecurityException | UnsupportedOperationException e) { + return () -> ALLOCATED_BYTES_UNAVAILABLE; + } + } + return threadMxBean::getCurrentThreadAllocatedBytes; + } + + static final class Round { + private final long startNanos; + private final long startAllocatedBytes; + + private Round(long startNanos, long startAllocatedBytes) { + this.startNanos = startNanos; + this.startAllocatedBytes = startAllocatedBytes; + } + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/metric/CloudMetrics.java b/fe/fe-core/src/main/java/org/apache/doris/metric/CloudMetrics.java index 7649b3b1c6780a..a663348a27f182 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/metric/CloudMetrics.java +++ b/fe/fe-core/src/main/java/org/apache/doris/metric/CloudMetrics.java @@ -55,6 +55,13 @@ public class CloudMetrics { protected static AutoMappedMetric CLUSTER_CLOUD_WARM_UP_CACHE_BALANCE_NUM; protected static AutoMappedMetric VIRTUAL_COMPUTE_GROUP_SWITCH_COUNTER; + protected static LongCounterMetric CLOUD_TABLET_REBALANCER_ROUND_TOTAL; + protected static LongCounterMetric CLOUD_TABLET_REBALANCER_ALLOCATED_BYTES_TOTAL; + protected static GaugeMetricImpl CLOUD_TABLET_REBALANCER_LAST_ROUND_ALLOCATED_BYTES; + protected static LongCounterMetric CLOUD_TABLET_REBALANCER_DURATION_MS_TOTAL; + protected static GaugeMetricImpl CLOUD_TABLET_REBALANCER_LAST_ROUND_DURATION_MS; + protected static LongCounterMetric CLOUD_TABLET_REBALANCER_TABLET_SCAN_TOTAL; + // Per-method meta-service RPC metrics public static AutoMappedMetric META_SERVICE_RPC_TOTAL; public static AutoMappedMetric META_SERVICE_RPC_FAILED; @@ -151,6 +158,8 @@ protected static void init() { "virtual_compute_group_switch_total", MetricUnit.NOUNIT, "virtual compute group active standby switch count")); + initCloudTabletRebalancerMetrics(); + // Per-method meta-service RPC metrics META_SERVICE_RPC_TOTAL = MetricRepo.addLabeledMetrics("method", () -> new LongCounterMetric("meta_service_rpc_total", MetricUnit.NOUNIT, @@ -202,4 +211,32 @@ protected static void init() { MetricUnit.NOUNIT, "meta service RPC requests per second (all methods)", 0.0); MetricRepo.DORIS_METRIC_REGISTER.addMetrics(META_SERVICE_RPC_ALL_PER_SECOND); } + + static void initCloudTabletRebalancerMetrics() { + CLOUD_TABLET_REBALANCER_ROUND_TOTAL = new LongCounterMetric( + "cloud_tablet_rebalancer_round_total", MetricUnit.OPERATIONS, + "total cloud tablet rebalancer rounds"); + CLOUD_TABLET_REBALANCER_ALLOCATED_BYTES_TOTAL = new LongCounterMetric( + "cloud_tablet_rebalancer_allocated_bytes_total", MetricUnit.BYTES, + "total bytes allocated by cloud tablet rebalancer rounds"); + CLOUD_TABLET_REBALANCER_LAST_ROUND_ALLOCATED_BYTES = new GaugeMetricImpl<>( + "cloud_tablet_rebalancer_last_round_allocated_bytes", MetricUnit.BYTES, + "bytes allocated by the last cloud tablet rebalancer round, or -1 when unavailable", -1L); + CLOUD_TABLET_REBALANCER_DURATION_MS_TOTAL = new LongCounterMetric( + "cloud_tablet_rebalancer_duration_ms_total", MetricUnit.MILLISECONDS, + "total cloud tablet rebalancer round duration in milliseconds"); + CLOUD_TABLET_REBALANCER_LAST_ROUND_DURATION_MS = new GaugeMetricImpl<>( + "cloud_tablet_rebalancer_last_round_duration_ms", MetricUnit.MILLISECONDS, + "duration of the last cloud tablet rebalancer round in milliseconds", 0L); + CLOUD_TABLET_REBALANCER_TABLET_SCAN_TOTAL = new LongCounterMetric( + "cloud_tablet_rebalancer_tablet_scan_total", MetricUnit.OPERATIONS, + "total tablet route entries scanned by cloud tablet rebalancer rounds"); + + MetricRepo.DORIS_METRIC_REGISTER.addMetrics(CLOUD_TABLET_REBALANCER_ROUND_TOTAL); + MetricRepo.DORIS_METRIC_REGISTER.addMetrics(CLOUD_TABLET_REBALANCER_ALLOCATED_BYTES_TOTAL); + MetricRepo.DORIS_METRIC_REGISTER.addMetrics(CLOUD_TABLET_REBALANCER_LAST_ROUND_ALLOCATED_BYTES); + MetricRepo.DORIS_METRIC_REGISTER.addMetrics(CLOUD_TABLET_REBALANCER_DURATION_MS_TOTAL); + MetricRepo.DORIS_METRIC_REGISTER.addMetrics(CLOUD_TABLET_REBALANCER_LAST_ROUND_DURATION_MS); + MetricRepo.DORIS_METRIC_REGISTER.addMetrics(CLOUD_TABLET_REBALANCER_TABLET_SCAN_TOTAL); + } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java b/fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java index c739957731061b..b3a40988e24db7 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java +++ b/fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java @@ -2253,6 +2253,21 @@ public static void updateClusterCloudBalanceNum(String clusterName, String clust MetricRepo.DORIS_METRIC_REGISTER.addMetrics(counter); } + public static void updateCloudTabletRebalancerMetrics(long durationMs, long allocatedBytes, + long tabletScanCount) { + if (!MetricRepo.isInit || Config.isNotCloudMode()) { + return; + } + CloudMetrics.CLOUD_TABLET_REBALANCER_ROUND_TOTAL.increase(1L); + CloudMetrics.CLOUD_TABLET_REBALANCER_DURATION_MS_TOTAL.increase(durationMs); + CloudMetrics.CLOUD_TABLET_REBALANCER_LAST_ROUND_DURATION_MS.setValue(durationMs); + CloudMetrics.CLOUD_TABLET_REBALANCER_TABLET_SCAN_TOTAL.increase(tabletScanCount); + CloudMetrics.CLOUD_TABLET_REBALANCER_LAST_ROUND_ALLOCATED_BYTES.setValue(allocatedBytes); + if (allocatedBytes >= 0L) { + CloudMetrics.CLOUD_TABLET_REBALANCER_ALLOCATED_BYTES_TOTAL.increase(allocatedBytes); + } + } + public static void increaseVirtualComputeGroupSwitch(String virtualComputeGroupId, String virtualComputeGroupName, String srcComputeGroupId, String srcComputeGroupName, String dstComputeGroupId, String dstComputeGroupName) { diff --git a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudTabletRebalancerMetricsTest.java b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudTabletRebalancerMetricsTest.java new file mode 100644 index 00000000000000..36e79c190b237d --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudTabletRebalancerMetricsTest.java @@ -0,0 +1,90 @@ +// 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.doris.cloud.catalog; + +import org.apache.doris.metric.MetricRepo; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.LongSupplier; + +public class CloudTabletRebalancerMetricsTest { + + @Test + public void testCreateUsesCurrentJvmThreadAllocationSupport() { + CloudTabletRebalancerMetrics metrics = CloudTabletRebalancerMetrics.create(); + Assertions.assertNotNull(metrics.startRound()); + } + + @Test + public void testAllocatedBytesSupplierHandlesJvmCapabilities() { + com.sun.management.ThreadMXBean unsupported = Mockito.mock(com.sun.management.ThreadMXBean.class); + Mockito.when(unsupported.isThreadAllocatedMemorySupported()).thenReturn(false); + LongSupplier unsupportedSupplier = CloudTabletRebalancerMetrics.createAllocatedBytesSupplier(unsupported); + Assertions.assertEquals(-1L, unsupportedSupplier.getAsLong()); + + com.sun.management.ThreadMXBean denied = Mockito.mock(com.sun.management.ThreadMXBean.class); + Mockito.when(denied.isThreadAllocatedMemorySupported()).thenReturn(true); + Mockito.when(denied.isThreadAllocatedMemoryEnabled()).thenReturn(false); + Mockito.doThrow(new SecurityException()).when(denied).setThreadAllocatedMemoryEnabled(true); + LongSupplier deniedSupplier = CloudTabletRebalancerMetrics.createAllocatedBytesSupplier(denied); + Assertions.assertEquals(-1L, deniedSupplier.getAsLong()); + + com.sun.management.ThreadMXBean enabled = Mockito.mock(com.sun.management.ThreadMXBean.class); + Mockito.when(enabled.isThreadAllocatedMemorySupported()).thenReturn(true); + Mockito.when(enabled.isThreadAllocatedMemoryEnabled()).thenReturn(true); + Mockito.when(enabled.getCurrentThreadAllocatedBytes()).thenReturn(1234L); + LongSupplier enabledSupplier = CloudTabletRebalancerMetrics.createAllocatedBytesSupplier(enabled); + Assertions.assertEquals(1234L, enabledSupplier.getAsLong()); + } + + @Test + public void testFinishRoundRecordsDurationAllocationAndWork() { + AtomicLong nanoTime = new AtomicLong(1_000_000_000L); + AtomicLong allocatedBytes = new AtomicLong(10_000L); + CloudTabletRebalancerMetrics metrics = + new CloudTabletRebalancerMetrics(nanoTime::get, allocatedBytes::get); + + CloudTabletRebalancerMetrics.Round round = metrics.startRound(); + nanoTime.set(1_012_000_000L); + allocatedBytes.set(10_777L); + + try (MockedStatic metricRepo = Mockito.mockStatic(MetricRepo.class)) { + metrics.finishRound(round, 42L); + metricRepo.verify(() -> MetricRepo.updateCloudTabletRebalancerMetrics(12L, 777L, 42L)); + } + } + + @Test + public void testFinishRoundMarksAllocationUnavailable() { + AtomicLong nanoTime = new AtomicLong(2_000_000_000L); + CloudTabletRebalancerMetrics metrics = new CloudTabletRebalancerMetrics(nanoTime::get, () -> -1L); + + CloudTabletRebalancerMetrics.Round round = metrics.startRound(); + nanoTime.set(2_003_000_000L); + + try (MockedStatic metricRepo = Mockito.mockStatic(MetricRepo.class)) { + metrics.finishRound(round, 7L); + metricRepo.verify(() -> MetricRepo.updateCloudTabletRebalancerMetrics(3L, -1L, 7L)); + } + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/metric/MetricsTest.java b/fe/fe-core/src/test/java/org/apache/doris/metric/MetricsTest.java index 19b73fcc03f41f..f3c2a403d1d7ab 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/metric/MetricsTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/metric/MetricsTest.java @@ -250,6 +250,48 @@ public void testVirtualComputeGroupSwitchMetricRename() { } } + @Test + public void testCloudTabletRebalancerMetrics() { + String originCloudUniqueId = Config.cloud_unique_id; + try { + Config.cloud_unique_id = "test_cloud_unique_id"; + CloudMetrics.initCloudTabletRebalancerMetrics(); + + MetricRepo.updateCloudTabletRebalancerMetrics(125L, 4096L, 200L); + + String metricResult = getPrometheusMetrics(); + Assert.assertTrue(metricResult.contains("# TYPE doris_fe_cloud_tablet_rebalancer_round_total counter")); + Assert.assertTrue(metricResult.contains("doris_fe_cloud_tablet_rebalancer_round_total 1")); + Assert.assertTrue(metricResult.contains( + "doris_fe_cloud_tablet_rebalancer_allocated_bytes_total 4096")); + Assert.assertTrue(metricResult.contains( + "doris_fe_cloud_tablet_rebalancer_last_round_allocated_bytes 4096")); + Assert.assertTrue(metricResult.contains( + "doris_fe_cloud_tablet_rebalancer_duration_ms_total 125")); + Assert.assertTrue(metricResult.contains( + "doris_fe_cloud_tablet_rebalancer_last_round_duration_ms 125")); + Assert.assertTrue(metricResult.contains( + "doris_fe_cloud_tablet_rebalancer_tablet_scan_total 200")); + + MetricRepo.updateCloudTabletRebalancerMetrics(25L, -1L, 50L); + + metricResult = getPrometheusMetrics(); + Assert.assertTrue(metricResult.contains("doris_fe_cloud_tablet_rebalancer_round_total 2")); + Assert.assertTrue(metricResult.contains( + "doris_fe_cloud_tablet_rebalancer_allocated_bytes_total 4096")); + Assert.assertTrue(metricResult.contains( + "doris_fe_cloud_tablet_rebalancer_last_round_allocated_bytes -1")); + Assert.assertTrue(metricResult.contains( + "doris_fe_cloud_tablet_rebalancer_duration_ms_total 150")); + Assert.assertTrue(metricResult.contains( + "doris_fe_cloud_tablet_rebalancer_last_round_duration_ms 25")); + Assert.assertTrue(metricResult.contains( + "doris_fe_cloud_tablet_rebalancer_tablet_scan_total 250")); + } finally { + Config.cloud_unique_id = originCloudUniqueId; + } + } + @Test public void testCloudWarmUpSyncJobMetricsReadStatsDirectlyFromJob() { String oldCloudUniqueId = Config.cloud_unique_id;