mxm commented on code in PR #698: URL: https://github.com/apache/flink-kubernetes-operator/pull/698#discussion_r1381602097
########## flink-autoscaler/src/main/java/org/apache/flink/autoscaler/standalone/JobListFetcher.java: ########## @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.autoscaler.standalone; + +import org.apache.flink.annotation.Experimental; +import org.apache.flink.autoscaler.JobAutoScalerContext; +import org.apache.flink.metrics.MetricGroup; + +import java.util.List; + +/** The JobListFetcher will fetch the jobContext of all jobs. */ +@Experimental +public interface JobListFetcher<KEY> { + + List<JobAutoScalerContext<KEY>> fetch(MetricGroup metricGroup) throws Exception; Review Comment: I'm also ok with keeping the interface. ########## flink-autoscaler-standalone/README.md: ########## @@ -0,0 +1,83 @@ +# Flink Autoscaler Standalone + +## What's the autoscaler standalone? + +`Flink Autoscaler Standalone` is an implementation of `Flink Autoscaler`, it runs as +a separate java process. It computes the reasonable parallelism of all job vertices +by monitoring the metrics, such as: processing rate, busy time, etc. Please see +[Autoscaler official doc](https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-main/docs/custom-resource/autoscaler/) +for an overview of how autoscaling works. + +`Flink Autoscaler Standalone` rescales flink job in-place by rest api of +[Externalized Declarative Resource Management](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/elastic_scaling/#externalized-declarative-resource-management). + +Kubernetes Operator is well integrated with Autoscaler, we strongly recommend using +Kubernetes Operator directly for the kubernetes flink jobs, and only flink jobs in +non-kubernetes environments use Autoscaler Standalone. + +## How To Use + +Currently, `Flink Autoscaler Standalone` only supports a single Flink cluster. +It can be any type of Flink cluster, includes: + +- Flink Standalone Cluster +- MiniCluster +- Flink yarn session cluster +- Flink yarn application cluster +- Flink kubernetes session cluster +- Flink kubernetes application cluster +- etc + +You can start a Flink Streaming job with the following ConfigOptions. + +``` +# Enable Adaptvie scheduler to play the in-place rescaling. +jobmanager.scheduler : adaptive + +# Enable autoscale and scaling +job.autoscaler.enabled : true +job.autoscaler.scaling.enabled : true +job.autoscaler.stabilization.interval : 1m +job.autoscaler.metrics.window : 3m +``` + +Note: In-place rescaling is only supported since Flink 1.18. Flink jobs before version +1.18 cannot be scaled automatically, but you can view the ScalingReport in Log. +ScalingReport will show the recommended parallelism for each vertex. + +After the flink job starts, please start the StandaloneAutoscaler process by the +following command. + +``` +java -cp flink-autoscaler-standalone-1.7-SNAPSHOT.jar \ +org.apache.flink.autoscaler.standalone.StandaloneAutoscalerEntrypoint \ +--flinkClusterHost localhost \ +--flinkClusterPort 8081 +``` + +Updating the `flinkClusterHost` and `flinkClusterPort` based on your flink cluster. +In general, the host and port are the same as Flink WebUI. + +## Extensibility of autoscaler standalone + +`Autoscaler` as a generic autoscaler component defines a series of generic interfaces: + +- **AutoScalerEventHandler** : Handling autoscaler events, such as: ScalingReport, + AutoscalerError, etc. It logs events by default. +- **AutoScalerStateStore** : Storing all state during scaling. `InMemoryAutoScalerStateStore` + is the default implementation, it's based on the Java Heap, so the state will be discarded + after process restarts. We will implement persistent State Store in the future, such as + : `JdbcAutoScalerStateStore`. +- **ScalingRealizer** : Applying scaling actions. `RescaleApiScalingRealizer` is the default + implementation, it uses the Rescale API to apply parallelism changes. +- **JobAutoScalerContext** : Including all details related to the current job. Review Comment: Line 61-73 should be part of the flink-autoscaler README. ########## flink-autoscaler/src/main/java/org/apache/flink/autoscaler/event/AutoScalerEventHandler.java: ########## @@ -32,32 +35,45 @@ import static org.apache.flink.autoscaler.metrics.ScalingMetric.TRUE_PROCESSING_RATE; /** - * Handler for autoscaler events. + * Handler for autoscaler events, it logs events by default. * * @param <KEY> The job key. * @param <Context> Instance of JobAutoScalerContext. */ @Experimental -public interface AutoScalerEventHandler<KEY, Context extends JobAutoScalerContext<KEY>> { - String SCALING_SUMMARY_ENTRY = +public class AutoScalerEventHandler<KEY, Context extends JobAutoScalerContext<KEY>> { Review Comment: I think this should remain an interface because this is meant to be implemented by other backends. We already have two implementation of this: Kubernetes, Standalone, and Logging only. ########## flink-autoscaler-standalone/README.md: ########## @@ -0,0 +1,83 @@ +# Flink Autoscaler Standalone + +## What's the autoscaler standalone? + +`Flink Autoscaler Standalone` is an implementation of `Flink Autoscaler`, it runs as +a separate java process. It computes the reasonable parallelism of all job vertices +by monitoring the metrics, such as: processing rate, busy time, etc. Please see +[Autoscaler official doc](https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-main/docs/custom-resource/autoscaler/) +for an overview of how autoscaling works. + +`Flink Autoscaler Standalone` rescales flink job in-place by rest api of +[Externalized Declarative Resource Management](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/elastic_scaling/#externalized-declarative-resource-management). + +Kubernetes Operator is well integrated with Autoscaler, we strongly recommend using +Kubernetes Operator directly for the kubernetes flink jobs, and only flink jobs in +non-kubernetes environments use Autoscaler Standalone. + +## How To Use + +Currently, `Flink Autoscaler Standalone` only supports a single Flink cluster. +It can be any type of Flink cluster, includes: + +- Flink Standalone Cluster +- MiniCluster +- Flink yarn session cluster +- Flink yarn application cluster +- Flink kubernetes session cluster +- Flink kubernetes application cluster +- etc + +You can start a Flink Streaming job with the following ConfigOptions. + +``` +# Enable Adaptvie scheduler to play the in-place rescaling. +jobmanager.scheduler : adaptive + +# Enable autoscale and scaling +job.autoscaler.enabled : true +job.autoscaler.scaling.enabled : true +job.autoscaler.stabilization.interval : 1m +job.autoscaler.metrics.window : 3m +``` + +Note: In-place rescaling is only supported since Flink 1.18. Flink jobs before version +1.18 cannot be scaled automatically, but you can view the ScalingReport in Log. +ScalingReport will show the recommended parallelism for each vertex. + +After the flink job starts, please start the StandaloneAutoscaler process by the +following command. + +``` +java -cp flink-autoscaler-standalone-1.7-SNAPSHOT.jar \ +org.apache.flink.autoscaler.standalone.StandaloneAutoscalerEntrypoint \ +--flinkClusterHost localhost \ +--flinkClusterPort 8081 +``` + +Updating the `flinkClusterHost` and `flinkClusterPort` based on your flink cluster. +In general, the host and port are the same as Flink WebUI. + +## Extensibility of autoscaler standalone + +`Autoscaler` as a generic autoscaler component defines a series of generic interfaces: + +- **AutoScalerEventHandler** : Handling autoscaler events, such as: ScalingReport, Review Comment: This should be an interface, currently a class. Additionally, we can provide a logging implementation. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
