This is an automated email from the ASF dual-hosted git repository. gyfora pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
commit 17fd0f7ead0d22795f3475c6634ce21068d189ba Author: Pradeepta Choudhury <[email protected]> AuthorDate: Tue Apr 21 16:18:24 2026 +0300 [FLINK-XXXXX][autoscaler] Custom Evaluator plugin for Flink Autoscaler Squashed from PR #953 (https://github.com/apache/flink-kubernetes-operator/pull/953). Original commits by Pradeepta Choudhury <[email protected]>: - Draft PR for plugin based approach for custom evaluator for scaling metric evaluation - Add sample custom evaluator for simple trend adjustor and fallback key support with delegating configuration - Add getName method to CustomEvaluator and javadocs - Refactor tests for the plugin and pass null for custom evaluators in standalone autoscaler entrypoint - Rename evaluator plugin to match naming convention for plugins Co-authored-by: Pradeepta Choudhury <[email protected]> --- docs/content.zh/docs/operations/plugins.md | 2 + .../standalone/StandaloneAutoscalerEntrypoint.java | 3 +- .../apache/flink/autoscaler/JobAutoScalerImpl.java | 35 ++- .../flink/autoscaler/ScalingMetricEvaluator.java | 127 +++++++++- .../flink/autoscaler/config/AutoScalerOptions.java | 18 ++ .../metrics/FlinkAutoscalerEvaluator.java | 118 ++++++++++ .../autoscaler/AutoScalerCustomEvaluatorTest.java | 173 ++++++++++++++ .../flink/autoscaler/BacklogBasedScalingTest.java | 4 +- .../autoscaler/DelayedScaleDownEndToEndTest.java | 3 +- .../flink/autoscaler/JobAutoScalerImplTest.java | 157 ++++++++++++- .../MetricsCollectionAndEvaluationTest.java | 9 +- .../autoscaler/RecommendedParallelismTest.java | 4 +- .../autoscaler/ScalingMetricEvaluatorTest.java | 258 ++++++++++++++++++++- .../autoscaler/metrics/TestCustomEvaluator.java | 51 ++++ .../flink/kubernetes/operator/FlinkOperator.java | 6 +- .../operator/autoscaler/AutoscalerFactory.java | 8 +- .../operator/autoscaler/AutoscalerUtils.java | 75 ++++++ .../src/test/assembly/test-plugins-assembly.xml | 3 +- .../operator/autoscaler/AutoscalerFactoryTest.java | 5 +- .../operator/autoscaler/AutoscalerUtilsTest.java | 65 ++++++ .../TestingFlinkDeploymentController.java | 3 +- ...ink.autoscaler.metrics.FlinkAutoscalerEvaluator | 19 ++ 22 files changed, 1113 insertions(+), 33 deletions(-) diff --git a/docs/content.zh/docs/operations/plugins.md b/docs/content.zh/docs/operations/plugins.md index 923569354..f420358a8 100644 --- a/docs/content.zh/docs/operations/plugins.md +++ b/docs/content.zh/docs/operations/plugins.md @@ -185,3 +185,5 @@ The following steps demonstrate how to develop and use a custom mutator. ```text 2023-12-12 06:26:56,667 o.a.f.k.o.u.MutatorUtils [INFO ] Discovered mutator from plugin directory[/opt/flink/plugins]: org.apache.flink.mutator.CustomFlinkMutator. ``` + +## Flink Autoscaler Custom Evaluator diff --git a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerEntrypoint.java b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerEntrypoint.java index c24972702..957af4c41 100644 --- a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerEntrypoint.java +++ b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerEntrypoint.java @@ -94,7 +94,8 @@ public class StandaloneAutoscalerEntrypoint { new ScalingExecutor<>(eventHandler, stateStore), eventHandler, new RescaleApiScalingRealizer<>(eventHandler), - stateStore); + stateStore, + null); } @VisibleForTesting diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java index d8ee54abf..ddfd78cf3 100644 --- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java +++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java @@ -19,14 +19,17 @@ package org.apache.flink.autoscaler; import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.api.common.JobStatus; +import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.autoscaler.config.AutoScalerOptions; import org.apache.flink.autoscaler.event.AutoScalerEventHandler; import org.apache.flink.autoscaler.exceptions.NotReadyException; import org.apache.flink.autoscaler.metrics.AutoscalerFlinkMetrics; import org.apache.flink.autoscaler.metrics.EvaluatedMetrics; +import org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator; import org.apache.flink.autoscaler.realizer.ScalingRealizer; import org.apache.flink.autoscaler.state.AutoScalerStateStore; import org.apache.flink.autoscaler.tuning.ConfigChanges; +import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.PipelineOptions; import org.apache.flink.util.Preconditions; @@ -36,9 +39,11 @@ import org.slf4j.LoggerFactory; import java.time.Clock; import java.util.HashMap; import java.util.Map; +import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import static org.apache.flink.autoscaler.config.AutoScalerOptions.AUTOSCALER_ENABLED; +import static org.apache.flink.autoscaler.config.AutoScalerOptions.CUSTOM_EVALUATOR_NAME; import static org.apache.flink.autoscaler.metrics.AutoscalerFlinkMetrics.initRecommendedParallelism; import static org.apache.flink.autoscaler.metrics.AutoscalerFlinkMetrics.resetRecommendedParallelism; import static org.apache.flink.autoscaler.metrics.ScalingHistoryUtils.getTrimmedScalingHistory; @@ -58,6 +63,7 @@ public class JobAutoScalerImpl<KEY, Context extends JobAutoScalerContext<KEY>> private final AutoScalerEventHandler<KEY, Context> eventHandler; private final ScalingRealizer<KEY, Context> scalingRealizer; private final AutoScalerStateStore<KEY, Context> stateStore; + private final Map<String, FlinkAutoscalerEvaluator> customEvaluators; private Clock clock = Clock.systemDefaultZone(); @@ -73,13 +79,15 @@ public class JobAutoScalerImpl<KEY, Context extends JobAutoScalerContext<KEY>> ScalingExecutor<KEY, Context> scalingExecutor, AutoScalerEventHandler<KEY, Context> eventHandler, ScalingRealizer<KEY, Context> scalingRealizer, - AutoScalerStateStore<KEY, Context> stateStore) { + AutoScalerStateStore<KEY, Context> stateStore, + Map<String, FlinkAutoscalerEvaluator> customEvaluators) { this.metricsCollector = metricsCollector; this.evaluator = evaluator; this.scalingExecutor = scalingExecutor; this.eventHandler = eventHandler; this.scalingRealizer = scalingRealizer; this.stateStore = stateStore; + this.customEvaluators = customEvaluators; } @Override @@ -203,8 +211,15 @@ public class JobAutoScalerImpl<KEY, Context extends JobAutoScalerContext<KEY>> // Scaling tracking data contains previous restart times that are taken into account var restartTime = scalingTracking.getMaxRestartTimeOrDefault(ctx.getConfiguration()); + + var customEvaluatorWithConfig = getCustomEvaluatorIfRequired(ctx.getConfiguration()); + var evaluatedMetrics = - evaluator.evaluate(ctx.getConfiguration(), collectedMetrics, restartTime); + evaluator.evaluate( + ctx.getConfiguration(), + collectedMetrics, + restartTime, + customEvaluatorWithConfig); LOG.debug("Evaluated metrics: {}", evaluatedMetrics); lastEvaluatedMetrics.put(ctx.getJobKey(), evaluatedMetrics); @@ -259,4 +274,20 @@ public class JobAutoScalerImpl<KEY, Context extends JobAutoScalerContext<KEY>> this.metricsCollector.setClock(clock); this.scalingExecutor.setClock(clock); } + + @VisibleForTesting + protected Tuple2<FlinkAutoscalerEvaluator, Configuration> getCustomEvaluatorIfRequired( + Configuration conf) { + return Optional.ofNullable(conf.get(CUSTOM_EVALUATOR_NAME)) + .map( + name -> { + FlinkAutoscalerEvaluator evaluator = customEvaluators.get(name); + return evaluator != null + ? new Tuple2<>( + evaluator, + AutoScalerOptions.forCustomEvaluator(conf, name)) + : null; + }) + .orElse(null); + } } diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java index 17956ec0c..9fb59ff6a 100644 --- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java +++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java @@ -18,16 +18,19 @@ package org.apache.flink.autoscaler; import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.autoscaler.config.AutoScalerOptions; import org.apache.flink.autoscaler.metrics.CollectedMetricHistory; import org.apache.flink.autoscaler.metrics.CollectedMetrics; import org.apache.flink.autoscaler.metrics.EvaluatedMetrics; import org.apache.flink.autoscaler.metrics.EvaluatedScalingMetric; +import org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator; import org.apache.flink.autoscaler.metrics.MetricAggregator; import org.apache.flink.autoscaler.metrics.ScalingMetric; import org.apache.flink.autoscaler.topology.JobTopology; import org.apache.flink.autoscaler.utils.AutoScalerUtils; import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.UnmodifiableConfiguration; import org.apache.flink.runtime.jobgraph.JobVertexID; import org.slf4j.Logger; @@ -38,6 +41,7 @@ import javax.annotation.Nullable; import java.time.Duration; import java.time.Instant; +import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.Optional; @@ -72,7 +76,12 @@ public class ScalingMetricEvaluator { private static final Logger LOG = LoggerFactory.getLogger(ScalingMetricEvaluator.class); public EvaluatedMetrics evaluate( - Configuration conf, CollectedMetricHistory collectedMetrics, Duration restartTime) { + Configuration conf, + CollectedMetricHistory collectedMetrics, + Duration restartTime, + @Nullable + Tuple2<FlinkAutoscalerEvaluator, Configuration> + customEvaluatorWithConfig) { LOG.debug("Restart time used in metrics evaluation: {}", restartTime); var scalingOutput = new HashMap<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>>(); var metricsHistory = collectedMetrics.getMetricHistory(); @@ -80,6 +89,23 @@ public class ScalingMetricEvaluator { boolean processingBacklog = isProcessingBacklog(topology, metricsHistory, conf); + var customEvaluationSession = + Optional.ofNullable(customEvaluatorWithConfig) + .map( + info -> + Tuple2.of( + info.f0, + new FlinkAutoscalerEvaluator.Context( + new UnmodifiableConfiguration(conf), + Collections.unmodifiableSortedMap( + metricsHistory), + Collections.unmodifiableMap(scalingOutput), + topology, + processingBacklog, + restartTime, + info.f1))) + .orElse(null); + for (var vertex : topology.getVerticesInTopologicalOrder()) { scalingOutput.put( vertex, @@ -90,7 +116,8 @@ public class ScalingMetricEvaluator { topology, vertex, processingBacklog, - restartTime)); + restartTime, + customEvaluationSession)); } var globalMetrics = evaluateGlobalMetrics(metricsHistory); @@ -132,7 +159,10 @@ public class ScalingMetricEvaluator { JobTopology topology, JobVertexID vertex, boolean processingBacklog, - Duration restartTime) { + Duration restartTime, + @Nullable + Tuple2<FlinkAutoscalerEvaluator, FlinkAutoscalerEvaluator.Context> + customEvaluationSession) { var latestVertexMetrics = metricsHistory.get(metricsHistory.lastKey()).getVertexMetrics().get(vertex); @@ -142,6 +172,7 @@ public class ScalingMetricEvaluator { double inputRateAvg = getRate(ScalingMetric.NUM_RECORDS_IN, vertex, metricsHistory); var evaluatedMetrics = new HashMap<ScalingMetric, EvaluatedScalingMetric>(); + computeTargetDataRate( topology, vertex, @@ -179,6 +210,24 @@ public class ScalingMetricEvaluator { EvaluatedScalingMetric.of(vertexInfo.getNumSourcePartitions())); computeProcessingRateThresholds(evaluatedMetrics, conf, processingBacklog, restartTime); + + Optional.ofNullable(customEvaluationSession) + .map( + session -> + runCustomEvaluator( + vertex, + Collections.unmodifiableMap(evaluatedMetrics), + session)) + .filter(customEvaluatedMetrics -> !customEvaluatedMetrics.isEmpty()) + .ifPresent( + customEvaluatedMetrics -> { + LOG.info( + "Merging custom evaluated metrics for vertex {}: {}", + vertex, + customEvaluatedMetrics); + mergeEvaluatedMetricsMaps(evaluatedMetrics, customEvaluatedMetrics); + }); + return evaluatedMetrics; } @@ -628,4 +677,76 @@ public class ScalingMetricEvaluator { to); return getRate(ScalingMetric.NUM_RECORDS_OUT, from, metricsHistory); } + + /** + * Executes the provided custom evaluator for the given job vertex. Calls {@link + * FlinkAutoscalerEvaluator#evaluateVertexMetrics} to evaluate scaling metrics. + * + * @param vertex The job vertex being evaluated. + * @param evaluatedMetrics Current evaluated metrics. + * @param customEvaluationSession A tuple containing the custom evaluator and evaluation + * context. + * @return A map of scaling metrics, with its corresponding evaluated scaling metric. + */ + @VisibleForTesting + protected static Map<ScalingMetric, EvaluatedScalingMetric> runCustomEvaluator( + JobVertexID vertex, + Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics, + Tuple2<FlinkAutoscalerEvaluator, FlinkAutoscalerEvaluator.Context> + customEvaluationSession) { + try { + return customEvaluationSession.f0.evaluateVertexMetrics( + vertex, evaluatedMetrics, customEvaluationSession.f1); + } catch (UnsupportedOperationException e) { + LOG.warn( + "Custom evaluator {} tried accessing an un-modifiable view.", + customEvaluationSession.f0.getClass(), + e); + } catch (Exception e) { + LOG.warn( + "Custom evaluator {} threw an exception.", + customEvaluationSession.f0.getClass(), + e); + } + + return Collections.emptyMap(); + } + + /** + * Merges the incoming evaluated metrics into actual evaluated metrics. + * + * @param actual The target evaluated metrics map to merge into. + * @param incoming The incoming map containing new evaluated metrics map to be merged + * (nullable). + */ + @VisibleForTesting + protected static void mergeEvaluatedMetricsMaps( + Map<ScalingMetric, EvaluatedScalingMetric> actual, + @Nullable Map<ScalingMetric, EvaluatedScalingMetric> incoming) { + Optional.ofNullable(incoming) + .ifPresent( + customEvaluatedMetric -> + customEvaluatedMetric.forEach( + (scalingMetric, evaluatedScalingMetric) -> + actual.merge( + scalingMetric, + evaluatedScalingMetric, + ScalingMetricEvaluator + ::mergeEvaluatedScalingMetric))); + } + + /** + * Merges two {@link EvaluatedScalingMetric} instances. + * + * @param actual The existing evaluated scaling metric. + * @param incoming The incoming evaluated scaling metric. + * @return A new {@link EvaluatedScalingMetric} instance with merged values. + */ + @VisibleForTesting + protected static EvaluatedScalingMetric mergeEvaluatedScalingMetric( + EvaluatedScalingMetric actual, EvaluatedScalingMetric incoming) { + return new EvaluatedScalingMetric( + !Double.isNaN(incoming.getCurrent()) ? incoming.getCurrent() : actual.getCurrent(), + !Double.isNaN(incoming.getAverage()) ? incoming.getAverage() : actual.getAverage()); + } } diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java index a67bfd505..8be3e0f10 100644 --- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java +++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java @@ -21,6 +21,8 @@ import org.apache.flink.autoscaler.JobVertexScaler; import org.apache.flink.autoscaler.metrics.MetricAggregator; import org.apache.flink.configuration.ConfigOption; import org.apache.flink.configuration.ConfigOptions; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.DelegatingConfiguration; import org.apache.flink.configuration.MemorySize; import java.time.Duration; @@ -31,6 +33,7 @@ public class AutoScalerOptions { public static final String OLD_K8S_OP_CONF_PREFIX = "kubernetes.operator."; public static final String AUTOSCALER_CONF_PREFIX = "job.autoscaler."; + public static final String CUSTOM_EVALUATOR_CONF_PREFIX = "metrics.custom-evaluator."; private static String oldOperatorConfigKey(String key) { return OLD_K8S_OP_CONF_PREFIX + AUTOSCALER_CONF_PREFIX + key; @@ -418,4 +421,19 @@ public class AutoScalerOptions { "Minimum allowed value for the observed scalability coefficient. " + "Prevents aggressive scaling by clamping low coefficient estimates. " + "If the estimated coefficient falls below this value, it is capped at the configured minimum."); + + public static final ConfigOption<String> CUSTOM_EVALUATOR_NAME = + autoScalerConfig(CUSTOM_EVALUATOR_CONF_PREFIX + "name") + .stringType() + .defaultValue(null) + .withFallbackKeys(oldOperatorConfigKey(CUSTOM_EVALUATOR_CONF_PREFIX + "name")) + .withDescription("Name of the custom evaluator to be used."); + + public static Configuration forCustomEvaluator( + Configuration configuration, String customEvaluatorName) { + // add support for fallBackKey with DelegatingConfiguration. + return new DelegatingConfiguration( + configuration, + AUTOSCALER_CONF_PREFIX + CUSTOM_EVALUATOR_CONF_PREFIX + customEvaluatorName + "."); + } } diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/FlinkAutoscalerEvaluator.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/FlinkAutoscalerEvaluator.java new file mode 100644 index 000000000..976f79b62 --- /dev/null +++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/FlinkAutoscalerEvaluator.java @@ -0,0 +1,118 @@ +/* + * 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.flink.autoscaler.metrics; + +import org.apache.flink.autoscaler.topology.JobTopology; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.core.plugin.Plugin; +import org.apache.flink.runtime.jobgraph.JobVertexID; + +import lombok.Getter; + +import java.time.Duration; +import java.time.Instant; +import java.util.Map; +import java.util.SortedMap; + +/** + * Interface for custom evaluators that allow tailored scaling metric evaluations. Implementations of + * this interface can provide custom logic to evaluate vertex metrics and merge them with internally + * evaluated metrics. + */ +public interface FlinkAutoscalerEvaluator extends Plugin { + + /** + * Returns the name of the custom evaluator. + * + * @return The name of the custom evaluator. + */ + String getName(); + + /** + * Evaluates scaling metrics for a given job vertex based on the internally evaluated metrics + * and context. + * + * @param vertex The {@link JobVertexID} identifying the vertex whose metrics are being + * evaluated. + * @param evaluatedMetrics An un-modifiable view of current vertex internally evaluated metrics. + * @param evaluationContext The evaluation context providing job-related configurations and + * historical metrics. + * @return A map of evaluated scaling metrics for the vertex which would get merged with + * internally evaluated metrics. + * @throws UnsupportedOperationException if an attempt is made to modify the {@code + * evaluatedMetrics}, {@code Context.jobConf}, {@code Context.metricsHistory}, {@code + * Context.evaluatedVertexMetrics}. + */ + Map<ScalingMetric, EvaluatedScalingMetric> evaluateVertexMetrics( + JobVertexID vertex, + Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics, + Context evaluationContext); + + /** + * Context providing relevant job and metric information to assist in custom metric evaluation. + */ + @Getter + class Context { + private final Configuration jobConf; + + private final SortedMap<Instant, CollectedMetrics> metricsHistory; + + private final Map<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>> + evaluatedVertexMetrics; + + private final JobTopology topology; + + private final boolean processingBacklog; + + private final Duration restartTime; + + private final Configuration customEvaluatorConf; + + /** + * Constructs a new {@link Context} instance. + * + * @param jobConf An un-modifiable view of job's configuration. + * @param metricsHistory An un-modifiable view of historical record of collected metrics, + * ordered by timestamp. + * @param evaluatedVertexMetrics This map contains an un-modifiable view of evaluated + * metrics for previously evaluated vertex. Note: evaluation of Vertex for scaling + * metrics happens topologically. + * @param topology The job topology representing the structure of the Flink job. + * @param processingBacklog Indicates whether the job is processing backlog. + * @param restartTime Maximum restart time based on scaling records. + * @param customEvaluatorConf The configuration associated with the custom evaluator. + */ + public Context( + Configuration jobConf, + SortedMap<Instant, CollectedMetrics> metricsHistory, + Map<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>> evaluatedVertexMetrics, + JobTopology topology, + boolean processingBacklog, + Duration restartTime, + Configuration customEvaluatorConf) { + this.jobConf = jobConf; + this.metricsHistory = metricsHistory; + this.evaluatedVertexMetrics = evaluatedVertexMetrics; + this.topology = topology; + this.processingBacklog = processingBacklog; + this.restartTime = restartTime; + this.customEvaluatorConf = customEvaluatorConf; + } + } +} diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/AutoScalerCustomEvaluatorTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/AutoScalerCustomEvaluatorTest.java new file mode 100644 index 000000000..00bf08e37 --- /dev/null +++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/AutoScalerCustomEvaluatorTest.java @@ -0,0 +1,173 @@ +/* + * 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.flink.autoscaler; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.autoscaler.config.AutoScalerOptions; +import org.apache.flink.autoscaler.event.TestingEventCollector; +import org.apache.flink.autoscaler.metrics.AutoscalerFlinkMetrics; +import org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator; +import org.apache.flink.autoscaler.metrics.TestCustomEvaluator; +import org.apache.flink.autoscaler.metrics.TestMetrics; +import org.apache.flink.autoscaler.realizer.TestingScalingRealizer; +import org.apache.flink.autoscaler.state.AutoScalerStateStore; +import org.apache.flink.autoscaler.state.InMemoryAutoScalerStateStore; +import org.apache.flink.autoscaler.topology.IOMetrics; +import org.apache.flink.autoscaler.topology.JobTopology; +import org.apache.flink.autoscaler.topology.VertexInfo; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.runtime.jobgraph.JobVertexID; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneId; +import java.util.Map; + +import static org.apache.flink.autoscaler.TestingAutoscalerUtils.createDefaultJobAutoScalerContext; +import static org.apache.flink.autoscaler.topology.ShipStrategy.REBALANCE; +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** Unit test for testing the integration of custom evaluators with the Flink job autoscaler. */ +public class AutoScalerCustomEvaluatorTest { + private JobAutoScalerContext<JobID> context; + private AutoScalerStateStore<JobID, JobAutoScalerContext<JobID>> stateStore; + + private TestingMetricsCollector<JobID, JobAutoScalerContext<JobID>> metricsCollector; + + private JobVertexID source1, sink; + + private JobAutoScalerImpl<JobID, JobAutoScalerContext<JobID>> autoscaler; + + @BeforeEach + public void setup() { + context = createDefaultJobAutoScalerContext(); + + TestingEventCollector<JobID, JobAutoScalerContext<JobID>> eventCollector = + new TestingEventCollector<>(); + stateStore = new InMemoryAutoScalerStateStore<>(); + + ScalingExecutor<JobID, JobAutoScalerContext<JobID>> scalingExecutor = + new ScalingExecutor<>(eventCollector, stateStore); + String testCustomEvaluatorName = "test-custom-evaluator"; + + var customEvaluators = createTestCustomEvaluator(); + + source1 = new JobVertexID(); + sink = new JobVertexID(); + + metricsCollector = + new TestingMetricsCollector<>( + new JobTopology( + new VertexInfo(source1, Map.of(), 1, 720, new IOMetrics(0, 0, 0)), + new VertexInfo( + sink, + Map.of(source1, REBALANCE), + 1, + 720, + new IOMetrics(0, 0, 0)))); + + var defaultConf = context.getConfiguration(); + defaultConf.set(AutoScalerOptions.AUTOSCALER_ENABLED, true); + defaultConf.set(AutoScalerOptions.STABILIZATION_INTERVAL, Duration.ZERO); + defaultConf.set(AutoScalerOptions.RESTART_TIME, Duration.ofSeconds(1)); + defaultConf.set(AutoScalerOptions.CATCH_UP_DURATION, Duration.ofSeconds(2)); + defaultConf.set(AutoScalerOptions.SCALING_ENABLED, true); + defaultConf.set(AutoScalerOptions.MAX_SCALE_DOWN_FACTOR, 1.); + defaultConf.set(AutoScalerOptions.MAX_SCALE_UP_FACTOR, (double) Integer.MAX_VALUE); + defaultConf.set(AutoScalerOptions.UTILIZATION_TARGET, 0.8); + defaultConf.set(AutoScalerOptions.UTILIZATION_MAX, 0.9); + defaultConf.set(AutoScalerOptions.UTILIZATION_MIN, 0.7); + defaultConf.set(AutoScalerOptions.SCALE_DOWN_INTERVAL, Duration.ZERO); + defaultConf.set(AutoScalerOptions.BACKLOG_PROCESSING_LAG_THRESHOLD, Duration.ofSeconds(1)); + + defaultConf.set(AutoScalerOptions.CUSTOM_EVALUATOR_NAME, testCustomEvaluatorName); + + autoscaler = + new JobAutoScalerImpl<>( + metricsCollector, + new ScalingMetricEvaluator(), + scalingExecutor, + eventCollector, + new TestingScalingRealizer<>(), + stateStore, + customEvaluators); + + // Reset custom window size to default + metricsCollector.setTestMetricWindowSize(null); + } + + @Test + public void test() throws Exception { + /* Test scaling up. */ + var now = Instant.ofEpochMilli(0); + setClocksTo(now); + metricsCollector.setJobUpdateTs(now); + // Adjust metric window size, so we can fill the metric window with two metrics + metricsCollector.setTestMetricWindowSize(Duration.ofSeconds(1)); + metricsCollector.updateMetrics( + source1, + TestMetrics.builder() + .numRecordsIn(0) + .numRecordsOut(0) + .numRecordsInPerSec(500.) + .maxBusyTimePerSec(8) + .pendingRecords(0L) + .build()); + metricsCollector.updateMetrics( + sink, TestMetrics.builder().numRecordsIn(0).maxBusyTimePerSec(850).build()); + + autoscaler.scale(context); + + now = now.plus(Duration.ofSeconds(1)); + setClocksTo(now); + + metricsCollector.updateMetrics( + source1, m -> m.setNumRecordsIn(500), m -> m.setNumRecordsOut(500)); + metricsCollector.updateMetrics(sink, m -> m.setNumRecordsIn(500)); + + autoscaler.scale(context); + + var scaledParallelism = ScalingExecutorTest.getScaledParallelism(stateStore, context); + assertEquals(3, scaledParallelism.get(source1)); + assertEquals(200, scaledParallelism.get(sink)); + assertFlinkMetricsCount(1, 0); + } + + private void setClocksTo(Instant time) { + var clock = Clock.fixed(time, ZoneId.systemDefault()); + autoscaler.setClock(clock); + } + + private void assertFlinkMetricsCount(int scalingCount, int balancedCount) { + AutoscalerFlinkMetrics autoscalerFlinkMetrics = + autoscaler.flinkMetrics.get(context.getJobKey()); + assertEquals(scalingCount, autoscalerFlinkMetrics.getNumScalingsCount()); + assertEquals(balancedCount, autoscalerFlinkMetrics.getNumBalancedCount()); + } + + private Map<String, FlinkAutoscalerEvaluator> createTestCustomEvaluator() { + var testCustomEvaluator = new TestCustomEvaluator(); + testCustomEvaluator.configure(new Configuration()); + return Map.of(testCustomEvaluator.getName(), testCustomEvaluator); + } +} diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/BacklogBasedScalingTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/BacklogBasedScalingTest.java index df950bb4c..956ecf32d 100644 --- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/BacklogBasedScalingTest.java +++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/BacklogBasedScalingTest.java @@ -40,6 +40,7 @@ import java.time.Clock; import java.time.Duration; import java.time.Instant; import java.time.ZoneId; +import java.util.Collections; import java.util.Map; import java.util.SortedMap; @@ -108,7 +109,8 @@ public class BacklogBasedScalingTest { scalingExecutor, eventCollector, new TestingScalingRealizer<>(), - stateStore); + stateStore, + Collections.emptyMap()); // Reset custom window size to default metricsCollector.setTestMetricWindowSize(null); diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/DelayedScaleDownEndToEndTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/DelayedScaleDownEndToEndTest.java index 420d9df3a..9eca91295 100644 --- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/DelayedScaleDownEndToEndTest.java +++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/DelayedScaleDownEndToEndTest.java @@ -122,7 +122,8 @@ public class DelayedScaleDownEndToEndTest { new ScalingExecutor<>(eventCollector, stateStore), eventCollector, scalingRealizer, - stateStore); + stateStore, + Collections.emptyMap()); // initially the last evaluated metrics are empty assertThat(autoscaler.lastEvaluatedMetrics.get(context.getJobKey())).isNull(); diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java index 334a27a48..1597ec5cc 100644 --- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java +++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java @@ -24,7 +24,9 @@ import org.apache.flink.autoscaler.event.TestingEventCollector; import org.apache.flink.autoscaler.exceptions.NotReadyException; import org.apache.flink.autoscaler.metrics.AutoscalerFlinkMetrics; import org.apache.flink.autoscaler.metrics.CollectedMetrics; +import org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator; import org.apache.flink.autoscaler.metrics.ScalingMetric; +import org.apache.flink.autoscaler.metrics.TestCustomEvaluator; import org.apache.flink.autoscaler.metrics.TestMetrics; import org.apache.flink.autoscaler.realizer.ScalingRealizer; import org.apache.flink.autoscaler.realizer.TestingScalingRealizer; @@ -33,6 +35,8 @@ import org.apache.flink.autoscaler.topology.JobTopology; import org.apache.flink.autoscaler.topology.VertexInfo; import org.apache.flink.autoscaler.tuning.ConfigChanges; import org.apache.flink.client.program.rest.RestClusterClient; +import org.apache.flink.configuration.ConfigOptions; +import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.PipelineOptions; import org.apache.flink.configuration.TaskManagerOptions; import org.apache.flink.metrics.Gauge; @@ -56,6 +60,7 @@ import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.Set; import java.util.TreeMap; import static java.util.Map.entry; @@ -66,6 +71,9 @@ import static org.apache.flink.autoscaler.config.AutoScalerOptions.VERTEX_SCALIN import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; /** Tests for JobAutoScalerImpl. */ @@ -114,7 +122,8 @@ public class JobAutoScalerImplTest { scalingExecutor, eventCollector, scalingRealizer, - stateStore); + stateStore, + Collections.emptyMap()); var now = Instant.now(); autoscaler.setClock(Clock.fixed(now, ZoneId.systemDefault())); @@ -153,7 +162,13 @@ public class JobAutoScalerImplTest { void testErrorReporting() throws Exception { var autoscaler = new JobAutoScalerImpl<>( - null, null, null, eventCollector, scalingRealizer, stateStore); + null, + null, + null, + eventCollector, + scalingRealizer, + stateStore, + Collections.emptyMap()); autoscaler.scale(context); Assertions.assertEquals( @@ -188,7 +203,8 @@ public class JobAutoScalerImplTest { null, eventCollector, scalingRealizer, - stateStore); + stateStore, + Collections.emptyMap()); // Should not produce an error autoscaler.scale(context); @@ -227,7 +243,8 @@ public class JobAutoScalerImplTest { null, eventCollector, realizeParallelismOverridesWithExceptionsScalingRealizer, - stateStore); + stateStore, + Collections.emptyMap()); // Should produce an error autoscaler.scale(context); @@ -244,7 +261,8 @@ public class JobAutoScalerImplTest { null, eventCollector, scalingRealizer, - stateStore); + stateStore, + Collections.emptyMap()); // Initially we should return empty overrides, do not crate any state assertThat(autoscaler.getParallelismOverrides(context)).isEmpty(); @@ -318,7 +336,13 @@ public class JobAutoScalerImplTest { var overrides = new HashMap<String, String>(); var autoscaler = new JobAutoScalerImpl<>( - null, null, null, eventCollector, scalingRealizer, stateStore) { + null, + null, + null, + eventCollector, + scalingRealizer, + stateStore, + Collections.emptyMap()) { public Map<String, String> getParallelismOverrides( JobAutoScalerContext<JobID> ctx) { return new HashMap<>(overrides); @@ -417,7 +441,13 @@ public class JobAutoScalerImplTest { context.getConfiguration().set(AutoScalerOptions.MEMORY_TUNING_ENABLED, true); var autoscaler = new JobAutoScalerImpl<>( - null, null, null, eventCollector, scalingRealizer, stateStore); + null, + null, + null, + eventCollector, + scalingRealizer, + stateStore, + Collections.emptyMap()); // Initially we should return empty overrides, do not crate any state assertThat(stateStore.getConfigChanges(context).getOverrides()).isEmpty(); @@ -465,7 +495,13 @@ public class JobAutoScalerImplTest { var autoscaler = new JobAutoScalerImpl<>( - null, null, null, eventCollector, scalingRealizer, stateStore); + null, + null, + null, + eventCollector, + scalingRealizer, + stateStore, + Collections.emptyMap()); autoscaler.scale(context); assertTrue(stateStore.getScalingHistory(context).isEmpty()); @@ -490,4 +526,109 @@ public class JobAutoScalerImplTest { private TestingScalingRealizer.Event<JobID, JobAutoScalerContext<JobID>> getEvent() { return scalingRealizer.events.poll(); } + + @Test + void testGetCustomEvaluatorIfRequired() { + FlinkAutoscalerEvaluator testCustomEvaluator = new TestCustomEvaluator(); + testCustomEvaluator.configure(new Configuration()); + var testCustomEvaluators = Map.of(testCustomEvaluator.getName(), testCustomEvaluator); + + var autoscalerWithCustomEvaluator = + new JobAutoScalerImpl<>( + null, + null, + null, + eventCollector, + scalingRealizer, + stateStore, + testCustomEvaluators); + + String testCustomEvaluatorName = "test-custom-evaluator"; + + var defaultConf = context.getConfiguration(); + + // Case 1: Custom evaluator configured. + defaultConf.set(AutoScalerOptions.CUSTOM_EVALUATOR_NAME, testCustomEvaluatorName); + + var customEvaluatorWithConfig = + autoscalerWithCustomEvaluator.getCustomEvaluatorIfRequired( + context.getConfiguration()); + assertNotNull(customEvaluatorWithConfig); + assertInstanceOf(FlinkAutoscalerEvaluator.class, customEvaluatorWithConfig.f0); + var customEvaluatorConfig = customEvaluatorWithConfig.f1; + assertNotNull(customEvaluatorConfig); + assertEquals(0, customEvaluatorConfig.keySet().size()); + + Set<String> expectedKeys = Set.of(); + assertEquals(expectedKeys, customEvaluatorConfig.keySet()); + + // Case 2: Custom evaluator configured with additional configs for the plugin. + defaultConf.set( + ConfigOptions.key( + AutoScalerOptions.AUTOSCALER_CONF_PREFIX + + AutoScalerOptions.CUSTOM_EVALUATOR_CONF_PREFIX + + testCustomEvaluatorName + + ".k1") + .stringType() + .noDefaultValue(), + "v1"); + + defaultConf.set( + ConfigOptions.key( + AutoScalerOptions.AUTOSCALER_CONF_PREFIX + + AutoScalerOptions.CUSTOM_EVALUATOR_CONF_PREFIX + + testCustomEvaluatorName + + ".k2") + .stringType() + .noDefaultValue(), + "v2"); + + var customEvaluatorWithConfigContainingAdditionalKeys = + autoscalerWithCustomEvaluator.getCustomEvaluatorIfRequired( + context.getConfiguration()); + assertNotNull(customEvaluatorWithConfigContainingAdditionalKeys); + assertInstanceOf( + FlinkAutoscalerEvaluator.class, + customEvaluatorWithConfigContainingAdditionalKeys.f0); + var customEvaluatorConfigContainingAdditionalKeys = + customEvaluatorWithConfigContainingAdditionalKeys.f1; + assertNotNull(customEvaluatorConfigContainingAdditionalKeys); + assertEquals(2, customEvaluatorConfigContainingAdditionalKeys.keySet().size()); + + expectedKeys = Set.of("k1", "k2"); + assertEquals(expectedKeys, customEvaluatorConfigContainingAdditionalKeys.keySet()); + + // Case 3: Custom evaluator configured but with a custom evaluator name that is not + // available. + defaultConf.set(AutoScalerOptions.CUSTOM_EVALUATOR_NAME, "test-custom-evaluator-no-match"); + + var customEvaluatorWithConfigNoMatch = + autoscalerWithCustomEvaluator.getCustomEvaluatorIfRequired( + context.getConfiguration()); + assertNull(customEvaluatorWithConfigNoMatch); + + // Case 4: Custom evaluator not configured. + defaultConf.removeConfig(AutoScalerOptions.CUSTOM_EVALUATOR_NAME); + + var customEvaluatorNotConfigured = + autoscalerWithCustomEvaluator.getCustomEvaluatorIfRequired( + context.getConfiguration()); + assertNull(customEvaluatorNotConfigured); + + // Case 5: No custom evaluators available. + var autoscalerWithoutCustomEvaluator = + new JobAutoScalerImpl<>( + null, + null, + null, + eventCollector, + scalingRealizer, + stateStore, + Collections.emptyMap()); + + var customEvaluatorConfigNoCustomEvaluators = + autoscalerWithoutCustomEvaluator.getCustomEvaluatorIfRequired( + context.getConfiguration()); + assertNull(customEvaluatorConfigNoCustomEvaluators); + } } diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MetricsCollectionAndEvaluationTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MetricsCollectionAndEvaluationTest.java index 5bca303f9..a046d3be3 100644 --- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MetricsCollectionAndEvaluationTest.java +++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MetricsCollectionAndEvaluationTest.java @@ -181,7 +181,7 @@ public class MetricsCollectionAndEvaluationTest { assertEquals(3, collectedMetrics.getMetricHistory().size()); assertTrue(collectedMetrics.isFullyCollected()); - var evaluation = evaluator.evaluate(conf, collectedMetrics, restartTime); + var evaluation = evaluator.evaluate(conf, collectedMetrics, restartTime, null); scalingExecutor.scaleResource( context, evaluation, @@ -401,7 +401,7 @@ public class MetricsCollectionAndEvaluationTest { var collectedMetrics = metricsCollector.updateMetrics(context, stateStore); var evaluation = - evaluator.evaluate(context.getConfiguration(), collectedMetrics, restartTime); + evaluator.evaluate(context.getConfiguration(), collectedMetrics, restartTime, null); assertEquals( 500., evaluation @@ -661,7 +661,7 @@ public class MetricsCollectionAndEvaluationTest { var collectedMetrics = collectMetrics(); var evaluation = - evaluator.evaluate(context.getConfiguration(), collectedMetrics, restartTime); + evaluator.evaluate(context.getConfiguration(), collectedMetrics, restartTime, null); assertEquals( 0, evaluation @@ -712,7 +712,8 @@ public class MetricsCollectionAndEvaluationTest { .getMetricHistory() .put(Instant.ofEpochSecond(1234), new CollectedMetrics(newMetrics, Map.of())); - evaluation = evaluator.evaluate(context.getConfiguration(), collectedMetrics, restartTime); + evaluation = + evaluator.evaluate(context.getConfiguration(), collectedMetrics, restartTime, null); assertEquals( 3., evaluation diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/RecommendedParallelismTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/RecommendedParallelismTest.java index f34aa1fc0..2902d25fa 100644 --- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/RecommendedParallelismTest.java +++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/RecommendedParallelismTest.java @@ -37,6 +37,7 @@ import java.time.Clock; import java.time.Duration; import java.time.Instant; import java.time.ZoneId; +import java.util.Collections; import java.util.Map; import static org.apache.flink.autoscaler.TestingAutoscalerUtils.createDefaultJobAutoScalerContext; @@ -98,7 +99,8 @@ public class RecommendedParallelismTest { new ScalingExecutor<>(eventCollector, stateStore), eventCollector, new TestingScalingRealizer<>(), - stateStore); + stateStore, + Collections.emptyMap()); // Reset custom window size to default metricsCollector.setTestMetricWindowSize(null); diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java index 6b530ddf8..b17295e1d 100644 --- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java +++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java @@ -22,11 +22,14 @@ import org.apache.flink.autoscaler.config.AutoScalerOptions; import org.apache.flink.autoscaler.metrics.CollectedMetricHistory; import org.apache.flink.autoscaler.metrics.CollectedMetrics; import org.apache.flink.autoscaler.metrics.EvaluatedScalingMetric; +import org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator; import org.apache.flink.autoscaler.metrics.MetricAggregator; import org.apache.flink.autoscaler.metrics.ScalingMetric; +import org.apache.flink.autoscaler.metrics.TestCustomEvaluator; import org.apache.flink.autoscaler.topology.JobTopology; import org.apache.flink.autoscaler.topology.VertexInfo; import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.UnmodifiableConfiguration; import org.apache.flink.runtime.jobgraph.JobVertexID; import org.junit.jupiter.api.Test; @@ -112,7 +115,8 @@ public class ScalingMetricEvaluatorTest { .evaluate( conf, new CollectedMetricHistory(topology, metricHistory, Instant.now()), - Duration.ZERO) + Duration.ZERO, + null) .getVertexMetrics(); assertEquals( @@ -146,7 +150,8 @@ public class ScalingMetricEvaluatorTest { .evaluate( conf, new CollectedMetricHistory(topology, metricHistory, Instant.now()), - Duration.ZERO) + Duration.ZERO, + null) .getVertexMetrics(); assertEquals( EvaluatedScalingMetric.avg(150), @@ -169,7 +174,8 @@ public class ScalingMetricEvaluatorTest { .evaluate( conf, new CollectedMetricHistory(topology, metricHistory, Instant.now()), - Duration.ZERO) + Duration.ZERO, + null) .getVertexMetrics(); assertEquals( EvaluatedScalingMetric.avg(150), @@ -191,7 +197,8 @@ public class ScalingMetricEvaluatorTest { .evaluate( conf, new CollectedMetricHistory(topology, metricHistory, Instant.now()), - Duration.ZERO) + Duration.ZERO, + null) .getVertexMetrics(); assertEquals( EvaluatedScalingMetric.avg(150), @@ -250,7 +257,8 @@ public class ScalingMetricEvaluatorTest { .evaluate( conf, new CollectedMetricHistory(topology, metricHistory, Instant.now()), - Duration.ZERO) + Duration.ZERO, + null) .getVertexMetrics(); assertEquals( EvaluatedScalingMetric.avg(100), @@ -260,6 +268,95 @@ public class ScalingMetricEvaluatorTest { evaluatedMetrics.get(sink).get(ScalingMetric.TARGET_DATA_RATE)); } + @Test + public void testEvaluateWithCustomEvaluator() { + var source = new JobVertexID(); + var sink = new JobVertexID(); + + var topology = + new JobTopology( + new VertexInfo(source, Collections.emptyMap(), 1, 1, null), + new VertexInfo(sink, Map.of(source, REBALANCE), 1, 1, null)); + + var metricHistory = new TreeMap<Instant, CollectedMetrics>(); + + metricHistory.put( + Instant.ofEpochMilli(1000), + new CollectedMetrics( + Map.of( + source, + Map.of( + ScalingMetric.LAG, + 0., + ScalingMetric.NUM_RECORDS_IN, + 0., + ScalingMetric.NUM_RECORDS_OUT, + 0., + ScalingMetric.LOAD, + .1), + sink, + Map.of(ScalingMetric.NUM_RECORDS_IN, 0., ScalingMetric.LOAD, .1)), + Map.of())); + + metricHistory.put( + Instant.ofEpochMilli(2000), + new CollectedMetrics( + Map.of( + source, + Map.of( + ScalingMetric.LAG, + 0., + ScalingMetric.NUM_RECORDS_IN, + 100., + ScalingMetric.NUM_RECORDS_OUT, + 200., + ScalingMetric.LOAD, + .4), + sink, + Map.of(ScalingMetric.NUM_RECORDS_IN, 200., ScalingMetric.LOAD, .2)), + Map.of())); + + var conf = new Configuration(); + + conf.set(CATCH_UP_DURATION, Duration.ofSeconds(2)); + FlinkAutoscalerEvaluator customEvaluator = new TestCustomEvaluator(); + var customEvaluatorWithConfig = new Tuple2<>(customEvaluator, new Configuration()); + + var evaluatedMetrics = + evaluator + .evaluate( + conf, + new CollectedMetricHistory(topology, metricHistory, Instant.now()), + Duration.ZERO, + customEvaluatorWithConfig) + .getVertexMetrics(); + + assertEquals( + EvaluatedScalingMetric.avg(.25), + evaluatedMetrics.get(source).get(ScalingMetric.LOAD)); + + assertEquals( + EvaluatedScalingMetric.avg(.15), + evaluatedMetrics.get(sink).get(ScalingMetric.LOAD)); + + assertEquals( + EvaluatedScalingMetric.avg(100000.0), + evaluatedMetrics.get(source).get(ScalingMetric.TARGET_DATA_RATE)); + assertEquals( + EvaluatedScalingMetric.of(0.0), + evaluatedMetrics.get(source).get(ScalingMetric.CATCH_UP_DATA_RATE)); + assertEquals( + EvaluatedScalingMetric.avg(200000.0), + evaluatedMetrics.get(sink).get(ScalingMetric.TARGET_DATA_RATE)); + assertEquals( + EvaluatedScalingMetric.of(0.0), + evaluatedMetrics.get(sink).get(ScalingMetric.CATCH_UP_DATA_RATE)); + assertEquals( + EvaluatedScalingMetric.of(0.0), + evaluatedMetrics.get(source).get(ScalingMetric.LAG)); + assertFalse(evaluatedMetrics.get(sink).containsKey(ScalingMetric.LAG)); + } + @Test public void testUtilizationBoundaryComputation() { @@ -884,4 +981,155 @@ public class ScalingMetricEvaluatorTest { map.get(ScalingMetric.SCALE_UP_RATE_THRESHOLD).getCurrent(), map.get(ScalingMetric.SCALE_DOWN_RATE_THRESHOLD).getCurrent()); } + + @Test + public void testRunCustomEvaluator() { + var source = new JobVertexID(); + var sink = new JobVertexID(); + + var topology = + new JobTopology( + new VertexInfo(source, Collections.emptyMap(), 1, 1, null), + new VertexInfo(sink, Map.of(source, REBALANCE), 1, 1, null)); + + var metricHistory = new TreeMap<Instant, CollectedMetrics>(); + + metricHistory.put( + Instant.ofEpochMilli(1000), + new CollectedMetrics( + Map.of( + source, + Map.of( + ScalingMetric.LAG, + 0., + ScalingMetric.NUM_RECORDS_IN, + 0., + ScalingMetric.NUM_RECORDS_OUT, + 0., + ScalingMetric.LOAD, + .1), + sink, + Map.of(ScalingMetric.NUM_RECORDS_IN, 0., ScalingMetric.LOAD, .1)), + Map.of())); + + metricHistory.put( + Instant.ofEpochMilli(2000), + new CollectedMetrics( + Map.of( + source, + Map.of( + ScalingMetric.LAG, + 0., + ScalingMetric.NUM_RECORDS_IN, + 100., + ScalingMetric.NUM_RECORDS_OUT, + 200., + ScalingMetric.LOAD, + .4), + sink, + Map.of(ScalingMetric.NUM_RECORDS_IN, 200., ScalingMetric.LOAD, .2)), + Map.of())); + + var conf = new Configuration(); + FlinkAutoscalerEvaluator customEvaluator = new TestCustomEvaluator(); + var evaluatedMetrics = new HashMap<ScalingMetric, EvaluatedScalingMetric>(); + + var testCustomEvaluationSession = + Tuple2.of( + customEvaluator, + new FlinkAutoscalerEvaluator.Context( + new UnmodifiableConfiguration(conf), + Collections.unmodifiableSortedMap(metricHistory), + Collections.unmodifiableMap(new HashMap<>()), + topology, + false, + Duration.ZERO, + new Configuration())); + + var testCustomEvaluatorResult = + ScalingMetricEvaluator.runCustomEvaluator( + source, evaluatedMetrics, testCustomEvaluationSession); + + assertFalse(testCustomEvaluatorResult.isEmpty()); + + assertTrue(testCustomEvaluatorResult.containsKey(ScalingMetric.TARGET_DATA_RATE)); + + assertEquals( + EvaluatedScalingMetric.avg(100000.0), + testCustomEvaluatorResult.get(ScalingMetric.TARGET_DATA_RATE)); + } + + @Test + public void testMergeEvaluatedMetricsMaps() { + Map<ScalingMetric, EvaluatedScalingMetric> actual = new HashMap<>(); + actual.put(ScalingMetric.TARGET_DATA_RATE, EvaluatedScalingMetric.avg(50000.0)); + actual.put(ScalingMetric.LOAD, EvaluatedScalingMetric.avg(0.5)); + + // Case 1: Merge with null (should not modify actual) + ScalingMetricEvaluator.mergeEvaluatedMetricsMaps(actual, null); + assertEquals(2, actual.size()); + assertEquals( + EvaluatedScalingMetric.avg(50000.0), actual.get(ScalingMetric.TARGET_DATA_RATE)); + assertEquals(EvaluatedScalingMetric.avg(0.5), actual.get(ScalingMetric.LOAD)); + + // Case 2: Merge with an empty map (should not modify actual) + ScalingMetricEvaluator.mergeEvaluatedMetricsMaps(actual, Collections.emptyMap()); + assertEquals(2, actual.size()); + assertEquals( + EvaluatedScalingMetric.avg(50000.0), actual.get(ScalingMetric.TARGET_DATA_RATE)); + assertEquals(EvaluatedScalingMetric.avg(0.5), actual.get(ScalingMetric.LOAD)); + + // Case 3: Merge with an incoming map + Map<ScalingMetric, EvaluatedScalingMetric> incoming = new HashMap<>(); + incoming.put(ScalingMetric.TARGET_DATA_RATE, EvaluatedScalingMetric.avg(100000.0)); + incoming.put(ScalingMetric.LAG, new EvaluatedScalingMetric(10.0, 10.0)); + + ScalingMetricEvaluator.mergeEvaluatedMetricsMaps(actual, incoming); + assertEquals(3, actual.size()); + + assertTrue(actual.containsKey(ScalingMetric.LAG)); + assertEquals(new EvaluatedScalingMetric(10.0, 10.0), actual.get(ScalingMetric.LAG)); + + assertTrue(actual.containsKey(ScalingMetric.TARGET_DATA_RATE)); + assertEquals( + EvaluatedScalingMetric.avg(100000.0), actual.get(ScalingMetric.TARGET_DATA_RATE)); + + assertTrue(actual.containsKey(ScalingMetric.LOAD)); + assertEquals(EvaluatedScalingMetric.avg(0.5), actual.get(ScalingMetric.LOAD)); + } + + @Test + public void testMergeEvaluatedScalingMetric() { + // Case 1 + EvaluatedScalingMetric actual = new EvaluatedScalingMetric(50.0, 100.0); + EvaluatedScalingMetric incoming = new EvaluatedScalingMetric(60.0, 120.0); + + EvaluatedScalingMetric result = + ScalingMetricEvaluator.mergeEvaluatedScalingMetric(actual, incoming); + + assertEquals(60.0, result.getCurrent()); + assertEquals(120.0, result.getAverage()); + + // Case 2 + EvaluatedScalingMetric incomingWithNaN = new EvaluatedScalingMetric(Double.NaN, Double.NaN); + result = ScalingMetricEvaluator.mergeEvaluatedScalingMetric(actual, incomingWithNaN); + + assertEquals(50.0, result.getCurrent()); + assertEquals(100.0, result.getAverage()); + + // Case 3 + EvaluatedScalingMetric incomingWithPartialNaN = + new EvaluatedScalingMetric(Double.NaN, 130.0); + result = ScalingMetricEvaluator.mergeEvaluatedScalingMetric(actual, incomingWithPartialNaN); + + assertEquals(50.0, result.getCurrent()); + assertEquals(130.0, result.getAverage()); + + // Case 4 + EvaluatedScalingMetric actualWithNaN = new EvaluatedScalingMetric(Double.NaN, Double.NaN); + result = ScalingMetricEvaluator.mergeEvaluatedScalingMetric(actualWithNaN, incoming); + + assertEquals(60.0, result.getCurrent()); + assertEquals(120.0, result.getAverage()); + } } diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/metrics/TestCustomEvaluator.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/metrics/TestCustomEvaluator.java new file mode 100644 index 000000000..b1426f6bc --- /dev/null +++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/metrics/TestCustomEvaluator.java @@ -0,0 +1,51 @@ +/* + * 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.flink.autoscaler.metrics; + +import org.apache.flink.runtime.jobgraph.JobVertexID; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +/** + * A test implementation of the {@link FlinkAutoscalerEvaluator} interface that provides + * custom scaling metric evaluations for job vertices in a Flink job. + */ +public class TestCustomEvaluator implements FlinkAutoscalerEvaluator { + @Override + public String getName() { + return "test-custom-evaluator"; + } + + @Override + public Map<ScalingMetric, EvaluatedScalingMetric> evaluateVertexMetrics( + JobVertexID vertex, + Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics, + Context evaluationContext) { + if (evaluationContext.getTopology().isSource(vertex)) { + var customEvaluatedMetrics = new HashMap<ScalingMetric, EvaluatedScalingMetric>(); + customEvaluatedMetrics.put( + ScalingMetric.TARGET_DATA_RATE, EvaluatedScalingMetric.avg(100000.0)); + + return customEvaluatedMetrics; + } + return Collections.emptyMap(); + } +} diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java index d3ad2d14b..1edd695f7 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java @@ -187,7 +187,9 @@ public class FlinkOperator { var statusRecorder = StatusRecorder.create(client, metricManager, listeners); var clusterResourceManager = ClusterResourceManager.of(configManager.getDefaultConfig(), client); - var autoscaler = AutoscalerFactory.create(client, eventRecorder, clusterResourceManager); + var autoscaler = + AutoscalerFactory.create( + client, eventRecorder, clusterResourceManager, configManager); var reconcilerFactory = new ReconcilerFactory(eventRecorder, statusRecorder, autoscaler); var observerFactory = new FlinkDeploymentObserverFactory(eventRecorder); var canaryResourceManager = new CanaryResourceManager<FlinkDeployment>(configManager); @@ -211,7 +213,7 @@ public class FlinkOperator { var metricManager = MetricManager.createFlinkSessionJobMetricManager(baseConfig, metricGroup); var statusRecorder = StatusRecorder.create(client, metricManager, listeners); - var autoscaler = AutoscalerFactory.create(client, eventRecorder, null); + var autoscaler = AutoscalerFactory.create(client, eventRecorder, null, configManager); var reconciler = new SessionJobReconciler(eventRecorder, statusRecorder, autoscaler); var observer = new FlinkSessionJobObserver(eventRecorder); var canaryResourceManager = new CanaryResourceManager<FlinkSessionJob>(configManager); diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerFactory.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerFactory.java index 7bbb5492e..d97d33938 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerFactory.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerFactory.java @@ -24,6 +24,7 @@ import org.apache.flink.autoscaler.ScalingExecutor; import org.apache.flink.autoscaler.ScalingMetricEvaluator; import org.apache.flink.kubernetes.operator.autoscaler.state.ConfigMapStore; import org.apache.flink.kubernetes.operator.autoscaler.state.KubernetesAutoScalerStateStore; +import org.apache.flink.kubernetes.operator.config.FlinkConfigManager; import org.apache.flink.kubernetes.operator.resources.ClusterResourceManager; import org.apache.flink.kubernetes.operator.utils.EventRecorder; @@ -36,10 +37,12 @@ public class AutoscalerFactory { public static JobAutoScaler<ResourceID, KubernetesJobAutoScalerContext> create( KubernetesClient client, EventRecorder eventRecorder, - ClusterResourceManager clusterResourceManager) { + ClusterResourceManager clusterResourceManager, + FlinkConfigManager configManager) { var stateStore = new KubernetesAutoScalerStateStore(new ConfigMapStore(client)); var eventHandler = new KubernetesAutoScalerEventHandler(eventRecorder); + var customEvaluators = AutoscalerUtils.discoverCustomEvaluators(configManager); return new JobAutoScalerImpl<>( new RestApiMetricsCollector<>(), @@ -47,6 +50,7 @@ public class AutoscalerFactory { new ScalingExecutor<>(eventHandler, stateStore, clusterResourceManager), eventHandler, new KubernetesScalingRealizer(), - stateStore); + stateStore, + customEvaluators); } } diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerUtils.java new file mode 100644 index 000000000..3ea5f077a --- /dev/null +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerUtils.java @@ -0,0 +1,75 @@ +/* + * 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.flink.kubernetes.operator.autoscaler; + +import org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator; +import org.apache.flink.configuration.ConfigConstants; +import org.apache.flink.core.plugin.PluginUtils; +import org.apache.flink.kubernetes.operator.config.FlinkConfigManager; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.HashMap; +import java.util.Map; + +/** Autoscaler related utility methods for Operator. */ +public class AutoscalerUtils { + private static final Logger LOG = LoggerFactory.getLogger(AutoscalerUtils.class); + + /** + * discovers custom evaluator's for autoscaler. + * + * @param configManager Flink Config manager + * @return A map of discovered custom evaluators, where the key is the evaluator name provided + * by {@link FlinkAutoscalerEvaluator#getName()}) and the value is the corresponding + * instance. + */ + public static Map<String, FlinkAutoscalerEvaluator> discoverCustomEvaluators( + FlinkConfigManager configManager) { + var conf = configManager.getDefaultConfig(); + Map<String, FlinkAutoscalerEvaluator> customEvaluators = new HashMap<>(); + PluginUtils.createPluginManagerFromRootFolder(conf) + .load(FlinkAutoscalerEvaluator.class) + .forEachRemaining( + customEvaluator -> { + String customEvaluatorName = customEvaluator.getName(); + LOG.info( + "Discovered custom evaluator for autoscaler from plugin directory[{}]: {}.", + System.getenv() + .getOrDefault( + ConfigConstants.ENV_FLINK_PLUGINS_DIR, + ConfigConstants.DEFAULT_FLINK_PLUGINS_DIRS), + customEvaluatorName); + customEvaluator.configure(conf); + if (customEvaluators.containsKey(customEvaluatorName)) { + LOG.warn( + "Duplicate custom evaluator name [{}] detected. Overwriting existing [{}] with [{}].", + customEvaluatorName, + customEvaluators + .get(customEvaluatorName) + .getClass() + .getName(), + customEvaluator.getClass().getName()); + } + customEvaluators.put(customEvaluatorName, customEvaluator); + }); + return customEvaluators; + } +} diff --git a/flink-kubernetes-operator/src/test/assembly/test-plugins-assembly.xml b/flink-kubernetes-operator/src/test/assembly/test-plugins-assembly.xml index 3cb8b5d17..67526784a 100644 --- a/flink-kubernetes-operator/src/test/assembly/test-plugins-assembly.xml +++ b/flink-kubernetes-operator/src/test/assembly/test-plugins-assembly.xml @@ -32,6 +32,7 @@ under the License. <include>org/apache/flink/kubernetes/operator/listener/TestingListener.class</include> <include>org/apache/flink/kubernetes/operator/autoscaler/TestingAutoscaler.class</include> <include>org/apache/flink/kubernetes/operator/mutator/TestMutator.java</include> + <include>org/apache/flink/autoscaler/metrics/TestCustomEvaluator.java</include> </includes> </fileSet> <fileSet> @@ -40,4 +41,4 @@ under the License. <outputDirectory>/META-INF/services</outputDirectory> </fileSet> </fileSets> -</assembly> +</assembly> \ No newline at end of file diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerFactoryTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerFactoryTest.java index f848d491a..6b884044b 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerFactoryTest.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerFactoryTest.java @@ -19,6 +19,8 @@ package org.apache.flink.kubernetes.operator.autoscaler; import org.apache.flink.autoscaler.JobAutoScaler; import org.apache.flink.autoscaler.JobAutoScalerImpl; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.kubernetes.operator.config.FlinkConfigManager; import org.apache.flink.kubernetes.operator.resources.ClusterResourceManager; import org.apache.flink.kubernetes.operator.utils.EventRecorder; import org.apache.flink.kubernetes.operator.utils.FlinkResourceEventCollector; @@ -47,7 +49,8 @@ public class AutoscalerFactoryTest { new EventRecorder( new FlinkResourceEventCollector(), new FlinkStateSnapshotEventCollector()), - new ClusterResourceManager(Duration.ZERO, kubernetesClient)); + new ClusterResourceManager(Duration.ZERO, kubernetesClient), + new FlinkConfigManager(new Configuration())); Assertions.assertTrue(autoScaler instanceof JobAutoScalerImpl); } } diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerUtilsTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerUtilsTest.java new file mode 100644 index 000000000..2d1f365da --- /dev/null +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerUtilsTest.java @@ -0,0 +1,65 @@ +/* + * 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.flink.kubernetes.operator.autoscaler; + +import org.apache.flink.configuration.ConfigConstants; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.kubernetes.operator.TestUtils; +import org.apache.flink.kubernetes.operator.config.FlinkConfigManager; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** Unit tests for {@link AutoscalerUtils}. */ +public class AutoscalerUtilsTest { + @TempDir public Path temporaryFolder; + + @Test + public void testDiscoverCustomEvaluators() throws IOException { + Map<String, String> originalEnv = System.getenv(); + try { + Map<String, String> systemEnv = new HashMap<>(originalEnv); + systemEnv.put( + ConfigConstants.ENV_FLINK_PLUGINS_DIR, + TestUtils.getTestPluginsRootDir(temporaryFolder)); + TestUtils.setEnv(systemEnv); + + // Discover evaluators + var discoveredEvaluators = + AutoscalerUtils.discoverCustomEvaluators( + new FlinkConfigManager(new Configuration())) + .keySet(); + // Expected evaluators + var expectedEvaluators = new HashSet<>(List.of("test-custom-evaluator")); + + assertEquals(expectedEvaluators, discoveredEvaluators); + } finally { + TestUtils.setEnv(originalEnv); + } + } +} diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java index 7c380727c..165fc697d 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java @@ -101,7 +101,8 @@ public class TestingFlinkDeploymentController flinkService.getKubernetesClient(), eventRecorder, new ClusterResourceManager( - Duration.ZERO, flinkService.getKubernetesClient()))); + Duration.ZERO, flinkService.getKubernetesClient()), + configManager)); canaryResourceManager = new CanaryResourceManager<>(configManager); flinkDeploymentController = new FlinkDeploymentController( diff --git a/flink-kubernetes-operator/src/test/resources/META-INF/services/org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator b/flink-kubernetes-operator/src/test/resources/META-INF/services/org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator new file mode 100644 index 000000000..a150d9783 --- /dev/null +++ b/flink-kubernetes-operator/src/test/resources/META-INF/services/org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator @@ -0,0 +1,19 @@ +# +# 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. +# + +org.apache.flink.autoscaler.metrics.TestCustomEvaluator \ No newline at end of file
