[FLINK-36018][autoscaler] Support lazy scale down to avoid frequent r…

…escaling (#875)

docs/layouts/shortcodes/generated/auto_scaler_configuration.html

@@ -146,18 +146,18 @@
             <td>Duration</td>
             <td>Maximum cap for the observed restart time when 'job.autoscaler.restart.time-tracking.enabled' is set to true.</td>
         </tr>
+        <tr>
+            <td><h5>job.autoscaler.scale-down.interval</h5></td>
+            <td style="word-wrap: break-word;">1 h</td>
+            <td>Duration</td>
+            <td>The delay time for scale down to be executed. If it is greater than 0, the scale down will be delayed. Delayed rescale can merge multiple scale downs within `scale-down.interval` into a scale down, thereby reducing the number of rescales. Reducing the frequency of job restarts can improve job availability. Scale down can be executed directly if it's less than or equal 0.</td>
+        </tr>
         <tr>
             <td><h5>job.autoscaler.scale-down.max-factor</h5></td>
             <td style="word-wrap: break-word;">0.6</td>
             <td>Double</td>
             <td>Max scale down factor. 1 means no limit on scale down, 0.6 means job can only be scaled down with 60% of the original parallelism.</td>
         </tr>
-        <tr>
-            <td><h5>job.autoscaler.scale-up.grace-period</h5></td>
-            <td style="word-wrap: break-word;">1 h</td>
-            <td>Duration</td>
-            <td>Duration in which no scale down of a vertex is allowed after it has been scaled up.</td>
-        </tr>
         <tr>
             <td><h5>job.autoscaler.scale-up.max-factor</h5></td>
             <td style="word-wrap: break-word;">100000.0</td>

e2e-tests/data/autoscaler.yaml

@@ -40,6 +40,7 @@ spec:
     job.autoscaler.scaling.enabled: "true"
     job.autoscaler.stabilization.interval: "5s"
     job.autoscaler.metrics.window: "1m"
+    job.autoscaler.scale-down.interval: "0m"
 
 #    Invalid Validations for testing autoscaler configurations
 #    kubernetes.operator.job.autoscaler.scale-down.max-factor: "-0.6"

flink-autoscaler-plugin-jdbc/src/main/java/org/apache/flink/autoscaler/jdbc/state/JdbcAutoScalerStateStore.java

@@ -19,6 +19,7 @@
 
 import org.apache.flink.annotation.Experimental;
 import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.autoscaler.DelayedScaleDown;
 import org.apache.flink.autoscaler.JobAutoScalerContext;
 import org.apache.flink.autoscaler.ScalingSummary;
 import org.apache.flink.autoscaler.ScalingTracking;
@@ -49,6 +50,7 @@
 
 import static org.apache.flink.autoscaler.jdbc.state.StateType.COLLECTED_METRICS;
 import static org.apache.flink.autoscaler.jdbc.state.StateType.CONFIG_OVERRIDES;
+import static org.apache.flink.autoscaler.jdbc.state.StateType.DELAYED_SCALE_DOWN;
 import static org.apache.flink.autoscaler.jdbc.state.StateType.PARALLELISM_OVERRIDES;
 import static org.apache.flink.autoscaler.jdbc.state.StateType.SCALING_HISTORY;
 import static org.apache.flink.autoscaler.jdbc.state.StateType.SCALING_TRACKING;
@@ -214,6 +216,35 @@ public void removeConfigChanges(Context jobContext) {
         jdbcStateStore.removeSerializedState(getSerializeKey(jobContext), CONFIG_OVERRIDES);
     }
 
+    @Override
+    public void storeDelayedScaleDown(Context jobContext, DelayedScaleDown delayedScaleDown)
+            throws Exception {
+        jdbcStateStore.putSerializedState(
+                getSerializeKey(jobContext),
+                DELAYED_SCALE_DOWN,
+                serializeDelayedScaleDown(delayedScaleDown));
+    }
+
+    @Nonnull
+    @Override
+    public DelayedScaleDown getDelayedScaleDown(Context jobContext) {
+        Optional<String> delayedScaleDown =
+                jdbcStateStore.getSerializedState(getSerializeKey(jobContext), DELAYED_SCALE_DOWN);
+        if (delayedScaleDown.isEmpty()) {
+            return new DelayedScaleDown();
+        }
+
+        try {
+            return deserializeDelayedScaleDown(delayedScaleDown.get());
+        } catch (JacksonException e) {
+            LOG.error(
+                    "Could not deserialize delayed scale down, possibly the format changed. Discarding...",
+                    e);
+            jdbcStateStore.removeSerializedState(getSerializeKey(jobContext), DELAYED_SCALE_DOWN);
+            return new DelayedScaleDown();
+        }
+    }
+
     @Override
     public void clearAll(Context jobContext) {
         jdbcStateStore.clearAll(getSerializeKey(jobContext));
@@ -296,4 +327,16 @@ private static ConfigChanges deserializeConfigOverrides(String configOverrides)
             return null;
         }
     }
