You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Dennis-Mircea (via GitHub)" <gi...@apache.org> on 2026/04/22 10:37:25 UTC
[PR] [FLINK-39511][flink-autoscaler] FLIP-514: Custom Evaluator plugin for Flink Autoscaler [flink-kubernetes-operator]
Dennis-Mircea opened a new pull request, #1099:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1099
<!--
*Thank you very much for contributing to the Apache Flink Kubernetes Operator - we are happy that you want to help us improve the project. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix][docs] Fix typo in event time introduction` or `[hotfix][javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can read more on how we use GitHub Actions for CI [here](https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-main/docs/development/guide/#cicd).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
This PR implements the [FLIP-514](https://cwiki.apache.org/confluence/display/FLINK/FLIP-514%3A+Custom+Evaluator+plugin+for+Flink+Autoscaler) and is a continuation of the work done as part of https://github.com/apache/flink-kubernetes-operator/pull/953 PR.
It introduces a pluggable `FlinkAutoscalerEvaluator` SPI that lets users provide custom scaling-metric evaluation logic on top of the metrics evaluated internally by the autoscaler, both in the Flink Kubernetes Operator (via the Flink Plugins mechanism) and in `flink-autoscaler-standalone` (via the standard Java `ServiceLoader`). The evaluator returned map is merged on top of the internally-evaluated metrics for each job vertex, so users can override or augment specific `ScalingMetric` values (e.g. `TARGET_DATA_RATE`, `TRUE_PROCESSING_RATE`, `CATCH_UP_DATA_RATE`) without forking the autoscaler.
## Brief change log
- Introduced the `FlinkAutoscalerEvaluator` SPI (`flink-autoscaler`) with `getName()`, `evaluateVertexMetrics(vertex, evaluatedMetrics, Context)` and a `Context` exposing an un-modifiable view of `jobConf`, `metricsHistory`, previously evaluated vertex metrics, `topology`, `processingBacklog`, `restartTime` and the evaluator-specific `customEvaluatorConf`.
- Added the `job.autoscaler.metrics.custom-evaluator.name` config option and `AutoScalerOptions#forCustomEvaluator(conf, name)` helper, which builds a `DelegatingConfiguration` scoped to `job.autoscaler.metrics.custom-evaluator.<name>.` for the evaluator.
- Wired custom-evaluator resolution into `JobAutoScalerImpl#getCustomEvaluatorIfRequired` and passed the resolved `Tuple2<FlinkAutoscalerEvaluator, Configuration>` down to `ScalingMetricEvaluator`, which now calls the evaluator per vertex in topological order and merges the returned metrics on top of the internally evaluated ones.
- Added operator-side discovery in `org.apache.flink.kubernetes.operator.autoscaler.AutoscalerUtils#discoverCustomEvaluators(Configuration)` using the Flink `PluginManager` (`/opt/flink/plugins`). The discovered evaluators are injected into `JobAutoScalerImpl` from `FlinkOperator`.
- Added standalone-side discovery in `org.apache.flink.autoscaler.standalone.AutoscalerUtils#discoverCustomEvaluators()` using `ServiceLoader.load(FlinkAutoscalerEvaluator.class)` and wired it into `StandaloneAutoscalerEntrypoint#createJobAutoscaler`.
- Documented the new plugin in both language variants of `docs/.../operations/plugins.md` under a new *Custom Flink Autoscaler Evaluator* section, including a warning hint that a single custom evaluator is supported per pipeline today.
## Verifying this change
<!--
Please make sure both new and modified tests in this PR follows the conventions defined in our code quality guide: https://flink.apache.org/contributing/code-style-and-quality-common.html#testing
-->
This change added tests and can be verified as follows:
- **Evaluator resolution and override semantics** (`flink-autoscaler`):
- `JobAutoScalerImplTest#testGetCustomEvaluatorIfRequired` covers the registry lookup (configured name present / absent / unknown) and the per-evaluator `DelegatingConfiguration` built by `AutoScalerOptions.forCustomEvaluator`.
- `MetricsCollectionAndEvaluationTest` exercises end-to-end evaluation with a `TestCustomEvaluator` override on `TARGET_DATA_RATE` for source vertices and asserts that the custom value wins over the internally computed one while the rest of the metrics remain untouched.
- **Operator-side plugin discovery** (`flink-kubernetes-operator`):
- `TestingFlinkDeploymentController` / `FlinkOperator` wiring verified by existing controller tests loading the `TestCustomEvaluator` via `META-INF/services`.
- **Standalone-side SPI discovery** (`flink-autoscaler-standalone`):
- New `AutoscalerUtilsTest#testDiscoverCustomEvaluators` registers `TestCustomEvaluator` through `META-INF/services/org.apache.flink.autoscaler.metrics.FlinkAutoscalerEvaluator` and asserts that `AutoscalerUtils.discoverCustomEvaluators()` finds it keyed by `getName()`.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changes to the `CustomResourceDescriptors`: no
- Core observer or reconciler logic that is regularly executed: no
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? docs & JavaDocs
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39511] FLIP-514: Custom Evaluator plugin for Flink Autoscaler [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #1099:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1099#discussion_r3426650480
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java:
##########
@@ -588,4 +635,76 @@ protected static double computeEdgeDataRate(
to);
return getRate(ScalingMetric.NUM_RECORDS_OUT, from, metricsHistory);
}
+
+ /**
+ * Executes the provided custom metric evaluator for the given job vertex. Calls {@link
+ * FlinkAutoscalerEvaluator#evaluateVertexMetrics} to evaluate scaling metrics.
+ *
+ * @param vertex The job vertex being evaluated.
+ * @param evaluatedMetrics Current evaluated metrics.
+ * @param customEvaluationSession A tuple containing the custom metric evaluator and evaluation
+ * context.
+ * @return A map of scaling metrics, with its corresponding evaluated scaling metric.
+ */
+ @VisibleForTesting
+ protected static Map<ScalingMetric, EvaluatedScalingMetric> runCustomEvaluator(
+ JobVertexID vertex,
+ Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
+ Tuple2<FlinkAutoscalerEvaluator, FlinkAutoscalerEvaluator.Context>
+ customEvaluationSession) {
+ try {
+ return customEvaluationSession.f0.evaluateVertexMetrics(
+ vertex, evaluatedMetrics, customEvaluationSession.f1);
+ } catch (UnsupportedOperationException e) {
+ LOG.warn(
+ "Custom metric evaluator {} tried accessing an un-modifiable view.",
+ customEvaluationSession.f0.getClass(),
+ e);
+ } catch (Exception e) {
+ LOG.warn(
+ "Custom metric evaluator {} threw an exception.",
+ customEvaluationSession.f0.getClass(),
+ e);
+ }
+
+ return Collections.emptyMap();
+ }
+
+ /**
+ * Merges the incoming evaluated metrics into actual evaluated metrics.
+ *
+ * @param actual The target evaluated metrics map to merge into.
+ * @param incoming The incoming map containing new evaluated metrics map to be merged
+ * (nullable).
+ */
+ @VisibleForTesting
+ protected static void mergeEvaluatedMetricsMaps(
+ Map<ScalingMetric, EvaluatedScalingMetric> actual,
+ @Nullable Map<ScalingMetric, EvaluatedScalingMetric> incoming) {
+ Optional.ofNullable(incoming)
+ .ifPresent(
+ customEvaluatedMetric ->
+ customEvaluatedMetric.forEach(
+ (scalingMetric, evaluatedScalingMetric) ->
+ actual.merge(
+ scalingMetric,
+ evaluatedScalingMetric,
+ ScalingMetricEvaluator
+ ::mergeEvaluatedScalingMetric)));
+ }
+
+ /**
+ * Merges two {@link EvaluatedScalingMetric} instances.
+ *
+ * @param actual The existing evaluated scaling metric.
+ * @param incoming The incoming evaluated scaling metric.
+ * @return A new {@link EvaluatedScalingMetric} instance with merged values.
+ */
Review Comment:
the comment is unnecessary I mean :D
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java:
##########
@@ -588,4 +635,76 @@ protected static double computeEdgeDataRate(
to);
return getRate(ScalingMetric.NUM_RECORDS_OUT, from, metricsHistory);
}
+
+ /**
+ * Executes the provided custom metric evaluator for the given job vertex. Calls {@link
+ * FlinkAutoscalerEvaluator#evaluateVertexMetrics} to evaluate scaling metrics.
+ *
+ * @param vertex The job vertex being evaluated.
+ * @param evaluatedMetrics Current evaluated metrics.
+ * @param customEvaluationSession A tuple containing the custom metric evaluator and evaluation
+ * context.
+ * @return A map of scaling metrics, with its corresponding evaluated scaling metric.
+ */
+ @VisibleForTesting
+ protected static Map<ScalingMetric, EvaluatedScalingMetric> runCustomEvaluator(
+ JobVertexID vertex,
+ Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
+ Tuple2<FlinkAutoscalerEvaluator, FlinkAutoscalerEvaluator.Context>
+ customEvaluationSession) {
+ try {
+ return customEvaluationSession.f0.evaluateVertexMetrics(
+ vertex, evaluatedMetrics, customEvaluationSession.f1);
+ } catch (UnsupportedOperationException e) {
+ LOG.warn(
+ "Custom metric evaluator {} tried accessing an un-modifiable view.",
+ customEvaluationSession.f0.getClass(),
+ e);
+ } catch (Exception e) {
+ LOG.warn(
+ "Custom metric evaluator {} threw an exception.",
+ customEvaluationSession.f0.getClass(),
+ e);
+ }
+
+ return Collections.emptyMap();
+ }
+
+ /**
+ * Merges the incoming evaluated metrics into actual evaluated metrics.
+ *
+ * @param actual The target evaluated metrics map to merge into.
+ * @param incoming The incoming map containing new evaluated metrics map to be merged
+ * (nullable).
+ */
Review Comment:
the comment is unnecessary I mean :D
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39511] FLIP-514: Custom Evaluator plugin for Flink Autoscaler [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on PR #1099:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1099#issuecomment-4324488395
> **Question & Follow-up: single-evaluator limitation vs. config namespace shape**
>
> As it stands, `JobAutoScalerImpl#getCustomEvaluatorIfRequired` reads a single `ConfigOption<String> CUSTOM_EVALUATOR_NAME` (`job.autoscaler.metrics.custom-evaluator.name`), looks up one `FlinkAutoscalerEvaluator` from the registry (keyed by `FlinkAutoscalerEvaluator#getName()`), and `ScalingMetricEvaluator` only knows how to invoke one evaluator per evaluation cycle. So at runtime **only one custom evaluator can be active per pipeline**, even though several may be registered via `META-INF/services`.
>
> At the same time, `AutoScalerOptions.forCustomEvaluator(conf, name)` builds the evaluator-specific view as a `DelegatingConfiguration` over `job.autoscaler.metrics.custom-evaluator.<name>.`, which means users have to write keys like:
>
> ```yaml
> job.autoscaler.metrics.custom-evaluator.name: custom-evaluator
> job.autoscaler.metrics.custom-evaluator.custom-evaluator.target-data-rate: 100000.0
> ```
>
> That repeated `<name>` segment is future-proofing for a multi-evaluator world, but today it's pure ceremony: there is never more than one evaluator reading from that namespace. It's confusing to document, easy to mis-type, and likely to trip users up.
>
> I see two coherent directions and would like the maintainers' preference before pushing further commits:
>
> **Option A - keep the single-evaluator contract, simplify the namespace.**
>
> Drop the `<name>` segment from `forCustomEvaluator` so `Context#getCustomEvaluatorConf()` is a delegating view over `job.autoscaler.metrics.custom-evaluator.` directly. To avoid the `name` key leaking into that view the selector would move out of that prefix (e.g. renamed to `job.autoscaler.metrics.custom-evaluator-name`, with a fallback key for the old one). User-facing config becomes:
>
> ```yaml
> job.autoscaler.metrics.custom-evaluator-name: custom-evaluator
> job.autoscaler.metrics.custom-evaluator.target-data-rate: 100000.0
> ```
>
> The **Option A** has the following pro & cons:
>
> * Pros: cleanest UX, matches the actual runtime semantics.
> * Cons: represents a big limitation in the custom evaluator flexibility.
>
> **Option B - lift the limitation, align the config shape with Flink metric reporters and with [FLIP-514](https://cwiki.apache.org/confluence/display/FLINK/FLIP-514).**
>
> Flink's metric-reporter config is the established idiom for "list of named instances, each with its own class and its own bag of options":
>
> ```ini
> metrics.reporters: my_jmx_reporter,my_other_reporter
>
> metrics.reporter.my_jmx_reporter.factory.class: org.apache.flink.metrics.jmx.JMXReporterFactory
> metrics.reporter.my_jmx_reporter.port: 9020-9040
>
> metrics.reporter.my_other_reporter.factory.class: org.apache.flink.metrics.graphite.GraphiteReporterFactory
> metrics.reporter.my_other_reporter.host: 192.168.1.1
> metrics.reporter.my_other_reporter.port: 10000
> ```
>
> Applied to custom evaluators, the mapping is:
>
> Reporter concept Evaluator equivalent
> `metrics.reporters: r1,r2` `job.autoscaler.metrics.custom-evaluators: custom-evaluator-1,custom-evaluator-2`
> `metrics.reporter.<r>.factory.class: FQN` `job.autoscaler.metrics.custom-evaluator.<instance>.class: FQN`
> `metrics.reporter.<r>.<opt>: v` `job.autoscaler.metrics.custom-evaluator.<instance>.<opt>: v`
> Concretely:
>
> ```yaml
> job.autoscaler.metrics.custom-evaluators: custom-evaluator-1,custom-evaluator-2
>
> job.autoscaler.metrics.custom-evaluator.custom-evaluator-1.class: org.apache.flink.autoscaler.custom.MyFirstEvaluator
> job.autoscaler.metrics.custom-evaluator.custom-evaluator-1.target-data-rate: 100000.0
>
> job.autoscaler.metrics.custom-evaluator.custom-evaluator-2.class: org.apache.flink.autoscaler.custom.MySecondEvaluator
> job.autoscaler.metrics.custom-evaluator.custom-evaluator-2.some-threshold: 0.8
> ```
>
> The consequence is that the `getName()` goes away from the SPI. The instance name (`custom-evaluator-1`) is purely a configuration-level handle. Evaluator resolution stops going through the `META-INF/services` -> `getName()` registry and instead becomes: for each entry in `custom-evaluators`, read its `...<instance>.class` property and instantiate the fully qualified name (via `Class#forName`, or via a factory SPI if we want DI/customization, analogous to `*ReporterFactory`). This is exactly the shape [FLIP-514](https://cwiki.apache.org/confluence/display/FLINK/FLIP-514) proposes in its [Job Level Config For Custom Evaluator](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=345377351#FLIP514:CustomEvaluatorpluginforFlinkAutoscaler-JobLevelConfigForCustomEvaluator) section, and it removes the oddity where two evaluators couldn't share the same `getName()` even though they were distinct implementations.
>
> To keep the ordering contract between multiple custom evaluators explicit and self-describing rather than implicit in the declaration order of `job.autoscaler.metrics.custom-evaluators`, we can adopt the same design already approved in [FLIP-575: Scaling Executor Plugin SPI for Flink Autoscaler](https://cwiki.apache.org/confluence/display/FLINK/FLIP-575%3A+Scaling+Executor+Plugin+SPI+for+Flink+Autoscaler) and implemented for [ScalingExecutorPlugin.java](https://github.com/apache/flink-kubernetes-operator/pull/1085/changes#diff-8bc87f00339ad429c6d66b8b57d0d7ec8358e5e0d95f542b2f7d2a1a19271769): a `default int priority()` method on the SPI, with evaluators executed in ascending priority and the returned metrics merged left-to-right (lower priority first, higher priority wins on conflicts). This gives us:
>
> * a single, well-defined ordering contract that doesn't depend on the user's config-key ordering or on `ServiceLoader/PluginManager` discovery order,
> * parity and composability with `ScalingExecutorPlugin`, so both autoscaler extension points share the same mental model
> * a natural extension path for Option B above as `priority()` simply becomes the tie-breaker once we allow more than one evaluator per pipeline.
>
> The **Option B** has the following pro & cons:
>
> * Pros: consistent with an idiom Flink users already know, aligns with FLIP-514, expressive (multiple evaluators per job), and drops the `getName()` coupling that is awkward today.
> * Cons: bigger change than A.
>
> My preference would be to land **Option B** so we match FLIP-514 directly and don't ship a config surface we know we'll have to migrate away from.
>
> Thoughts?
@Dennis-Mircea let's go for option 2 but to limit the change/complexity we can also start with supporting only a single evaluator plugin and add the priority / ordering logic as a follow up jira
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39511][flink-autoscaler] FLIP-514: Custom Evaluator plugin for Flink Autoscaler [flink-kubernetes-operator]
Posted by "Dennis-Mircea (via GitHub)" <gi...@apache.org>.
Dennis-Mircea commented on PR #1099:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1099#issuecomment-4295606059
**Question & Follow-up: single-evaluator limitation vs. config namespace shape**
As it stands, `JobAutoScalerImpl#getCustomEvaluatorIfRequired` reads a single `ConfigOption<String> CUSTOM_EVALUATOR_NAME` (`job.autoscaler.metrics.custom-evaluator.name`), looks up one `FlinkAutoscalerEvaluator` from the registry (keyed by `FlinkAutoscalerEvaluator#getName()`), and `ScalingMetricEvaluator` only knows how to invoke one evaluator per evaluation cycle. So at runtime **only one custom evaluator can be active per pipeline**, even though several may be registered via `META-INF/services`.
At the same time, `AutoScalerOptions.forCustomEvaluator(conf, name)` builds the evaluator-specific view as a `DelegatingConfiguration` over `job.autoscaler.metrics.custom-evaluator.<name>.`, which means users have to write keys like:
```yaml
job.autoscaler.metrics.custom-evaluator.name: custom-evaluator
job.autoscaler.metrics.custom-evaluator.custom-evaluator.target-data-rate: 100000.0
```
That repeated `<name>` segment is future-proofing for a multi-evaluator world, but today it's pure ceremony: there is never more than one evaluator reading from that namespace. It's confusing to document, easy to mis-type, and likely to trip users up.
I see two coherent directions and would like the maintainers' preference before pushing further commits:
**Option A - keep the single-evaluator contract, simplify the namespace.**
Drop the `<name>` segment from `forCustomEvaluator` so `Context#getCustomEvaluatorConf()` is a delegating view over `job.autoscaler.metrics.custom-evaluator.` directly. To avoid the `name` key leaking into that view the selector would move out of that prefix (e.g. renamed to `job.autoscaler.metrics.custom-evaluator-name`, with a fallback key for the old one). User-facing config becomes:
```yaml
job.autoscaler.metrics.custom-evaluator-name: custom-evaluator
job.autoscaler.metrics.custom-evaluator.target-data-rate: 100000.0
```
The **Option A** has the following pro & cons:
- Pros: cleanest UX, matches the actual runtime semantics.
- Cons: represents a big limitation in the custom evaluator flexibility.
**Option B - lift the limitation, align the config shape with Flink metric reporters and with [FLIP-514](https://cwiki.apache.org/confluence/display/FLINK/FLIP-514).**
Flink's metric-reporter config is the established idiom for "list of named instances, each with its own class and its own bag of options":
```properties
metrics.reporters: my_jmx_reporter,my_other_reporter
metrics.reporter.my_jmx_reporter.factory.class: org.apache.flink.metrics.jmx.JMXReporterFactory
metrics.reporter.my_jmx_reporter.port: 9020-9040
metrics.reporter.my_other_reporter.factory.class: org.apache.flink.metrics.graphite.GraphiteReporterFactory
metrics.reporter.my_other_reporter.host: 192.168.1.1
metrics.reporter.my_other_reporter.port: 10000
```
Applied to custom evaluators, the mapping is:
| Reporter concept | Evaluator equivalent |
| ------------------------------------------- | ------------------------------------------------------------------------------------------- |
| `metrics.reporters: r1,r2` | `job.autoscaler.metrics.custom-evaluators: custom-evaluator-1,custom-evaluator-2` |
| `metrics.reporter.<r>.factory.class: FQN` | `job.autoscaler.metrics.custom-evaluator.<instance>.class: FQN` |
| `metrics.reporter.<r>.<opt>: v` | `job.autoscaler.metrics.custom-evaluator.<instance>.<opt>: v` |
Concretely:
```yaml
job.autoscaler.metrics.custom-evaluators: custom-evaluator-1,custom-evaluator-2
job.autoscaler.metrics.custom-evaluator.custom-evaluator-1.class: org.apache.flink.autoscaler.custom.MyFirstEvaluator
job.autoscaler.metrics.custom-evaluator.custom-evaluator-1.target-data-rate: 100000.0
job.autoscaler.metrics.custom-evaluator.custom-evaluator-2.class: org.apache.flink.autoscaler.custom.MySecondEvaluator
job.autoscaler.metrics.custom-evaluator.custom-evaluator-2.some-threshold: 0.8
```
The consequence is that the `getName()` goes away from the SPI. The instance name (`custom-evaluator-1`) is purely a configuration-level handle. Evaluator resolution stops going through the `META-INF/services` -> `getName()` registry and instead becomes: for each entry in `custom-evaluators`, read its `...<instance>.class` property and instantiate the fully qualified name (via `Class#forName`, or via a factory SPI if we want DI/customization, analogous to `*ReporterFactory`). This is exactly the shape [FLIP-514](https://cwiki.apache.org/confluence/display/FLINK/FLIP-514) proposes in its [Job Level Config For Custom Evaluator](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=345377351#FLIP514:CustomEvaluatorpluginforFlinkAutoscaler-JobLevelConfigForCustomEvaluator) section, and it removes the oddity where two evaluators couldn't share the same `getName()` even though they were distinct implementations.
To keep the ordering contract between multiple custom evaluators explicit and self-describing rather than implicit in the declaration order of `job.autoscaler.metrics.custom-evaluators`, we can adopt the same design already approved in [FLIP-575: Scaling Executor Plugin SPI for Flink Autoscaler](https://cwiki.apache.org/confluence/display/FLINK/FLIP-575%3A+Scaling+Executor+Plugin+SPI+for+Flink+Autoscaler) and implemented for [ScalingExecutorPlugin.java](https://github.com/apache/flink-kubernetes-operator/pull/1085/changes#diff-8bc87f00339ad429c6d66b8b57d0d7ec8358e5e0d95f542b2f7d2a1a19271769): a `default int priority()` method on the SPI, with evaluators executed in ascending priority and the returned metrics merged left-to-right (lower priority first, higher priority wins on conflicts). This gives us:
- a single, well-defined ordering contract that doesn't depend on the user's config-key ordering or on `ServiceLoader/PluginManager` discovery order,
- parity and composability with `ScalingExecutorPlugin`, so both autoscaler extension points share the same mental model
- a natural extension path for Option B above as `priority()` simply becomes the tie-breaker once we allow more than one evaluator per pipeline.
The **Option B** has the following pro & cons:
- Pros: consistent with an idiom Flink users already know, aligns with FLIP-514, expressive (multiple evaluators per job), and drops the `getName()` coupling that is awkward today.
- Cons: bigger change than A.
My preference would be to land **Option B** so we match FLIP-514 directly and don't ship a config surface we know we'll have to migrate away from.
Thoughts?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39511] FLIP-514: Custom Evaluator plugin for Flink Autoscaler [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #1099:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1099#discussion_r3426650480
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java:
##########
@@ -588,4 +635,76 @@ protected static double computeEdgeDataRate(
to);
return getRate(ScalingMetric.NUM_RECORDS_OUT, from, metricsHistory);
}
+
+ /**
+ * Executes the provided custom metric evaluator for the given job vertex. Calls {@link
+ * FlinkAutoscalerEvaluator#evaluateVertexMetrics} to evaluate scaling metrics.
+ *
+ * @param vertex The job vertex being evaluated.
+ * @param evaluatedMetrics Current evaluated metrics.
+ * @param customEvaluationSession A tuple containing the custom metric evaluator and evaluation
+ * context.
+ * @return A map of scaling metrics, with its corresponding evaluated scaling metric.
+ */
+ @VisibleForTesting
+ protected static Map<ScalingMetric, EvaluatedScalingMetric> runCustomEvaluator(
+ JobVertexID vertex,
+ Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
+ Tuple2<FlinkAutoscalerEvaluator, FlinkAutoscalerEvaluator.Context>
+ customEvaluationSession) {
+ try {
+ return customEvaluationSession.f0.evaluateVertexMetrics(
+ vertex, evaluatedMetrics, customEvaluationSession.f1);
+ } catch (UnsupportedOperationException e) {
+ LOG.warn(
+ "Custom metric evaluator {} tried accessing an un-modifiable view.",
+ customEvaluationSession.f0.getClass(),
+ e);
+ } catch (Exception e) {
+ LOG.warn(
+ "Custom metric evaluator {} threw an exception.",
+ customEvaluationSession.f0.getClass(),
+ e);
+ }
+
+ return Collections.emptyMap();
+ }
+
+ /**
+ * Merges the incoming evaluated metrics into actual evaluated metrics.
+ *
+ * @param actual The target evaluated metrics map to merge into.
+ * @param incoming The incoming map containing new evaluated metrics map to be merged
+ * (nullable).
+ */
+ @VisibleForTesting
+ protected static void mergeEvaluatedMetricsMaps(
+ Map<ScalingMetric, EvaluatedScalingMetric> actual,
+ @Nullable Map<ScalingMetric, EvaluatedScalingMetric> incoming) {
+ Optional.ofNullable(incoming)
+ .ifPresent(
+ customEvaluatedMetric ->
+ customEvaluatedMetric.forEach(
+ (scalingMetric, evaluatedScalingMetric) ->
+ actual.merge(
+ scalingMetric,
+ evaluatedScalingMetric,
+ ScalingMetricEvaluator
+ ::mergeEvaluatedScalingMetric)));
+ }
+
+ /**
+ * Merges two {@link EvaluatedScalingMetric} instances.
+ *
+ * @param actual The existing evaluated scaling metric.
+ * @param incoming The incoming evaluated scaling metric.
+ * @return A new {@link EvaluatedScalingMetric} instance with merged values.
+ */
Review Comment:
the comment/javadoc is unnecessary I mean :D
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java:
##########
@@ -588,4 +635,76 @@ protected static double computeEdgeDataRate(
to);
return getRate(ScalingMetric.NUM_RECORDS_OUT, from, metricsHistory);
}
+
+ /**
+ * Executes the provided custom metric evaluator for the given job vertex. Calls {@link
+ * FlinkAutoscalerEvaluator#evaluateVertexMetrics} to evaluate scaling metrics.
+ *
+ * @param vertex The job vertex being evaluated.
+ * @param evaluatedMetrics Current evaluated metrics.
+ * @param customEvaluationSession A tuple containing the custom metric evaluator and evaluation
+ * context.
+ * @return A map of scaling metrics, with its corresponding evaluated scaling metric.
+ */
+ @VisibleForTesting
+ protected static Map<ScalingMetric, EvaluatedScalingMetric> runCustomEvaluator(
+ JobVertexID vertex,
+ Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
+ Tuple2<FlinkAutoscalerEvaluator, FlinkAutoscalerEvaluator.Context>
+ customEvaluationSession) {
+ try {
+ return customEvaluationSession.f0.evaluateVertexMetrics(
+ vertex, evaluatedMetrics, customEvaluationSession.f1);
+ } catch (UnsupportedOperationException e) {
+ LOG.warn(
+ "Custom metric evaluator {} tried accessing an un-modifiable view.",
+ customEvaluationSession.f0.getClass(),
+ e);
+ } catch (Exception e) {
+ LOG.warn(
+ "Custom metric evaluator {} threw an exception.",
+ customEvaluationSession.f0.getClass(),
+ e);
+ }
+
+ return Collections.emptyMap();
+ }
+
+ /**
+ * Merges the incoming evaluated metrics into actual evaluated metrics.
+ *
+ * @param actual The target evaluated metrics map to merge into.
+ * @param incoming The incoming map containing new evaluated metrics map to be merged
+ * (nullable).
+ */
Review Comment:
the comment/javadoc is unnecessary I mean :D
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39511] FLIP-514: Custom Evaluator plugin for Flink Autoscaler [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora merged PR #1099:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1099
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39511] FLIP-514: Custom Evaluator plugin for Flink Autoscaler [flink-kubernetes-operator]
Posted by "Dennis-Mircea (via GitHub)" <gi...@apache.org>.
Dennis-Mircea commented on code in PR #1099:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1099#discussion_r3426487262
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java:
##########
@@ -588,4 +635,76 @@ protected static double computeEdgeDataRate(
to);
return getRate(ScalingMetric.NUM_RECORDS_OUT, from, metricsHistory);
}
+
+ /**
+ * Executes the provided custom metric evaluator for the given job vertex. Calls {@link
+ * FlinkAutoscalerEvaluator#evaluateVertexMetrics} to evaluate scaling metrics.
+ *
+ * @param vertex The job vertex being evaluated.
+ * @param evaluatedMetrics Current evaluated metrics.
+ * @param customEvaluationSession A tuple containing the custom metric evaluator and evaluation
+ * context.
+ * @return A map of scaling metrics, with its corresponding evaluated scaling metric.
+ */
+ @VisibleForTesting
+ protected static Map<ScalingMetric, EvaluatedScalingMetric> runCustomEvaluator(
+ JobVertexID vertex,
+ Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
+ Tuple2<FlinkAutoscalerEvaluator, FlinkAutoscalerEvaluator.Context>
+ customEvaluationSession) {
+ try {
+ return customEvaluationSession.f0.evaluateVertexMetrics(
+ vertex, evaluatedMetrics, customEvaluationSession.f1);
+ } catch (UnsupportedOperationException e) {
+ LOG.warn(
+ "Custom metric evaluator {} tried accessing an un-modifiable view.",
+ customEvaluationSession.f0.getClass(),
+ e);
+ } catch (Exception e) {
+ LOG.warn(
+ "Custom metric evaluator {} threw an exception.",
+ customEvaluationSession.f0.getClass(),
+ e);
+ }
+
+ return Collections.emptyMap();
+ }
+
+ /**
+ * Merges the incoming evaluated metrics into actual evaluated metrics.
+ *
+ * @param actual The target evaluated metrics map to merge into.
+ * @param incoming The incoming map containing new evaluated metrics map to be merged
+ * (nullable).
+ */
+ @VisibleForTesting
+ protected static void mergeEvaluatedMetricsMaps(
+ Map<ScalingMetric, EvaluatedScalingMetric> actual,
+ @Nullable Map<ScalingMetric, EvaluatedScalingMetric> incoming) {
+ Optional.ofNullable(incoming)
+ .ifPresent(
+ customEvaluatedMetric ->
+ customEvaluatedMetric.forEach(
+ (scalingMetric, evaluatedScalingMetric) ->
+ actual.merge(
+ scalingMetric,
+ evaluatedScalingMetric,
+ ScalingMetricEvaluator
+ ::mergeEvaluatedScalingMetric)));
+ }
+
+ /**
+ * Merges two {@link EvaluatedScalingMetric} instances.
+ *
+ * @param actual The existing evaluated scaling metric.
+ * @param incoming The incoming evaluated scaling metric.
+ * @return A new {@link EvaluatedScalingMetric} instance with merged values.
+ */
Review Comment:
These two are the actual override step, so I'd keep them. The custom evaluator returns only the metrics it wants to change, and `mergeEvaluatedScalingMetric` does a NaN-aware, per-field merge (`current`/`average`) of the returned `EvaluatedScalingMetric` onto the internally computed one, while `mergeEvaluatedMetricsMaps` applies that across the returned map. A plain `putAll` would force evaluators to re-supply every field (and both `current` and `average`) just to override one, and would break internal values with NaN where they only meant to set one side. This is what makes "augment/override specific `ScalingMetric` values" work as documented.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39511] FLIP-514: Custom Evaluator plugin for Flink Autoscaler [flink-kubernetes-operator]
Posted by "Dennis-Mircea (via GitHub)" <gi...@apache.org>.
Dennis-Mircea commented on code in PR #1099:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1099#discussion_r3427233583
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java:
##########
@@ -588,4 +635,76 @@ protected static double computeEdgeDataRate(
to);
return getRate(ScalingMetric.NUM_RECORDS_OUT, from, metricsHistory);
}
+
+ /**
+ * Executes the provided custom metric evaluator for the given job vertex. Calls {@link
+ * FlinkAutoscalerEvaluator#evaluateVertexMetrics} to evaluate scaling metrics.
+ *
+ * @param vertex The job vertex being evaluated.
+ * @param evaluatedMetrics Current evaluated metrics.
+ * @param customEvaluationSession A tuple containing the custom metric evaluator and evaluation
+ * context.
+ * @return A map of scaling metrics, with its corresponding evaluated scaling metric.
+ */
+ @VisibleForTesting
+ protected static Map<ScalingMetric, EvaluatedScalingMetric> runCustomEvaluator(
+ JobVertexID vertex,
+ Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
+ Tuple2<FlinkAutoscalerEvaluator, FlinkAutoscalerEvaluator.Context>
+ customEvaluationSession) {
+ try {
+ return customEvaluationSession.f0.evaluateVertexMetrics(
+ vertex, evaluatedMetrics, customEvaluationSession.f1);
+ } catch (UnsupportedOperationException e) {
+ LOG.warn(
+ "Custom metric evaluator {} tried accessing an un-modifiable view.",
+ customEvaluationSession.f0.getClass(),
+ e);
+ } catch (Exception e) {
+ LOG.warn(
+ "Custom metric evaluator {} threw an exception.",
+ customEvaluationSession.f0.getClass(),
+ e);
+ }
+
+ return Collections.emptyMap();
+ }
+
+ /**
+ * Merges the incoming evaluated metrics into actual evaluated metrics.
+ *
+ * @param actual The target evaluated metrics map to merge into.
+ * @param incoming The incoming map containing new evaluated metrics map to be merged
+ * (nullable).
+ */
Review Comment:
Fixed!
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java:
##########
@@ -588,4 +635,76 @@ protected static double computeEdgeDataRate(
to);
return getRate(ScalingMetric.NUM_RECORDS_OUT, from, metricsHistory);
}
+
+ /**
+ * Executes the provided custom metric evaluator for the given job vertex. Calls {@link
+ * FlinkAutoscalerEvaluator#evaluateVertexMetrics} to evaluate scaling metrics.
+ *
+ * @param vertex The job vertex being evaluated.
+ * @param evaluatedMetrics Current evaluated metrics.
+ * @param customEvaluationSession A tuple containing the custom metric evaluator and evaluation
+ * context.
+ * @return A map of scaling metrics, with its corresponding evaluated scaling metric.
+ */
+ @VisibleForTesting
+ protected static Map<ScalingMetric, EvaluatedScalingMetric> runCustomEvaluator(
+ JobVertexID vertex,
+ Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
+ Tuple2<FlinkAutoscalerEvaluator, FlinkAutoscalerEvaluator.Context>
+ customEvaluationSession) {
+ try {
+ return customEvaluationSession.f0.evaluateVertexMetrics(
+ vertex, evaluatedMetrics, customEvaluationSession.f1);
+ } catch (UnsupportedOperationException e) {
+ LOG.warn(
+ "Custom metric evaluator {} tried accessing an un-modifiable view.",
+ customEvaluationSession.f0.getClass(),
+ e);
+ } catch (Exception e) {
+ LOG.warn(
+ "Custom metric evaluator {} threw an exception.",
+ customEvaluationSession.f0.getClass(),
+ e);
+ }
+
+ return Collections.emptyMap();
+ }
+
+ /**
+ * Merges the incoming evaluated metrics into actual evaluated metrics.
+ *
+ * @param actual The target evaluated metrics map to merge into.
+ * @param incoming The incoming map containing new evaluated metrics map to be merged
+ * (nullable).
+ */
+ @VisibleForTesting
+ protected static void mergeEvaluatedMetricsMaps(
+ Map<ScalingMetric, EvaluatedScalingMetric> actual,
+ @Nullable Map<ScalingMetric, EvaluatedScalingMetric> incoming) {
+ Optional.ofNullable(incoming)
+ .ifPresent(
+ customEvaluatedMetric ->
+ customEvaluatedMetric.forEach(
+ (scalingMetric, evaluatedScalingMetric) ->
+ actual.merge(
+ scalingMetric,
+ evaluatedScalingMetric,
+ ScalingMetricEvaluator
+ ::mergeEvaluatedScalingMetric)));
+ }
+
+ /**
+ * Merges two {@link EvaluatedScalingMetric} instances.
+ *
+ * @param actual The existing evaluated scaling metric.
+ * @param incoming The incoming evaluated scaling metric.
+ * @return A new {@link EvaluatedScalingMetric} instance with merged values.
+ */
Review Comment:
Fixed!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39511] FLIP-514: Custom Evaluator plugin for Flink Autoscaler [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #1099:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1099#discussion_r3426646628
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/FlinkAutoscalerEvaluator.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.runtime.jobgraph.JobVertexID;
+
+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.FlinkAutoscalerEvaluator}.
+ */
+public interface FlinkAutoscalerEvaluator {
+
+ /**
+ * 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.
+ */
+ class Context {
+ private final Configuration jobConf;
Review Comment:
We can merge it without the prefix on top of the original (which was what I had in mind), I really don't see the value in having 2 configs. To me this is unnecessary complexity.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39511] FLIP-514: Custom Evaluator plugin for Flink Autoscaler [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #1099:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1099#discussion_r3427542071
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/FlinkAutoscalerEvaluator.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.runtime.jobgraph.JobVertexID;
+
+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.FlinkAutoscalerEvaluator}.
+ */
+public interface FlinkAutoscalerEvaluator {
+
+ /**
+ * 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.
+ */
+ class Context {
+ private final Configuration jobConf;
Review Comment:
I think this is perfect and exactly how it's expected to work
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39511] FLIP-514: Custom Evaluator plugin for Flink Autoscaler [flink-kubernetes-operator]
Posted by "Dennis-Mircea (via GitHub)" <gi...@apache.org>.
Dennis-Mircea commented on PR #1099:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1099#issuecomment-4587088668
Coming back to this after stepping back to look at the broader module structure. Wanted to raise a design question before merge, because it affects what custom-plugin authors will compile against. This applies to #1085 PR as well.
### The asymmetry today
The operator's plugin SPIs are split across two modules:
| Module | SPIs it carries |
|---|---|
| `flink-kubernetes-operator-api` (thin) | `FlinkResourceListener` |
| `flink-kubernetes-operator` (implementation, heavy) | `FlinkResourceValidator`, `FlinkResourceMutator` |
A custom-validator or custom-mutator author has to depend on the heavy implementation module (which transitively pulls JOSDK, fabric8, the autoscaler, etc.) just to compile against one interface. A custom-listener author depends on a much thinner `-api` artifact. The asymmetry is historical, not designed.
### The question
This PR adds new plugin-style SPIs to `flink-autoscaler`, which is the implementation module. Would it be cleaner to land them in a dedicated `flink-autoscaler-api` module instead, so the user-facing answer to "what do I import to write an autoscaler plugin?" is a single thin artifact?
The plugin author's first question when implementing a new SPI is always "which module do I import?". Today that answer is inconsistent across SPIs and points at heavy implementation artifacts in most cases. Landing these new SPIs in a thin `flink-autoscaler-api` would set a cleaner precedent.
### Why now, not after merge
Once the new SPIs are in `flink-autoscaler`, the project implicitly commits to "autoscaler plugin SPIs live in the implementation module" as the pattern. A later restructuring becomes a multi-PR migration that breaks plugin-author coordinates a second time. The cost of asking the question now is a slight delay on this PR. The cost of asking it later is significantly higher.
### The broader implication
If we want to standardize on "plugin SPIs live in `-api` modules" going forward, that also implies:
- Moving `FlinkResourceValidator` and `FlinkResourceMutator` into `flink-kubernetes-operator-api` to restore symmetry with `FlinkResourceListener`.
- Creating `flink-autoscaler-api` carrying both the existing `JobAutoScaler` / `ScalingRealizer` / `AutoScalerStateStore` and the new SPIs from this PR.
That's a project-wide structural decision and would warrant a `[DISCUSS]` thread on dev@flink.apache.org before all of it lands. I do not want to block this PR with that broader conversation, but the immediate question (where the new SPIs in this PR should live) shapes the email and is worth aligning on first.
@gyfora, what do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39511] FLIP-514: Custom Evaluator plugin for Flink Autoscaler [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #1099:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1099#discussion_r3425465811
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/FlinkAutoscalerEvaluator.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.runtime.jobgraph.JobVertexID;
+
+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.FlinkAutoscalerEvaluator}.
+ */
+public interface FlinkAutoscalerEvaluator {
+
+ /**
+ * 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.
+ */
+ class Context {
Review Comment:
Use `@Value` to reduce boilerplate
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java:
##########
@@ -72,14 +76,34 @@ public class ScalingMetricEvaluator {
private static final Logger LOG = LoggerFactory.getLogger(ScalingMetricEvaluator.class);
public EvaluatedMetrics evaluate(
- Configuration conf, CollectedMetricHistory collectedMetrics, Duration restartTime) {
+ Configuration conf,
+ CollectedMetricHistory collectedMetrics,
+ Duration restartTime,
+ @Nullable Tuple2<FlinkAutoscalerEvaluator, Configuration> customEvaluatorWithConfig) {
Review Comment:
As I said in another comment, maybe we can do this once during initialization? Unless we expect this to change and be reconfigured at runtime
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java:
##########
@@ -588,4 +635,76 @@ protected static double computeEdgeDataRate(
to);
return getRate(ScalingMetric.NUM_RECORDS_OUT, from, metricsHistory);
}
+
+ /**
+ * Executes the provided custom metric evaluator for the given job vertex. Calls {@link
+ * FlinkAutoscalerEvaluator#evaluateVertexMetrics} to evaluate scaling metrics.
+ *
+ * @param vertex The job vertex being evaluated.
+ * @param evaluatedMetrics Current evaluated metrics.
+ * @param customEvaluationSession A tuple containing the custom metric evaluator and evaluation
+ * context.
+ * @return A map of scaling metrics, with its corresponding evaluated scaling metric.
+ */
+ @VisibleForTesting
+ protected static Map<ScalingMetric, EvaluatedScalingMetric> runCustomEvaluator(
+ JobVertexID vertex,
+ Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
+ Tuple2<FlinkAutoscalerEvaluator, FlinkAutoscalerEvaluator.Context>
+ customEvaluationSession) {
+ try {
+ return customEvaluationSession.f0.evaluateVertexMetrics(
+ vertex, evaluatedMetrics, customEvaluationSession.f1);
+ } catch (UnsupportedOperationException e) {
+ LOG.warn(
+ "Custom metric evaluator {} tried accessing an un-modifiable view.",
+ customEvaluationSession.f0.getClass(),
+ e);
+ } catch (Exception e) {
+ LOG.warn(
+ "Custom metric evaluator {} threw an exception.",
+ customEvaluationSession.f0.getClass(),
+ e);
+ }
+
+ return Collections.emptyMap();
+ }
+
+ /**
+ * Merges the incoming evaluated metrics into actual evaluated metrics.
+ *
+ * @param actual The target evaluated metrics map to merge into.
+ * @param incoming The incoming map containing new evaluated metrics map to be merged
+ * (nullable).
+ */
+ @VisibleForTesting
+ protected static void mergeEvaluatedMetricsMaps(
+ Map<ScalingMetric, EvaluatedScalingMetric> actual,
+ @Nullable Map<ScalingMetric, EvaluatedScalingMetric> incoming) {
+ Optional.ofNullable(incoming)
+ .ifPresent(
+ customEvaluatedMetric ->
+ customEvaluatedMetric.forEach(
+ (scalingMetric, evaluatedScalingMetric) ->
+ actual.merge(
+ scalingMetric,
+ evaluatedScalingMetric,
+ ScalingMetricEvaluator
+ ::mergeEvaluatedScalingMetric)));
+ }
+
+ /**
+ * Merges two {@link EvaluatedScalingMetric} instances.
+ *
+ * @param actual The existing evaluated scaling metric.
+ * @param incoming The incoming evaluated scaling metric.
+ * @return A new {@link EvaluatedScalingMetric} instance with merged values.
+ */
Review Comment:
unnecessary
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java:
##########
@@ -259,4 +276,55 @@ void setClock(Clock clock) {
this.metricsCollector.setClock(clock);
this.scalingExecutor.setClock(clock);
}
+
+ @VisibleForTesting
+ protected Tuple2<FlinkAutoscalerEvaluator, Configuration> getCustomEvaluatorIfRequired(
Review Comment:
I think this method should go to a utility class if possible. We should aim to keep this class minimal with the core algorithm flow
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java:
##########
@@ -588,4 +635,76 @@ protected static double computeEdgeDataRate(
to);
return getRate(ScalingMetric.NUM_RECORDS_OUT, from, metricsHistory);
}
+
+ /**
+ * Executes the provided custom metric evaluator for the given job vertex. Calls {@link
+ * FlinkAutoscalerEvaluator#evaluateVertexMetrics} to evaluate scaling metrics.
+ *
+ * @param vertex The job vertex being evaluated.
+ * @param evaluatedMetrics Current evaluated metrics.
+ * @param customEvaluationSession A tuple containing the custom metric evaluator and evaluation
+ * context.
+ * @return A map of scaling metrics, with its corresponding evaluated scaling metric.
+ */
+ @VisibleForTesting
+ protected static Map<ScalingMetric, EvaluatedScalingMetric> runCustomEvaluator(
+ JobVertexID vertex,
+ Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
+ Tuple2<FlinkAutoscalerEvaluator, FlinkAutoscalerEvaluator.Context>
+ customEvaluationSession) {
+ try {
+ return customEvaluationSession.f0.evaluateVertexMetrics(
+ vertex, evaluatedMetrics, customEvaluationSession.f1);
+ } catch (UnsupportedOperationException e) {
+ LOG.warn(
+ "Custom metric evaluator {} tried accessing an un-modifiable view.",
+ customEvaluationSession.f0.getClass(),
+ e);
Review Comment:
This feels unnecessary
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/FlinkAutoscalerEvaluator.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.runtime.jobgraph.JobVertexID;
+
+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.FlinkAutoscalerEvaluator}.
+ */
+public interface FlinkAutoscalerEvaluator {
+
+ /**
+ * 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.
+ */
+ class Context {
+ private final Configuration jobConf;
Review Comment:
I think it's a bit strange to have both jobConf and customEvaluatorConf. I think we should have a single `config` field that merges these two. Take the evaluator conf and apply the jobConf on top of it. This would allow CRs to control evaluator behaviour easily. This is the pattern followed in other places in general
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java:
##########
@@ -259,4 +276,55 @@ void setClock(Clock clock) {
this.metricsCollector.setClock(clock);
this.scalingExecutor.setClock(clock);
}
+
+ @VisibleForTesting
+ protected Tuple2<FlinkAutoscalerEvaluator, Configuration> getCustomEvaluatorIfRequired(
Review Comment:
Maybe this can happen inside the ScalingMetricEvaluator constructor/initialization so we don't have to update the evaluate method signature
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java:
##########
@@ -588,4 +635,76 @@ protected static double computeEdgeDataRate(
to);
return getRate(ScalingMetric.NUM_RECORDS_OUT, from, metricsHistory);
}
+
+ /**
+ * Executes the provided custom metric evaluator for the given job vertex. Calls {@link
+ * FlinkAutoscalerEvaluator#evaluateVertexMetrics} to evaluate scaling metrics.
+ *
+ * @param vertex The job vertex being evaluated.
+ * @param evaluatedMetrics Current evaluated metrics.
+ * @param customEvaluationSession A tuple containing the custom metric evaluator and evaluation
+ * context.
+ * @return A map of scaling metrics, with its corresponding evaluated scaling metric.
+ */
+ @VisibleForTesting
+ protected static Map<ScalingMetric, EvaluatedScalingMetric> runCustomEvaluator(
+ JobVertexID vertex,
+ Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
+ Tuple2<FlinkAutoscalerEvaluator, FlinkAutoscalerEvaluator.Context>
+ customEvaluationSession) {
+ try {
+ return customEvaluationSession.f0.evaluateVertexMetrics(
+ vertex, evaluatedMetrics, customEvaluationSession.f1);
+ } catch (UnsupportedOperationException e) {
+ LOG.warn(
+ "Custom metric evaluator {} tried accessing an un-modifiable view.",
+ customEvaluationSession.f0.getClass(),
+ e);
+ } catch (Exception e) {
+ LOG.warn(
+ "Custom metric evaluator {} threw an exception.",
+ customEvaluationSession.f0.getClass(),
+ e);
+ }
+
+ return Collections.emptyMap();
+ }
+
+ /**
+ * Merges the incoming evaluated metrics into actual evaluated metrics.
+ *
+ * @param actual The target evaluated metrics map to merge into.
+ * @param incoming The incoming map containing new evaluated metrics map to be merged
+ * (nullable).
+ */
Review Comment:
unnecessary
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39511] FLIP-514: Custom Evaluator plugin for Flink Autoscaler [flink-kubernetes-operator]
Posted by "Dennis-Mircea (via GitHub)" <gi...@apache.org>.
Dennis-Mircea commented on code in PR #1099:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1099#discussion_r3426456075
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/FlinkAutoscalerEvaluator.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.runtime.jobgraph.JobVertexID;
+
+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.FlinkAutoscalerEvaluator}.
+ */
+public interface FlinkAutoscalerEvaluator {
+
+ /**
+ * 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.
+ */
+ class Context {
+ private final Configuration jobConf;
Review Comment:
`customEvaluatorConf` is already derived from `jobConf`: `AutoScalerOptions.customEvaluatorConfiguration(conf, name)` takes the `job.autoscaler.metrics.custom-evaluator.<name>.` subtree and strips the prefix, so evaluator-specific keys are handed over ready to read by their short name. Merging the two back into one map would re-introduce exactly the prefix-handling boilerplate this helper removes, and authors would have to re-derive their own scope. `jobConf` is still there (unmodifiable) for evaluators that need the full job view, so both use cases are covered.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39511] FLIP-514: Custom Evaluator plugin for Flink Autoscaler [flink-kubernetes-operator]
Posted by "Dennis-Mircea (via GitHub)" <gi...@apache.org>.
Dennis-Mircea commented on code in PR #1099:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1099#discussion_r3426470571
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java:
##########
@@ -588,4 +635,76 @@ protected static double computeEdgeDataRate(
to);
return getRate(ScalingMetric.NUM_RECORDS_OUT, from, metricsHistory);
}
+
+ /**
+ * Executes the provided custom metric evaluator for the given job vertex. Calls {@link
+ * FlinkAutoscalerEvaluator#evaluateVertexMetrics} to evaluate scaling metrics.
+ *
+ * @param vertex The job vertex being evaluated.
+ * @param evaluatedMetrics Current evaluated metrics.
+ * @param customEvaluationSession A tuple containing the custom metric evaluator and evaluation
+ * context.
+ * @return A map of scaling metrics, with its corresponding evaluated scaling metric.
+ */
+ @VisibleForTesting
+ protected static Map<ScalingMetric, EvaluatedScalingMetric> runCustomEvaluator(
+ JobVertexID vertex,
+ Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
+ Tuple2<FlinkAutoscalerEvaluator, FlinkAutoscalerEvaluator.Context>
+ customEvaluationSession) {
+ try {
+ return customEvaluationSession.f0.evaluateVertexMetrics(
+ vertex, evaluatedMetrics, customEvaluationSession.f1);
+ } catch (UnsupportedOperationException e) {
+ LOG.warn(
+ "Custom metric evaluator {} tried accessing an un-modifiable view.",
+ customEvaluationSession.f0.getClass(),
+ e);
Review Comment:
Right, that's correct. I removed it!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39511] FLIP-514: Custom Evaluator plugin for Flink Autoscaler [flink-kubernetes-operator]
Posted by "Dennis-Mircea (via GitHub)" <gi...@apache.org>.
Dennis-Mircea commented on code in PR #1099:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1099#discussion_r3427232241
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/FlinkAutoscalerEvaluator.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.runtime.jobgraph.JobVertexID;
+
+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.FlinkAutoscalerEvaluator}.
+ */
+public interface FlinkAutoscalerEvaluator {
+
+ /**
+ * 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.
+ */
+ class Context {
+ private final Configuration jobConf;
Review Comment:
Ok. I addressed it. I merged them into a single `config`. The evaluator now gets one `Context.getConfig()` that is the full job config with its namespaced options (prefix stripped) overlaid on top, so the two-config split is gone and the prefix-stripping convenience stays.
I went with evaluator-keys-win (the namespaced value overrides the job-level value of the same key), for two reasons:
- It's the only order that makes the per-evaluator namespace useful. The point of `job.autoscaler.metrics.custom-evaluator.<name>.<key>` is to let a CR tune this evaluator independently. If the job-level value won, the override would silently be a no-op whenever the key also exists at job level.
- It's safe, because this merged config is only ever handed to the evaluator as a read-only view. The override changes what the evaluator sees, never the autoscaler's own decisions, and the keys are namespaced so a collision is deliberate, not accidental.
One thing to be aware of on the shape: the result is a mixed config. It keeps the original job-config keys (including the long `job.autoscaler.metrics.custom-evaluator.<name>.<key>` entries) and adds the prefix-stripped short keys on top, so a given evaluator option shows up both as its full namespaced key and as its stripped short name. Evaluators are expected to read the stripped short name. I kept the original keys rather than removing them so we don't have to walk the whole config every cycle.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39511] FLIP-514: Custom Evaluator plugin for Flink Autoscaler [flink-kubernetes-operator]
Posted by "Dennis-Mircea (via GitHub)" <gi...@apache.org>.
Dennis-Mircea commented on PR #1099:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1099#issuecomment-4327090302
> > **Question & Follow-up: single-evaluator limitation vs. config namespace shape**
> > As it stands, `JobAutoScalerImpl#getCustomEvaluatorIfRequired` reads a single `ConfigOption<String> CUSTOM_EVALUATOR_NAME` (`job.autoscaler.metrics.custom-evaluator.name`), looks up one `FlinkAutoscalerEvaluator` from the registry (keyed by `FlinkAutoscalerEvaluator#getName()`), and `ScalingMetricEvaluator` only knows how to invoke one evaluator per evaluation cycle. So at runtime **only one custom evaluator can be active per pipeline**, even though several may be registered via `META-INF/services`.
> > At the same time, `AutoScalerOptions.forCustomEvaluator(conf, name)` builds the evaluator-specific view as a `DelegatingConfiguration` over `job.autoscaler.metrics.custom-evaluator.<name>.`, which means users have to write keys like:
> > ```
> > job.autoscaler.metrics.custom-evaluator.name: custom-evaluator
> > job.autoscaler.metrics.custom-evaluator.custom-evaluator.target-data-rate: 100000.0
> > ```
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> > That repeated `<name>` segment is future-proofing for a multi-evaluator world, but today it's pure ceremony: there is never more than one evaluator reading from that namespace. It's confusing to document, easy to mis-type, and likely to trip users up.
> > I see two coherent directions and would like the maintainers' preference before pushing further commits:
> > **Option A - keep the single-evaluator contract, simplify the namespace.**
> > Drop the `<name>` segment from `forCustomEvaluator` so `Context#getCustomEvaluatorConf()` is a delegating view over `job.autoscaler.metrics.custom-evaluator.` directly. To avoid the `name` key leaking into that view the selector would move out of that prefix (e.g. renamed to `job.autoscaler.metrics.custom-evaluator-name`, with a fallback key for the old one). User-facing config becomes:
> > ```
> > job.autoscaler.metrics.custom-evaluator-name: custom-evaluator
> > job.autoscaler.metrics.custom-evaluator.target-data-rate: 100000.0
> > ```
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> > The **Option A** has the following pro & cons:
> >
> > * Pros: cleanest UX, matches the actual runtime semantics.
> > * Cons: represents a big limitation in the custom evaluator flexibility.
> >
> > **Option B - lift the limitation, align the config shape with Flink metric reporters and with [FLIP-514](https://cwiki.apache.org/confluence/display/FLINK/FLIP-514).**
> > Flink's metric-reporter config is the established idiom for "list of named instances, each with its own class and its own bag of options":
> > ```
> > metrics.reporters: my_jmx_reporter,my_other_reporter
> >
> > metrics.reporter.my_jmx_reporter.factory.class: org.apache.flink.metrics.jmx.JMXReporterFactory
> > metrics.reporter.my_jmx_reporter.port: 9020-9040
> >
> > metrics.reporter.my_other_reporter.factory.class: org.apache.flink.metrics.graphite.GraphiteReporterFactory
> > metrics.reporter.my_other_reporter.host: 192.168.1.1
> > metrics.reporter.my_other_reporter.port: 10000
> > ```
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> > Applied to custom evaluators, the mapping is:
> > Reporter concept Evaluator equivalent
> > `metrics.reporters: r1,r2` `job.autoscaler.metrics.custom-evaluators: custom-evaluator-1,custom-evaluator-2`
> > `metrics.reporter.<r>.factory.class: FQN` `job.autoscaler.metrics.custom-evaluator.<instance>.class: FQN`
> > `metrics.reporter.<r>.<opt>: v` `job.autoscaler.metrics.custom-evaluator.<instance>.<opt>: v`
> > Concretely:
> > ```
> > job.autoscaler.metrics.custom-evaluators: custom-evaluator-1,custom-evaluator-2
> >
> > job.autoscaler.metrics.custom-evaluator.custom-evaluator-1.class: org.apache.flink.autoscaler.custom.MyFirstEvaluator
> > job.autoscaler.metrics.custom-evaluator.custom-evaluator-1.target-data-rate: 100000.0
> >
> > job.autoscaler.metrics.custom-evaluator.custom-evaluator-2.class: org.apache.flink.autoscaler.custom.MySecondEvaluator
> > job.autoscaler.metrics.custom-evaluator.custom-evaluator-2.some-threshold: 0.8
> > ```
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> > The consequence is that the `getName()` goes away from the SPI. The instance name (`custom-evaluator-1`) is purely a configuration-level handle. Evaluator resolution stops going through the `META-INF/services` -> `getName()` registry and instead becomes: for each entry in `custom-evaluators`, read its `...<instance>.class` property and instantiate the fully qualified name (via `Class#forName`, or via a factory SPI if we want DI/customization, analogous to `*ReporterFactory`). This is exactly the shape [FLIP-514](https://cwiki.apache.org/confluence/display/FLINK/FLIP-514) proposes in its [Job Level Config For Custom Evaluator](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=345377351#FLIP514:CustomEvaluatorpluginforFlinkAutoscaler-JobLevelConfigForCustomEvaluator) section, and it removes the oddity where two evaluators couldn't share the same `getName()` even though they were distinct implementations.
> > To keep the ordering contract between multiple custom evaluators explicit and self-describing rather than implicit in the declaration order of `job.autoscaler.metrics.custom-evaluators`, we can adopt the same design already approved in [FLIP-575: Scaling Executor Plugin SPI for Flink Autoscaler](https://cwiki.apache.org/confluence/display/FLINK/FLIP-575%3A+Scaling+Executor+Plugin+SPI+for+Flink+Autoscaler) and implemented for [ScalingExecutorPlugin.java](https://github.com/apache/flink-kubernetes-operator/pull/1085/changes#diff-8bc87f00339ad429c6d66b8b57d0d7ec8358e5e0d95f542b2f7d2a1a19271769): a `default int priority()` method on the SPI, with evaluators executed in ascending priority and the returned metrics merged left-to-right (lower priority first, higher priority wins on conflicts). This gives us:
> >
> > * a single, well-defined ordering contract that doesn't depend on the user's config-key ordering or on `ServiceLoader/PluginManager` discovery order,
> > * parity and composability with `ScalingExecutorPlugin`, so both autoscaler extension points share the same mental model
> > * a natural extension path for Option B above as `priority()` simply becomes the tie-breaker once we allow more than one evaluator per pipeline.
> >
> > The **Option B** has the following pro & cons:
> >
> > * Pros: consistent with an idiom Flink users already know, aligns with FLIP-514, expressive (multiple evaluators per job), and drops the `getName()` coupling that is awkward today.
> > * Cons: bigger change than A.
> >
> > My preference would be to land **Option B** so we match FLIP-514 directly and don't ship a config surface we know we'll have to migrate away from.
> > Thoughts?
>
> @Dennis-Mircea let's go for option 2 but to limit the change/complexity we can also start with supporting only a single evaluator plugin and add the priority / ordering logic as a follow up jira
> > **Question & Follow-up: single-evaluator limitation vs. config namespace shape**
> > As it stands, `JobAutoScalerImpl#getCustomEvaluatorIfRequired` reads a single `ConfigOption<String> CUSTOM_EVALUATOR_NAME` (`job.autoscaler.metrics.custom-evaluator.name`), looks up one `FlinkAutoscalerEvaluator` from the registry (keyed by `FlinkAutoscalerEvaluator#getName()`), and `ScalingMetricEvaluator` only knows how to invoke one evaluator per evaluation cycle. So at runtime **only one custom evaluator can be active per pipeline**, even though several may be registered via `META-INF/services`.
> > At the same time, `AutoScalerOptions.forCustomEvaluator(conf, name)` builds the evaluator-specific view as a `DelegatingConfiguration` over `job.autoscaler.metrics.custom-evaluator.<name>.`, which means users have to write keys like:
> > ```
> > job.autoscaler.metrics.custom-evaluator.name: custom-evaluator
> > job.autoscaler.metrics.custom-evaluator.custom-evaluator.target-data-rate: 100000.0
> > ```
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> > That repeated `<name>` segment is future-proofing for a multi-evaluator world, but today it's pure ceremony: there is never more than one evaluator reading from that namespace. It's confusing to document, easy to mis-type, and likely to trip users up.
> > I see two coherent directions and would like the maintainers' preference before pushing further commits:
> > **Option A - keep the single-evaluator contract, simplify the namespace.**
> > Drop the `<name>` segment from `forCustomEvaluator` so `Context#getCustomEvaluatorConf()` is a delegating view over `job.autoscaler.metrics.custom-evaluator.` directly. To avoid the `name` key leaking into that view the selector would move out of that prefix (e.g. renamed to `job.autoscaler.metrics.custom-evaluator-name`, with a fallback key for the old one). User-facing config becomes:
> > ```
> > job.autoscaler.metrics.custom-evaluator-name: custom-evaluator
> > job.autoscaler.metrics.custom-evaluator.target-data-rate: 100000.0
> > ```
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> > The **Option A** has the following pro & cons:
> >
> > * Pros: cleanest UX, matches the actual runtime semantics.
> > * Cons: represents a big limitation in the custom evaluator flexibility.
> >
> > **Option B - lift the limitation, align the config shape with Flink metric reporters and with [FLIP-514](https://cwiki.apache.org/confluence/display/FLINK/FLIP-514).**
> > Flink's metric-reporter config is the established idiom for "list of named instances, each with its own class and its own bag of options":
> > ```
> > metrics.reporters: my_jmx_reporter,my_other_reporter
> >
> > metrics.reporter.my_jmx_reporter.factory.class: org.apache.flink.metrics.jmx.JMXReporterFactory
> > metrics.reporter.my_jmx_reporter.port: 9020-9040
> >
> > metrics.reporter.my_other_reporter.factory.class: org.apache.flink.metrics.graphite.GraphiteReporterFactory
> > metrics.reporter.my_other_reporter.host: 192.168.1.1
> > metrics.reporter.my_other_reporter.port: 10000
> > ```
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> > Applied to custom evaluators, the mapping is:
> > Reporter concept Evaluator equivalent
> > `metrics.reporters: r1,r2` `job.autoscaler.metrics.custom-evaluators: custom-evaluator-1,custom-evaluator-2`
> > `metrics.reporter.<r>.factory.class: FQN` `job.autoscaler.metrics.custom-evaluator.<instance>.class: FQN`
> > `metrics.reporter.<r>.<opt>: v` `job.autoscaler.metrics.custom-evaluator.<instance>.<opt>: v`
> > Concretely:
> > ```
> > job.autoscaler.metrics.custom-evaluators: custom-evaluator-1,custom-evaluator-2
> >
> > job.autoscaler.metrics.custom-evaluator.custom-evaluator-1.class: org.apache.flink.autoscaler.custom.MyFirstEvaluator
> > job.autoscaler.metrics.custom-evaluator.custom-evaluator-1.target-data-rate: 100000.0
> >
> > job.autoscaler.metrics.custom-evaluator.custom-evaluator-2.class: org.apache.flink.autoscaler.custom.MySecondEvaluator
> > job.autoscaler.metrics.custom-evaluator.custom-evaluator-2.some-threshold: 0.8
> > ```
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> > The consequence is that the `getName()` goes away from the SPI. The instance name (`custom-evaluator-1`) is purely a configuration-level handle. Evaluator resolution stops going through the `META-INF/services` -> `getName()` registry and instead becomes: for each entry in `custom-evaluators`, read its `...<instance>.class` property and instantiate the fully qualified name (via `Class#forName`, or via a factory SPI if we want DI/customization, analogous to `*ReporterFactory`). This is exactly the shape [FLIP-514](https://cwiki.apache.org/confluence/display/FLINK/FLIP-514) proposes in its [Job Level Config For Custom Evaluator](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=345377351#FLIP514:CustomEvaluatorpluginforFlinkAutoscaler-JobLevelConfigForCustomEvaluator) section, and it removes the oddity where two evaluators couldn't share the same `getName()` even though they were distinct implementations.
> > To keep the ordering contract between multiple custom evaluators explicit and self-describing rather than implicit in the declaration order of `job.autoscaler.metrics.custom-evaluators`, we can adopt the same design already approved in [FLIP-575: Scaling Executor Plugin SPI for Flink Autoscaler](https://cwiki.apache.org/confluence/display/FLINK/FLIP-575%3A+Scaling+Executor+Plugin+SPI+for+Flink+Autoscaler) and implemented for [ScalingExecutorPlugin.java](https://github.com/apache/flink-kubernetes-operator/pull/1085/changes#diff-8bc87f00339ad429c6d66b8b57d0d7ec8358e5e0d95f542b2f7d2a1a19271769): a `default int priority()` method on the SPI, with evaluators executed in ascending priority and the returned metrics merged left-to-right (lower priority first, higher priority wins on conflicts). This gives us:
> >
> > * a single, well-defined ordering contract that doesn't depend on the user's config-key ordering or on `ServiceLoader/PluginManager` discovery order,
> > * parity and composability with `ScalingExecutorPlugin`, so both autoscaler extension points share the same mental model
> > * a natural extension path for Option B above as `priority()` simply becomes the tie-breaker once we allow more than one evaluator per pipeline.
> >
> > The **Option B** has the following pro & cons:
> >
> > * Pros: consistent with an idiom Flink users already know, aligns with FLIP-514, expressive (multiple evaluators per job), and drops the `getName()` coupling that is awkward today.
> > * Cons: bigger change than A.
> >
> > My preference would be to land **Option B** so we match FLIP-514 directly and don't ship a config surface we know we'll have to migrate away from.
> > Thoughts?
>
> @Dennis-Mircea let's go for option 2 but to limit the change/complexity we can also start with supporting only a single evaluator plugin and add the priority / ordering logic as a follow up jira
The option 2 is now fully implemented and the [FLINK-39554](https://issues.apache.org/jira/browse/FLINK-39554) JIRA was created as follow up. The PR is now ready for final review.
One last question and clarification about the `FlinkAutoscalerEvaluator`: In order to keep the alignment between the autoscaler plugins, as the [FLIP-575](https://cwiki.apache.org/confluence/display/FLINK/FLIP-575%3A+Scaling+Executor+Plugin+SPI+for+Flink+Autoscaler) will introduce the `ScalingExecutorPlugin`, should we batter rename this SPI `ScalingMetricsEvaluatorPlugin` or `AutoscalerMetricsEvaluatorPlugin`? I'd lean toward `ScalingMetricsEvaluatorPlugin` for symmetry, but happy to defer to reviewers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39511] FLIP-514: Custom Evaluator plugin for Flink Autoscaler [flink-kubernetes-operator]
Posted by "Dennis-Mircea (via GitHub)" <gi...@apache.org>.
Dennis-Mircea commented on code in PR #1099:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1099#discussion_r3426451422
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/FlinkAutoscalerEvaluator.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.runtime.jobgraph.JobVertexID;
+
+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.FlinkAutoscalerEvaluator}.
+ */
+public interface FlinkAutoscalerEvaluator {
+
+ /**
+ * 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.
+ */
+ class Context {
Review Comment:
I'd prefer to keep the explicit constructor/getters here. `Context` is part of the public SPI that user plugins import and compile against, and Lombok-generated members aren't in the source the way they are in the bytecode. For example, we already hit this with `IOMetrics` (which uses `@Value`): the IDE flags *"Library source does not match the bytecode for class IOMetrics"* on it. I'd rather not expose evaluator authors to that, so I kept the boilerplate explicit on this API type.
<img width="1181" height="761" alt="img_4" src="https://github.com/user-attachments/assets/c85a4dac-f686-4205-bf37-5428565d79b2" />
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39511] FLIP-514: Custom Evaluator plugin for Flink Autoscaler [flink-kubernetes-operator]
Posted by "Dennis-Mircea (via GitHub)" <gi...@apache.org>.
Dennis-Mircea commented on code in PR #1099:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1099#discussion_r3426461322
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java:
##########
@@ -259,4 +276,55 @@ void setClock(Clock clock) {
this.metricsCollector.setClock(clock);
this.scalingExecutor.setClock(clock);
}
+
+ @VisibleForTesting
+ protected Tuple2<FlinkAutoscalerEvaluator, Configuration> getCustomEvaluatorIfRequired(
Review Comment:
Good point. I addressed this. The registry now lives on `ScalingMetricEvaluator`: it takes the discovered evaluators via its constructor and resolves the configured one internally. `JobAutoScalerImpl` no longer references custom evaluators at all (constructor param, field, and method removed), so the core flow stays minimal. The discovery wiring in `AutoscalerFactory` / `StandaloneAutoscalerEntrypoint` now feeds `new ScalingMetricEvaluator(customEvaluators)`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39511] FLIP-514: Custom Evaluator plugin for Flink Autoscaler [flink-kubernetes-operator]
Posted by "Dennis-Mircea (via GitHub)" <gi...@apache.org>.
Dennis-Mircea commented on code in PR #1099:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1099#discussion_r3426489076
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java:
##########
@@ -588,4 +635,76 @@ protected static double computeEdgeDataRate(
to);
return getRate(ScalingMetric.NUM_RECORDS_OUT, from, metricsHistory);
}
+
+ /**
+ * Executes the provided custom metric evaluator for the given job vertex. Calls {@link
+ * FlinkAutoscalerEvaluator#evaluateVertexMetrics} to evaluate scaling metrics.
+ *
+ * @param vertex The job vertex being evaluated.
+ * @param evaluatedMetrics Current evaluated metrics.
+ * @param customEvaluationSession A tuple containing the custom metric evaluator and evaluation
+ * context.
+ * @return A map of scaling metrics, with its corresponding evaluated scaling metric.
+ */
+ @VisibleForTesting
+ protected static Map<ScalingMetric, EvaluatedScalingMetric> runCustomEvaluator(
+ JobVertexID vertex,
+ Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
+ Tuple2<FlinkAutoscalerEvaluator, FlinkAutoscalerEvaluator.Context>
+ customEvaluationSession) {
+ try {
+ return customEvaluationSession.f0.evaluateVertexMetrics(
+ vertex, evaluatedMetrics, customEvaluationSession.f1);
+ } catch (UnsupportedOperationException e) {
+ LOG.warn(
+ "Custom metric evaluator {} tried accessing an un-modifiable view.",
+ customEvaluationSession.f0.getClass(),
+ e);
+ } catch (Exception e) {
+ LOG.warn(
+ "Custom metric evaluator {} threw an exception.",
+ customEvaluationSession.f0.getClass(),
+ e);
+ }
+
+ return Collections.emptyMap();
+ }
+
+ /**
+ * Merges the incoming evaluated metrics into actual evaluated metrics.
+ *
+ * @param actual The target evaluated metrics map to merge into.
+ * @param incoming The incoming map containing new evaluated metrics map to be merged
+ * (nullable).
+ */
Review Comment:
Same explanation as the previous one!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39511] FLIP-514: Custom Evaluator plugin for Flink Autoscaler [flink-kubernetes-operator]
Posted by "Dennis-Mircea (via GitHub)" <gi...@apache.org>.
Dennis-Mircea commented on code in PR #1099:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1099#discussion_r3426464881
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java:
##########
@@ -72,14 +76,34 @@ public class ScalingMetricEvaluator {
private static final Logger LOG = LoggerFactory.getLogger(ScalingMetricEvaluator.class);
public EvaluatedMetrics evaluate(
- Configuration conf, CollectedMetricHistory collectedMetrics, Duration restartTime) {
+ Configuration conf,
+ CollectedMetricHistory collectedMetrics,
+ Duration restartTime,
+ @Nullable Tuple2<FlinkAutoscalerEvaluator, Configuration> customEvaluatorWithConfig) {
Review Comment:
Addressed!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org