diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java b/amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java index 2e9aa4a283..eb943e6057 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java @@ -41,11 +41,13 @@ import org.apache.amoro.server.dashboard.model.OptimizerResourceInfo; import org.apache.amoro.server.ha.HighAvailabilityContainer; import org.apache.amoro.server.manager.AbstractOptimizerContainer; +import org.apache.amoro.server.manager.MetricManager; import org.apache.amoro.server.optimizing.OptimizingProcess; import org.apache.amoro.server.optimizing.OptimizingQueue; import org.apache.amoro.server.optimizing.OptimizingStatus; import org.apache.amoro.server.optimizing.TaskRuntime; import org.apache.amoro.server.optimizing.dra.DynamicAllocationConfig; +import org.apache.amoro.server.optimizing.dra.DynamicAllocationMetrics; import org.apache.amoro.server.optimizing.dra.DynamicAllocationState; import org.apache.amoro.server.optimizing.dra.PendingRegistrations; import org.apache.amoro.server.persistence.StatedPersistentBase; @@ -1136,21 +1138,81 @@ private class OptimizerScaleKeeper extends AbstractKeeper { new ConcurrentHashMap<>(); private final Set watchedGroups = ConcurrentHashMap.newKeySet(); private final Map planningBoundStreaks = new ConcurrentHashMap<>(); + private final Map metricsByGroup = new ConcurrentHashMap<>(); public OptimizerScaleKeeper(String threadName) { super(threadName); } - /** Start watching a group if dynamic allocation is effectively enabled on it. Idempotent. */ - public void watch(ResourceGroup resourceGroup) { + /** + * Start watching a group if dynamic allocation is effectively enabled on it. Idempotent. + * + *

Watch and unwatch are serialized: metric registration is not idempotent (re-registering a + * live key throws), so an unlocked watch/unwatch interleaving could strand a registration that + * the other side never saw — after which every re-watch of the group throws before queueing its + * scale task, leaving it watched-but-dead until a restart. These are rare control-plane calls; + * the lock costs nothing on the scaling hot path. + */ + public synchronized void watch(ResourceGroup resourceGroup) { + if (stopped) { + // A watch arriving after dispose — an in-flight config-sync run or the round's re-check + // racing a leader hand-off — must not register metrics from a dead service: the keys + // would outlive it in the global registry and fail the next leader's watch. + return; + } if (!DynamicAllocationConfig.isEffectivelyEnabled(resourceGroup)) { + // Propagate a disable on the config-entry path itself: the round-driven unwatch runs on + // the leader only, so a follower relying on it would keep the group's drain blocks and + // exported metrics until failover. + unwatch(resourceGroup.getName()); return; } - if (watchedGroups.add(resourceGroup.getName())) { + if (!watchedGroups.contains(resourceGroup.getName())) { + // Register metrics before marking the group watched: a failed registration must leave + // the group rewatchable, not watched-but-dead with every retry swallowed by the entry. + registerMetrics(resourceGroup.getName()); + watchedGroups.add(resourceGroup.getName()); suspendingQueue.add(new DraScaleTask(resourceGroup.getName(), 0)); } } + /** + * Register the group's DRA gauges and counters, keyed by the keeper's own state: unlike the + * queue-scoped {@code OptimizerGroupMetrics} they live with the watch, so a group handed back + * to the legacy floor keeper stops exporting scaling metrics it no longer produces. + */ + private void registerMetrics(String groupName) { + DynamicAllocationMetrics metrics = + new DynamicAllocationMetrics( + groupName, + MetricManager.getInstance().getGlobalRegistry(), + new DynamicAllocationMetrics.Source() { + @Override + public int pendingRemovalOptimizers() { + return (int) + pendingRemovalTokens.stream() + .map(authOptimizers::get) + .filter( + optimizer -> + optimizer != null && groupName.equals(optimizer.getGroupName())) + .count(); + } + + @Override + public int effectiveThreads() { + return getTotalQuota(groupName) + pendingThreads(groupName); + } + + @Override + public long backlogDurationMs() { + DynamicAllocationState state = scaleStates.get(groupName); + return state == null ? 0 : state.backlogDurationMs(System.currentTimeMillis()); + } + }); + metrics.register(); + metricsByGroup.put(groupName, metrics); + } + /** Clear the boot-window accounting of a registered optimizer (AMS-launched ones only). */ public void onOptimizerRegistered(OptimizerInstance optimizer) { if (optimizer.getResourceId() == null) { @@ -1162,10 +1224,14 @@ public void onOptimizerRegistered(OptimizerInstance optimizer) { } } - private void unwatch(String groupName) { + private synchronized void unwatch(String groupName) { watchedGroups.remove(groupName); scaleStates.remove(groupName); planningBoundStreaks.remove(groupName); + DynamicAllocationMetrics metrics = metricsByGroup.remove(groupName); + if (metrics != null) { + metrics.unregister(); + } // A drain block left behind would starve the group's pods forever once the legacy floor // keeper resumes duty for the disabled group: re-admit them to task assignment. authOptimizers.values().stream() @@ -1182,11 +1248,23 @@ private void unwatch(String groupName) { * must go too: leaving it would leak the entry and, if a group with the same name is created * before the next evaluation, suppress its scale-up with the old group's phantom capacity. */ - public void onGroupDeleted(String groupName) { + public synchronized void onGroupDeleted(String groupName) { unwatch(groupName); pendingRegistrations.remove(groupName); } + /** + * The global metric registry outlives this service: on a leader hand-off the next leader's + * fresh service watches the same groups, and any keys left behind here would make that watch + * throw, leaving the group watched-but-dead until a JVM restart. + */ + @Override + public synchronized void dispose() { + super.dispose(); + metricsByGroup.values().forEach(DynamicAllocationMetrics::unregister); + metricsByGroup.clear(); + } + @Override protected void processTask(DraScaleTask task) { ResourceGroup resourceGroup; @@ -1303,6 +1381,11 @@ private void evaluateScaleDown( if (victim == null) { return; } + // The drain start is the scale-down action; the eventual removal only completes it. + DynamicAllocationMetrics metrics = metricsByGroup.get(groupName); + if (metrics != null) { + metrics.incScaleDown(); + } beginGracefulDrain(victim, now + config.getDrainTimeout().toMillis()); // Only a snapshot taken after the token entered the pending-removal set can prove idleness: // the pre-insert one may miss a task fetched by a long-poll racing the drain start. @@ -1400,6 +1483,10 @@ private void scaleIfNeeded( } return; } + DynamicAllocationMetrics metrics = metricsByGroup.get(groupName); + if (metrics != null) { + metrics.incScaleUp(); + } int threadsPerInstance = config.getExecutorParallelism(); LOG.info( "Dynamic allocation scaling out group {}: {} instance(s) of {} thread(s), effective threads {}", diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizerGroupMetrics.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizerGroupMetrics.java index 5b70a4e83d..4b640d4da6 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizerGroupMetrics.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizerGroupMetrics.java @@ -32,13 +32,17 @@ import org.apache.amoro.metrics.MetricDefine; import org.apache.amoro.metrics.MetricKey; import org.apache.amoro.metrics.MetricRegistry; +import org.apache.amoro.server.optimizing.dra.DynamicAllocationConfig; +import org.apache.amoro.server.optimizing.dra.DynamicAllocationState; import org.apache.amoro.server.resource.OptimizerInstance; import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableMap; import org.apache.amoro.shade.guava32.com.google.common.collect.Lists; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; /** Metrics manager for an optimizer group. */ public class OptimizerGroupMetrics { @@ -104,6 +108,21 @@ public class OptimizerGroupMetrics { .withTags(GROUP_TAG) .build(); + public static final MetricDefine OPTIMIZER_GROUP_IDLE_OPTIMIZERS = + defineGauge("optimizer_group_idle_optimizers") + .withDescription( + "Number of optimizer instances with no in-flight task in optimizer group") + .withTags(GROUP_TAG) + .build(); + + public static final MetricDefine OPTIMIZER_GROUP_CONFIG_INVALID = + defineGauge("optimizer_group_config_invalid") + .withDescription( + "1 while the group's dynamic allocation configuration is invalid and the " + + "fail-safe fallback is active, else 0") + .withTags(GROUP_TAG) + .build(); + private final String groupName; private final MetricRegistry registry; private final OptimizingQueue optimizingQueue; @@ -199,6 +218,29 @@ public void register() { optimizerInstances.values().stream() .mapToLong(OptimizerInstance::getThreadCount) .sum()); + registerMetric( + registry, + OPTIMIZER_GROUP_IDLE_OPTIMIZERS, + (Gauge) + () -> { + Set busyTokens = + optimizingQueue + .collectTasks(task -> DynamicAllocationState.occupiesThread(task.getStatus())) + .stream() + .map(TaskRuntime::getToken) + .collect(Collectors.toSet()); + return optimizerInstances.keySet().stream() + .filter(token -> !busyTokens.contains(token)) + .count(); + }); + registerMetric( + registry, + OPTIMIZER_GROUP_CONFIG_INVALID, + (Gauge) + () -> + DynamicAllocationConfig.isConfigInvalid(optimizingQueue.getOptimizerGroup()) + ? 1 + : 0); } public void unregister() { diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationConfig.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationConfig.java index cfa0e2f1db..813a3f5dbf 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationConfig.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationConfig.java @@ -218,6 +218,31 @@ public static boolean isEffectivelyEnabled(ResourceGroup group) { } } + /** + * Whether the group opted into dynamic allocation but its configuration does not hold, i.e. the + * fail-safe fallback of {@link #isEffectivelyEnabled(ResourceGroup)} is active. Read side of the + * config-invalid gauge. A group that never opted in is not invalid, whatever its other properties + * parse to — its leftover values are inert and must not alarm. + */ + public static boolean isConfigInvalid(ResourceGroup group) { + boolean enabled = + PropertyUtil.propertyAsBoolean( + group.getProperties(), + OptimizerProperties.DYNAMIC_ALLOCATION_ENABLED, + OptimizerProperties.DYNAMIC_ALLOCATION_ENABLED_DEFAULT); + if (!enabled) { + return false; + } + try { + parse(group).validate(); + return false; + } catch (RuntimeException e) { + // Not just IllegalArgumentException: duration parsing can throw ArithmeticException on + // overflow, and a gauge read must never propagate — it would abort the whole scrape. + return true; + } + } + /** * The min-parallelism property key that {@link #resolveMinParallelism(ResourceGroup)} actually * reads for this group. Writers updating the effective value (e.g. the keeper's auto-reset) must diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationMetrics.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationMetrics.java new file mode 100644 index 0000000000..40ddf3cc30 --- /dev/null +++ b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationMetrics.java @@ -0,0 +1,142 @@ +/* + * 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.amoro.server.optimizing.dra; + +import static org.apache.amoro.metrics.MetricDefine.defineCounter; +import static org.apache.amoro.metrics.MetricDefine.defineGauge; + +import org.apache.amoro.metrics.Counter; +import org.apache.amoro.metrics.Gauge; +import org.apache.amoro.metrics.Metric; +import org.apache.amoro.metrics.MetricDefine; +import org.apache.amoro.metrics.MetricKey; +import org.apache.amoro.metrics.MetricRegistry; +import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableMap; +import org.apache.amoro.shade.guava32.com.google.common.collect.Lists; + +import java.util.List; + +/** + * Per-group metrics of dynamic allocation (AIP-5). Their sources — the pending-removal set, the + * boot-window accounting and the backlog timer — are owned by the scale keeper, so unlike the + * queue-scoped {@code OptimizerGroupMetrics} these are registered when the keeper starts watching a + * group and unregistered when it stops; gauge reads go through the injected {@link Source}. + */ +public class DynamicAllocationMetrics { + + static final String GROUP_TAG = "group"; + + public static final MetricDefine OPTIMIZER_GROUP_PENDING_REMOVAL_OPTIMIZERS = + defineGauge("optimizer_group_pending_removal_optimizers") + .withDescription("Number of optimizer instances in graceful drain in optimizer group") + .withTags(GROUP_TAG) + .build(); + + public static final MetricDefine OPTIMIZER_GROUP_EFFECTIVE_THREADS = + defineGauge("optimizer_group_effective_threads") + .withDescription( + "Number of registered threads plus threads of optimizers pending registration " + + "in optimizer group") + .withTags(GROUP_TAG) + .build(); + + public static final MetricDefine OPTIMIZER_GROUP_BACKLOG_DURATION_MS = + defineGauge("optimizer_group_backlog_duration_ms") + .withDescription( + "Duration in milliseconds since demand first exceeded capacity in optimizer " + + "group, 0 while there is no backlog") + .withTags(GROUP_TAG) + .build(); + + public static final MetricDefine OPTIMIZER_GROUP_SCALE_UP_TOTAL = + defineCounter("optimizer_group_scale_up_total") + .withDescription( + "Cumulative count of attempted scale-up actions in optimizer group, one per " + + "scale-out round regardless of instance count or request outcome") + .withTags(GROUP_TAG) + .build(); + + public static final MetricDefine OPTIMIZER_GROUP_SCALE_DOWN_TOTAL = + defineCounter("optimizer_group_scale_down_total") + .withDescription( + "Cumulative count of scale-down actions in optimizer group, one per drain start") + .withTags(GROUP_TAG) + .build(); + + /** Read side of the gauges, implemented by the owner of the scaling state. */ + public interface Source { + + /** Number of this group's optimizer instances currently in graceful drain. */ + int pendingRemovalOptimizers(); + + /** Registered threads plus threads of optimizers pending registration. */ + int effectiveThreads(); + + /** Duration since demand first exceeded capacity, {@code 0} while there is no backlog. */ + long backlogDurationMs(); + } + + private final String groupName; + private final MetricRegistry registry; + private final Source source; + private final Counter scaleUpTotal = new Counter(); + private final Counter scaleDownTotal = new Counter(); + private final List registeredMetricKeys = Lists.newArrayList(); + + public DynamicAllocationMetrics(String groupName, MetricRegistry registry, Source source) { + this.groupName = groupName; + this.registry = registry; + this.source = source; + } + + public void register() { + try { + registerMetric( + OPTIMIZER_GROUP_PENDING_REMOVAL_OPTIMIZERS, + (Gauge) source::pendingRemovalOptimizers); + registerMetric(OPTIMIZER_GROUP_EFFECTIVE_THREADS, (Gauge) source::effectiveThreads); + registerMetric(OPTIMIZER_GROUP_BACKLOG_DURATION_MS, (Gauge) source::backlogDurationMs); + registerMetric(OPTIMIZER_GROUP_SCALE_UP_TOTAL, scaleUpTotal); + registerMetric(OPTIMIZER_GROUP_SCALE_DOWN_TOTAL, scaleDownTotal); + } catch (Exception e) { + // Roll back any metrics that were partially registered before the failure so that a retry + // finds a clean state. + unregister(); + throw e; + } + } + + public void unregister() { + registeredMetricKeys.forEach(registry::unregister); + registeredMetricKeys.clear(); + } + + public void incScaleUp() { + scaleUpTotal.inc(); + } + + public void incScaleDown() { + scaleDownTotal.inc(); + } + + private void registerMetric(MetricDefine define, Metric metric) { + registeredMetricKeys.add( + registry.register(define, ImmutableMap.of(GROUP_TAG, groupName), metric)); + } +} diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java index e0cd450573..034d7c92c1 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/dra/DynamicAllocationState.java @@ -145,6 +145,15 @@ public boolean wasDemandActive() { return lastEvalDemandActive; } + /** + * How long demand has exceeded capacity as of {@code nowMs}, or {@code 0} while there is no + * backlog. Read side of the backlog-duration gauge: derived at scrape time rather than pushed by + * the keeper, whose rounds are seconds apart. + */ + public long backlogDurationMs(long nowMs) { + return backlogSinceMs < 0 ? 0 : nowMs - backlogSinceMs; + } + /** * Update per-token idle observations from this round's snapshot. A token with in-flight tasks has * its busy timestamp refreshed; a token seen for the first time is seeded with {@code nowMs} diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerScaleKeeper.java b/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerScaleKeeper.java index ab913232c6..2522d212ad 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerScaleKeeper.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/TestOptimizerScaleKeeper.java @@ -24,12 +24,20 @@ import org.apache.amoro.TableTestHelper; import org.apache.amoro.catalog.BasicCatalogTestHelper; import org.apache.amoro.catalog.CatalogTestHelper; +import org.apache.amoro.config.Configurations; +import org.apache.amoro.metrics.Counter; +import org.apache.amoro.metrics.Metric; +import org.apache.amoro.metrics.MetricDefine; +import org.apache.amoro.metrics.MetricKey; import org.apache.amoro.resource.ResourceContainer; import org.apache.amoro.resource.ResourceGroup; +import org.apache.amoro.server.manager.MetricManager; +import org.apache.amoro.server.optimizing.dra.DynamicAllocationMetrics; import org.apache.amoro.server.resource.ContainerMetadata; import org.apache.amoro.server.resource.Containers; import org.apache.amoro.server.resource.OptimizerInstance; import org.apache.amoro.server.table.AMSTableTestBase; +import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableMap; import org.apache.amoro.shade.guava32.com.google.common.collect.Maps; import org.apache.iceberg.common.DynFields; import org.junit.After; @@ -233,6 +241,14 @@ public void testBootWindowPreventsDuplicateScaleOuts() throws InterruptedExcepti 2, scaleOutCallCount.get(), "the deficit must be requested exactly once while the pods are still booting"); + @SuppressWarnings("unchecked") + org.apache.amoro.metrics.Gauge effectiveThreads = + (org.apache.amoro.metrics.Gauge) + draMetric(DynamicAllocationMetrics.OPTIMIZER_GROUP_EFFECTIVE_THREADS, group.getName()); + Assertions.assertEquals( + 2, + effectiveThreads.getValue().intValue(), + "requested-but-unregistered threads must count as effective"); } /** @@ -492,6 +508,272 @@ public void testScaleDownRemovesOneInstancePerCooldown() { "the cooldown expiry should admit the next removal"); } + private Metric draMetric(MetricDefine define, String groupName) { + return MetricManager.getInstance() + .getGlobalRegistry() + .getMetrics() + .get(new MetricKey(define, ImmutableMap.of("group", groupName))); + } + + /** + * The keeper owns the DRA metric lifecycle: watching a group registers its gauges and counters, + * and a disable — which hands the group back to the legacy floor keeper — removes them. + */ + @Test + public void testDraMetricsRegisteredOnWatchAndRemovedOnDisable() throws InterruptedException { + resourceAvailable.set(true); + scaleOutCallCount.set(0); + ResourceGroup group = buildDraResourceGroup(TEST_GROUP_NAME + "-12", 0, 1); + optimizerManager().createResourceGroup(group); + optimizingService().createResourceGroup(group); + + Assertions.assertNotNull( + draMetric( + DynamicAllocationMetrics.OPTIMIZER_GROUP_PENDING_REMOVAL_OPTIMIZERS, group.getName())); + Assertions.assertNotNull( + draMetric(DynamicAllocationMetrics.OPTIMIZER_GROUP_EFFECTIVE_THREADS, group.getName())); + Assertions.assertNotNull( + draMetric(DynamicAllocationMetrics.OPTIMIZER_GROUP_BACKLOG_DURATION_MS, group.getName())); + Assertions.assertNotNull( + draMetric(DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_UP_TOTAL, group.getName())); + Assertions.assertNotNull( + draMetric(DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_DOWN_TOTAL, group.getName())); + + Map legacyProps = Maps.newHashMap(); + legacyProps.put("memory", "1024"); + ResourceGroup disabled = + new ResourceGroup.Builder(group.getName(), MOCK_CONTAINER_NAME) + .addProperties(legacyProps) + .build(); + optimizerManager().updateResourceGroup(disabled); + optimizingService().updateResourceGroup(disabled); + Thread.sleep(500); + + Assertions.assertNull( + draMetric(DynamicAllocationMetrics.OPTIMIZER_GROUP_EFFECTIVE_THREADS, group.getName()), + "a disabled group's DRA metrics must go with its watch"); + Assertions.assertNull( + draMetric(DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_UP_TOTAL, group.getName())); + } + + /** + * A disable unwatches on the config-entry path itself, not on the keeper's next round: the + * round-driven unwatch runs on the leader only, so without this a follower would keep exporting + * the group's DRA metrics until failover. + */ + @Test + public void testDisableUnwatchesOnUpdatePathWithoutKeeperRound() throws InterruptedException { + resourceAvailable.set(true); + scaleOutCallCount.set(0); + ResourceGroup group = buildSlowDraResourceGroup(TEST_GROUP_NAME + "-15", 0); + optimizerManager().createResourceGroup(group); + optimizingService().createResourceGroup(group); + Assertions.assertNotNull( + draMetric(DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_UP_TOTAL, group.getName())); + // Let the delay-0 first round pass; the next one is minutes away (600s cadence), so any + // removal observed below must come from the update path, not from a keeper round. + Thread.sleep(200); + + Map legacyProps = Maps.newHashMap(); + legacyProps.put("memory", "1024"); + ResourceGroup disabled = + new ResourceGroup.Builder(group.getName(), MOCK_CONTAINER_NAME) + .addProperties(legacyProps) + .build(); + optimizerManager().updateResourceGroup(disabled); + optimizingService().updateResourceGroup(disabled); + + Assertions.assertNull( + draMetric(DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_UP_TOTAL, group.getName()), + "the update path must unwatch a disabled group on every node"); + } + + /** A scale-out round counts as one scale-up action, and the gauges read the keeper's state. */ + @Test + public void testScaleUpRoundIncrementsCounterAndGaugesReadState() throws InterruptedException { + resourceAvailable.set(true); + scaleOutCallCount.set(0); + ResourceGroup group = buildDraResourceGroup(TEST_GROUP_NAME + "-13", 2, 1); + optimizerManager().createResourceGroup(group); + optimizingService().createResourceGroup(group); + Thread.sleep(500); + + Counter scaleUp = + (Counter) + draMetric(DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_UP_TOTAL, group.getName()); + Assertions.assertEquals( + 1, scaleUp.getCount(), "one floor-deficit round is exactly one scale-up action"); + @SuppressWarnings("unchecked") + org.apache.amoro.metrics.Gauge effectiveThreads = + (org.apache.amoro.metrics.Gauge) + draMetric(DynamicAllocationMetrics.OPTIMIZER_GROUP_EFFECTIVE_THREADS, group.getName()); + Assertions.assertEquals( + 2, + effectiveThreads.getValue().intValue(), + "registered floor capacity should be visible as effective threads"); + } + + /** The pending-removal gauge tracks instances through drain start, retry, and removal. */ + @Test + public void testPendingRemovalGaugeCountsDrainingInstances() { + resourceAvailable.set(true); + scaleOutCallCount.set(0); + ResourceGroup group = buildSlowDraResourceGroup(TEST_GROUP_NAME + "-16", 0); + optimizerManager().createResourceGroup(group); + optimizingService().createResourceGroup(group); + OptimizerInstance optimizer = registerOptimizer(group.getName(), 1); + @SuppressWarnings("unchecked") + org.apache.amoro.metrics.Gauge pendingRemoval = + (org.apache.amoro.metrics.Gauge) + draMetric( + DynamicAllocationMetrics.OPTIMIZER_GROUP_PENDING_REMOVAL_OPTIMIZERS, + group.getName()); + Assertions.assertEquals(0, pendingRemoval.getValue().intValue()); + + // A failing release keeps the instance draining, exactly the stuck state the gauge is for. + mockContainer.setReleaseAvailable(false); + optimizingService().beginGracefulDrain(optimizer.getToken(), Long.MAX_VALUE); + optimizingService().executeRemoval(optimizer.getToken()); + Assertions.assertEquals(1, pendingRemoval.getValue().intValue()); + + mockContainer.setReleaseAvailable(true); + optimizingService().executeRemoval(optimizer.getToken()); + Assertions.assertEquals(0, pendingRemoval.getValue().intValue()); + } + + /** Starting a drain counts as one scale-down action. */ + @Test + public void testScaleDownIncrementsCounter() { + resourceAvailable.set(true); + scaleOutCallCount.set(0); + ResourceGroup group = buildSlowDraResourceGroup(TEST_GROUP_NAME + "-14", 0); + optimizerManager().createResourceGroup(group); + optimizingService().createResourceGroup(group); + registerOptimizer(group.getName(), 1); + + long t0 = System.currentTimeMillis(); + optimizingService().evaluateDynamicAllocation(group.getName(), t0); + Counter scaleDown = + (Counter) + draMetric(DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_DOWN_TOTAL, group.getName()); + Assertions.assertEquals(0, scaleDown.getCount()); + + optimizingService().evaluateDynamicAllocation(group.getName(), t0 + 1_300_000L); + Assertions.assertEquals( + 1, scaleDown.getCount(), "the drain start is the scale-down action, counted once"); + } + + /** + * A watch whose metric registration fails must leave no residue: the group stays unwatched, so + * the next config pass can watch it again instead of finding a watched-but-dead entry that + * swallows every retry until a restart. + */ + @Test + public void testWatchFailureLeavesGroupRewatchable() { + ResourceGroup group = buildSlowDraResourceGroup(TEST_GROUP_NAME + "-17", 0); + optimizerManager().createResourceGroup(group); + // Occupy one of the group's DRA keys so the watch's registration fails midway. + MetricKey conflict = + MetricManager.getInstance() + .getGlobalRegistry() + .register( + DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_UP_TOTAL, + ImmutableMap.of("group", group.getName()), + new Counter()); + Assertions.assertThrows( + RuntimeException.class, () -> optimizingService().createResourceGroup(group)); + Assertions.assertNull( + draMetric( + DynamicAllocationMetrics.OPTIMIZER_GROUP_PENDING_REMOVAL_OPTIMIZERS, group.getName()), + "a failed watch must roll back the metrics it managed to register"); + MetricManager.getInstance().getGlobalRegistry().unregister(conflict); + + optimizingService().updateResourceGroup(group); + Assertions.assertNotNull( + draMetric(DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_UP_TOTAL, group.getName()), + "the group must be rewatchable after a failed watch"); + Assertions.assertNotNull( + draMetric( + DynamicAllocationMetrics.OPTIMIZER_GROUP_PENDING_REMOVAL_OPTIMIZERS, group.getName())); + } + + /** + * The global metric registry outlives the service on a leader hand-off: a disposed service must + * unregister its DRA metrics, or the next leader's watch of the same group throws on the leftover + * keys and the group goes watched-but-dead until a JVM restart. + */ + @Test + public void testDisposeUnregistersDraMetrics() { + ResourceGroup group = buildSlowDraResourceGroup(TEST_GROUP_NAME + "-18", 0); + optimizerManager().createResourceGroup(group); + DefaultOptimizingService formerLeader = + new DefaultOptimizingService( + new Configurations(), catalogManager(), optimizerManager(), tableService(), null, null); + formerLeader.createResourceGroup(group); + Assertions.assertNotNull( + draMetric(DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_UP_TOTAL, group.getName())); + + formerLeader.dispose(); + Assertions.assertNull( + draMetric(DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_UP_TOTAL, group.getName()), + "a disposed service must not leave DRA metrics behind for the next leader"); + + // The next leader takes over the group without colliding with leftover keys. + optimizingService().createResourceGroup(group); + Assertions.assertNotNull( + draMetric(DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_UP_TOTAL, group.getName())); + } + + /** + * A watch arriving after dispose — an in-flight config-sync run or the keeper round's re-check + * racing a leader hand-off — must not register metrics from the dead service: the keys would + * outlive it in the global registry and fail the next leader's watch. Only never-watched groups + * are exposed (a watched group's entry survives in watchedGroups and swallows the call), which is + * exactly the racing paths' state: a new or re-enabled group, or one the round just unwatched. + */ + @Test + public void testWatchAfterDisposeDoesNotRegisterMetrics() { + ResourceGroup group = buildSlowDraResourceGroup(TEST_GROUP_NAME + "-19", 0); + DefaultOptimizingService formerLeader = + new DefaultOptimizingService( + new Configurations(), catalogManager(), optimizerManager(), tableService(), null, null); + formerLeader.dispose(); + + formerLeader.updateResourceGroup(group); + Assertions.assertNull( + draMetric(DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_UP_TOTAL, group.getName()), + "a watch arriving after dispose must not register metrics from a dead service"); + } + + /** + * watch() is documented as idempotent: an update of an already-watched group with a still-enabled + * config — routine property tuning — must re-enter watch() as a no-op instead of colliding with + * the group's live metric keys. + */ + @Test + public void testEnabledUpdateReentersWatchIdempotently() { + ResourceGroup group = buildSlowDraResourceGroup(TEST_GROUP_NAME + "-20", 0); + optimizerManager().createResourceGroup(group); + optimizingService().createResourceGroup(group); + Metric before = + draMetric(DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_UP_TOTAL, group.getName()); + Assertions.assertNotNull(before); + + Map properties = Maps.newHashMap(group.getProperties()); + properties.put(OptimizerProperties.DYNAMIC_ALLOCATION_MAX_PARALLELISM, "6"); + ResourceGroup updated = + new ResourceGroup.Builder(group.getName(), MOCK_CONTAINER_NAME) + .addProperties(properties) + .build(); + optimizerManager().updateResourceGroup(updated); + optimizingService().updateResourceGroup(updated); + + Assertions.assertSame( + before, + draMetric(DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_UP_TOTAL, group.getName()), + "an enabled-to-enabled update must keep the group's live metrics, not re-register them"); + } + /** * Disabling dynamic allocation mid-drain re-admits the draining pod to task assignment: once the * legacy floor keeper resumes duty for the group, a leftover drain block would starve the pod diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java index 938b97783a..3955e335e3 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java @@ -20,8 +20,10 @@ import static org.apache.amoro.server.optimizing.OptimizerGroupMetrics.GROUP_TAG; import static org.apache.amoro.server.optimizing.OptimizerGroupMetrics.OPTIMIZER_GROUP_COMMITTING_TABLES; +import static org.apache.amoro.server.optimizing.OptimizerGroupMetrics.OPTIMIZER_GROUP_CONFIG_INVALID; import static org.apache.amoro.server.optimizing.OptimizerGroupMetrics.OPTIMIZER_GROUP_EXECUTING_TABLES; import static org.apache.amoro.server.optimizing.OptimizerGroupMetrics.OPTIMIZER_GROUP_EXECUTING_TASKS; +import static org.apache.amoro.server.optimizing.OptimizerGroupMetrics.OPTIMIZER_GROUP_IDLE_OPTIMIZERS; import static org.apache.amoro.server.optimizing.OptimizerGroupMetrics.OPTIMIZER_GROUP_IDLE_TABLES; import static org.apache.amoro.server.optimizing.OptimizerGroupMetrics.OPTIMIZER_GROUP_MEMORY_BYTES_ALLOCATED; import static org.apache.amoro.server.optimizing.OptimizerGroupMetrics.OPTIMIZER_GROUP_OPTIMIZER_INSTANCES; @@ -31,6 +33,7 @@ import static org.apache.amoro.server.optimizing.OptimizerGroupMetrics.OPTIMIZER_GROUP_THREADS; import org.apache.amoro.BasicTableTestHelper; +import org.apache.amoro.OptimizerProperties; import org.apache.amoro.ServerTableIdentifier; import org.apache.amoro.TableFormat; import org.apache.amoro.TableTestHelper; @@ -58,6 +61,7 @@ import org.apache.amoro.server.table.DefaultTableRuntime; import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableMap; import org.apache.amoro.shade.guava32.com.google.common.collect.Lists; +import org.apache.amoro.shade.guava32.com.google.common.collect.Maps; import org.apache.amoro.table.MixedTable; import org.apache.amoro.table.TableProperties; import org.apache.amoro.table.UnkeyedTable; @@ -778,6 +782,63 @@ public void testAddAndRemoveOptimizers() { queue.dispose(); } + /** An optimizer is idle while it has no in-flight (SCHEDULED/ACKED) task. */ + @Test + public void testIdleOptimizersMetric() { + DefaultTableRuntime tableRuntime = initTableWithFiles(); + OptimizingQueue queue = buildOptimizingGroupService(tableRuntime); + MetricRegistry registry = MetricManager.getInstance().getGlobalRegistry(); + Map tagValues = ImmutableMap.of(GROUP_TAG, testResourceGroup().getName()); + Gauge idleOptimizersGauge = + (Gauge) + registry.getMetrics().get(new MetricKey(OPTIMIZER_GROUP_IDLE_OPTIMIZERS, tagValues)); + + OptimizerRegisterInfo registerInfo = + new OptimizerRegisterInfo( + 2, 2048, System.currentTimeMillis(), testResourceGroup().getName()); + final OptimizerInstance optimizer = new OptimizerInstance(registerInfo, "test_container"); + queue.addOptimizer(optimizer); + Assert.assertEquals(1, idleOptimizersGauge.getValue().longValue()); + + OptimizerThread thread = + new OptimizerThread(1, null) { + @Override + public String getToken() { + return optimizer.getToken(); + } + }; + Assert.assertNotNull(queue.pollTask(thread, MAX_POLLING_TIME)); + Assert.assertEquals( + "an optimizer holding an in-flight task is not idle", + 0, + idleOptimizersGauge.getValue().longValue()); + + queue.removeOptimizer(optimizer); + queue.dispose(); + } + + /** The gauge flips when a config update leaves an opted-in group with an invalid DRA config. */ + @Test + public void testConfigInvalidMetric() { + OptimizingQueue queue = buildOptimizingGroupService(); + MetricRegistry registry = MetricManager.getInstance().getGlobalRegistry(); + Map tagValues = ImmutableMap.of(GROUP_TAG, testResourceGroup().getName()); + Gauge configInvalidGauge = + (Gauge) + registry.getMetrics().get(new MetricKey(OPTIMIZER_GROUP_CONFIG_INVALID, tagValues)); + Assert.assertEquals(0, configInvalidGauge.getValue().intValue()); + + Map props = Maps.newHashMap(); + props.put(OptimizerProperties.DYNAMIC_ALLOCATION_ENABLED, "true"); + // Enabled without max-parallelism: invalid, running under the startup fail-safe fallback. + queue.updateOptimizerGroup( + new ResourceGroup.Builder(testResourceGroup().getName(), "local") + .addProperties(props) + .build()); + Assert.assertEquals(1, configInvalidGauge.getValue().intValue()); + queue.dispose(); + } + @Test public void testProcessCloseKeepsLastOptimizedSnapshotId() { DefaultTableRuntime tableRuntime = initTableWithFiles(); diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestComputeScaleUp.java b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestComputeScaleUp.java index 58e852f0f5..75ae08c902 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestComputeScaleUp.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestComputeScaleUp.java @@ -82,6 +82,31 @@ void idleCapacityCoveringBacklogReturnsZero() { Assertions.assertEquals(0, state.computeScaleUp(8, 2, 3, 0, config(0, 100, 2), T0)); } + // --- backlogDurationMs: read-side accessor for the backlog gauge --- + + @Test + void backlogDurationIsZeroWithoutDemand() { + DynamicAllocationState state = new DynamicAllocationState(); + Assertions.assertEquals(0, state.backlogDurationMs(T0)); + } + + @Test + void backlogDurationTracksSinceDemandFirstObserved() { + DynamicAllocationState state = new DynamicAllocationState(); + DynamicAllocationConfig config = config(0, 100, 2); + state.computeScaleUp(2, 2, 5, 0, config, T0); + Assertions.assertEquals(BACKLOG_MS - 1, state.backlogDurationMs(T0 + BACKLOG_MS - 1)); + } + + @Test + void backlogDurationResetsWhenDemandClears() { + DynamicAllocationState state = new DynamicAllocationState(); + DynamicAllocationConfig config = config(0, 100, 2); + state.computeScaleUp(2, 2, 5, 0, config, T0); + state.computeScaleUp(8, 2, 0, 0, config, T0 + 10_000); + Assertions.assertEquals(0, state.backlogDurationMs(T0 + 10_000)); + } + // --- immediate demand (Layer 1): backlog timer, ramp, clamp --- @Test diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestDynamicAllocationConfig.java b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestDynamicAllocationConfig.java index e0eec2a037..fefef83f6f 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestDynamicAllocationConfig.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestDynamicAllocationConfig.java @@ -304,6 +304,53 @@ void notEffectivelyEnabledWhenConfigInvalid() { Assertions.assertFalse(DynamicAllocationConfig.isEffectivelyEnabled(group(props))); } + // --- isConfigInvalid: read side of the optimizer_group_config_invalid gauge --- + + @Test + void validEnabledConfigIsNotInvalid() { + Assertions.assertFalse(DynamicAllocationConfig.isConfigInvalid(group(enabledProps()))); + } + + @Test + void groupThatNeverOptedInIsNotInvalid() { + Assertions.assertFalse(DynamicAllocationConfig.isConfigInvalid(group(new HashMap<>()))); + } + + @Test + void enabledConfigFailingValidationIsInvalid() { + Map props = new HashMap<>(); + props.put(OptimizerProperties.DYNAMIC_ALLOCATION_ENABLED, "true"); + // Enabled without max-parallelism: the startup fail-safe silently disables DRA, so this is + // exactly the state the gauge must surface. + Assertions.assertTrue(DynamicAllocationConfig.isConfigInvalid(group(props))); + } + + @Test + void enabledConfigFailingParseIsInvalid() { + Map props = enabledProps(); + props.put(OptimizerProperties.DYNAMIC_ALLOCATION_DRAIN_TIMEOUT, "not-a-duration"); + Assertions.assertTrue(DynamicAllocationConfig.isConfigInvalid(group(props))); + } + + @Test + void enabledConfigOverflowingDurationIsInvalid() { + Map props = enabledProps(); + // Duration parsing multiplies the unit out (Math.multiplyExact), so an absurd value throws + // ArithmeticException instead of IllegalArgumentException. The gauge read must report it as + // invalid, not propagate: a gauge exception aborts the whole metrics scrape. + props.put(OptimizerProperties.DYNAMIC_ALLOCATION_DRAIN_TIMEOUT, Long.MAX_VALUE + "min"); + Assertions.assertTrue(DynamicAllocationConfig.isConfigInvalid(group(props))); + } + + @Test + void disabledGroupWithMalformedPropertiesIsNotInvalid() { + Map props = new HashMap<>(); + props.put(OptimizerProperties.DYNAMIC_ALLOCATION_DRAIN_TIMEOUT, "not-a-duration"); + // Whatever its leftover properties parse to, a group that has not opted in is not in the + // fail-safe fallback and must not alarm. + Assertions.assertFalse(DynamicAllocationConfig.isConfigInvalid(group(props))); + } + @Test void effectiveMinParallelismKeyPrefersNamespacedWhenPresent() { Map props = new HashMap<>(); diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestDynamicAllocationMetrics.java b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestDynamicAllocationMetrics.java new file mode 100644 index 0000000000..c5240b2de9 --- /dev/null +++ b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/dra/TestDynamicAllocationMetrics.java @@ -0,0 +1,134 @@ +/* + * 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.amoro.server.optimizing.dra; + +import org.apache.amoro.metrics.Counter; +import org.apache.amoro.metrics.Gauge; +import org.apache.amoro.metrics.Metric; +import org.apache.amoro.metrics.MetricDefine; +import org.apache.amoro.metrics.MetricKey; +import org.apache.amoro.metrics.MetricRegistry; +import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableMap; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +/** + * Tests for {@link DynamicAllocationMetrics}: gauge values are read through the injected {@link + * DynamicAllocationMetrics.Source}, counters accumulate through the inc hooks, and unregister + * removes every metric of the group. + */ +public class TestDynamicAllocationMetrics { + + private static class FakeSource implements DynamicAllocationMetrics.Source { + int pendingRemovalOptimizers; + int effectiveThreads; + long backlogDurationMs; + + @Override + public int pendingRemovalOptimizers() { + return pendingRemovalOptimizers; + } + + @Override + public int effectiveThreads() { + return effectiveThreads; + } + + @Override + public long backlogDurationMs() { + return backlogDurationMs; + } + } + + private final MetricRegistry registry = new MetricRegistry(); + private final FakeSource source = new FakeSource(); + private final DynamicAllocationMetrics metrics = + new DynamicAllocationMetrics("group1", registry, source); + + private Metric metric(MetricDefine define) { + return registry.getMetrics().get(new MetricKey(define, ImmutableMap.of("group", "group1"))); + } + + @Test + void gaugesReflectSourceValues() { + metrics.register(); + source.pendingRemovalOptimizers = 2; + source.effectiveThreads = 7; + source.backlogDurationMs = 45_000L; + + Gauge pendingRemoval = + (Gauge) metric(DynamicAllocationMetrics.OPTIMIZER_GROUP_PENDING_REMOVAL_OPTIMIZERS); + Gauge effectiveThreads = + (Gauge) metric(DynamicAllocationMetrics.OPTIMIZER_GROUP_EFFECTIVE_THREADS); + Gauge backlogDuration = + (Gauge) metric(DynamicAllocationMetrics.OPTIMIZER_GROUP_BACKLOG_DURATION_MS); + Assertions.assertEquals(2, ((Number) pendingRemoval.getValue()).intValue()); + Assertions.assertEquals(7, ((Number) effectiveThreads.getValue()).intValue()); + Assertions.assertEquals(45_000L, ((Number) backlogDuration.getValue()).longValue()); + } + + @Test + void countersStartAtZeroAndAccumulate() { + metrics.register(); + Counter scaleUp = (Counter) metric(DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_UP_TOTAL); + Counter scaleDown = (Counter) metric(DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_DOWN_TOTAL); + Assertions.assertEquals(0, scaleUp.getCount()); + Assertions.assertEquals(0, scaleDown.getCount()); + + metrics.incScaleUp(); + metrics.incScaleUp(); + metrics.incScaleDown(); + Assertions.assertEquals(2, scaleUp.getCount()); + Assertions.assertEquals(1, scaleDown.getCount()); + } + + @Test + void registerRollsBackPartialRegistrationOnFailure() { + // Occupy one of the group's keys so register() fails midway through its five metrics. + MetricKey conflict = + registry.register( + DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_UP_TOTAL, + ImmutableMap.of("group", "group1"), + new Counter()); + Assertions.assertThrows(RuntimeException.class, metrics::register); + Assertions.assertNull( + metric(DynamicAllocationMetrics.OPTIMIZER_GROUP_PENDING_REMOVAL_OPTIMIZERS), + "a failed register() must not leave partially registered metrics behind"); + Assertions.assertNull(metric(DynamicAllocationMetrics.OPTIMIZER_GROUP_EFFECTIVE_THREADS)); + Assertions.assertNull(metric(DynamicAllocationMetrics.OPTIMIZER_GROUP_BACKLOG_DURATION_MS)); + + registry.unregister(conflict); + metrics.register(); + Assertions.assertNotNull(metric(DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_UP_TOTAL)); + Assertions.assertNotNull( + metric(DynamicAllocationMetrics.OPTIMIZER_GROUP_PENDING_REMOVAL_OPTIMIZERS)); + } + + @Test + void unregisterRemovesAllGroupMetrics() { + metrics.register(); + metrics.unregister(); + Assertions.assertNull( + metric(DynamicAllocationMetrics.OPTIMIZER_GROUP_PENDING_REMOVAL_OPTIMIZERS)); + Assertions.assertNull(metric(DynamicAllocationMetrics.OPTIMIZER_GROUP_EFFECTIVE_THREADS)); + Assertions.assertNull(metric(DynamicAllocationMetrics.OPTIMIZER_GROUP_BACKLOG_DURATION_MS)); + Assertions.assertNull(metric(DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_UP_TOTAL)); + Assertions.assertNull(metric(DynamicAllocationMetrics.OPTIMIZER_GROUP_SCALE_DOWN_TOTAL)); + } +} diff --git a/docs/user-guides/metrics.md b/docs/user-guides/metrics.md index ee6c11e21a..31c1e10c3f 100644 --- a/docs/user-guides/metrics.md +++ b/docs/user-guides/metrics.md @@ -77,6 +77,23 @@ Amoro has supported built-in metrics to measure status of table self-optimizing | optimizer_group_optimizer_instances | Gauge | group | Number of optimizer instances in optimizer group | | optimizer_group_memory_bytes_allocated | Gauge | group | Memory bytes allocated in optimizer group | | optimizer_group_threads | Gauge | group | Number of total threads in optimizer group | +| optimizer_group_idle_optimizers | Gauge | group | Number of optimizer instances with no in-flight task in optimizer group | +| optimizer_group_config_invalid | Gauge | group | 1 while the group's dynamic allocation configuration is invalid and the fail-safe fallback is active, else 0 | + +The following metrics are exported only while dynamic allocation is effectively enabled on the group +(see [Managing optimizers](../managing-optimizers/)); the counters count attempted scaling actions, so +one scale-out round — whatever its instance count and whether its resource requests succeed — and one +drain start each count 1. A rising `scale_up_total` without a matching rise of +`optimizer_group_optimizer_instances`, sustained beyond the pod boot window, signals failing +resource requests; short gaps are normal while requested pods are still booting. + +| Metric Name | Type | Tags | Description | +|-------------------------------------------|---------|-------|----------------------------------------------------| +| optimizer_group_pending_removal_optimizers | Gauge | group | Number of optimizer instances in graceful drain in optimizer group | +| optimizer_group_effective_threads | Gauge | group | Number of registered threads plus threads of optimizers pending registration in optimizer group | +| optimizer_group_backlog_duration_ms | Gauge | group | Duration in milliseconds since demand first exceeded capacity in optimizer group, 0 while there is no backlog | +| optimizer_group_scale_up_total | Counter | group | Cumulative count of attempted scale-up actions in optimizer group | +| optimizer_group_scale_down_total | Counter | group | Cumulative count of scale-down actions in optimizer group | ## Orphan Files Cleaning metrics