This is an automated email from the ASF dual-hosted git repository.
zhoujinsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new 56c34f61e [AMORO-4313] AIP-5 Phase 5: observability metrics for
dynamic allocation (#4314)
56c34f61e is described below
commit 56c34f61ece921d83ab6d32ca0a8f6715f0447fd
Author: Jiwon Park <[email protected]>
AuthorDate: Mon Aug 17 12:46:58 2026 +0900
[AMORO-4313] AIP-5 Phase 5: observability metrics for dynamic allocation
(#4314)
* [AMORO-4313] Expose backlog duration from DynamicAllocationState for the
backlog gauge
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4313] Add per-group dynamic allocation metrics with source-backed
gauges and scale counters
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4313] Wire dynamic allocation metrics into the scale keeper watch
lifecycle
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4313] Add idle-optimizer and config-invalid gauges to the
optimizer group metrics
Signed-off-by: Jiwon Park <[email protected]>
* [AMORO-4313] Document dynamic allocation metrics in the metrics table
Signed-off-by: Jiwon Park <[email protected]>
---------
Signed-off-by: Jiwon Park <[email protected]>
---
.../amoro/server/DefaultOptimizingService.java | 97 ++++++-
.../server/optimizing/OptimizerGroupMetrics.java | 42 +++
.../optimizing/dra/DynamicAllocationConfig.java | 25 ++
.../optimizing/dra/DynamicAllocationMetrics.java | 142 +++++++++++
.../optimizing/dra/DynamicAllocationState.java | 9 +
.../amoro/server/TestOptimizerScaleKeeper.java | 282 +++++++++++++++++++++
.../server/optimizing/TestOptimizingQueue.java | 61 +++++
.../server/optimizing/dra/TestComputeScaleUp.java | 25 ++
.../dra/TestDynamicAllocationConfig.java | 47 ++++
.../dra/TestDynamicAllocationMetrics.java | 134 ++++++++++
docs/user-guides/metrics.md | 17 ++
11 files changed, 876 insertions(+), 5 deletions(-)
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 2e9aa4a28..eb943e605 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.catalog.CatalogManager;
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 @@ public class DefaultOptimizingService extends
StatedPersistentBase
new ConcurrentHashMap<>();
private final Set<String> watchedGroups = ConcurrentHashMap.newKeySet();
private final Map<String, Integer> planningBoundStreaks = new
ConcurrentHashMap<>();
+ private final Map<String, DynamicAllocationMetrics> 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.
+ *
+ * <p>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 class DefaultOptimizingService extends
StatedPersistentBase
}
}
- 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 @@ public class DefaultOptimizingService extends
StatedPersistentBase
* 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 @@ public class DefaultOptimizingService extends
StatedPersistentBase
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 @@ public class DefaultOptimizingService extends
StatedPersistentBase
}
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 5b70a4e83..4b640d4da 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.Metric;
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 class OptimizerGroupMetrics {
optimizerInstances.values().stream()
.mapToLong(OptimizerInstance::getThreadCount)
.sum());
+ registerMetric(
+ registry,
+ OPTIMIZER_GROUP_IDLE_OPTIMIZERS,
+ (Gauge<Long>)
+ () -> {
+ Set<String> 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<Integer>)
+ () ->
+
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 cfa0e2f1d..813a3f5db 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 class DynamicAllocationConfig {
}
}
+ /**
+ * 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 000000000..40ddf3cc3
--- /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<MetricKey> 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<Integer>) source::pendingRemovalOptimizers);
+ registerMetric(OPTIMIZER_GROUP_EFFECTIVE_THREADS, (Gauge<Integer>)
source::effectiveThreads);
+ registerMetric(OPTIMIZER_GROUP_BACKLOG_DURATION_MS, (Gauge<Long>)
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 e0cd45057..034d7c92c 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 final class DynamicAllocationState {
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 ab913232c..2522d212a 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.TableFormat;
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 class TestOptimizerScaleKeeper extends
AMSTableTestBase {
2,
scaleOutCallCount.get(),
"the deficit must be requested exactly once while the pods are still
booting");
+ @SuppressWarnings("unchecked")
+ org.apache.amoro.metrics.Gauge<Integer> effectiveThreads =
+ (org.apache.amoro.metrics.Gauge<Integer>)
+
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 class TestOptimizerScaleKeeper extends
AMSTableTestBase {
"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<String, String> 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<String, String> 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<Integer> effectiveThreads =
+ (org.apache.amoro.metrics.Gauge<Integer>)
+
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<Integer> pendingRemoval =
+ (org.apache.amoro.metrics.Gauge<Integer>)
+ 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<String, String> 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 938b97783..3955e335e 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 @@ package org.apache.amoro.server.optimizing;
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
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.AMSTableTestBase;
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 class TestOptimizingQueue extends AMSTableTestBase {
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<String, String> tagValues = ImmutableMap.of(GROUP_TAG,
testResourceGroup().getName());
+ Gauge<Long> idleOptimizersGauge =
+ (Gauge<Long>)
+ 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<String, String> tagValues = ImmutableMap.of(GROUP_TAG,
testResourceGroup().getName());
+ Gauge<Integer> configInvalidGauge =
+ (Gauge<Integer>)
+ registry.getMetrics().get(new
MetricKey(OPTIMIZER_GROUP_CONFIG_INVALID, tagValues));
+ Assert.assertEquals(0, configInvalidGauge.getValue().intValue());
+
+ Map<String, String> 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 58e852f0f..75ae08c90 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 @@ public class TestComputeScaleUp {
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 e0eec2a03..fefef83f6 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 @@ public class TestDynamicAllocationConfig {
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<String, String> 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<String, String> props = enabledProps();
+ props.put(OptimizerProperties.DYNAMIC_ALLOCATION_DRAIN_TIMEOUT,
"not-a-duration");
+
Assertions.assertTrue(DynamicAllocationConfig.isConfigInvalid(group(props)));
+ }
+
+ @Test
+ void enabledConfigOverflowingDurationIsInvalid() {
+ Map<String, String> 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<String, String> 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<String, String> 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 000000000..c5240b2de
--- /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 ee6c11e21..31c1e10c3 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
