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 30928cbc29348ed01d7461668a0deb48c12adc35 Author: Dennis-Mircea Ciupitu <[email protected]> AuthorDate: Wed Jun 17 15:27:02 2026 +0300 [FLINK-39511][autoscaler] FLIP-514: Custom Evaluator plugin for Flink Autoscaler --- docs/content.zh/docs/operations/plugins.md | 92 +++++++++++- docs/content/docs/operations/plugins.md | 92 ++++++++++++ .../generated/auto_scaler_configuration.html | 18 +++ .../standalone/StandaloneAutoscalerEntrypoint.java | 11 +- .../standalone/utils/AutoscalerUtils.java | 55 +++++++ .../standalone/utils/AutoscalerUtilsTest.java | 43 ++++++ ...utoscaler.metrics.ScalingMetricsEvaluatorPlugin | 3 +- .../apache/flink/autoscaler/JobAutoScalerImpl.java | 34 +---- .../flink/autoscaler/ScalingMetricEvaluator.java | 129 ++++++++++++----- .../flink/autoscaler/config/AutoScalerOptions.java | 120 +++++++++++++-- .../metrics/FlinkAutoscalerEvaluator.java | 118 --------------- .../metrics/ScalingMetricsEvaluatorPlugin.java | 114 +++++++++++++++ .../autoscaler/AutoScalerCustomEvaluatorTest.java | 118 +++++++++++++-- .../flink/autoscaler/BacklogBasedScalingTest.java | 7 +- .../autoscaler/DelayedScaleDownEndToEndTest.java | 5 +- .../flink/autoscaler/JobAutoScalerImplTest.java | 159 ++------------------ .../MetricsCollectionAndEvaluationTest.java | 11 +- .../autoscaler/RecommendedParallelismTest.java | 7 +- .../autoscaler/ScalingMetricEvaluatorTest.java | 161 ++++++++++++++++++--- .../autoscaler/metrics/TestCustomEvaluator.java | 10 +- .../flink/kubernetes/operator/FlinkOperator.java | 5 +- .../operator/autoscaler/AutoscalerFactory.java | 12 +- .../operator/autoscaler/AutoscalerUtils.java | 75 ---------- .../kubernetes/operator/utils/AutoscalerUtils.java | 61 ++++++++ .../operator/autoscaler/AutoscalerFactoryTest.java | 3 +- .../operator/autoscaler/AutoscalerUtilsTest.java | 22 ++- .../TestingFlinkDeploymentController.java | 20 +-- ...toscaler.metrics.ScalingMetricsEvaluatorPlugin} | 0 28 files changed, 993 insertions(+), 512 deletions(-) diff --git a/docs/content.zh/docs/operations/plugins.md b/docs/content.zh/docs/operations/plugins.md index f420358a8..b5ae5e1a2 100644 --- a/docs/content.zh/docs/operations/plugins.md +++ b/docs/content.zh/docs/operations/plugins.md @@ -186,4 +186,94 @@ The following steps demonstrate how to develop and use a custom mutator. 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 +## Custom Scaling Metric Evaluators + +`ScalingMetricsEvaluatorPlugin` is a pluggable component that allows users to provide custom scaling-metric evaluation logic on top of the metrics evaluated internally by the autoscaler. Custom metric evaluators are discovered through the [Plugins](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/filesystems/plugins) mechanism when running inside the Kubernetes operator, and through the standard Java `ServiceLoader` mechanism when running with `flink-autoscaler-standa [...] + +For each evaluation cycle, the autoscaler invokes the custom metric evaluator selected via the `job.autoscaler.metrics.custom-evaluators` configuration option once per job vertex. The metrics returned by the custom evaluator are merged on top of the internally evaluated metrics, allowing users to override or augment specific `ScalingMetric` values (e.g. `TARGET_DATA_RATE`, `TRUE_PROCESSING_RATE`, `CATCH_UP_DATA_RATE`). + +All `job.autoscaler.*` keys related to custom metric evaluators also support the legacy `kubernetes.operator.`-prefixed form as a fallback (for example, `kubernetes.operator.job.autoscaler.metrics.custom-evaluators`). The canonical key takes precedence on overlap. + +The following steps demonstrate how to develop and use a custom metric evaluator. + +1. Implement the `ScalingMetricsEvaluatorPlugin` interface: + ```java + package org.apache.flink.autoscaler.custom; + + import org.apache.flink.autoscaler.metrics.EvaluatedScalingMetric; + import org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin; + import org.apache.flink.autoscaler.metrics.ScalingMetric; + import org.apache.flink.runtime.jobgraph.JobVertexID; + + import java.util.HashMap; + import java.util.Map; + + /** Custom metric evaluator implementation of {@link ScalingMetricsEvaluatorPlugin}. */ + public class CustomEvaluator implements ScalingMetricsEvaluatorPlugin { + + @Override + public Map<ScalingMetric, EvaluatedScalingMetric> evaluateVertexMetrics( + JobVertexID vertex, + Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics, + Context context) { + Map<ScalingMetric, EvaluatedScalingMetric> overrides = new HashMap<>(); + // Example: override the target data rate for source vertices based + // on an evaluator-specific option read from the merged configuration. + double target = context.getConfig().getDouble("target-data-rate", 0.0); + if (target > 0 && context.getTopology().isSource(vertex)) { + overrides.put( + ScalingMetric.TARGET_DATA_RATE, + EvaluatedScalingMetric.avg(target)); + } + return overrides; + } + } + ``` + + The `Context` object exposes an un-modifiable view of the configuration, the metrics history, previously evaluated vertex metrics (evaluation happens topologically), the job topology, backlog status, and max restart time. The configuration is the job configuration with the evaluator-specific options (prefix stripped) merged on top, so an evaluator-specific key overrides the job-level value of the same key. + +2. Create the service definition file `org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin` in `META-INF/services` with the fully-qualified class name of your implementation: + ```text + org.apache.flink.autoscaler.custom.CustomEvaluator + ``` + +3. Use the Maven tool to package the project and generate the custom metric evaluator JAR. + +4. Select the custom metric evaluator via configuration. The evaluator whose implementation class FQN matches the configured `job.autoscaler.metrics.custom-evaluator.<name>.class` value will be invoked, and any other `job.autoscaler.metrics.custom-evaluator.<name>.*` entries are merged on top of the job configuration with the `job.autoscaler.metrics.custom-evaluator.<instance>.` prefix stripped, and exposed via `Context.getConfig()`, taking precedence over the job-level value of the same [...] + ```yaml + job.autoscaler.metrics.custom-evaluators: my-evaluator + job.autoscaler.metrics.custom-evaluator.my-evaluator.class: org.apache.flink.autoscaler.custom.CustomEvaluator + job.autoscaler.metrics.custom-evaluator.my-evaluator.target-data-rate: 100000.0 + ``` + {{< hint warning >}} + **Only one custom metric evaluator per pipeline is supported for now**. Currently, `job.autoscaler.metrics.custom-evaluators` is parsed as a list, but if more than one entry is configured the autoscaler logs a warning and falls back to the first entry, ignoring the rest. Registering multiple implementations via `META-INF/services` is fine as they form a registry that different jobs can select from by class FQN, but a single job cannot chain or compose more than one evaluator. Multi-in [...] + {{< /hint >}} + +5. Deploy the evaluator. + + - **Operator deployment** - create a Dockerfile to build a custom image from the `apache/flink-kubernetes-operator` official image and copy the generated JAR to a custom metric evaluator plugin directory under `/opt/flink/plugins` (the value of the `FLINK_PLUGINS_DIR` environment variable in the flink-kubernetes-operator helm chart). The structure of the custom evaluator directory under `/opt/flink/plugins` is as follows: + ```text + /opt/flink/plugins + ├── custom-evaluator + │ ├── custom-evaluator.jar + └── ... + ``` + + With the custom metric evaluator directory location, the Dockerfile is defined as follows: + ```shell script + FROM apache/flink-kubernetes-operator + ENV FLINK_PLUGINS_DIR=/opt/flink/plugins + ENV CUSTOM_EVALUATOR_DIR=custom-evaluator + RUN mkdir $FLINK_PLUGINS_DIR/$CUSTOM_EVALUATOR_DIR + COPY custom-evaluator.jar $FLINK_PLUGINS_DIR/$CUSTOM_EVALUATOR_DIR/ + ``` + + Install the flink-kubernetes-operator helm chart with the custom image and verify the `deploy/flink-kubernetes-operator` log has: + ```text + o.a.f.k.o.a.AutoscalerUtils [INFO ] Discovered custom metric evaluator from plugin directory[/opt/flink/plugins]: org.apache.flink.autoscaler.custom.CustomEvaluator. + ``` + + - **Standalone autoscaler** - simply place the custom metric evaluator JAR on the classpath of the `flink-autoscaler-standalone` process. It will be picked up automatically via Java's `ServiceLoader` and discovery will be logged: + ```text + o.a.f.a.s.AutoscalerUtils [INFO ] Discovered custom metric evaluator via ServiceLoader: org.apache.flink.autoscaler.custom.CustomEvaluator. + ``` diff --git a/docs/content/docs/operations/plugins.md b/docs/content/docs/operations/plugins.md index 4f387add6..21ef284d4 100644 --- a/docs/content/docs/operations/plugins.md +++ b/docs/content/docs/operations/plugins.md @@ -185,3 +185,95 @@ 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. ``` + +## Custom Scaling Metric Evaluators + +`ScalingMetricsEvaluatorPlugin` is a pluggable component that allows users to provide custom scaling-metric evaluation logic on top of the metrics evaluated internally by the autoscaler. Custom metric evaluators are discovered through the [Plugins](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/filesystems/plugins) mechanism when running inside the Kubernetes operator, and through the standard Java `ServiceLoader` mechanism when running with `flink-autoscaler-standa [...] + +For each evaluation cycle, the autoscaler invokes the custom metric evaluator selected via the `job.autoscaler.metrics.custom-evaluators` configuration option once per job vertex. The metrics returned by the custom evaluator are merged on top of the internally evaluated metrics, allowing users to override or augment specific `ScalingMetric` values (e.g. `TARGET_DATA_RATE`, `TRUE_PROCESSING_RATE`, `CATCH_UP_DATA_RATE`). + +All `job.autoscaler.*` keys related to custom metric evaluators also support the legacy `kubernetes.operator.`-prefixed form as a fallback (for example, `kubernetes.operator.job.autoscaler.metrics.custom-evaluators`). The canonical key takes precedence on overlap. + +The following steps demonstrate how to develop and use a custom metric evaluator. + +1. Implement the `ScalingMetricsEvaluatorPlugin` interface: + ```java + package org.apache.flink.autoscaler.custom; + + import org.apache.flink.autoscaler.metrics.EvaluatedScalingMetric; + import org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin; + import org.apache.flink.autoscaler.metrics.ScalingMetric; + import org.apache.flink.runtime.jobgraph.JobVertexID; + + import java.util.HashMap; + import java.util.Map; + + /** Custom metric evaluator implementation of {@link ScalingMetricsEvaluatorPlugin}. */ + public class CustomEvaluator implements ScalingMetricsEvaluatorPlugin { + + @Override + public Map<ScalingMetric, EvaluatedScalingMetric> evaluateVertexMetrics( + JobVertexID vertex, + Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics, + Context context) { + Map<ScalingMetric, EvaluatedScalingMetric> overrides = new HashMap<>(); + // Example: override the target data rate for source vertices based + // on an evaluator-specific option read from the merged configuration. + double target = context.getConfig().getDouble("target-data-rate", 0.0); + if (target > 0 && context.getTopology().isSource(vertex)) { + overrides.put( + ScalingMetric.TARGET_DATA_RATE, + EvaluatedScalingMetric.avg(target)); + } + return overrides; + } + } + ``` + + The `Context` object exposes an un-modifiable view of the configuration, the metrics history, previously evaluated vertex metrics (evaluation happens topologically), the job topology, backlog status, and max restart time. The configuration is the job configuration with the evaluator-specific options (prefix stripped) merged on top, so an evaluator-specific key overrides the job-level value of the same key. + +2. Create the service definition file `org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin` in `META-INF/services` with the fully-qualified class name of your implementation: + ```text + org.apache.flink.autoscaler.custom.CustomEvaluator + ``` + +3. Use the Maven tool to package the project and generate the custom metric evaluator JAR. + +4. Select the custom metric evaluator via configuration. The evaluator whose implementation class FQN matches the configured `job.autoscaler.metrics.custom-evaluator.<name>.class` value will be invoked, and any other `job.autoscaler.metrics.custom-evaluator.<name>.*` entries are merged on top of the job configuration with the `job.autoscaler.metrics.custom-evaluator.<instance>.` prefix stripped, and exposed via `Context.getConfig()`, taking precedence over the job-level value of the same [...] + ```yaml + job.autoscaler.metrics.custom-evaluators: my-evaluator + job.autoscaler.metrics.custom-evaluator.my-evaluator.class: org.apache.flink.autoscaler.custom.CustomEvaluator + job.autoscaler.metrics.custom-evaluator.my-evaluator.target-data-rate: 100000.0 + ``` + {{< hint warning >}} + **Only one custom metric evaluator per pipeline is supported for now**. Currently, `job.autoscaler.metrics.custom-evaluators` is parsed as a list, but if more than one entry is configured the autoscaler logs a warning and falls back to the first entry, ignoring the rest. Registering multiple implementations via `META-INF/services` is fine as they form a registry that different jobs can select from by class FQN, but a single job cannot chain or compose more than one evaluator. Multi-in [...] + {{< /hint >}} + +5. Deploy the evaluator. + + - **Operator deployment** - create a Dockerfile to build a custom image from the `apache/flink-kubernetes-operator` official image and copy the generated JAR to a custom metric evaluator plugin directory under `/opt/flink/plugins` (the value of the `FLINK_PLUGINS_DIR` environment variable in the flink-kubernetes-operator helm chart). The structure of the custom evaluator directory under `/opt/flink/plugins` is as follows: + ```text + /opt/flink/plugins + ├── custom-evaluator + │ ├── custom-evaluator.jar + └── ... + ``` + + With the custom metric evaluator directory location, the Dockerfile is defined as follows: + ```shell script + FROM apache/flink-kubernetes-operator + ENV FLINK_PLUGINS_DIR=/opt/flink/plugins + ENV CUSTOM_EVALUATOR_DIR=custom-evaluator + RUN mkdir $FLINK_PLUGINS_DIR/$CUSTOM_EVALUATOR_DIR + COPY custom-evaluator.jar $FLINK_PLUGINS_DIR/$CUSTOM_EVALUATOR_DIR/ + ``` + + Install the flink-kubernetes-operator helm chart with the custom image and verify the `deploy/flink-kubernetes-operator` log has: + ```text + o.a.f.k.o.a.AutoscalerUtils [INFO ] Discovered custom metric evaluator from plugin directory[/opt/flink/plugins]: org.apache.flink.autoscaler.custom.CustomEvaluator. + ``` + + - **Standalone autoscaler** - simply place the custom metric evaluator JAR on the classpath of the `flink-autoscaler-standalone` process. It will be picked up automatically via Java's `ServiceLoader` and discovery will be logged: + ```text + o.a.f.a.s.AutoscalerUtils [INFO ] Discovered custom metric evaluator via ServiceLoader: org.apache.flink.autoscaler.custom.CustomEvaluator. + ``` diff --git a/docs/layouts/shortcodes/generated/auto_scaler_configuration.html b/docs/layouts/shortcodes/generated/auto_scaler_configuration.html index 3c12ee108..2844f5a2f 100644 --- a/docs/layouts/shortcodes/generated/auto_scaler_configuration.html +++ b/docs/layouts/shortcodes/generated/auto_scaler_configuration.html @@ -92,6 +92,24 @@ <td><p>Enum</p></td> <td>Metric aggregator to use for busyTime metrics. This affects how true processing/output rate will be computed. Using max allows us to handle jobs with data skew more robustly, while avg may provide better stability when we know that the load distribution is even.<br /><br />Possible values:<ul><li>"AVG"</li><li>"MAX"</li><li>"MIN"</li></ul></td> </tr> + <tr> + <td><h5>job.autoscaler.metrics.custom-evaluator.<name>.<parameter></h5></td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>Free-form, evaluator-specific options for the custom metric evaluator instance <name>. All keys with this prefix are passed to the evaluator at runtime with the 'job.autoscaler.metrics.custom-evaluator.<name>.' prefix stripped.</td> + </tr> + <tr> + <td><h5>job.autoscaler.metrics.custom-evaluator.<name>.class</h5></td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>Fully-qualified class name of the implementation to instantiate for the custom metric evaluator instance <name> declared in 'job.autoscaler.metrics.custom-evaluators'. The class must be discovered via the Plugin / ServiceLoader mechanism and registered under META-INF/services/org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin.</td> + </tr> + <tr> + <td><h5>job.autoscaler.metrics.custom-evaluators</h5></td> + <td style="word-wrap: break-word;"></td> + <td>List<String></td> + <td>List of named custom metric evaluator instances to register with the autoscaler. For each <name> entry, the implementation class to instantiate must be set via 'job.autoscaler.metrics.custom-evaluator.<name>.class', and additional per-instance options live under 'job.autoscaler.metrics.custom-evaluator.<name>.<option>'. Only a single instance is honored for now: if more than one is configured, the autoscaler logs a warning and uses the first entry, [...] + </tr> <tr> <td><h5>job.autoscaler.metrics.window</h5></td> <td style="word-wrap: break-word;">15 min</td> 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 957af4c41..443c296e4 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 @@ -27,8 +27,10 @@ import org.apache.flink.autoscaler.RestApiMetricsCollector; import org.apache.flink.autoscaler.ScalingExecutor; import org.apache.flink.autoscaler.ScalingMetricEvaluator; import org.apache.flink.autoscaler.event.AutoScalerEventHandler; +import org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin; import org.apache.flink.autoscaler.standalone.flinkcluster.FlinkClusterJobListFetcher; import org.apache.flink.autoscaler.standalone.realizer.RescaleApiScalingRealizer; +import org.apache.flink.autoscaler.standalone.utils.AutoscalerUtils; import org.apache.flink.autoscaler.state.AutoScalerStateStore; import org.apache.flink.client.program.rest.RestClusterClient; import org.apache.flink.configuration.Configuration; @@ -38,6 +40,8 @@ import org.apache.flink.runtime.highavailability.nonha.standalone.StandaloneClie import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.Collection; + import static org.apache.flink.autoscaler.config.AutoScalerOptions.FLINK_CLIENT_TIMEOUT; import static org.apache.flink.autoscaler.standalone.config.AutoscalerStandaloneOptions.FETCHER_FLINK_CLUSTER_HOST; import static org.apache.flink.autoscaler.standalone.config.AutoscalerStandaloneOptions.FETCHER_FLINK_CLUSTER_PORT; @@ -88,14 +92,15 @@ public class StandaloneAutoscalerEntrypoint { JobAutoScaler<KEY, Context> createJobAutoscaler( AutoScalerEventHandler<KEY, Context> eventHandler, AutoScalerStateStore<KEY, Context> stateStore) { + Collection<ScalingMetricsEvaluatorPlugin> customEvaluators = + AutoscalerUtils.discoverCustomEvaluators(); return new JobAutoScalerImpl<>( new RestApiMetricsCollector<>(), - new ScalingMetricEvaluator(), + new ScalingMetricEvaluator(customEvaluators), new ScalingExecutor<>(eventHandler, stateStore), eventHandler, new RescaleApiScalingRealizer<>(eventHandler), - stateStore, - null); + stateStore); } @VisibleForTesting diff --git a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/utils/AutoscalerUtils.java b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/utils/AutoscalerUtils.java new file mode 100644 index 000000000..39f4f0539 --- /dev/null +++ b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/utils/AutoscalerUtils.java @@ -0,0 +1,55 @@ +/* + * 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.standalone.utils; + +import org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.ServiceLoader; + +/** Autoscaler related utility methods for the standalone autoscaler. */ +public class AutoscalerUtils { + + private static final Logger LOG = LoggerFactory.getLogger(AutoscalerUtils.class); + + /** + * Discovers custom metric evaluators for the standalone autoscaler via Java's {@link + * ServiceLoader} mechanism. Implementations must be registered under {@code + * META-INF/services/org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin} on the + * classpath. + * + * @return The list of discovered custom metric evaluators. + */ + public static Collection<ScalingMetricsEvaluatorPlugin> discoverCustomEvaluators() { + List<ScalingMetricsEvaluatorPlugin> customEvaluators = new ArrayList<>(); + ServiceLoader.load(ScalingMetricsEvaluatorPlugin.class) + .forEach( + customEvaluator -> { + LOG.info( + "Discovered custom metric evaluator via ServiceLoader: {}.", + customEvaluator.getClass().getName()); + customEvaluators.add(customEvaluator); + }); + return customEvaluators; + } +} diff --git a/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/utils/AutoscalerUtilsTest.java b/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/utils/AutoscalerUtilsTest.java new file mode 100644 index 000000000..d720dbdac --- /dev/null +++ b/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/utils/AutoscalerUtilsTest.java @@ -0,0 +1,43 @@ +/* + * 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.standalone.utils; + +import org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin; +import org.apache.flink.autoscaler.metrics.TestCustomEvaluator; + +import org.junit.jupiter.api.Test; + +import java.util.Collection; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Unit tests for {@link AutoscalerUtils}. */ +class AutoscalerUtilsTest { + + @Test + void testDiscoverCustomEvaluators() { + Collection<ScalingMetricsEvaluatorPlugin> evaluators = + AutoscalerUtils.discoverCustomEvaluators(); + + assertThat(evaluators) + .as( + "Expected to discover the TestCustomEvaluator registered under META-INF/services") + .hasSize(1) + .hasOnlyElementsOfType(TestCustomEvaluator.class); + } +} diff --git a/flink-kubernetes-operator/src/test/resources/META-INF/services/org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator b/flink-autoscaler-standalone/src/test/resources/META-INF/services/org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin similarity index 93% copy from flink-kubernetes-operator/src/test/resources/META-INF/services/org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator copy to flink-autoscaler-standalone/src/test/resources/META-INF/services/org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin index a150d9783..a0deb60ed 100644 --- a/flink-kubernetes-operator/src/test/resources/META-INF/services/org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator +++ b/flink-autoscaler-standalone/src/test/resources/META-INF/services/org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin @@ -16,4 +16,5 @@ # limitations under the License. # -org.apache.flink.autoscaler.metrics.TestCustomEvaluator \ No newline at end of file +org.apache.flink.autoscaler.metrics.TestCustomEvaluator + 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 ddfd78cf3..b99f4da36 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,17 +19,14 @@ 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; @@ -39,11 +36,9 @@ 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; @@ -63,7 +58,6 @@ 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(); @@ -79,15 +73,13 @@ public class JobAutoScalerImpl<KEY, Context extends JobAutoScalerContext<KEY>> ScalingExecutor<KEY, Context> scalingExecutor, AutoScalerEventHandler<KEY, Context> eventHandler, ScalingRealizer<KEY, Context> scalingRealizer, - AutoScalerStateStore<KEY, Context> stateStore, - Map<String, FlinkAutoscalerEvaluator> customEvaluators) { + AutoScalerStateStore<KEY, Context> stateStore) { this.metricsCollector = metricsCollector; this.evaluator = evaluator; this.scalingExecutor = scalingExecutor; this.eventHandler = eventHandler; this.scalingRealizer = scalingRealizer; this.stateStore = stateStore; - this.customEvaluators = customEvaluators; } @Override @@ -212,14 +204,8 @@ 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, - customEvaluatorWithConfig); + evaluator.evaluate(ctx.getConfiguration(), collectedMetrics, restartTime); LOG.debug("Evaluated metrics: {}", evaluatedMetrics); lastEvaluatedMetrics.put(ctx.getJobKey(), evaluatedMetrics); @@ -274,20 +260,4 @@ 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 9fb59ff6a..4a1083742 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 @@ -24,9 +24,9 @@ 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.metrics.ScalingMetricsEvaluatorPlugin; import org.apache.flink.autoscaler.topology.JobTopology; import org.apache.flink.autoscaler.utils.AutoScalerUtils; import org.apache.flink.configuration.Configuration; @@ -41,13 +41,17 @@ import javax.annotation.Nullable; import java.time.Duration; import java.time.Instant; +import java.util.Collection; import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Optional; import java.util.SortedMap; +import java.util.stream.Collectors; import static org.apache.flink.autoscaler.config.AutoScalerOptions.BACKLOG_PROCESSING_LAG_THRESHOLD; +import static org.apache.flink.autoscaler.config.AutoScalerOptions.CUSTOM_EVALUATORS; import static org.apache.flink.autoscaler.config.AutoScalerOptions.TARGET_UTILIZATION_BOUNDARY; import static org.apache.flink.autoscaler.config.AutoScalerOptions.UTILIZATION_MAX; import static org.apache.flink.autoscaler.config.AutoScalerOptions.UTILIZATION_MIN; @@ -75,13 +79,14 @@ public class ScalingMetricEvaluator { private static final Logger LOG = LoggerFactory.getLogger(ScalingMetricEvaluator.class); + private final Collection<ScalingMetricsEvaluatorPlugin> customEvaluators; + + public ScalingMetricEvaluator(Collection<ScalingMetricsEvaluatorPlugin> customEvaluators) { + this.customEvaluators = customEvaluators; + } + public EvaluatedMetrics evaluate( - Configuration conf, - CollectedMetricHistory collectedMetrics, - Duration restartTime, - @Nullable - Tuple2<FlinkAutoscalerEvaluator, Configuration> - customEvaluatorWithConfig) { + Configuration conf, CollectedMetricHistory collectedMetrics, Duration restartTime) { LOG.debug("Restart time used in metrics evaluation: {}", restartTime); var scalingOutput = new HashMap<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>>(); var metricsHistory = collectedMetrics.getMetricHistory(); @@ -89,21 +94,21 @@ public class ScalingMetricEvaluator { boolean processingBacklog = isProcessingBacklog(topology, metricsHistory, conf); + var customEvaluatorWithConfig = getCustomEvaluatorIfRequired(conf); var customEvaluationSession = Optional.ofNullable(customEvaluatorWithConfig) .map( info -> Tuple2.of( info.f0, - new FlinkAutoscalerEvaluator.Context( - new UnmodifiableConfiguration(conf), + new ScalingMetricsEvaluatorPlugin.Context( + mergeEvaluatorConfig(conf, info.f1), Collections.unmodifiableSortedMap( metricsHistory), Collections.unmodifiableMap(scalingOutput), topology, processingBacklog, - restartTime, - info.f1))) + restartTime))) .orElse(null); for (var vertex : topology.getVerticesInTopologicalOrder()) { @@ -161,7 +166,7 @@ public class ScalingMetricEvaluator { boolean processingBacklog, Duration restartTime, @Nullable - Tuple2<FlinkAutoscalerEvaluator, FlinkAutoscalerEvaluator.Context> + Tuple2<ScalingMetricsEvaluatorPlugin, ScalingMetricsEvaluatorPlugin.Context> customEvaluationSession) { var latestVertexMetrics = @@ -679,12 +684,12 @@ public class ScalingMetricEvaluator { } /** - * Executes the provided custom evaluator for the given job vertex. Calls {@link - * FlinkAutoscalerEvaluator#evaluateVertexMetrics} to evaluate scaling metrics. + * Executes the provided custom metric evaluator for the given job vertex. Calls {@link + * ScalingMetricsEvaluatorPlugin#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 + * @param customEvaluationSession A tuple containing the custom metric evaluator and evaluation * context. * @return A map of scaling metrics, with its corresponding evaluated scaling metric. */ @@ -692,19 +697,14 @@ public class ScalingMetricEvaluator { protected static Map<ScalingMetric, EvaluatedScalingMetric> runCustomEvaluator( JobVertexID vertex, Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics, - Tuple2<FlinkAutoscalerEvaluator, FlinkAutoscalerEvaluator.Context> + Tuple2<ScalingMetricsEvaluatorPlugin, ScalingMetricsEvaluatorPlugin.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.", + "Custom metric evaluator {} threw an exception.", customEvaluationSession.f0.getClass(), e); } @@ -713,12 +713,82 @@ public class ScalingMetricEvaluator { } /** - * Merges the incoming evaluated metrics into actual evaluated metrics. + * Builds the configuration handed to a custom evaluator: the full job configuration with the + * evaluator-specific options (prefix already stripped) merged on top, so an evaluator-specific + * key takes precedence over the job-level value of the same key. The result is an un-modifiable + * view. + * + * @param jobConf The full job configuration. + * @param evaluatorConf The evaluator-specific options with their prefix already stripped. + * @return An un-modifiable merged configuration. + */ + @VisibleForTesting + protected static Configuration mergeEvaluatorConfig( + Configuration jobConf, Configuration evaluatorConf) { + Configuration merged = new Configuration(jobConf); + merged.addAll(evaluatorConf); + return new UnmodifiableConfiguration(merged); + } + + /** + * Resolves the configured custom metric evaluator (if any) from the registered evaluators. * - * @param actual The target evaluated metrics map to merge into. - * @param incoming The incoming map containing new evaluated metrics map to be merged - * (nullable). + * @param conf The job configuration. + * @return A tuple of the resolved {@link ScalingMetricsEvaluatorPlugin} and its + * evaluator-specific configuration, or {@code null} if no custom evaluator is configured or + * resolvable. */ + @VisibleForTesting + protected Tuple2<ScalingMetricsEvaluatorPlugin, Configuration> getCustomEvaluatorIfRequired( + Configuration conf) { + List<String> instances = conf.get(CUSTOM_EVALUATORS); + if (instances == null || instances.isEmpty()) { + return null; + } + if (instances.size() > 1) { + LOG.warn( + "Only a single custom metric evaluator is currently supported, but {} were configured via '{}': {}. " + + "Falling back to the first entry ('{}'); the remaining entries will be ignored. " + + "Multi-instance support (with a priority/ordering contract) will be added as a follow-up.", + instances.size(), + CUSTOM_EVALUATORS.key(), + instances, + instances.get(0)); + } + String instance = instances.get(0); + String classKey = AutoScalerOptions.customEvaluatorClassKey(instance); + String configuredClassName = + conf.get(AutoScalerOptions.customEvaluatorClassOption(instance)); + if (configuredClassName == null || configuredClassName.isBlank()) { + LOG.warn( + "Custom metric evaluator instance '{}' is configured in '{}' but no implementation class is set via '{}'. " + + "No custom metric evaluator will be applied.", + instance, + CUSTOM_EVALUATORS.key(), + classKey); + return null; + } + ScalingMetricsEvaluatorPlugin match = null; + for (ScalingMetricsEvaluatorPlugin evaluator : customEvaluators) { + if (evaluator.getClass().getName().equals(configuredClassName)) { + match = evaluator; + break; + } + } + if (match == null) { + LOG.warn( + "No registered custom metric evaluator matches class '{}' configured for instance '{}' via '{}'. Discovered evaluators: {}.", + configuredClassName, + instance, + classKey, + customEvaluators.stream() + .map(e -> e.getClass().getName()) + .collect(Collectors.toList())); + return null; + } + return new Tuple2<>(match, AutoScalerOptions.customEvaluatorConfiguration(conf, instance)); + } + @VisibleForTesting protected static void mergeEvaluatedMetricsMaps( Map<ScalingMetric, EvaluatedScalingMetric> actual, @@ -735,13 +805,6 @@ public class 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) { 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 8be3e0f10..c1d723e2e 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 @@ -22,7 +22,6 @@ 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; @@ -34,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."; + public static final String CUSTOM_EVALUATOR_CLASS_SUFFIX = "class"; private static String oldOperatorConfigKey(String key) { return OLD_K8S_OP_CONF_PREFIX + AUTOSCALER_CONF_PREFIX + key; @@ -422,18 +422,116 @@ public class AutoScalerOptions { + "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") + public static final ConfigOption<List<String>> CUSTOM_EVALUATORS = + autoScalerConfig("metrics.custom-evaluators") .stringType() - .defaultValue(null) - .withFallbackKeys(oldOperatorConfigKey(CUSTOM_EVALUATOR_CONF_PREFIX + "name")) - .withDescription("Name of the custom evaluator to be used."); + .asList() + .defaultValues() + .withFallbackKeys(oldOperatorConfigKey("metrics.custom-evaluators")) + .withDescription( + "List of named custom metric evaluator instances to register with the autoscaler. " + + "For each <name> entry, the implementation class to instantiate must be set via " + + "'job.autoscaler.metrics.custom-evaluator.<name>.class', and additional per-instance " + + "options live under 'job.autoscaler.metrics.custom-evaluator.<name>.<option>'. " + + "Only a single instance is honored for now: if more than one is configured, the autoscaler " + + "logs a warning and uses the first entry, ignoring the rest. " + + "Multi-instance support, including a priority/ordering contract, will be added as a follow-up."); + + /** + * Documentation-only template option describing the per-instance class FQN key. Not read at + * runtime. The {@link #customEvaluatorClassKey(String)} is used to derive the actual key for a + * given evaluator instance. + */ + @SuppressWarnings("unused") + public static final ConfigOption<String> CUSTOM_EVALUATOR_CLASS_TEMPLATE = + autoScalerConfig("metrics.custom-evaluator.<name>.class") + .stringType() + .noDefaultValue() + .withFallbackKeys(oldOperatorConfigKey("metrics.custom-evaluator.<name>.class")) + .withDescription( + "Fully-qualified class name of the implementation to instantiate for the custom metric evaluator " + + "instance <name> declared in 'job.autoscaler.metrics.custom-evaluators'. " + + "The class must be discovered via the Plugin / ServiceLoader mechanism and registered " + + "under META-INF/services/org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin."); + + /** + * Documentation-only template option describing the per-instance free-form parameter keys. Not + * read at runtime. The actual evaluator parameters are surfaced to the evaluator via {@link + * #customEvaluatorConfiguration(Configuration, String)}. + */ + @SuppressWarnings("unused") + public static final ConfigOption<String> CUSTOM_EVALUATOR_PARAMETER_TEMPLATE = + autoScalerConfig("metrics.custom-evaluator.<name>.<parameter>") + .stringType() + .noDefaultValue() + .withFallbackKeys( + oldOperatorConfigKey("metrics.custom-evaluator.<name>.<parameter>")) + .withDescription( + "Free-form, evaluator-specific options for the custom metric evaluator instance <name>. " + + "All keys with this prefix are passed to the evaluator at runtime with the " + + "'job.autoscaler.metrics.custom-evaluator.<name>.' prefix stripped."); + + public static String customEvaluatorClassKey(String instanceName) { + return AUTOSCALER_CONF_PREFIX + + CUSTOM_EVALUATOR_CONF_PREFIX + + instanceName + + "." + + CUSTOM_EVALUATOR_CLASS_SUFFIX; + } + + public static String customEvaluatorClassFallbackKey(String instanceName) { + return OLD_K8S_OP_CONF_PREFIX + customEvaluatorClassKey(instanceName); + } + + /** + * Builds the {@link ConfigOption} used to read the implementation class FQN for the given + * custom metric evaluator instance, including the legacy {@code kubernetes.operator.} prefix as + * a fallback key (canonical key takes precedence on overlap). + */ + public static ConfigOption<String> customEvaluatorClassOption(String instanceName) { + return ConfigOptions.key(customEvaluatorClassKey(instanceName)) + .stringType() + .noDefaultValue() + .withFallbackKeys(customEvaluatorClassFallbackKey(instanceName)); + } - public static Configuration forCustomEvaluator( + /** + * Builds a fresh, prefix-stripped {@link Configuration} that exposes the per-instance options + * configured for the scaling custom metric evaluator instance identified by {@code + * instanceName} under both the canonical ({@code + * job.autoscaler.metrics.custom-evaluator.<name>.}) and the legacy ({@code + * kubernetes.operator.job.autoscaler.metrics.custom-evaluator.<name>.}) namespaces. The legacy + * prefix is applied first so that, on overlap, values written under the canonical prefix take + * precedence, matching the semantics of {@link + * org.apache.flink.configuration.ConfigOption#withFallbackKeys} used elsewhere in this class. + */ + public static Configuration customEvaluatorConfiguration( Configuration configuration, String customEvaluatorName) { - // add support for fallBackKey with DelegatingConfiguration. - return new DelegatingConfiguration( - configuration, - AUTOSCALER_CONF_PREFIX + CUSTOM_EVALUATOR_CONF_PREFIX + customEvaluatorName + "."); + String canonicalPrefix = + AUTOSCALER_CONF_PREFIX + CUSTOM_EVALUATOR_CONF_PREFIX + customEvaluatorName + "."; + String legacyPrefix = + OLD_K8S_OP_CONF_PREFIX + + AUTOSCALER_CONF_PREFIX + + CUSTOM_EVALUATOR_CONF_PREFIX + + customEvaluatorName + + "."; + Configuration merged = new Configuration(); + configuration + .toMap() + .forEach( + (k, v) -> { + if (k.startsWith(legacyPrefix)) { + merged.setString(k.substring(legacyPrefix.length()), v); + } + }); + configuration + .toMap() + .forEach( + (k, v) -> { + if (k.startsWith(canonicalPrefix)) { + merged.setString(k.substring(canonicalPrefix.length()), v); + } + }); + return merged; } } 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 deleted file mode 100644 index 976f79b62..000000000 --- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/FlinkAutoscalerEvaluator.java +++ /dev/null @@ -1,118 +0,0 @@ -/* - * 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/main/java/org/apache/flink/autoscaler/metrics/ScalingMetricsEvaluatorPlugin.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetricsEvaluatorPlugin.java new file mode 100644 index 000000000..59d9978de --- /dev/null +++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetricsEvaluatorPlugin.java @@ -0,0 +1,114 @@ +/* + * 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.annotation.Experimental; +import org.apache.flink.autoscaler.topology.JobTopology; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.runtime.jobgraph.JobVertexID; + +import lombok.Value; + +import java.time.Duration; +import java.time.Instant; +import java.util.Map; +import java.util.SortedMap; + +/** + * A pluggable plugin that allows users to provide custom scaling-metric evaluation logic on top of + * the metrics evaluated internally by the autoscaler. Implementations are invoked once per job + * vertex during each evaluation cycle, and the metrics they return are merged on top of the + * internally evaluated metrics, allowing users to override or augment specific {@link + * ScalingMetric} values. + * + * <p>Only one custom metric evaluator per pipeline is supported for now. If multiple instances are + * configured, the autoscaler logs a warning and falls back to the first entry, ignoring the rest. + * Registering multiple implementations via {@code META-INF/services} is fine as they form a + * registry that different jobs can select from by class FQN, but a single job cannot chain or + * compose more than one evaluator. + * + * <p>This was introduced as part of <a + * href="https://cwiki.apache.org/confluence/display/FLINK/FLIP-514%3A+Custom+Evaluator+plugin+for+Flink+Autoscaler";>FLIP-514: + * Custom Evaluator plugin for Flink Autoscaler</a> and is complementary to <a + * href="https://cwiki.apache.org/confluence/display/FLINK/FLIP-575%3A+Scaling+Executor+Plugin+SPI+for+Flink+Autoscaler";>FLIP-575: + * Scaling Executor Plugin SPI for Flink Autoscaler</a> which provides extensibility at the scaling + * decision execution layer. + * + * <p>Implementations are discovered via Java's {@link java.util.ServiceLoader} mechanism. To + * register a custom metric evaluator, add the fully qualified class name of the implementation to + * {@code META-INF/services/org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin}. + */ +@Experimental +public interface ScalingMetricsEvaluatorPlugin { + + /** + * 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.config}, {@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. + */ + @Value + class Context { + /** + * An un-modifiable view of the job configuration, with the evaluator-specific options + * (configured under {@code job.autoscaler.metrics.custom-evaluator.<name>.}) merged on top + * with their prefix stripped. Evaluator-specific keys take precedence over the job-level + * value of the same key, so a CR can override behaviour for this evaluator alone. + */ + Configuration config; + + /** + * An un-modifiable view of the historical record of collected metrics, ordered by + * timestamp. + */ + SortedMap<Instant, CollectedMetrics> metricsHistory; + + /** + * An un-modifiable view of evaluated metrics for previously evaluated vertices. Note: + * evaluation of a vertex for scaling metrics happens topologically. + */ + Map<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>> evaluatedVertexMetrics; + + /** The job topology representing the structure of the Flink job. */ + JobTopology topology; + + /** Indicates whether the job is processing backlog. */ + boolean processingBacklog; + + /** Maximum restart time based on scaling records. */ + Duration restartTime; + } +} 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 index 00bf08e37..a7b5418cf 100644 --- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/AutoScalerCustomEvaluatorTest.java +++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/AutoScalerCustomEvaluatorTest.java @@ -22,7 +22,7 @@ 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.ScalingMetricsEvaluatorPlugin; import org.apache.flink.autoscaler.metrics.TestCustomEvaluator; import org.apache.flink.autoscaler.metrics.TestMetrics; import org.apache.flink.autoscaler.realizer.TestingScalingRealizer; @@ -31,6 +31,7 @@ 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.ConfigOptions; import org.apache.flink.configuration.Configuration; import org.apache.flink.runtime.jobgraph.JobVertexID; @@ -41,13 +42,18 @@ import java.time.Clock; import java.time.Duration; import java.time.Instant; import java.time.ZoneId; +import java.util.Collection; +import java.util.List; import java.util.Map; import static org.apache.flink.autoscaler.TestingAutoscalerUtils.createDefaultJobAutoScalerContext; import static org.apache.flink.autoscaler.topology.ShipStrategy.REBALANCE; +import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; -/** Unit test for testing the integration of custom evaluators with the Flink job autoscaler. */ +/** + * Unit test for testing the integration of custom metric evaluators with the Flink job autoscaler. + */ public class AutoScalerCustomEvaluatorTest { private JobAutoScalerContext<JobID> context; private AutoScalerStateStore<JobID, JobAutoScalerContext<JobID>> stateStore; @@ -100,17 +106,22 @@ public class AutoScalerCustomEvaluatorTest { 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); + defaultConf.set(AutoScalerOptions.CUSTOM_EVALUATORS, List.of(testCustomEvaluatorName)); + defaultConf.set( + ConfigOptions.key( + AutoScalerOptions.customEvaluatorClassKey(testCustomEvaluatorName)) + .stringType() + .noDefaultValue(), + TestCustomEvaluator.class.getName()); autoscaler = new JobAutoScalerImpl<>( metricsCollector, - new ScalingMetricEvaluator(), + new ScalingMetricEvaluator(customEvaluators), scalingExecutor, eventCollector, new TestingScalingRealizer<>(), - stateStore, - customEvaluators); + stateStore); // Reset custom window size to default metricsCollector.setTestMetricWindowSize(null); @@ -153,6 +164,95 @@ public class AutoScalerCustomEvaluatorTest { assertFlinkMetricsCount(1, 0); } + @Test + void testForCustomEvaluatorStripsCanonicalPrefix() { + Configuration conf = new Configuration(); + conf.setString( + AutoScalerOptions.AUTOSCALER_CONF_PREFIX + + AutoScalerOptions.CUSTOM_EVALUATOR_CONF_PREFIX + + "my-evaluator.target-data-rate", + "12345.0"); + // Unrelated keys must not leak into the per-evaluator view. + conf.setString( + AutoScalerOptions.AUTOSCALER_CONF_PREFIX + + AutoScalerOptions.CUSTOM_EVALUATOR_CONF_PREFIX + + "other-evaluator.foo", + "bar"); + conf.setString("unrelated.key", "should-not-leak"); + + Configuration scoped = AutoScalerOptions.customEvaluatorConfiguration(conf, "my-evaluator"); + assertThat(scoped.toMap()) + .containsExactlyInAnyOrderEntriesOf(Map.of("target-data-rate", "12345.0")); + } + + @Test + void testForCustomEvaluatorHonorsLegacyFallbackPrefix() { + Configuration conf = new Configuration(); + // Only the legacy 'kubernetes.operator.' prefix is set, both for a free-form parameter + // and for the per-instance class FQN. + conf.setString( + AutoScalerOptions.OLD_K8S_OP_CONF_PREFIX + + AutoScalerOptions.AUTOSCALER_CONF_PREFIX + + AutoScalerOptions.CUSTOM_EVALUATOR_CONF_PREFIX + + "my-evaluator.target-data-rate", + "999.0"); + conf.setString( + AutoScalerOptions.customEvaluatorClassFallbackKey("my-evaluator"), + "org.example.LegacyEvaluator"); + + Configuration scoped = AutoScalerOptions.customEvaluatorConfiguration(conf, "my-evaluator"); + assertThat(scoped.toMap()) + .containsExactlyInAnyOrderEntriesOf( + Map.of( + "target-data-rate", "999.0", + "class", "org.example.LegacyEvaluator")); + + // The class option must resolve via withFallbackKeys() against the legacy prefix. + assertThat(conf.get(AutoScalerOptions.customEvaluatorClassOption("my-evaluator"))) + .isEqualTo("org.example.LegacyEvaluator"); + } + + @Test + void testForCustomEvaluatorCanonicalPrefixWinsOverLegacyOnOverlap() { + Configuration conf = new Configuration(); + // Both prefixes set for the same free-form option and for the class FQN -> canonical + // takes precedence in both cases. + conf.setString( + AutoScalerOptions.OLD_K8S_OP_CONF_PREFIX + + AutoScalerOptions.AUTOSCALER_CONF_PREFIX + + AutoScalerOptions.CUSTOM_EVALUATOR_CONF_PREFIX + + "my-evaluator.target-data-rate", + "from-legacy"); + conf.setString( + AutoScalerOptions.AUTOSCALER_CONF_PREFIX + + AutoScalerOptions.CUSTOM_EVALUATOR_CONF_PREFIX + + "my-evaluator.target-data-rate", + "from-canonical"); + conf.setString( + AutoScalerOptions.customEvaluatorClassFallbackKey("my-evaluator"), + "org.example.LegacyEvaluator"); + conf.setString( + AutoScalerOptions.customEvaluatorClassKey("my-evaluator"), + "org.example.CanonicalEvaluator"); + + Configuration scoped = AutoScalerOptions.customEvaluatorConfiguration(conf, "my-evaluator"); + assertThat(scoped.toMap().get("target-data-rate")).isEqualTo("from-canonical"); + assertThat(scoped.toMap().get("class")).isEqualTo("org.example.CanonicalEvaluator"); + + // The class option must prefer the canonical key over the legacy fallback on overlap. + assertThat(conf.get(AutoScalerOptions.customEvaluatorClassOption("my-evaluator"))) + .isEqualTo("org.example.CanonicalEvaluator"); + } + + @Test + void testForCustomEvaluatorReturnsEmptyConfigWhenNothingMatches() { + Configuration conf = new Configuration(); + conf.setString("unrelated.key", "value"); + + Configuration scoped = AutoScalerOptions.customEvaluatorConfiguration(conf, "my-evaluator"); + assertThat(scoped.toMap()).isEmpty(); + } + private void setClocksTo(Instant time) { var clock = Clock.fixed(time, ZoneId.systemDefault()); autoscaler.setClock(clock); @@ -165,9 +265,7 @@ public class AutoScalerCustomEvaluatorTest { assertEquals(balancedCount, autoscalerFlinkMetrics.getNumBalancedCount()); } - private Map<String, FlinkAutoscalerEvaluator> createTestCustomEvaluator() { - var testCustomEvaluator = new TestCustomEvaluator(); - testCustomEvaluator.configure(new Configuration()); - return Map.of(testCustomEvaluator.getName(), testCustomEvaluator); + private Collection<ScalingMetricsEvaluatorPlugin> createTestCustomEvaluator() { + return List.of(new 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 956ecf32d..9ff58ab21 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,7 +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.List; import java.util.Map; import java.util.SortedMap; @@ -105,12 +105,11 @@ public class BacklogBasedScalingTest { autoscaler = new JobAutoScalerImpl<>( metricsCollector, - new ScalingMetricEvaluator(), + new ScalingMetricEvaluator(List.of()), scalingExecutor, eventCollector, new TestingScalingRealizer<>(), - stateStore, - Collections.emptyMap()); + stateStore); // 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 9eca91295..43c9f8a1b 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 @@ -118,12 +118,11 @@ public class DelayedScaleDownEndToEndTest { autoscaler = new JobAutoScalerImpl<>( metricsCollector, - new ScalingMetricEvaluator(), + new ScalingMetricEvaluator(List.of()), new ScalingExecutor<>(eventCollector, stateStore), eventCollector, scalingRealizer, - stateStore, - Collections.emptyMap()); + stateStore); // 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 1597ec5cc..384d97631 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,9 +24,7 @@ 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; @@ -35,8 +33,6 @@ 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; @@ -60,7 +56,6 @@ 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; @@ -71,9 +66,6 @@ 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. */ @@ -111,7 +103,7 @@ public class JobAutoScalerImplTest { .pendingRecords(0L) .build()); - ScalingMetricEvaluator evaluator = new ScalingMetricEvaluator(); + ScalingMetricEvaluator evaluator = new ScalingMetricEvaluator(Collections.emptyList()); ScalingExecutor<JobID, JobAutoScalerContext<JobID>> scalingExecutor = new ScalingExecutor<>(eventCollector, stateStore); @@ -122,8 +114,7 @@ public class JobAutoScalerImplTest { scalingExecutor, eventCollector, scalingRealizer, - stateStore, - Collections.emptyMap()); + stateStore); var now = Instant.now(); autoscaler.setClock(Clock.fixed(now, ZoneId.systemDefault())); @@ -162,13 +153,7 @@ public class JobAutoScalerImplTest { void testErrorReporting() throws Exception { var autoscaler = new JobAutoScalerImpl<>( - null, - null, - null, - eventCollector, - scalingRealizer, - stateStore, - Collections.emptyMap()); + null, null, null, eventCollector, scalingRealizer, stateStore); autoscaler.scale(context); Assertions.assertEquals( @@ -203,8 +188,7 @@ public class JobAutoScalerImplTest { null, eventCollector, scalingRealizer, - stateStore, - Collections.emptyMap()); + stateStore); // Should not produce an error autoscaler.scale(context); @@ -243,8 +227,7 @@ public class JobAutoScalerImplTest { null, eventCollector, realizeParallelismOverridesWithExceptionsScalingRealizer, - stateStore, - Collections.emptyMap()); + stateStore); // Should produce an error autoscaler.scale(context); @@ -261,8 +244,7 @@ public class JobAutoScalerImplTest { null, eventCollector, scalingRealizer, - stateStore, - Collections.emptyMap()); + stateStore); // Initially we should return empty overrides, do not crate any state assertThat(autoscaler.getParallelismOverrides(context)).isEmpty(); @@ -336,13 +318,7 @@ public class JobAutoScalerImplTest { var overrides = new HashMap<String, String>(); var autoscaler = new JobAutoScalerImpl<>( - null, - null, - null, - eventCollector, - scalingRealizer, - stateStore, - Collections.emptyMap()) { + null, null, null, eventCollector, scalingRealizer, stateStore) { public Map<String, String> getParallelismOverrides( JobAutoScalerContext<JobID> ctx) { return new HashMap<>(overrides); @@ -441,13 +417,7 @@ public class JobAutoScalerImplTest { context.getConfiguration().set(AutoScalerOptions.MEMORY_TUNING_ENABLED, true); var autoscaler = new JobAutoScalerImpl<>( - null, - null, - null, - eventCollector, - scalingRealizer, - stateStore, - Collections.emptyMap()); + null, null, null, eventCollector, scalingRealizer, stateStore); // Initially we should return empty overrides, do not crate any state assertThat(stateStore.getConfigChanges(context).getOverrides()).isEmpty(); @@ -495,13 +465,7 @@ public class JobAutoScalerImplTest { var autoscaler = new JobAutoScalerImpl<>( - null, - null, - null, - eventCollector, - scalingRealizer, - stateStore, - Collections.emptyMap()); + null, null, null, eventCollector, scalingRealizer, stateStore); autoscaler.scale(context); assertTrue(stateStore.getScalingHistory(context).isEmpty()); @@ -526,109 +490,4 @@ 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 a046d3be3..c347cff8a 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 @@ -80,7 +80,7 @@ public class MetricsCollectionAndEvaluationTest { @BeforeEach public void setup() { context = createDefaultJobAutoScalerContext(); - evaluator = new ScalingMetricEvaluator(); + evaluator = new ScalingMetricEvaluator(List.of()); stateStore = new InMemoryAutoScalerStateStore<>(); scalingExecutor = new ScalingExecutor<>(new TestingEventCollector<>(), stateStore); @@ -181,7 +181,7 @@ public class MetricsCollectionAndEvaluationTest { assertEquals(3, collectedMetrics.getMetricHistory().size()); assertTrue(collectedMetrics.isFullyCollected()); - var evaluation = evaluator.evaluate(conf, collectedMetrics, restartTime, null); + var evaluation = evaluator.evaluate(conf, collectedMetrics, restartTime); 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, null); + evaluator.evaluate(context.getConfiguration(), collectedMetrics, restartTime); assertEquals( 500., evaluation @@ -661,7 +661,7 @@ public class MetricsCollectionAndEvaluationTest { var collectedMetrics = collectMetrics(); var evaluation = - evaluator.evaluate(context.getConfiguration(), collectedMetrics, restartTime, null); + evaluator.evaluate(context.getConfiguration(), collectedMetrics, restartTime); assertEquals( 0, evaluation @@ -712,8 +712,7 @@ public class MetricsCollectionAndEvaluationTest { .getMetricHistory() .put(Instant.ofEpochSecond(1234), new CollectedMetrics(newMetrics, Map.of())); - evaluation = - evaluator.evaluate(context.getConfiguration(), collectedMetrics, restartTime, null); + evaluation = evaluator.evaluate(context.getConfiguration(), collectedMetrics, restartTime); 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 2902d25fa..99cdc8ead 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,7 +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.List; import java.util.Map; import static org.apache.flink.autoscaler.TestingAutoscalerUtils.createDefaultJobAutoScalerContext; @@ -95,12 +95,11 @@ public class RecommendedParallelismTest { autoscaler = new JobAutoScalerImpl<>( metricsCollector, - new ScalingMetricEvaluator(), + new ScalingMetricEvaluator(List.of()), new ScalingExecutor<>(eventCollector, stateStore), eventCollector, new TestingScalingRealizer<>(), - stateStore, - Collections.emptyMap()); + stateStore); // 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 b17295e1d..d914337f2 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,12 +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.ScalingMetricsEvaluatorPlugin; 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.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.UnmodifiableConfiguration; import org.apache.flink.runtime.jobgraph.JobVertexID; @@ -38,7 +40,9 @@ import java.time.Duration; import java.time.Instant; import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; +import java.util.Set; import java.util.TreeMap; import static org.apache.flink.autoscaler.config.AutoScalerOptions.CATCH_UP_DURATION; @@ -48,12 +52,15 @@ import static org.apache.flink.autoscaler.config.AutoScalerOptions.UTILIZATION_T import static org.apache.flink.autoscaler.topology.ShipStrategy.REBALANCE; 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; /** Scaling evaluator test. */ public class ScalingMetricEvaluatorTest { - private ScalingMetricEvaluator evaluator = new ScalingMetricEvaluator(); + private ScalingMetricEvaluator evaluator = new ScalingMetricEvaluator(List.of()); @Test public void testLagBasedSourceScaling() { @@ -115,8 +122,7 @@ public class ScalingMetricEvaluatorTest { .evaluate( conf, new CollectedMetricHistory(topology, metricHistory, Instant.now()), - Duration.ZERO, - null) + Duration.ZERO) .getVertexMetrics(); assertEquals( @@ -150,8 +156,7 @@ public class ScalingMetricEvaluatorTest { .evaluate( conf, new CollectedMetricHistory(topology, metricHistory, Instant.now()), - Duration.ZERO, - null) + Duration.ZERO) .getVertexMetrics(); assertEquals( EvaluatedScalingMetric.avg(150), @@ -174,8 +179,7 @@ public class ScalingMetricEvaluatorTest { .evaluate( conf, new CollectedMetricHistory(topology, metricHistory, Instant.now()), - Duration.ZERO, - null) + Duration.ZERO) .getVertexMetrics(); assertEquals( EvaluatedScalingMetric.avg(150), @@ -197,8 +201,7 @@ public class ScalingMetricEvaluatorTest { .evaluate( conf, new CollectedMetricHistory(topology, metricHistory, Instant.now()), - Duration.ZERO, - null) + Duration.ZERO) .getVertexMetrics(); assertEquals( EvaluatedScalingMetric.avg(150), @@ -257,8 +260,7 @@ public class ScalingMetricEvaluatorTest { .evaluate( conf, new CollectedMetricHistory(topology, metricHistory, Instant.now()), - Duration.ZERO, - null) + Duration.ZERO) .getVertexMetrics(); assertEquals( EvaluatedScalingMetric.avg(100), @@ -319,16 +321,20 @@ public class ScalingMetricEvaluatorTest { var conf = new Configuration(); conf.set(CATCH_UP_DURATION, Duration.ofSeconds(2)); - FlinkAutoscalerEvaluator customEvaluator = new TestCustomEvaluator(); - var customEvaluatorWithConfig = new Tuple2<>(customEvaluator, new Configuration()); + ScalingMetricsEvaluatorPlugin customEvaluator = new TestCustomEvaluator(); + var customEvaluatorName = "test-custom-evaluator"; + conf.set(AutoScalerOptions.CUSTOM_EVALUATORS, List.of(customEvaluatorName)); + conf.set( + AutoScalerOptions.customEvaluatorClassOption(customEvaluatorName), + TestCustomEvaluator.class.getName()); + var customEvaluatorAwareEvaluator = new ScalingMetricEvaluator(List.of(customEvaluator)); var evaluatedMetrics = - evaluator + customEvaluatorAwareEvaluator .evaluate( conf, new CollectedMetricHistory(topology, metricHistory, Instant.now()), - Duration.ZERO, - customEvaluatorWithConfig) + Duration.ZERO) .getVertexMetrics(); assertEquals( @@ -1031,20 +1037,19 @@ public class ScalingMetricEvaluatorTest { Map.of())); var conf = new Configuration(); - FlinkAutoscalerEvaluator customEvaluator = new TestCustomEvaluator(); + ScalingMetricsEvaluatorPlugin customEvaluator = new TestCustomEvaluator(); var evaluatedMetrics = new HashMap<ScalingMetric, EvaluatedScalingMetric>(); var testCustomEvaluationSession = Tuple2.of( customEvaluator, - new FlinkAutoscalerEvaluator.Context( + new ScalingMetricsEvaluatorPlugin.Context( new UnmodifiableConfiguration(conf), Collections.unmodifiableSortedMap(metricHistory), Collections.unmodifiableMap(new HashMap<>()), topology, false, - Duration.ZERO, - new Configuration())); + Duration.ZERO)); var testCustomEvaluatorResult = ScalingMetricEvaluator.runCustomEvaluator( @@ -1132,4 +1137,118 @@ public class ScalingMetricEvaluatorTest { assertEquals(60.0, result.getCurrent()); assertEquals(120.0, result.getAverage()); } + + @Test + void testGetCustomEvaluatorIfRequired() { + ScalingMetricsEvaluatorPlugin testCustomEvaluator = new TestCustomEvaluator(); + var customEvaluatorAwareEvaluator = + new ScalingMetricEvaluator(List.of(testCustomEvaluator)); + + String testCustomEvaluatorName = "test-custom-evaluator"; + String testCustomEvaluatorClassName = TestCustomEvaluator.class.getName(); + ConfigOption<String> classOpt = + ConfigOptions.key( + AutoScalerOptions.customEvaluatorClassKey(testCustomEvaluatorName)) + .stringType() + .noDefaultValue(); + + var conf = new Configuration(); + + // Case 1: Single custom evaluator instance configured with its FQN class. + conf.set(AutoScalerOptions.CUSTOM_EVALUATORS, List.of(testCustomEvaluatorName)); + conf.set(classOpt, testCustomEvaluatorClassName); + + var customEvaluatorWithConfig = + customEvaluatorAwareEvaluator.getCustomEvaluatorIfRequired(conf); + assertNotNull(customEvaluatorWithConfig); + assertInstanceOf(ScalingMetricsEvaluatorPlugin.class, customEvaluatorWithConfig.f0); + assertEquals( + testCustomEvaluatorClassName, customEvaluatorWithConfig.f0.getClass().getName()); + var customEvaluatorConfig = customEvaluatorWithConfig.f1; + assertNotNull(customEvaluatorConfig); + // Only the .class key lives under the instance namespace at this point. + assertEquals(Set.of("class"), customEvaluatorConfig.keySet()); + + // Case 2: Custom evaluator with additional per-instance options. + conf.set( + ConfigOptions.key( + AutoScalerOptions.AUTOSCALER_CONF_PREFIX + + AutoScalerOptions.CUSTOM_EVALUATOR_CONF_PREFIX + + testCustomEvaluatorName + + ".k1") + .stringType() + .noDefaultValue(), + "v1"); + conf.set( + ConfigOptions.key( + AutoScalerOptions.AUTOSCALER_CONF_PREFIX + + AutoScalerOptions.CUSTOM_EVALUATOR_CONF_PREFIX + + testCustomEvaluatorName + + ".k2") + .stringType() + .noDefaultValue(), + "v2"); + + var customEvaluatorWithConfigContainingAdditionalKeys = + customEvaluatorAwareEvaluator.getCustomEvaluatorIfRequired(conf); + assertNotNull(customEvaluatorWithConfigContainingAdditionalKeys); + assertInstanceOf( + ScalingMetricsEvaluatorPlugin.class, + customEvaluatorWithConfigContainingAdditionalKeys.f0); + var customEvaluatorConfigContainingAdditionalKeys = + customEvaluatorWithConfigContainingAdditionalKeys.f1; + assertNotNull(customEvaluatorConfigContainingAdditionalKeys); + assertEquals( + Set.of("class", "k1", "k2"), + customEvaluatorConfigContainingAdditionalKeys.keySet()); + + // Case 3: Configured class FQN does not match any registered evaluator -> null + warn. + conf.set(classOpt, "org.apache.flink.autoscaler.metrics.UnknownEvaluator"); + assertNull(customEvaluatorAwareEvaluator.getCustomEvaluatorIfRequired(conf)); + + // Case 4: Instance listed but no .class option set -> null + warn. + conf.removeConfig(classOpt); + assertNull(customEvaluatorAwareEvaluator.getCustomEvaluatorIfRequired(conf)); + + // Case 5: Custom evaluators list not configured at all. + conf.removeConfig(AutoScalerOptions.CUSTOM_EVALUATORS); + assertNull(customEvaluatorAwareEvaluator.getCustomEvaluatorIfRequired(conf)); + + // Case 6: More than one instance configured -> warn + first-wins (single-instance + // constraint). + conf.set( + AutoScalerOptions.CUSTOM_EVALUATORS, + List.of(testCustomEvaluatorName, "another-instance")); + conf.set(classOpt, testCustomEvaluatorClassName); + var firstWinsResult = customEvaluatorAwareEvaluator.getCustomEvaluatorIfRequired(conf); + assertNotNull(firstWinsResult); + assertEquals(testCustomEvaluatorClassName, firstWinsResult.f0.getClass().getName()); + conf.removeConfig(AutoScalerOptions.CUSTOM_EVALUATORS); + conf.removeConfig(classOpt); + + // Case 7: No custom evaluators registered at all -> null even when configured. + var noCustomEvaluator = new ScalingMetricEvaluator(Collections.emptyList()); + conf.set(AutoScalerOptions.CUSTOM_EVALUATORS, List.of(testCustomEvaluatorName)); + conf.set(classOpt, testCustomEvaluatorClassName); + assertNull(noCustomEvaluator.getCustomEvaluatorIfRequired(conf)); + } + + @Test + void testMergeEvaluatorConfig() { + var jobConf = new Configuration(); + jobConf.setString("only-job", "j"); + jobConf.setString("shared", "job-value"); + + var evaluatorConf = new Configuration(); + evaluatorConf.setString("only-evaluator", "e"); + evaluatorConf.setString("shared", "evaluator-value"); + + var merged = ScalingMetricEvaluator.mergeEvaluatorConfig(jobConf, evaluatorConf); + + // Job-only and evaluator-only keys are both visible. + assertEquals("j", merged.getString("only-job", null)); + assertEquals("e", merged.getString("only-evaluator", null)); + // On a key present in both, the evaluator-specific value wins. + assertEquals("evaluator-value", merged.getString("shared", null)); + } } 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 index b1426f6bc..4db951294 100644 --- 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 @@ -25,14 +25,10 @@ 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. + * A test implementation of the {@link ScalingMetricsEvaluatorPlugin} 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"; - } +public class TestCustomEvaluator implements ScalingMetricsEvaluatorPlugin { @Override public Map<ScalingMetric, EvaluatedScalingMetric> evaluateVertexMetrics( 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 1edd695f7..f437ed9c9 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 @@ -188,8 +188,7 @@ public class FlinkOperator { var clusterResourceManager = ClusterResourceManager.of(configManager.getDefaultConfig(), client); var autoscaler = - AutoscalerFactory.create( - client, eventRecorder, clusterResourceManager, configManager); + AutoscalerFactory.create(client, eventRecorder, clusterResourceManager, baseConfig); var reconcilerFactory = new ReconcilerFactory(eventRecorder, statusRecorder, autoscaler); var observerFactory = new FlinkDeploymentObserverFactory(eventRecorder); var canaryResourceManager = new CanaryResourceManager<FlinkDeployment>(configManager); @@ -213,7 +212,7 @@ public class FlinkOperator { var metricManager = MetricManager.createFlinkSessionJobMetricManager(baseConfig, metricGroup); var statusRecorder = StatusRecorder.create(client, metricManager, listeners); - var autoscaler = AutoscalerFactory.create(client, eventRecorder, null, configManager); + var autoscaler = AutoscalerFactory.create(client, eventRecorder, null, baseConfig); 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 d97d33938..506187946 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 @@ -22,10 +22,11 @@ import org.apache.flink.autoscaler.JobAutoScalerImpl; import org.apache.flink.autoscaler.RestApiMetricsCollector; import org.apache.flink.autoscaler.ScalingExecutor; import org.apache.flink.autoscaler.ScalingMetricEvaluator; +import org.apache.flink.configuration.Configuration; 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.AutoscalerUtils; import org.apache.flink.kubernetes.operator.utils.EventRecorder; import io.fabric8.kubernetes.client.KubernetesClient; @@ -38,19 +39,18 @@ public class AutoscalerFactory { KubernetesClient client, EventRecorder eventRecorder, ClusterResourceManager clusterResourceManager, - FlinkConfigManager configManager) { + Configuration conf) { var stateStore = new KubernetesAutoScalerStateStore(new ConfigMapStore(client)); var eventHandler = new KubernetesAutoScalerEventHandler(eventRecorder); - var customEvaluators = AutoscalerUtils.discoverCustomEvaluators(configManager); + var customEvaluators = AutoscalerUtils.discoverCustomEvaluators(conf); return new JobAutoScalerImpl<>( new RestApiMetricsCollector<>(), - new ScalingMetricEvaluator(), + new ScalingMetricEvaluator(customEvaluators), new ScalingExecutor<>(eventHandler, stateStore, clusterResourceManager), eventHandler, new KubernetesScalingRealizer(), - stateStore, - customEvaluators); + stateStore); } } 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 deleted file mode 100644 index 3ea5f077a..000000000 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerUtils.java +++ /dev/null @@ -1,75 +0,0 @@ -/* - * 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/main/java/org/apache/flink/kubernetes/operator/utils/AutoscalerUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/AutoscalerUtils.java new file mode 100644 index 000000000..a960994c9 --- /dev/null +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/AutoscalerUtils.java @@ -0,0 +1,61 @@ +/* + * 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.utils; + +import org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin; +import org.apache.flink.configuration.ConfigConstants; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.core.plugin.PluginUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +/** Autoscaler related utility methods for Operator. */ +public class AutoscalerUtils { + private static final Logger LOG = LoggerFactory.getLogger(AutoscalerUtils.class); + + /** + * Discovers custom metric evaluators for autoscaler. + * + * @param conf Base FlinkConfigManager configuration + * @return The list of discovered custom metric evaluators. + */ + public static Collection<ScalingMetricsEvaluatorPlugin> discoverCustomEvaluators( + Configuration conf) { + List<ScalingMetricsEvaluatorPlugin> customEvaluators = new ArrayList<>(); + PluginUtils.createPluginManagerFromRootFolder(conf) + .load(ScalingMetricsEvaluatorPlugin.class) + .forEachRemaining( + customEvaluator -> { + LOG.info( + "Discovered custom metric evaluator for autoscaler from plugin directory[{}]: {}.", + System.getenv() + .getOrDefault( + ConfigConstants.ENV_FLINK_PLUGINS_DIR, + ConfigConstants.DEFAULT_FLINK_PLUGINS_DIRS), + customEvaluator.getClass().getName()); + customEvaluators.add(customEvaluator); + }); + return customEvaluators; + } +} 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 6b884044b..28aa942ce 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 @@ -20,7 +20,6 @@ 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; @@ -50,7 +49,7 @@ public class AutoscalerFactoryTest { new FlinkResourceEventCollector(), new FlinkStateSnapshotEventCollector()), new ClusterResourceManager(Duration.ZERO, kubernetesClient), - new FlinkConfigManager(new Configuration())); + 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 index 2d1f365da..b18a31732 100644 --- 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 @@ -18,22 +18,23 @@ package org.apache.flink.kubernetes.operator.autoscaler; +import org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin; +import org.apache.flink.autoscaler.metrics.TestCustomEvaluator; 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.apache.flink.kubernetes.operator.utils.AutoscalerUtils; 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.Collection; import java.util.HashMap; -import java.util.HashSet; -import java.util.List; import java.util.Map; -import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.assertj.core.api.Assertions.assertThat; /** Unit tests for {@link AutoscalerUtils}. */ public class AutoscalerUtilsTest { @@ -49,15 +50,12 @@ public class AutoscalerUtilsTest { 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")); + Collection<ScalingMetricsEvaluatorPlugin> discoveredEvaluators = + AutoscalerUtils.discoverCustomEvaluators(new Configuration()); - assertEquals(expectedEvaluators, discoveredEvaluators); + assertThat(discoveredEvaluators) + .hasSize(1) + .hasOnlyElementsOfType(TestCustomEvaluator.class); } 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 165fc697d..5c9f3a41f 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 @@ -64,20 +64,20 @@ import java.util.function.BiConsumer; public class TestingFlinkDeploymentController implements Reconciler<FlinkDeployment>, Cleaner<FlinkDeployment> { - @Getter private ReconcilerFactory reconcilerFactory; - private FlinkDeploymentController flinkDeploymentController; - @Getter private StatusUpdateCounter statusUpdateCounter = new StatusUpdateCounter(); - private FlinkResourceEventCollector flinkResourceEventCollector = + @Getter private final ReconcilerFactory reconcilerFactory; + private final FlinkDeploymentController flinkDeploymentController; + @Getter private final StatusUpdateCounter statusUpdateCounter = new StatusUpdateCounter(); + private final FlinkResourceEventCollector flinkResourceEventCollector = new FlinkResourceEventCollector(); private EventRecorder eventRecorder; - @Getter private TestingFlinkResourceContextFactory contextFactory; + @Getter private final TestingFlinkResourceContextFactory contextFactory; + @Getter private final StatusRecorder<FlinkDeployment, FlinkDeploymentStatus> statusRecorder; + @Getter private final CanaryResourceManager<FlinkDeployment> canaryResourceManager; - @Getter private StatusRecorder<FlinkDeployment, FlinkDeploymentStatus> statusRecorder; - @Getter private CanaryResourceManager<FlinkDeployment> canaryResourceManager; - - private Map<ResourceID, Tuple2<FlinkDeploymentSpec, Long>> currentGenerations = new HashMap<>(); + private final Map<ResourceID, Tuple2<FlinkDeploymentSpec, Long>> currentGenerations = + new HashMap<>(); public TestingFlinkDeploymentController( FlinkConfigManager configManager, TestingFlinkService flinkService) { @@ -102,7 +102,7 @@ public class TestingFlinkDeploymentController eventRecorder, new ClusterResourceManager( Duration.ZERO, flinkService.getKubernetesClient()), - configManager)); + new Configuration())); 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.ScalingMetricsEvaluatorPlugin similarity index 100% rename from flink-kubernetes-operator/src/test/resources/META-INF/services/org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator rename to flink-kubernetes-operator/src/test/resources/META-INF/services/org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin
