You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2023/06/25 14:15:10 UTC
[flink-kubernetes-operator] 07/07: [FLINK-32057][tests] Add e2e test for autoscaler
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
commit 3bb4fcc6a0247f63177a2720f54d71834dd25c49
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Tue Jun 13 15:07:58 2023 +0200
[FLINK-32057][tests] Add e2e test for autoscaler
---
.github/workflows/ci.yml | 13 ++++++
.../data/{sessionjob-cr.yaml => autoscaler.yaml} | 39 +++++++---------
e2e-tests/data/flinkdep-cr.yaml | 4 --
e2e-tests/data/multi-sessionjob.yaml | 4 --
e2e-tests/data/sessionjob-cr.yaml | 4 --
e2e-tests/test_autoscaler.sh | 53 ++++++++++++++++++++++
e2e-tests/utils.sh | 27 +++++++++++
examples/autoscaling/autoscaling.yaml | 2 +-
8 files changed, 110 insertions(+), 36 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 4125c1d4..c4572b54 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -76,6 +76,7 @@ jobs:
- test_sessionjob_kubernetes_ha.sh
- test_sessionjob_operations.sh
- test_multi_sessionjob.sh
+ - test_autoscaler.sh
include:
- namespace: flink
extraArgs: '--create-namespace --set "watchNamespaces={default,flink}"'
@@ -94,6 +95,18 @@ jobs:
exclude:
- namespace: default
test: test_multi_sessionjob.sh
+ - namespace: default
+ test: test_autoscaler.sh
+ - mode: standalone
+ test: test_autoscaler.sh
+ - version: v1_13
+ test: test_autoscaler.sh
+ - version: v1_14
+ test: test_autoscaler.sh
+ - version: v1_15
+ test: test_autoscaler.sh
+ - version: v1_16
+ test: test_autoscaler.sh
name: e2e_ci
steps:
- uses: actions/checkout@v2
diff --git a/e2e-tests/data/sessionjob-cr.yaml b/e2e-tests/data/autoscaler.yaml
similarity index 77%
copy from e2e-tests/data/sessionjob-cr.yaml
copy to e2e-tests/data/autoscaler.yaml
index d10e1cd3..ce41f325 100644
--- a/e2e-tests/data/sessionjob-cr.yaml
+++ b/e2e-tests/data/autoscaler.yaml
@@ -20,10 +20,10 @@ apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
namespace: default
- name: session-cluster-1
+ name: flink-autoscaler-e2e
spec:
- image: flink:1.17
- flinkVersion: v1_17
+ image: flink:1.18
+ flinkVersion: v1_18
ingress:
template: "/{{namespace}}/{{name}}(/|$)(.*)"
className: "nginx"
@@ -31,16 +31,19 @@ spec:
nginx.ingress.kubernetes.io/rewrite-target: "/$2"
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
- high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory
+ high-availability.type: kubernetes
high-availability.storageDir: file:///opt/flink/volume/flink-ha
state.checkpoints.dir: file:///opt/flink/volume/flink-cp
state.savepoints.dir: file:///opt/flink/volume/flink-sp
+
+ kubernetes.operator.job.autoscaler.enabled: "true"
+ kubernetes.operator.job.autoscaler.scaling.enabled: "true"
+ kubernetes.operator.job.autoscaler.stabilization.interval: "5s"
+ kubernetes.operator.job.autoscaler.metrics.window: "1m"
+
+ jobmanager.scheduler: adaptive
serviceAccount: flink
podTemplate:
- apiVersion: v1
- kind: Pod
- metadata:
- name: pod-template
spec:
containers:
# Do not change the main container name
@@ -56,7 +59,7 @@ spec:
volumes:
- name: flink-volume
persistentVolumeClaim:
- claimName: session-cluster-1-pvc
+ claimName: flink-autoscaler-e2e
jobManager:
resource:
memory: "1024m"
@@ -65,27 +68,17 @@ spec:
resource:
memory: "1024m"
cpu: 0.5
- mode: native
-
----
-apiVersion: flink.apache.org/v1beta1
-kind: FlinkSessionJob
-metadata:
- namespace: default
- name: flink-example-statemachine
-spec:
- deploymentName: session-cluster-1
job:
- jarURI: https://repo1.maven.org/maven2/org/apache/flink/flink-examples-streaming_2.12/1.14.4/flink-examples-streaming_2.12-1.14.4.jar
+ jarURI: local:///opt/flink/examples/streaming/StateMachineExample.jar
parallelism: 2
- upgradeMode: savepoint
- entryClass: org.apache.flink.streaming.examples.statemachine.StateMachineExample
+ upgradeMode: last-state
+ mode: native
---
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
- name: session-cluster-1-pvc
+ name: flink-autoscaler-e2e
spec:
accessModes:
- ReadWriteOnce
diff --git a/e2e-tests/data/flinkdep-cr.yaml b/e2e-tests/data/flinkdep-cr.yaml
index e4273787..05048c13 100644
--- a/e2e-tests/data/flinkdep-cr.yaml
+++ b/e2e-tests/data/flinkdep-cr.yaml
@@ -37,10 +37,6 @@ spec:
state.savepoints.dir: file:///opt/flink/volume/flink-sp
serviceAccount: flink
podTemplate:
- apiVersion: v1
- kind: Pod
- metadata:
- name: pod-template
spec:
initContainers:
- name: artifacts-fetcher
diff --git a/e2e-tests/data/multi-sessionjob.yaml b/e2e-tests/data/multi-sessionjob.yaml
index 160dae68..fd30c838 100644
--- a/e2e-tests/data/multi-sessionjob.yaml
+++ b/e2e-tests/data/multi-sessionjob.yaml
@@ -37,10 +37,6 @@ spec:
state.savepoints.dir: file:///opt/flink/volume/flink-sp
serviceAccount: flink
podTemplate:
- apiVersion: v1
- kind: Pod
- metadata:
- name: pod-template
spec:
containers:
# Do not change the main container name
diff --git a/e2e-tests/data/sessionjob-cr.yaml b/e2e-tests/data/sessionjob-cr.yaml
index d10e1cd3..ff469ab2 100644
--- a/e2e-tests/data/sessionjob-cr.yaml
+++ b/e2e-tests/data/sessionjob-cr.yaml
@@ -37,10 +37,6 @@ spec:
state.savepoints.dir: file:///opt/flink/volume/flink-sp
serviceAccount: flink
podTemplate:
- apiVersion: v1
- kind: Pod
- metadata:
- name: pod-template
spec:
containers:
# Do not change the main container name
diff --git a/e2e-tests/test_autoscaler.sh b/e2e-tests/test_autoscaler.sh
new file mode 100644
index 00000000..49767d64
--- /dev/null
+++ b/e2e-tests/test_autoscaler.sh
@@ -0,0 +1,53 @@
+#!/usr/bin/env bash
+################################################################################
+# 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.
+################################################################################
+
+SCRIPT_DIR=$(dirname "$(readlink -f "$0")")
+source "${SCRIPT_DIR}/utils.sh"
+
+CLUSTER_ID="flink-autoscaler-e2e"
+APPLICATION_YAML="${SCRIPT_DIR}/data/autoscaler.yaml"
+APPLICATION_IDENTIFIER="flinkdep/$CLUSTER_ID"
+TIMEOUT=300
+
+on_exit cleanup_and_exit "$APPLICATION_YAML" $TIMEOUT $CLUSTER_ID
+
+retry_times 5 30 "kubectl apply -f $APPLICATION_YAML" || exit 1
+
+wait_for_jobmanager_running $CLUSTER_ID $TIMEOUT
+jm_pod_name=$(get_jm_pod_name $CLUSTER_ID)
+
+wait_for_logs $jm_pod_name "Completed checkpoint [0-9]+ for job" ${TIMEOUT} || exit 1
+wait_for_status $APPLICATION_IDENTIFIER '.status.lifecycleState' STABLE ${TIMEOUT} || exit 1
+wait_for_event FlinkDeployment $CLUSTER_ID .reason==\"ScalingReport\" ${TIMEOUT} || exit 1
+wait_for_event FlinkDeployment $CLUSTER_ID .reason==\"SpecChanged\" ${TIMEOUT} || exit 1
+
+FLINK_VERSION=$(get_flink_version $APPLICATION_IDENTIFIER)
+echo "Flink version: $FLINK_VERSION"
+if [ "$FLINK_VERSION" = "v1_17" ]; then
+ echo "Verifying full upgrade triggered"
+ wait_for_event FlinkDeployment $CLUSTER_ID .reason==\"Suspended\" ${TIMEOUT} || exit 1
+else
+ echo "Verifying inplace scaling triggered"
+ wait_for_event FlinkDeployment $CLUSTER_ID .reason==\"Scaling\" ${TIMEOUT} || exit 1
+fi
+wait_for_status $APPLICATION_IDENTIFIER '.status.lifecycleState' STABLE ${TIMEOUT} || exit 1
+
+check_operator_log_for_errors || exit 1
+
+echo "Successfully run the autoscaler test"
diff --git a/e2e-tests/utils.sh b/e2e-tests/utils.sh
index 9d2d0566..adaba65f 100755
--- a/e2e-tests/utils.sh
+++ b/e2e-tests/utils.sh
@@ -57,6 +57,26 @@ function wait_for_status {
exit 1
}
+function wait_for_event {
+ local kind=$1
+ local resource=$2
+ local event_filter=$3
+ local timeout=$4
+
+ echo "Waiting for $resource event matching $event_filter..."
+ for i in $(seq 1 ${timeout}); do
+ test=$(kubectl get events --field-selector involvedObject.kind=$kind,involvedObject.name=$resource -oyaml | yq ".items.[] | select($event_filter)")
+ if [ "$test" ]; then
+ echo "Successfully verified that $resource event exists."
+ return 0
+ fi
+
+ sleep 1
+ done
+ echo "Event verification for $resource failed with timeout of ${timeout}."
+ exit 1
+}
+
function assert_available_slots() {
expected=$1
CLUSTER_ID=$2
@@ -130,6 +150,13 @@ function retry_times() {
return 1
}
+function get_flink_version() {
+ local resource=$1
+
+ kubectl get -oyaml $resource | yq ".spec.flinkVersion"
+}
+
+
function check_operator_log_for_errors {
local ignore=$1
echo "Checking for operator log errors..."
diff --git a/examples/autoscaling/autoscaling.yaml b/examples/autoscaling/autoscaling.yaml
index 40de8a01..26edaaad 100644
--- a/examples/autoscaling/autoscaling.yaml
+++ b/examples/autoscaling/autoscaling.yaml
@@ -31,7 +31,7 @@ spec:
taskmanager.numberOfTaskSlots: "4"
state.savepoints.dir: file:///flink-data/savepoints
state.checkpoints.dir: file:///flink-data/checkpoints
- high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory
+ high-availability.type: kubernetes
high-availability.storageDir: file:///flink-data/ha
execution.checkpointing.interval: "1m"
jobmanager.scheduler: adaptive