+
+    private static String serializeDelayedScaleDown(DelayedScaleDown delayedScaleDown)
+            throws JacksonException {
+        return YAML_MAPPER.writeValueAsString(delayedScaleDown.getFirstTriggerTime());
+    }
+
+    private static DelayedScaleDown deserializeDelayedScaleDown(String delayedScaleDown)
+            throws JacksonException {
+        Map<JobVertexID, Instant> firstTriggerTime =
+                YAML_MAPPER.readValue(delayedScaleDown, new TypeReference<>() {});
+        return new DelayedScaleDown(firstTriggerTime);
+    }
 }

flink-autoscaler-plugin-jdbc/src/main/java/org/apache/flink/autoscaler/jdbc/state/StateType.java

@@ -27,7 +27,8 @@ public enum StateType {
     SCALING_TRACKING("scalingTracking"),
     COLLECTED_METRICS("collectedMetrics"),
     PARALLELISM_OVERRIDES("parallelismOverrides"),
-    CONFIG_OVERRIDES("configOverrides");
+    CONFIG_OVERRIDES("configOverrides"),
+    DELAYED_SCALE_DOWN("delayedScaleDown");
 
     /**
      * The identifier of each state type, it will be used to store. Please ensure the identifier is

flink-autoscaler/src/main/java/org/apache/flink/autoscaler/DelayedScaleDown.java

@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.autoscaler;
+
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+
+import lombok.Getter;
+
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+/** All delayed scale down requests. */
+public class DelayedScaleDown {
+
+    @Getter private final Map<JobVertexID, Instant> firstTriggerTime;
+
+    // Have any scale down request been updated? It doesn't need to be stored, it is only used to
+    // determine whether DelayedScaleDown needs to be stored.
+    @Getter private boolean isUpdated = false;
+
+    public DelayedScaleDown() {
+        this.firstTriggerTime = new HashMap<>();
+    }
+
+    public DelayedScaleDown(Map<JobVertexID, Instant> firstTriggerTime) {
+        this.firstTriggerTime = firstTriggerTime;
+    }
+
+    Optional<Instant> getFirstTriggerTimeForVertex(JobVertexID vertex) {
+        return Optional.ofNullable(firstTriggerTime.get(vertex));
+    }
+
+    void updateTriggerTime(JobVertexID vertex, Instant instant) {
+        firstTriggerTime.put(vertex, instant);
+        isUpdated = true;
+    }
+
+    void clearVertex(JobVertexID vertex) {
+        Instant removed = firstTriggerTime.remove(vertex);
+        if (removed != null) {
+            isUpdated = true;
+        }
+    }
+
+    void clearAll() {
+        if (firstTriggerTime.isEmpty()) {
+            return;
+        }
+        firstTriggerTime.clear();
+        isUpdated = true;
+    }
+}

flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java

@@ -214,9 +214,20 @@ private void runScalingLogic(Context ctx, AutoscalerFlinkMetrics autoscalerMetri
             return;
         }
 
+        var delayedScaleDown = stateStore.getDelayedScaleDown(ctx);
         var parallelismChanged =
                 scalingExecutor.scaleResource(
-                        ctx, evaluatedMetrics, scalingHistory, scalingTracking, now, jobTopology);
+                        ctx,
+                        evaluatedMetrics,
+                        scalingHistory,
+                        scalingTracking,
+                        now,
+                        jobTopology,
+                        delayedScaleDown);
+
+        if (delayedScaleDown.isUpdated()) {
+            stateStore.storeDelayedScaleDown(ctx, delayedScaleDown);
+        }
 
         if (parallelismChanged) {
             autoscalerMetrics.incrementScaling();