[FLINK-39511][flink-autoscaler] FLIP-514: Custom Evaluator plugin for Flink Autoscaler

Author: Dennis-Mircea

docs/content.zh/docs/operations/plugins.md

@@ -189,4 +189,96 @@ 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 Flink Autoscaler Evaluator
+
+`FlinkAutoscalerEvaluator` 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 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-standalone`. In both cases the implementation class must be registered in `META-INF/services`.
+
+For each evaluation cycle, the autoscaler invokes the custom evaluator selected via the `job.autoscaler.metrics.custom-evaluator.name` 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`).
+
+The following steps demonstrate how to develop and use a custom evaluator.
+
+1. Implement the `FlinkAutoscalerEvaluator` interface:
+    ```java
+    package org.apache.flink.autoscaler.custom;
+
+    import org.apache.flink.autoscaler.metrics.EvaluatedScalingMetric;
+    import org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator;
+    import org.apache.flink.autoscaler.metrics.ScalingMetric;
+    import org.apache.flink.runtime.jobgraph.JobVertexID;
+
+    import java.util.HashMap;
+    import java.util.Map;
+
+    /** Custom evaluator implementation of {@link FlinkAutoscalerEvaluator}. */
+    public class CustomEvaluator implements FlinkAutoscalerEvaluator {
+
+        @Override
+        public String getName() {
+            return "custom-evaluator";
+        }
+
+        @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 a value read from the evaluator-specific configuration.
+            double target = context.getCustomEvaluatorConf().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 job configuration, the metrics history, previously evaluated vertex metrics (evaluation happens topologically), the job topology, backlog status, max restart time, and the evaluator-specific configuration.
+
+2. Create the service definition file `org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator` 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 evaluator JAR.
+
+4. Select the custom evaluator via configuration. The evaluator whose `getName()` matches the configured name will be invoked; any `job.autoscaler.metrics.custom-evaluator.<name>.*` entries are surfaced to the evaluator via `Context#getCustomEvaluatorConf()` (with the `job.autoscaler.metrics.custom-evaluator.<name>.` prefix stripped):
+    ```yaml
+    job.autoscaler.metrics.custom-evaluator.name: custom-evaluator
+    job.autoscaler.metrics.custom-evaluator.custom-evaluator.target-data-rate: 100000.0
+    ```
+   {{< hint warning >}}
+   **Only one custom evaluator per pipeline is supported**. The `job.autoscaler.metrics.custom-evaluator.name` is a single-valued option and the autoscaler resolves and invokes exactly one evaluator per evaluation cycle. Registering multiple implementations via `META-INF/services` is fine as they form a registry that different jobs can select from by name, but a single job cannot chain or compose more than one evaluator.
+   {{< /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 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 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 evaluator from plugin directory[/opt/flink/plugins]: org.apache.flink.autoscaler.custom.CustomEvaluator.
+        ```
+
+    - **Standalone autoscaler** – simply place the custom 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 evaluator via ServiceLoader: org.apache.flink.autoscaler.custom.CustomEvaluator (name=custom-evaluator).
+        ```

docs/content/docs/operations/plugins.md

@@ -188,3 +188,97 @@ 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 Flink Autoscaler Evaluator
+
+`FlinkAutoscalerEvaluator` 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 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-standalone`. In both cases the implementation class must be registered in `META-INF/services`.
+
+For each evaluation cycle, the autoscaler invokes the custom evaluator selected via the `job.autoscaler.metrics.custom-evaluator.name` 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`).
+
+The following steps demonstrate how to develop and use a custom evaluator.
+
+1. Implement the `FlinkAutoscalerEvaluator` interface:
+    ```java
+    package org.apache.flink.autoscaler.custom;
+
+    import org.apache.flink.autoscaler.metrics.EvaluatedScalingMetric;
+    import org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator;
+    import org.apache.flink.autoscaler.metrics.ScalingMetric;
+    import org.apache.flink.runtime.jobgraph.JobVertexID;
+
+    import java.util.HashMap;
+    import java.util.Map;
+
+    /** Custom evaluator implementation of {@link FlinkAutoscalerEvaluator}. */
+    public class CustomEvaluator implements FlinkAutoscalerEvaluator {
+
+        @Override
+        public String getName() {
+            return "custom-evaluator";
+        }
+
+        @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 a value read from the evaluator-specific configuration.
+            double target = context.getCustomEvaluatorConf().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 job configuration, the metrics history, previously evaluated vertex metrics (evaluation happens topologically), the job topology, backlog status, max restart time, and the evaluator-specific configuration.
+
+2. Create the service definition file `org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator` 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 evaluator JAR.
+
+4. Select the custom evaluator via configuration. The evaluator whose `getName()` matches the configured name will be invoked; any `job.autoscaler.metrics.custom-evaluator.<name>.*` entries are surfaced to the evaluator via `Context#getCustomEvaluatorConf()` (with the `job.autoscaler.metrics.custom-evaluator.<name>.` prefix stripped):
+    ```yaml
+    job.autoscaler.metrics.custom-evaluator.name: custom-evaluator
+    job.autoscaler.metrics.custom-evaluator.custom-evaluator.target-data-rate: 100000.0
+    ```
+   {{< hint warning >}}
+   **Only one custom evaluator per pipeline is supported**. The `job.autoscaler.metrics.custom-evaluator.name` is a single-valued option and the autoscaler resolves and invokes exactly one evaluator per evaluation cycle. Registering multiple implementations via `META-INF/services` is fine as they form a registry that different jobs can select from by name, but a single job cannot chain or compose more than one evaluator.
+   {{< /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 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 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 evaluator from plugin directory[/opt/flink/plugins]: org.apache.flink.autoscaler.custom.CustomEvaluator.
+        ```
+
+    - **Standalone autoscaler** - simply place the custom 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 evaluator via ServiceLoader: org.apache.flink.autoscaler.custom.CustomEvaluator (name=custom-evaluator).
+        ```

docs/layouts/shortcodes/generated/auto_scaler_configuration.html

@@ -92,6 +92,12 @@
             <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</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>String</td>
+            <td>Name of the custom evaluator to be used.</td>
+        </tr>
         <tr>
             <td><h5>job.autoscaler.metrics.window</h5></td>
             <td style="word-wrap: break-word;">15 min</td>

flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/AutoscalerUtils.java

@@ -0,0 +1,68 @@
+/*
+ * 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;
+
+import org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+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 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.FlinkAutoscalerEvaluator} on the
+     * classpath.
+     *
+     * @return A map of discovered custom evaluators keyed by the name returned by {@link
+     *     FlinkAutoscalerEvaluator#getName()}. If two implementations report the same name, the
+     *     later one overwrites the earlier one and a warning is logged.
+     */
+    public static Map<String, FlinkAutoscalerEvaluator> discoverCustomEvaluators() {
+        Map<String, FlinkAutoscalerEvaluator> customEvaluators = new HashMap<>();
+        ServiceLoader.load(FlinkAutoscalerEvaluator.class)
+                .forEach(
+                        customEvaluator -> {
+                            String customEvaluatorName = customEvaluator.getName();
+                            LOG.info(
+                                    "Discovered custom evaluator via ServiceLoader: {} (name={}).",
+                                    customEvaluator.getClass().getName(),
+                                    customEvaluatorName);
+                            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;
+    }
+}

flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerEntrypoint.java

@@ -27,6 +27,7 @@
 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.FlinkAutoscalerEvaluator;
 import org.apache.flink.autoscaler.standalone.flinkcluster.FlinkClusterJobListFetcher;
 import org.apache.flink.autoscaler.standalone.realizer.RescaleApiScalingRealizer;
 import org.apache.flink.autoscaler.state.AutoScalerStateStore;
@@ -38,6 +39,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Map;
+
 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 +91,16 @@ JobListFetcher<KEY, Context> createJobListFetcher(Configuration conf) {
             JobAutoScaler<KEY, Context> createJobAutoscaler(
                     AutoScalerEventHandler<KEY, Context> eventHandler,
                     AutoScalerStateStore<KEY, Context> stateStore) {
+        Map<String, FlinkAutoscalerEvaluator> customEvaluators =
+                AutoscalerUtils.discoverCustomEvaluators();
         return new JobAutoScalerImpl<>(
                 new RestApiMetricsCollector<>(),
                 new ScalingMetricEvaluator(),
                 new ScalingExecutor<>(eventHandler, stateStore),
                 eventHandler,
                 new RescaleApiScalingRealizer<>(eventHandler),
                 stateStore,
-                null);
+                customEvaluators);
     }
 
     @VisibleForTesting