You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by po...@apache.org on 2020/11/14 15:07:49 UTC
[airflow] branch v1-10-test updated (01eb5d4 -> 430c047)
This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a change to branch v1-10-test
in repository https://gitbox.apache.org/repos/asf/airflow.git.
omit 01eb5d4 Simplifies check whether the CI image should be rebuilt (#12181)
omit 42b3491 For v1-10-test PRs and pushes, use target branch scripts for images (#12339)
omit 23151fd Deploy was not working from Breeze (#12319)
omit f1f58f7 Python base image is shared between CI and PROD image (#12280)
omit 533ea3c Added k9s as integrated tool to help with kubernetes testing (#12163)
new 2807875 Added k9s as integrated tool to help with kubernetes testing (#12163)
new b284205 Python base image is shared between CI and PROD image (#12280)
new 0a51dfd Deploy was not working from Breeze (#12319)
new a90131c For v1-10-test PRs and pushes, use target branch scripts for images (#12339)
new 430c047 Simplifies check whether the CI image should be rebuilt (#12181)
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (01eb5d4)
\
N -- N -- N refs/heads/v1-10-test (430c047)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
scripts/ci/libraries/_kind.sh | 2 --
1 file changed, 2 deletions(-)
[airflow] 02/05: Python base image is shared between CI and PROD
image (#12280)
Posted by po...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v1-10-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit b28420576048a6bedaa89e5bde9aec4277595cf1
Author: Jarek Potiuk <ja...@polidea.com>
AuthorDate: Thu Nov 12 12:31:14 2020 +0100
Python base image is shared between CI and PROD image (#12280)
When you are building CI images locally you use the CI
base images from apache:airflow/python* now to maintain
consistency and avoid often rebuilds. But when you build
prod images, you would accidentaly override it with the
python base image available in python repo which might be
different (newer and not yet tested in CI). This PR
changes it to use the same base image which is now
tagged in Apache Airflow's dockerhub repository.
(cherry picked from commit 3c2c29187afffb1a1dd1327803097651cdc079ee)
---
scripts/ci/libraries/_push_pull_remove_images.sh | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/scripts/ci/libraries/_push_pull_remove_images.sh b/scripts/ci/libraries/_push_pull_remove_images.sh
index 4c51f0a..7c65db1 100644
--- a/scripts/ci/libraries/_push_pull_remove_images.sh
+++ b/scripts/ci/libraries/_push_pull_remove_images.sh
@@ -154,7 +154,8 @@ function push_pull_remove_images::pull_prod_images_if_needed() {
fi
push_pull_remove_images::pull_image_github_dockerhub "${PYTHON_BASE_IMAGE}" "${GITHUB_REGISTRY_PYTHON_BASE_IMAGE}${PYTHON_TAG_SUFFIX}"
else
- docker pull "${PYTHON_BASE_IMAGE}"
+ docker pull "${AIRFLOW_CI_PYTHON_IMAGE}"
+ docker tag "${AIRFLOW_CI_PYTHON_IMAGE}" "${PYTHON_BASE_IMAGE}"
fi
echo
fi
[airflow] 05/05: Simplifies check whether the CI image should be
rebuilt (#12181)
Posted by po...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v1-10-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 430c047a0605125e463fdd3b45e7a848e2d26c13
Author: Jarek Potiuk <ja...@polidea.com>
AuthorDate: Fri Nov 13 22:21:39 2020 +0100
Simplifies check whether the CI image should be rebuilt (#12181)
Rather than counting changed layers in the image (which was
enigmatic, difficult and prone to some magic number) we rely now
on random file generated while building the image.
We are using the docker image caching mechanism here. The random
file will be regenerated only when the previous layer (which is
about installling Airflow dependencies for the first time) gets
rebuild. And for us this is the indication, that the building
the image will take quite some time. This layer should be
relatively static - even if setup.py changes the CI image is
designed in the way that the first time installation of Airflow
dependencies is not invalidated.
This should lead to faster and less frequent rebuild for people
using Breeze and static checks.
(cherry picked from commit 167b9b9889ac5481b21cb35c6cdef5869b8ab713)
---
Dockerfile.ci | 9 +-
breeze | 6 +-
manifests/.gitignore | 2 +-
scripts/ci/libraries/_build_images.sh | 163 ++++++++++++++++----------------
scripts/ci/libraries/_initialization.sh | 19 +++-
5 files changed, 106 insertions(+), 93 deletions(-)
diff --git a/Dockerfile.ci b/Dockerfile.ci
index 7699706..f06087b 100644
--- a/Dockerfile.ci
+++ b/Dockerfile.ci
@@ -132,7 +132,9 @@ ARG RUNTIME_APT_DEPS="\
sqlite3 \
tmux \
unzip \
- vim"
+ vim \
+ xxd"
+ENV RUNTIME_APT_DEP=${RUNTIME_APT_DEPS}
ARG ADDITIONAL_RUNTIME_APT_DEPS=""
ENV ADDITIONAL_RUNTIME_APT_DEPS=${ADDITIONAL_RUNTIME_APT_DEPS}
@@ -275,6 +277,11 @@ RUN if [[ ${AIRFLOW_PRE_CACHED_PIP_PACKAGES} == "true" ]]; then \
fi
+# Generate random hex dump file so that we can determine whether it's faster to rebuild the image
+# using current cache (when our dump is the same as the remote onb) or better to pull
+# the new image (when it is different)
+RUN head -c 30 /dev/urandom | xxd -ps >/build-cache-hash
+
# Link dumb-init for backwards compatibility (so that older images also work)
RUN ln -sf /usr/bin/dumb-init /usr/local/bin/dumb-init
diff --git a/breeze b/breeze
index 55412d0..175a4ab 100755
--- a/breeze
+++ b/breeze
@@ -2873,11 +2873,9 @@ function breeze::run_build_command() {
build_images::prepare_prod_build
build_images::build_prod_images
else
+
build_images::prepare_ci_build
- md5sum::calculate_md5sum_for_all_files
- build_images::build_ci_image
- md5sum::update_all_md5
- build_images::build_ci_image_manifest
+ build_images::rebuild_ci_image_if_needed
fi
;;
cleanup_image | run_exec)
diff --git a/manifests/.gitignore b/manifests/.gitignore
index a6c57f5..72e8ffc 100644
--- a/manifests/.gitignore
+++ b/manifests/.gitignore
@@ -1 +1 @@
-*.json
+*
diff --git a/scripts/ci/libraries/_build_images.sh b/scripts/ci/libraries/_build_images.sh
index a2577cb..52b8a84 100644
--- a/scripts/ci/libraries/_build_images.sh
+++ b/scripts/ci/libraries/_build_images.sh
@@ -136,10 +136,18 @@ function build_images::confirm_image_rebuild() {
;;
esac
elif [[ -t 0 ]]; then
+ echo
+ echo
+ echo "Make sure that you rebased to latest master before rebuilding!"
+ echo
# Check if this script is run interactively with stdin open and terminal attached
"${AIRFLOW_SOURCES}/confirm" "${ACTION} image ${THE_IMAGE_TYPE}-python${PYTHON_MAJOR_MINOR_VERSION}"
RES=$?
elif [[ ${DETECTED_TERMINAL:=$(tty)} != "not a tty" ]]; then
+ echo > "${DETECTED_TERMINAL}"
+ echo > "${DETECTED_TERMINAL}"
+ echo "Make sure that you rebased to latest master before rebuilding!" > "${DETECTED_TERMINAL}"
+ echo > "${DETECTED_TERMINAL}"
# Make sure to use output of tty rather than stdin/stdout when available - this way confirm
# will works also in case of pre-commits (git does not pass stdin/stdout to pre-commit hooks)
# shellcheck disable=SC2094
@@ -151,6 +159,10 @@ function build_images::confirm_image_rebuild() {
export DETECTED_TERMINAL=/dev/tty
# Make sure to use /dev/tty first rather than stdin/stdout when available - this way confirm
# will works also in case of pre-commits (git does not pass stdin/stdout to pre-commit hooks)
+ echo > "${DETECTED_TERMINAL}"
+ echo > "${DETECTED_TERMINAL}"
+ echo "Make sure that you rebased to latest master before rebuilding!" > "${DETECTED_TERMINAL}"
+ echo > "${DETECTED_TERMINAL}"
# shellcheck disable=SC2094
"${AIRFLOW_SOURCES}/confirm" "${ACTION} image ${THE_IMAGE_TYPE}-python${PYTHON_MAJOR_MINOR_VERSION}" \
<"${DETECTED_TERMINAL}" >"${DETECTED_TERMINAL}"
@@ -193,113 +205,95 @@ function build_images::confirm_image_rebuild() {
# We cannot use docker registry APIs as they are available only with authorisation
# But this image can be pulled without authentication
function build_images::build_ci_image_manifest() {
- docker inspect "${AIRFLOW_CI_IMAGE}" >"manifests/${AIRFLOW_CI_BASE_TAG}.json"
docker build \
- --build-arg AIRFLOW_CI_BASE_TAG="${AIRFLOW_CI_BASE_TAG}" \
--tag="${AIRFLOW_CI_LOCAL_MANIFEST_IMAGE}" \
-f- . <<EOF
-ARG AIRFLOW_CI_BASE_TAG
FROM scratch
-COPY "manifests/${AIRFLOW_CI_BASE_TAG}.json" .
+COPY "manifests/local-build-cache-hash" /build-cache-hash
CMD ""
EOF
}
#
-# Retrieves information about layers in the local IMAGE
-# it stores list of SHAs of image layers in the file pointed at by TMP_MANIFEST_LOCAL_SHA
+# Retrieves information about build cache hash random file from the local image
#
-function build_images::get_local_image_info() {
- TMP_MANIFEST_LOCAL_JSON=$(mktemp)
- TMP_MANIFEST_LOCAL_SHA=$(mktemp)
+function build_images::get_local_build_cache_hash() {
+
set +e
# Remove the container just in case
- docker rm --force "local-airflow-manifest" 2>/dev/null >/dev/null
- # Create manifest from the local manifest image
- if ! docker create --name "local-airflow-manifest" "${AIRFLOW_CI_LOCAL_MANIFEST_IMAGE}" 2>/dev/null; then
- echo
- echo "Local manifest image not available"
- echo
+ docker rm --force "local-airflow-ci-container" 2>/dev/null >/dev/null
+ if ! docker create --name "local-airflow-ci-container" "${AIRFLOW_CI_IMAGE}" 2>/dev/null; then
+ >&2 echo
+ >&2 echo "Local airflow CI image not available"
+ >&2 echo
LOCAL_MANIFEST_IMAGE_UNAVAILABLE="true"
+ export LOCAL_MANIFEST_IMAGE_UNAVAILABLE
+ touch "${LOCAL_IMAGE_BUILD_CACHE_HASH_FILE}"
return
fi
- # Create manifest from the local manifest image
- docker cp "local-airflow-manifest:${AIRFLOW_CI_BASE_TAG}.json" "${TMP_MANIFEST_LOCAL_JSON}"
- sed 's/ *//g' "${TMP_MANIFEST_LOCAL_JSON}" | grep '^"sha256:' >"${TMP_MANIFEST_LOCAL_SHA}"
- docker rm --force "local-airflow-manifest" 2>/dev/null >/dev/null
+ docker cp "local-airflow-ci-container:/build-cache-hash" \
+ "${LOCAL_IMAGE_BUILD_CACHE_HASH_FILE}" 2> /dev/null \
+ || touch "${LOCAL_IMAGE_BUILD_CACHE_HASH_FILE}"
set -e
+ echo
+ echo "Local build cache hash: '$(cat "${LOCAL_IMAGE_BUILD_CACHE_HASH_FILE}")'"
+ echo
}
-#
-# Retrieves information about layers in the remote IMAGE
-# it stores list of SHAs of image layers in the file pointed at by TMP_MANIFEST_REMOTE_SHA
-# This cannot be done easily with existing APIs of Dockerhub because they require additional authentication
-# even for public images. Therefore instead we are downloading a specially prepared manifest image
-# which is built together with the main image. This special manifest image is prepared during
-# building of the main image and contains single JSON file being result of docker inspect on that image
-# This image is from scratch so it is very tiny
-function build_images::get_remote_image_info() {
+# Retrieves information about the build cache hash random file from the remote image.
+# We actually use manifest image for that, which is a really, really small image to pull!
+# The problem is that inspecting information about remote image cannot be done easily with existing APIs
+# of Dockerhub because they require additional authentication even for public images.
+# Therefore instead we are downloading a specially prepared manifest image
+# which is built together with the main image and pushed with it. This special manifest image is prepared
+# during building of the main image and contains single file which is randomly built during the docker
+# build in the right place in the image (right after installing all dependencies of Apache Airflow
+# for the first time). When this random file gets regenerated it means that either base image has
+# changed or some of the earlier layers was modified - which means that it is usually faster to pull
+# that image first and then rebuild it - because this will likely be faster
+function build_images::get_remote_image_build_cache_hash() {
set +e
# Pull remote manifest image
if ! docker pull "${AIRFLOW_CI_REMOTE_MANIFEST_IMAGE}" 2>/dev/null >/dev/null; then
- echo >&2
- echo >&2 "Remote docker registry unreachable"
- echo >&2
+ >&2 echo
+ >&2 echo "Remote docker registry unreachable"
+ >&2 echo
REMOTE_DOCKER_REGISTRY_UNREACHABLE="true"
+ export REMOTE_DOCKER_REGISTRY_UNREACHABLE
+ touch "${REMOTE_IMAGE_BUILD_CACHE_HASH_FILE}"
return
fi
set -e
-
- # Docker needs the file passed to --cidfile to not exist, so we can't use mktemp
- TMP_CONTAINER_DIR="$(mktemp -d)"
- TMP_CONTAINER_ID="${TMP_CONTAINER_DIR}/remote-airflow-manifest-$$.container_id"
- FILES_TO_CLEANUP_ON_EXIT+=("$TMP_CONTAINER_ID")
-
- TMP_MANIFEST_REMOTE_JSON=$(mktemp)
- TMP_MANIFEST_REMOTE_SHA=$(mktemp)
- # Create container out of the manifest image without running it
- docker create --cidfile "${TMP_CONTAINER_ID}" "${AIRFLOW_CI_REMOTE_MANIFEST_IMAGE}" 2>/dev/null >/dev/null
+ # Create container dump out of the manifest image without actually running it
+ docker create --cidfile "${REMOTE_IMAGE_CONTAINER_ID_FILE}" "${AIRFLOW_CI_REMOTE_MANIFEST_IMAGE}" \
+ 2>/dev/null >/dev/null || true
# Extract manifest and store it in local file
- docker cp "$(cat "${TMP_CONTAINER_ID}"):${AIRFLOW_CI_BASE_TAG}.json" "${TMP_MANIFEST_REMOTE_JSON}"
- # Filter everything except SHAs of image layers
- sed 's/ *//g' "${TMP_MANIFEST_REMOTE_JSON}" | grep '^"sha256:' >"${TMP_MANIFEST_REMOTE_SHA}"
- docker rm --force "$(cat "${TMP_CONTAINER_ID}")" 2>/dev/null >/dev/null
+ docker cp "$(cat "${REMOTE_IMAGE_CONTAINER_ID_FILE}"):/build-cache-hash" \
+ "${REMOTE_IMAGE_BUILD_CACHE_HASH_FILE}" 2> /dev/null \
+ || touch "${REMOTE_IMAGE_BUILD_CACHE_HASH_FILE}"
+ docker rm --force "$(cat "${REMOTE_IMAGE_CONTAINER_ID_FILE}")" 2>/dev/null || true
+ echo
+ echo "Remote build cache hash: '$(cat "${REMOTE_IMAGE_BUILD_CACHE_HASH_FILE}")'"
+ echo
}
-# The Number determines the cut-off between local building time and pull + build time.
-# It is a bit experimental and it will have to be kept
-# updated as we keep on changing layers. The cut-off point is at the moment when we do first
-# pip install "https://github.com/apache/airflow/archive/${AIRFLOW_BRANCH}.tar...
-# you can get it via this command:
-# docker history --no-trunc apache/airflow:master-python3.6-ci | \
-# grep ^sha256 | grep -n "pip uninstall" | awk 'BEGIN {FS=":"} {print $1 }'
-#
-# This command returns the number of layer in docker history where pip uninstall is called. This is the
-# line that will take a lot of time to run and at this point it's worth to pull the image from repo
-# if there are at least NN changed layers in your docker file, you should pull the image.
-#
-# Note that this only matters if you have any of the important files changed since the last build
-# of your image such as Dockerfile.ci, setup.py etc.
-#
-MAGIC_CUT_OFF_NUMBER_OF_LAYERS=36
-
# Compares layers from both remote and local image and set FORCE_PULL_IMAGES to true in case
# More than the last NN layers are different.
-function build_images::compare_layers() {
- NUM_DIFF=$(diff -y --suppress-common-lines "${TMP_MANIFEST_REMOTE_SHA}" "${TMP_MANIFEST_LOCAL_SHA}" |
- wc -l || true)
- rm -f "${TMP_MANIFEST_REMOTE_JSON}" "${TMP_MANIFEST_REMOTE_SHA}" "${TMP_MANIFEST_LOCAL_JSON}" "${TMP_MANIFEST_LOCAL_SHA}"
- echo
- echo "Number of layers different between the local and remote image: ${NUM_DIFF}"
- echo
- # This is where setup py is rebuilt - it will usually take a looooot of time to build it, so it is
- # Better to pull here
- if ((NUM_DIFF >= MAGIC_CUT_OFF_NUMBER_OF_LAYERS)); then
+function build_images::compare_local_and_remote_build_cache_hash() {
+ set +e
+ local remote_hash
+ remote_hash=$(cat "${REMOTE_IMAGE_BUILD_CACHE_HASH_FILE}")
+ local local_hash
+ local_hash=$(cat "${LOCAL_IMAGE_BUILD_CACHE_HASH_FILE}")
+
+ if [[ ${remote_hash} != "${local_hash}" ||
+ ${local_hash} == "" ]]; then
echo
echo
- echo "WARNING! Your image and the dockerhub image differ significantly"
+ echo "Your image and the dockerhub have different or missing build cache hashes."
+ echo "Local hash: '${local_hash}'. Remote hash: '${remote_hash}'."
echo
echo "Forcing pulling the images. It will be faster than rebuilding usually."
echo "You can avoid it by setting SKIP_CHECK_REMOTE_IMAGE to true"
@@ -307,9 +301,10 @@ function build_images::compare_layers() {
export FORCE_PULL_IMAGES="true"
else
echo
- echo "No need to pull the image. Local rebuild will be faster"
+ echo "No need to pull the image. Yours and remote cache hashes are the same!"
echo
fi
+ set -e
}
# Prints summary of the build parameters
@@ -335,7 +330,6 @@ function build_images::get_docker_image_names() {
# CI image to build
export AIRFLOW_CI_IMAGE="${DOCKERHUB_USER}/${DOCKERHUB_REPO}:${AIRFLOW_CI_BASE_TAG}"
-
# Base production image tag - used to build kubernetes tag as well
if [[ ${FORCE_AIRFLOW_PROD_BASE_TAG=} == "" ]]; then
export AIRFLOW_PROD_BASE_TAG="${BRANCH_NAME}-python${PYTHON_MAJOR_MINOR_VERSION}"
@@ -396,6 +390,9 @@ function build_images::prepare_ci_build() {
# In case rebuild is needed, it determines (by comparing layers in local and remote image)
# Whether pull is needed before rebuild.
function build_images::rebuild_ci_image_if_needed() {
+ verbosity::print_info
+ verbosity::print_info "Checking if pull or just build for ${THE_IMAGE_TYPE} is needed."
+ verbosity::print_info
if [[ -f "${BUILT_CI_IMAGE_FLAG_FILE}" ]]; then
verbosity::print_info
verbosity::print_info "${THE_IMAGE_TYPE} image already built locally."
@@ -418,6 +415,7 @@ function build_images::rebuild_ci_image_if_needed() {
local needs_docker_build="false"
md5sum::check_if_docker_build_is_needed
+ build_images::get_local_build_cache_hash
if [[ ${needs_docker_build} == "true" ]]; then
if [[ ${SKIP_CHECK_REMOTE_IMAGE:=} != "true" && ${DOCKER_CACHE} == "pulled" ]]; then
# Check if remote image is different enough to force pull
@@ -427,14 +425,12 @@ function build_images::rebuild_ci_image_if_needed() {
echo
echo "Checking if the remote image needs to be pulled"
echo
- build_images::get_remote_image_info
- if [[ ${REMOTE_DOCKER_REGISTRY_UNREACHABLE:=} != "true" ]]; then
- build_images::get_local_image_info
- if [[ ${LOCAL_MANIFEST_IMAGE_UNAVAILABLE:=} != "true" ]]; then
- build_images::compare_layers
- else
- FORCE_PULL_IMAGES="true"
- fi
+ build_images::get_remote_image_build_cache_hash
+ if [[ ${REMOTE_DOCKER_REGISTRY_UNREACHABLE:=} != "true" && \
+ ${LOCAL_MANIFEST_IMAGE_UNAVAILABLE:=} != "true" ]]; then
+ build_images::compare_local_and_remote_build_cache_hash
+ else
+ FORCE_PULL_IMAGES="true"
fi
fi
SKIP_REBUILD="false"
@@ -453,6 +449,7 @@ function build_images::rebuild_ci_image_if_needed() {
verbosity::print_info "Build start: ${THE_IMAGE_TYPE} image."
verbosity::print_info
build_images::build_ci_image
+ build_images::get_local_build_cache_hash
md5sum::update_all_md5
build_images::build_ci_image_manifest
verbosity::print_info
diff --git a/scripts/ci/libraries/_initialization.sh b/scripts/ci/libraries/_initialization.sh
index b58fa2e..396bded 100644
--- a/scripts/ci/libraries/_initialization.sh
+++ b/scripts/ci/libraries/_initialization.sh
@@ -186,9 +186,9 @@ function initialization::initialize_files_for_rebuild_check() {
"Dockerfile.ci"
".dockerignore"
"airflow/version.py"
- "airflow/www/package.json"
- "airflow/www/yarn.lock"
- "airflow/www/webpack.config.js"
+ "airflow/www_rbac/package.json"
+ "airflow/www_rbac/yarn.lock"
+ "airflow/www_rbac/webpack.config.js"
)
}
@@ -452,6 +452,12 @@ function initialization::initialize_test_variables() {
export TEST_TYPE=${TEST_TYPE:=""}
}
+function initialization::initialize_build_image_variables() {
+ REMOTE_IMAGE_CONTAINER_ID_FILE="${AIRFLOW_SOURCES}/manifests/remote-airflow-manifest-image"
+ LOCAL_IMAGE_BUILD_CACHE_HASH_FILE="${AIRFLOW_SOURCES}/manifests/local-build-cache-hash"
+ REMOTE_IMAGE_BUILD_CACHE_HASH_FILE="${AIRFLOW_SOURCES}/manifests/remote-build-cache-hash"
+}
+
# Common environment that is initialized by both Breeze and CI scripts
function initialization::initialize_common_environment() {
initialization::create_directories
@@ -469,6 +475,7 @@ function initialization::initialize_common_environment() {
initialization::initialize_git_variables
initialization::initialize_github_variables
initialization::initialize_test_variables
+ initialization::initialize_build_image_variables
}
function initialization::set_default_python_version_if_empty() {
@@ -750,6 +757,10 @@ function initialization::make_constants_read_only() {
readonly BUILT_CI_IMAGE_FLAG_FILE
readonly INIT_SCRIPT_FILE
+ readonly REMOTE_IMAGE_CONTAINER_ID_FILE
+ readonly LOCAL_IMAGE_BUILD_CACHE_HASH_FILE
+ readonly REMOTE_IMAGE_BUILD_CACHE_HASH_FILE
+
}
# converts parameters to json array
@@ -772,7 +783,7 @@ function initialization::ga_output() {
function initialization::ga_env() {
if [[ ${GITHUB_ENV=} != "" ]]; then
- echo "${1}=${2}" >> "${GITHUB_ENV}"
+ echo "${1}=${2}" >>"${GITHUB_ENV}"
fi
}
[airflow] 03/05: Deploy was not working from Breeze (#12319)
Posted by po...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v1-10-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 0a51dfd78c3bf58eaacf4c1136228bb50388a7ce
Author: Jarek Potiuk <ja...@polidea.com>
AuthorDate: Thu Nov 12 19:53:35 2020 +0100
Deploy was not working from Breeze (#12319)
The get_cluster_name was called twice resulting in redonly
error after rebasing/fixing CI failure in #12163.
This PR is fxing it.
(cherry picked from commit af19b126e94876c371553f6a7cfae6b1102f79fd)
---
scripts/ci/libraries/_kind.sh | 1 -
1 file changed, 1 deletion(-)
diff --git a/scripts/ci/libraries/_kind.sh b/scripts/ci/libraries/_kind.sh
index defa4de..3a170cb 100644
--- a/scripts/ci/libraries/_kind.sh
+++ b/scripts/ci/libraries/_kind.sh
@@ -173,7 +173,6 @@ function kind::perform_kind_cluster_operation() {
echo "Deploying Airflow to KinD"
echo
kind::build_image_for_kubernetes_tests
- kind::get_kind_cluster_name
kind::load_image_to_kind_cluster
kind::deploy_airflow_with_helm
kind::deploy_test_kubernetes_resources
[airflow] 01/05: Added k9s as integrated tool to help with
kubernetes testing (#12163)
Posted by po...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v1-10-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 280787593252d6cf9e60c2cd078f9b6abb3d57cd
Author: Jarek Potiuk <ja...@polidea.com>
AuthorDate: Wed Nov 11 17:15:02 2020 +0100
Added k9s as integrated tool to help with kubernetes testing (#12163)
The K9s is fantastic tool that helps to debug a running k8s
instance. It is terminal-based windowed CLI that makes you
several times more productive comparing to using kubectl
commands. We've integrated k9s (it is run as a docker container
and downloaded on demand). We've also separated out KUBECONFIG
of the integrated kind cluster so that it does not mess with
kubernetes configuration you might already have.
Also - together with that the "surrounding" of the kubernetes
tests were simplified and improved so that the k9s integration
can be utilized well. Instead of kubectl port forwarding (which
caused multitude of problems) we are now utilizing kind's
portMapping feature + custom NodePort resource that maps
port 8080 to 30007 NodePort which in turn maps it to 8080
port of the Webserver. This way we do not have to establish
an external kubectl port forward which is prone to error and
management - everything is brought up when Airflow gets
deployed to the Kind Cluster and shuts down when the Kind
cluster is stopped.
Yet another problem fixed was killing of postgres by one of the
kubernetes tests ('test_integration_run_dag_with_scheduler_failure').
Instead of just killing the scheduler it killed all pods - including
the Postgres one (it was named 'airflow-postgres.*'). That caused
various problems, as the database could be left in a strange state.
I changed the tests to do what it claimed was doing - so killing only the
scheduler during the test. This seemed to improve the stability
of tests immensely in my local setup.
(cherry picked from commit 21999dd56e9dbe9f7f9e25961954c5677c3c7c58)
---
.github/workflows/ci.yml | 17 +-
BREEZE.rst | 11 +-
TESTING.rst | 151 ++++-
breeze | 9 +
breeze-complete | 5 +-
chart/requirements.lock | 4 +-
images/testing/k9s.png | Bin 0 -> 238713 bytes
images/testing/kubeconfig-env.png | Bin 0 -> 231280 bytes
images/testing/kubernetes-virtualenv.png | Bin 0 -> 110011 bytes
images/testing/pytest-runner.png | Bin 0 -> 131589 bytes
images/testing/run-test.png | Bin 0 -> 140728 bytes
kubernetes_tests/test_kubernetes_executor.py | 7 +-
kubernetes_tests/test_kubernetes_pod_operator.py | 673 ++++++++-------------
scripts/ci/kubernetes/ci_run_kubernetes_tests.sh | 2 -
...up_cluster_and_deploy_airflow_to_kubernetes.sh} | 3 +-
scripts/ci/kubernetes/kind-cluster-conf.yaml | 5 +
.../{kind-cluster-conf.yaml => nodeport.yaml} | 30 +-
...oy_app_to_kubernetes.sh => redeploy_airflow.sh} | 6 +-
scripts/ci/libraries/_kind.sh | 126 ++--
19 files changed, 512 insertions(+), 537 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 6c854a1..81890a7 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -636,23 +636,14 @@ jobs:
python-version: ${{ env.PYTHON_MAJOR_MINOR_VERSION }}
- name: "Free space"
run: ./scripts/ci/tools/ci_free_space_on_ci.sh
- - name: "Setup Kind Cluster ${{ env.KIND_VERSION }}"
- uses: engineerd/setup-kind@v0.4.0
- with:
- version: "${{ env.KIND_VERSION }}"
- name: airflow-python-${{matrix.python-version}}-${{matrix.kubernetes-version}}
- config: "scripts/ci/kubernetes/kind-cluster-conf.yaml"
- name: "Prepare PROD Image"
run: ./scripts/ci/images/ci_prepare_prod_image_on_ci.sh
- - name: "Deploy airflow to cluster"
- id: deploy-app
- run: ./scripts/ci/kubernetes/ci_deploy_app_to_kubernetes.sh
+ - name: "Setup cluster and deploy Airflow"
+ id: setp-cluster-deploy-app
+ run: ./scripts/ci/kubernetes/ci_setup_cluster_and_deploy_airflow_to_kubernetes.sh
env:
# We have the right image pulled already by the previous step
SKIP_BUILDING_PROD_IMAGE: "true"
- # due to some instabilities, in CI we try to increase port numbers when trying to establish
- # port forwarding
- INCREASE_PORT_NUMBER_FOR_KUBERNETES: "true"
- name: "Cache virtualenv for kubernetes testing"
uses: actions/cache@v2
env:
@@ -669,7 +660,7 @@ jobs:
key: "${{ env.cache-name }}-${{ github.job }}-${{ hashFiles('setup.py') }}\
-${{ needs.build-info.outputs.defaultKindVersion }}\
-${{ needs.build-info.outputs.defaultHelmVersion }}\
--$${{ matrix.kubernetes-version }}"
+-${{ matrix.kubernetes-version }}"
- name: "Kubernetes Tests"
run: ./scripts/ci/kubernetes/ci_run_kubernetes_tests.sh
- name: "Upload KinD logs"
diff --git a/BREEZE.rst b/BREEZE.rst
index c3c2d95..ce7dc6a 100644
--- a/BREEZE.rst
+++ b/BREEZE.rst
@@ -1188,6 +1188,7 @@ This is the current syntax for `./breeze <./breeze>`_:
image building time in production image and at container entering time for CI image. One of:
1.10.12 1.10.11 1.10.10 1.10.9 1.10.8 1.10.7 1.10.6 1.10.5 1.10.4 1.10.3 1.10.2
+ wheel
-t, --install-airflow-reference INSTALL_AIRFLOW_REFERENCE
If specified, installs Airflow directly from reference in GitHub. This happens at
@@ -1712,7 +1713,14 @@ This is the current syntax for `./breeze <./breeze>`_:
to the cluster so you can also pass appropriate build image flags that will influence
rebuilding the production image. Operation is one of:
- start stop restart status deploy test shell
+ start stop restart status deploy test shell k9s
+
+ The last two operations - shell and k9s allow you to perform interactive testing with
+ kubernetes tests. You can enter the shell from which you can run kubernetes tests and in
+ another terminal you can start the k9s CLI to debug kubernetes instance. It is an easy
+ way to debug the kubernetes deployments.
+
+ You can read more about k9s at https://k9scli.io/
Flags:
@@ -2087,6 +2095,7 @@ This is the current syntax for `./breeze <./breeze>`_:
image building time in production image and at container entering time for CI image. One of:
1.10.12 1.10.11 1.10.10 1.10.9 1.10.8 1.10.7 1.10.6 1.10.5 1.10.4 1.10.3 1.10.2
+ wheel
-t, --install-airflow-reference INSTALL_AIRFLOW_REFERENCE
If specified, installs Airflow directly from reference in GitHub. This happens at
diff --git a/TESTING.rst b/TESTING.rst
index c8b170b..f6c6c10 100644
--- a/TESTING.rst
+++ b/TESTING.rst
@@ -418,6 +418,14 @@ can also decide to only run tests with ``-m quarantined`` flag to run only those
Running Tests with Kubernetes
=============================
+Airflow has tests that are run against real kubernetes cluster. We are using
+`Kind <https://kind.sigs.k8s.io/>`_ to create and run the cluster. We integrated the tools to start/stop/
+deploy and run the cluster tests in our repository and into Breeze development environment.
+
+Configuration for the cluster is kept in ``./build/.kube/config`` file in your Airflow source repository and
+our scripts set the ``KUBECONFIG`` variable to it. If you want to interact with the Kind cluster created
+you can do it from outside of the scripts by exporting this variable and point it to this file.
+
Starting Kubernetes Cluster
---------------------------
@@ -425,7 +433,7 @@ For your testing you manage Kind cluster with ``kind-cluster`` breeze command:
.. code-block:: bash
- ./breeze kind-cluster [ start | stop | recreate | status | deploy | test | shell ]
+ ./breeze kind-cluster [ start | stop | recreate | status | deploy | test | shell | k9s ]
The command allows you to start/stop/recreate/status Kind Kubernetes cluster, deploy Airflow via Helm
chart as well as interact with the cluster (via test and shell commands).
@@ -444,11 +452,11 @@ Deploying Airflow to Kubernetes Cluster
Deploying Airflow to the Kubernetes cluster created is also done via ``kind-cluster deploy`` breeze command:
-.. code-block:: bash`
+.. code-block:: bash
./breeze kind-cluster deploy
-The deploy commands performs tthose steps:
+The deploy commands performs those steps:
1. It rebuilds the latest ``apache/airflow:master-pythonX.Y`` production images using the
latest sources using local cachine. It also adds example DAGs to the image, so that they do not
@@ -465,20 +473,63 @@ Running tests with Kubernetes Cluster
You can either run all tests or you can select which tests to run. You can also enter interactive virtualenv
to run the tests manually one by one.
-.. code-block:: bash
+Running kubernetes tests via shell:
- Running kubernetes tests
+.. code-block:: bash
./scripts/ci/kubernetes/ci_run_kubernetes_tests.sh - runs all kubernetes tests
./scripts/ci/kubernetes/ci_run_kubernetes_tests.sh TEST [TEST ...] - runs selected kubernetes tests (from kubernetes_tests folder)
+
+
+Running kubernetes tests via breeze:
+
+.. code-block:: bash
+
+ ./breeze kind-cluster test
+ ./breeze kind-cluster test -- TEST TEST [TEST ...]
+
+
+Entering shell with Kubernetes Cluster
+--------------------------------------
+
+This shell is prepared to run kubernetes tests interactively. It has ``kubectl`` and ``kind`` cli tools
+available in the path, it has also activated virtualenv environment that allows you to run tests via pytest.
+
+You can enter the shell via those scripts
+
./scripts/ci/kubernetes/ci_run_kubernetes_tests.sh [-i|--interactive] - Activates virtual environment ready to run tests and drops you in
./scripts/ci/kubernetes/ci_run_kubernetes_tests.sh [--help] - Prints this help message
-You can also run the same tests command with Breeze, using ``kind-cluster test`` command (to run all
-kubernetes tests) and with ``kind-cluster shell`` command you can enter interactive shell when you can
-run tests.
+.. code-block:: bash
+
+ ./breeze kind-cluster shell
+
+K9s CLI - debug kubernetes in style!
+------------------------------------
+
+Breeze has built-in integration with fantastic k9s CLI tool, that allows you to debug the kubernetes
+installation effortlessly and in style. K9S provides terminal (but windowed) CLI that allows you to
+easily observe what's going on in the kubernetes instance, observe the resources defined (pods, secrets,
+custom resource definitions), enter shell for the Pods/Containers running, see the log files and more.
+
+You can read more about k9s at `https://k9scli.io/ <https://k9scli.io/>`_
+
+Here is the screenshot of k9s tools in operation:
+
+.. image:: images/testing/k9s.png
+ :align: center
+ :alt: K9S tool
+
+
+You can enter the k9s tool via breeze (after you deployed Airflow):
+
+.. code-block:: bash
+
+ ./breeze kind-cluster k9s
+
+You can exit k9s by pressing Ctrl-C.
Typical testing pattern for Kubernetes tests
--------------------------------------------
@@ -578,7 +629,6 @@ This prepares and enters the virtualenv in ``.build/.kubernetes_venv`` folder:
./breeze kind-cluster shell
-
Once you enter the environment you receive this information:
@@ -595,12 +645,67 @@ Once you enter the environment you receive this information:
You are entering the virtualenv now. Type exit to exit back to the original shell
+In a separate terminal you can open the k9s CLI:
+
+.. code-block:: bash
+
+ ./breeze kind-cluster k9s
+
+Use it to observe what's going on in your cluster.
+
+6. Debugging in IntelliJ/PyCharm
+
+It is very easy to running/debug Kubernetes tests with IntelliJ/PyCharm. Unlike the regular tests they are
+in ``kubernetes_tests`` folder and if you followed the previous steps and entered the shell using
+``./breeze kind-cluster shell`` command, you can setup your IDE very easily to run (and debug) your
+tests using the standard IntelliJ Run/Debug feature. You just need a few steps:
+
+a) Add the virtualenv as interpreter for the project:
+
+.. image:: images/testing/kubernetes-virtualenv.png
+ :align: center
+ :alt: Kubernetes testing virtualenv
+
+The virtualenv is created in your "Airflow" source directory in ``.build/.kubernetes_venv/`` folder and you
+have to find ``python`` binary and choose it when selecting interpreter.
+
+b) Choose pytest as test runner:
+
+.. image:: images/testing/pytest-runner.png
+ :align: center
+ :alt: Pytest runner
+
+c) Run/Debug tests using standard "Run/Debug" feature of IntelliJ
+
+.. image:: images/testing/run-tests.png
+ :align: center
+ :alt: Run/Debug tests
+
+
+NOTE! The first time you run it, it will likely fail with
+``kubernetes.config.config_exception.ConfigException``:
+``Invalid kube-config file. Expected key current-context in kube-config``. You need to add KUBECONFIG
+environment variabl copying it from the result of "./breeze kind-cluster test":
+
+.. code-block:: bash
+
+ echo ${KUBECONFIG}
+
+ /home/jarek/code/airflow/.build/.kube/config
+
+
+.. image:: images/testing/kubeconfig-env.png
+ :align: center
+ :alt: Run/Debug tests
+
+
+The configuration for kubernetes is stored in your "Airflow" source directory in ".build/.kube/config" file
+and this is where KUBECONFIG env should point to.
You can iterate with tests while you are in the virtualenv. All the tests requiring kubernetes cluster
are in "kubernetes_tests" folder. You can add extra ``pytest`` parameters then (for example ``-s`` will
print output generated test logs and print statements to the terminal immediately.
-
.. code-block:: bash
pytest kubernetes_tests/test_kubernetes_executor.py::TestKubernetesExecutor::test_integration_run_dag_with_scheduler_failure -s
@@ -609,6 +714,30 @@ print output generated test logs and print statements to the terminal immediatel
You can modify the tests or KubernetesPodOperator and re-run them without re-deploying
airflow to KinD cluster.
+
+Sometimes there are side effects from running tests. You can run ``redeploy_airflow.sh`` without
+recreating the whole cluster. This will delete the whole namespace, including the database data
+and start a new Airflow deployment in the cluster.
+
+.. code-block:: bash
+
+ ./scripts/ci/redeploy_airflow.sh
+
+If needed you can also delete the cluster manually:
+
+
+.. code-block:: bash
+
+ kind get clusters
+ kind delete clusters <NAME_OF_THE_CLUSTER>
+
+Kind has also useful commands to inspect your running cluster:
+
+.. code-block:: text
+
+ kind --help
+
+
However, when you change Airflow Kubernetes executor implementation you need to redeploy
Airflow to the cluster.
@@ -617,7 +746,7 @@ Airflow to the cluster.
./breeze kind-cluster deploy
-5. Stop KinD cluster when you are done
+7. Stop KinD cluster when you are done
.. code-block:: bash
diff --git a/breeze b/breeze
index 3498c64..55412d0 100755
--- a/breeze
+++ b/breeze
@@ -1684,6 +1684,13 @@ ${CMDNAME} kind-cluster [FLAGS] OPERATION
${FORMATTED_KIND_OPERATIONS}
+ The last two operations - shell and k9s allow you to perform interactive testing with
+ kubernetes tests. You can enter the shell from which you can run kubernetes tests and in
+ another terminal you can start the k9s CLI to debug kubernetes instance. It is an easy
+ way to debug the kubernetes deployments.
+
+ You can read more about k9s at https://k9scli.io/
+
Flags:
$(breeze::flag_airflow_variants)
$(breeze::flag_build_docker_images)
@@ -2901,6 +2908,8 @@ function breeze::run_build_command() {
echo "Run Kubernetes tests with the KinD cluster "
elif [[ ${KIND_CLUSTER_OPERATION} == "shell" ]]; then
echo "Enter an interactive shell for kubernetes testing"
+ elif [[ ${KIND_CLUSTER_OPERATION} == "k9s" ]]; then
+ echo "Run k9s cli to debug in style"
elif [[ -z ${KIND_CLUSTER_OPERATION=} ]]; then
echo
echo "Please provide an operation to run"
diff --git a/breeze-complete b/breeze-complete
index 4855c86..94854ad 100644
--- a/breeze-complete
+++ b/breeze-complete
@@ -32,7 +32,7 @@ _breeze_allowed_helm_versions="v3.2.4"
_breeze_allowed_kind_versions="v0.8.0"
_breeze_allowed_mysql_versions="5.6 5.7"
_breeze_allowed_postgres_versions="9.6 10 11 12 13"
-_breeze_allowed_kind_operations="start stop restart status deploy test shell"
+_breeze_allowed_kind_operations="start stop restart status deploy test shell k9s"
_breeze_allowed_test_types="All Core Integration Heisentests Postgres MySQL Helm"
# shellcheck disable=SC2034
@@ -60,6 +60,7 @@ _breeze_allowed_install_airflow_versions=$(cat <<-EOF
1.10.4
1.10.3
1.10.2
+wheel
EOF
)
@@ -134,7 +135,7 @@ _breeze_long_options="
help python: backend: integration:
kubernetes-mode: kubernetes-version: helm-version: kind-version:
skip-mounting-local-sources install-airflow-version: install-airflow-reference: db-reset
-verbose assume-yes assume-no assume-quit forward-credentials rbac-ui init-script:
+verbose assume-yes assume-no assume-quit forward-credentials init-script:
force-build-images force-pull-images production-image extras: force-clean-images skip-rebuild-check
build-cache-local build-cache-pulled build-cache-disabled disable-pip-cache
dockerhub-user: dockerhub-repo: github-registry github-repository: github-image-id:
diff --git a/chart/requirements.lock b/chart/requirements.lock
index 3f3c34a..e460e9f 100644
--- a/chart/requirements.lock
+++ b/chart/requirements.lock
@@ -2,5 +2,5 @@ dependencies:
- name: postgresql
repository: https://charts.helm.sh/stable/
version: 6.3.12
-digest: sha256:58d88cf56e78b2380091e9e16cc6ccf58b88b3abe4a1886dd47cd9faef5309af
-generated: "2020-11-04T15:59:36.967913-08:00"
+digest: sha256:1748aa702050d4e72ffba1b18960f49bfe5368757cf976116afeffbdedda1c98
+generated: "2020-11-07T17:40:45.418723358+01:00"
diff --git a/images/testing/k9s.png b/images/testing/k9s.png
new file mode 100644
index 0000000..a8eec97
Binary files /dev/null and b/images/testing/k9s.png differ
diff --git a/images/testing/kubeconfig-env.png b/images/testing/kubeconfig-env.png
new file mode 100644
index 0000000..b2ebfd5
Binary files /dev/null and b/images/testing/kubeconfig-env.png differ
diff --git a/images/testing/kubernetes-virtualenv.png b/images/testing/kubernetes-virtualenv.png
new file mode 100644
index 0000000..6e208d6
Binary files /dev/null and b/images/testing/kubernetes-virtualenv.png differ
diff --git a/images/testing/pytest-runner.png b/images/testing/pytest-runner.png
new file mode 100644
index 0000000..fdb48cc
Binary files /dev/null and b/images/testing/pytest-runner.png differ
diff --git a/images/testing/run-test.png b/images/testing/run-test.png
new file mode 100644
index 0000000..21a5c9d
Binary files /dev/null and b/images/testing/run-test.png differ
diff --git a/kubernetes_tests/test_kubernetes_executor.py b/kubernetes_tests/test_kubernetes_executor.py
index 694cf75..bb89cb7 100644
--- a/kubernetes_tests/test_kubernetes_executor.py
+++ b/kubernetes_tests/test_kubernetes_executor.py
@@ -64,10 +64,11 @@ class TestKubernetesExecutor(unittest.TestCase):
return len(names)
@staticmethod
- def _delete_airflow_pod():
+ def _delete_airflow_pod(name=''):
+ suffix = '-' + name if name else ''
air_pod = check_output(['kubectl', 'get', 'pods']).decode()
air_pod = air_pod.split('\n')
- names = [re.compile(r'\s+').split(x)[0] for x in air_pod if 'airflow' in x]
+ names = [re.compile(r'\s+').split(x)[0] for x in air_pod if 'airflow' + suffix in x]
if names:
check_call(['kubectl', 'delete', 'pod', names[0]])
@@ -233,7 +234,7 @@ class TestKubernetesExecutor(unittest.TestCase):
execution_date = self.start_job_in_kubernetes(dag_id, host)
- self._delete_airflow_pod()
+ self._delete_airflow_pod("scheduler")
time.sleep(10) # give time for pod to restart
diff --git a/kubernetes_tests/test_kubernetes_pod_operator.py b/kubernetes_tests/test_kubernetes_pod_operator.py
index 7a8674a..39b7900 100644
--- a/kubernetes_tests/test_kubernetes_pod_operator.py
+++ b/kubernetes_tests/test_kubernetes_pod_operator.py
@@ -15,43 +15,38 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-
import json
import logging
import os
+import random
import shutil
import sys
-import unittest
import textwrap
+import unittest
+from unittest import mock
+from unittest.mock import ANY
-import kubernetes.client.models as k8s
import pendulum
+from kubernetes.client import models as k8s
from kubernetes.client.api_client import ApiClient
from kubernetes.client.rest import ApiException
-from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator
from airflow.exceptions import AirflowException
from airflow.kubernetes import kube_client
-from airflow.kubernetes.pod import Port
from airflow.kubernetes.pod_generator import PodDefaults
from airflow.kubernetes.pod_launcher import PodLauncher
-from airflow.kubernetes.pod_runtime_info_env import PodRuntimeInfoEnv
from airflow.kubernetes.secret import Secret
-from airflow.kubernetes.volume import Volume
-from airflow.kubernetes.volume_mount import VolumeMount
from airflow.models import DAG, TaskInstance
+from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from airflow.utils import timezone
from airflow.version import version as airflow_version
-from tests.compat import mock, patch
-# noinspection DuplicatedCode
def create_context(task):
dag = DAG(dag_id="dag")
tzinfo = pendulum.timezone("Europe/Amsterdam")
execution_date = timezone.datetime(2016, 1, 1, 1, 0, 0, tzinfo=tzinfo)
- task_instance = TaskInstance(task=task,
- execution_date=execution_date)
+ task_instance = TaskInstance(task=task, execution_date=execution_date)
return {
"dag": dag,
"ts": execution_date.isoformat(),
@@ -60,7 +55,11 @@ def create_context(task):
}
-# noinspection DuplicatedCode,PyUnusedLocal
+def get_kubeconfig_path():
+ kubeconfig_path = os.environ.get('KUBECONFIG')
+ return kubeconfig_path if kubeconfig_path else os.path.expanduser('~/.kube/config')
+
+
class TestKubernetesPodOperatorSystem(unittest.TestCase):
def get_current_task_name(self):
# reverse test name to make pod name unique (it has limited length)
@@ -74,29 +73,33 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
'kind': 'Pod',
'metadata': {
'namespace': 'default',
- 'name': mock.ANY,
+ 'name': ANY,
'annotations': {},
'labels': {
- 'foo': 'bar', 'kubernetes_pod_operator': 'True',
+ 'foo': 'bar',
+ 'kubernetes_pod_operator': 'True',
'airflow_version': airflow_version.replace('+', '-'),
'execution_date': '2016-01-01T0100000100-a2f50a31f',
'dag_id': 'dag',
- 'task_id': 'task',
- 'try_number': '1'},
+ 'task_id': ANY,
+ 'try_number': '1',
+ },
},
'spec': {
'affinity': {},
- 'containers': [{
- 'image': 'ubuntu:16.04',
- 'args': ["echo 10"],
- 'command': ["bash", "-cx"],
- 'env': [],
- 'imagePullPolicy': 'IfNotPresent',
- 'envFrom': [],
- 'name': 'base',
- 'ports': [],
- 'volumeMounts': [],
- }],
+ 'containers': [
+ {
+ 'image': 'ubuntu:16.04',
+ 'args': ["echo 10"],
+ 'command': ["bash", "-cx"],
+ 'env': [],
+ 'envFrom': [],
+ 'resources': {},
+ 'name': 'base',
+ 'ports': [],
+ 'volumeMounts': [],
+ }
+ ],
'hostNetwork': False,
'imagePullSecrets': [],
'initContainers': [],
@@ -106,29 +109,19 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
'serviceAccountName': 'default',
'tolerations': [],
'volumes': [],
- }
+ },
}
def tearDown(self):
client = kube_client.get_kube_client(in_cluster=False)
client.delete_collection_namespaced_pod(namespace="default")
+ import time
- def create_context(self, task):
- dag = DAG(dag_id="dag")
- tzinfo = pendulum.timezone("Europe/Amsterdam")
- execution_date = timezone.datetime(2016, 1, 1, 1, 0, 0, tzinfo=tzinfo)
- task_instance = TaskInstance(task=task,
- execution_date=execution_date)
- return {
- "dag": dag,
- "ts": execution_date.isoformat(),
- "task": task,
- "ti": task_instance,
- }
+ time.sleep(1)
def test_do_xcom_push_defaults_false(self):
new_config_path = '/tmp/kube_config'
- old_config_path = os.path.expanduser('~/.kube/config')
+ old_config_path = get_kubeconfig_path()
shutil.copy(old_config_path, new_config_path)
k = KubernetesPodOperator(
@@ -137,8 +130,8 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
config_file=new_config_path,
@@ -147,7 +140,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
def test_config_path_move(self):
new_config_path = '/tmp/kube_config'
- old_config_path = os.path.expanduser('~/.kube/config')
+ old_config_path = get_kubeconfig_path()
shutil.copy(old_config_path, new_config_path)
k = KubernetesPodOperator(
@@ -157,103 +150,16 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
arguments=["echo 10"],
labels={"foo": "bar"},
name="test1",
- task_id="task",
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
config_file=new_config_path,
)
- context = self.create_context(k)
+ context = create_context(k)
k.execute(context)
actual_pod = self.api_client.sanitize_for_serialization(k.pod)
self.assertEqual(self.expected_pod, actual_pod)
- @mock.patch("airflow.kubernetes.pod_launcher.PodLauncher.start_pod")
- @mock.patch("airflow.kubernetes.pod_launcher.PodLauncher.monitor_pod")
- @mock.patch("airflow.kubernetes.kube_client.get_kube_client")
- def test_config_path(self, client_mock, monitor_mock, start_mock): # pylint: disable=unused-argument
- from airflow.utils.state import State
-
- file_path = "/tmp/fake_file"
- k = KubernetesPodOperator(
- namespace='default',
- image="ubuntu:16.04",
- cmds=["bash", "-cx"],
- arguments=["echo 10"],
- labels={"foo": "bar"},
- name="test",
- task_id="task",
- in_cluster=False,
- do_xcom_push=False,
- config_file=file_path,
- cluster_context='default',
- )
- monitor_mock.return_value = (State.SUCCESS, None)
- client_mock.list_namespaced_pod.return_value = []
- context = self.create_context(k)
- k.execute(context=context)
- client_mock.assert_called_once_with(
- in_cluster=False,
- cluster_context='default',
- config_file=file_path,
- )
-
- @mock.patch("airflow.kubernetes.pod_launcher.PodLauncher.start_pod")
- @mock.patch("airflow.kubernetes.pod_launcher.PodLauncher.monitor_pod")
- @mock.patch("airflow.kubernetes.kube_client.get_kube_client")
- def test_image_pull_secrets_correctly_set(self, mock_client, monitor_mock, start_mock):
- from airflow.utils.state import State
-
- fake_pull_secrets = "fakeSecret"
- k = KubernetesPodOperator(
- namespace='default',
- image="ubuntu:16.04",
- cmds=["bash", "-cx"],
- arguments=["echo 10"],
- labels={"foo": "bar"},
- name="test",
- task_id="task",
- in_cluster=False,
- do_xcom_push=False,
- image_pull_secrets=fake_pull_secrets,
- cluster_context='default',
- )
- monitor_mock.return_value = (State.SUCCESS, None)
- context = self.create_context(k)
- k.execute(context=context)
- self.assertEqual(
- start_mock.call_args[0][0].spec.image_pull_secrets,
- [k8s.V1LocalObjectReference(name=fake_pull_secrets)]
- )
-
- @mock.patch("airflow.kubernetes.pod_launcher.PodLauncher.start_pod")
- @mock.patch("airflow.kubernetes.pod_launcher.PodLauncher.monitor_pod")
- @mock.patch("airflow.kubernetes.pod_launcher.PodLauncher.delete_pod")
- @mock.patch("airflow.kubernetes.kube_client.get_kube_client")
- def test_pod_delete_even_on_launcher_error(
- self,
- mock_client,
- delete_pod_mock,
- monitor_pod_mock,
- start_pod_mock): # pylint: disable=unused-argument
- k = KubernetesPodOperator(
- namespace='default',
- image="ubuntu:16.04",
- cmds=["bash", "-cx"],
- arguments=["echo 10"],
- labels={"foo": "bar"},
- name="test",
- task_id="task",
- in_cluster=False,
- do_xcom_push=False,
- cluster_context='default',
- is_delete_operator_pod=True,
- )
- monitor_pod_mock.side_effect = AirflowException('fake failure')
- with self.assertRaises(AirflowException):
- context = self.create_context(k)
- k.execute(context=context)
- assert delete_pod_mock.called
-
def test_working_pod(self):
k = KubernetesPodOperator(
namespace='default',
@@ -261,8 +167,8 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
)
@@ -279,49 +185,15 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
is_delete_operator_pod=True,
)
- context = self.create_context(k)
- k.execute(context)
- actual_pod = self.api_client.sanitize_for_serialization(k.pod)
- self.assertEqual(self.expected_pod['spec'], actual_pod['spec'])
- self.assertEqual(self.expected_pod['metadata']['labels'], actual_pod['metadata']['labels'])
-
- def test_pod_with_volume_secret(self):
- k = KubernetesPodOperator(
- namespace='default',
- image="ubuntu:16.04",
- cmds=["bash", "-cx"],
- in_cluster=False,
- labels={"foo": "bar"},
- arguments=["echo 10"],
- secrets=[Secret(
- deploy_type="volume",
- deploy_target="/var/location",
- secret="my-secret",
- key="content.json",
- )],
- name="airflow-test-pod",
- task_id="task",
- get_logs=True,
- is_delete_operator_pod=True,
- )
-
- context = self.create_context(k)
+ context = create_context(k)
k.execute(context)
actual_pod = self.api_client.sanitize_for_serialization(k.pod)
- self.expected_pod['spec']['containers'][0]['volumeMounts'] = [
- {'mountPath': '/var/location',
- 'name': mock.ANY,
- 'readOnly': True}]
- self.expected_pod['spec']['volumes'] = [
- {'name': mock.ANY,
- 'secret': {'secretName': 'my-secret'}}
- ]
self.assertEqual(self.expected_pod['spec'], actual_pod['spec'])
self.assertEqual(self.expected_pod['metadata']['labels'], actual_pod['metadata']['labels'])
@@ -332,13 +204,13 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
hostnetwork=True,
)
- context = self.create_context(k)
+ context = create_context(k)
k.execute(context)
actual_pod = self.api_client.sanitize_for_serialization(k.pod)
self.expected_pod['spec']['hostNetwork'] = True
@@ -353,14 +225,14 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
hostnetwork=True,
- dnspolicy=dns_policy
+ dnspolicy=dns_policy,
)
- context = self.create_context(k)
+ context = create_context(k)
k.execute(context)
actual_pod = self.api_client.sanitize_for_serialization(k.pod)
self.expected_pod['spec']['hostNetwork'] = True
@@ -376,32 +248,28 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
- schedulername=scheduler_name
+ schedulername=scheduler_name,
)
- context = self.create_context(k)
+ context = create_context(k)
k.execute(context)
actual_pod = self.api_client.sanitize_for_serialization(k.pod)
self.expected_pod['spec']['schedulerName'] = scheduler_name
self.assertEqual(self.expected_pod, actual_pod)
- self.assertEqual(self.expected_pod['spec'], actual_pod['spec'])
- self.assertEqual(self.expected_pod['metadata']['labels'], actual_pod['metadata']['labels'])
def test_pod_node_selectors(self):
- node_selectors = {
- 'beta.kubernetes.io/os': 'linux'
- }
+ node_selectors = {'beta.kubernetes.io/os': 'linux'}
k = KubernetesPodOperator(
namespace='default',
image="ubuntu:16.04",
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
node_selectors=node_selectors,
@@ -413,40 +281,28 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
self.assertEqual(self.expected_pod, actual_pod)
def test_pod_resources(self):
- resources = {
- 'limit_cpu': 0.25,
- 'limit_memory': '64Mi',
- 'limit_ephemeral_storage': '2Gi',
- 'request_cpu': '250m',
- 'request_memory': '64Mi',
- 'request_ephemeral_storage': '1Gi',
- }
+ resources = k8s.V1ResourceRequirements(
+ requests={'memory': '64Mi', 'cpu': '250m', 'ephemeral-storage': '1Gi'},
+ limits={'memory': '64Mi', 'cpu': 0.25, 'nvidia.com/gpu': None, 'ephemeral-storage': '2Gi'},
+ )
k = KubernetesPodOperator(
namespace='default',
image="ubuntu:16.04",
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
resources=resources,
)
- context = self.create_context(k)
+ context = create_context(k)
k.execute(context)
actual_pod = self.api_client.sanitize_for_serialization(k.pod)
self.expected_pod['spec']['containers'][0]['resources'] = {
- 'requests': {
- 'memory': '64Mi',
- 'cpu': '250m',
- 'ephemeral-storage': '1Gi'
- },
- 'limits': {
- 'memory': '64Mi',
- 'cpu': 0.25,
- 'ephemeral-storage': '2Gi'
- }
+ 'requests': {'memory': '64Mi', 'cpu': '250m', 'ephemeral-storage': '1Gi'},
+ 'limits': {'memory': '64Mi', 'cpu': 0.25, 'nvidia.com/gpu': None, 'ephemeral-storage': '2Gi'},
}
self.assertEqual(self.expected_pod, actual_pod)
@@ -457,11 +313,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
'nodeSelectorTerms': [
{
'matchExpressions': [
- {
- 'key': 'beta.kubernetes.io/os',
- 'operator': 'In',
- 'values': ['linux']
- }
+ {'key': 'beta.kubernetes.io/os', 'operator': 'In', 'values': ['linux']}
]
}
]
@@ -474,8 +326,8 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
affinity=affinity,
@@ -487,7 +339,10 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
self.assertEqual(self.expected_pod, actual_pod)
def test_port(self):
- port = Port('http', 80)
+ port = k8s.V1ContainerPort(
+ name='http',
+ container_port=80,
+ )
k = KubernetesPodOperator(
namespace='default',
@@ -495,37 +350,33 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
ports=[port],
)
- context = self.create_context(k)
+ context = create_context(k)
k.execute(context=context)
actual_pod = self.api_client.sanitize_for_serialization(k.pod)
- self.expected_pod['spec']['containers'][0]['ports'] = [{
- 'name': 'http',
- 'containerPort': 80
- }]
+ self.expected_pod['spec']['containers'][0]['ports'] = [{'name': 'http', 'containerPort': 80}]
self.assertEqual(self.expected_pod, actual_pod)
def test_volume_mount(self):
- with patch.object(PodLauncher, 'log') as mock_logger:
- volume_mount = VolumeMount('test-volume',
- mount_path='/tmp/test_volume',
- sub_path=None,
- read_only=False)
-
- volume_config = {
- 'persistentVolumeClaim':
- {
- 'claimName': 'test-volume'
- }
- }
- volume = Volume(name='test-volume', configs=volume_config)
- args = ["echo \"retrieved from mount\" > /tmp/test_volume/test.txt "
- "&& cat /tmp/test_volume/test.txt"]
+ with mock.patch.object(PodLauncher, 'log') as mock_logger:
+ volume_mount = k8s.V1VolumeMount(
+ name='test-volume', mount_path='/tmp/test_volume', sub_path=None, read_only=False
+ )
+
+ volume = k8s.V1Volume(
+ name='test-volume',
+ persistent_volume_claim=k8s.V1PersistentVolumeClaimVolumeSource(claim_name='test-volume'),
+ )
+
+ args = [
+ "echo \"retrieved from mount\" > /tmp/test_volume/test.txt "
+ "&& cat /tmp/test_volume/test.txt"
+ ]
k = KubernetesPodOperator(
namespace='default',
image="ubuntu:16.04",
@@ -534,27 +385,22 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
labels={"foo": "bar"},
volume_mounts=[volume_mount],
volumes=[volume],
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
)
context = create_context(k)
k.execute(context=context)
- mock_logger.info.assert_any_call(b"retrieved from mount\n")
+ mock_logger.info.assert_any_call('retrieved from mount')
actual_pod = self.api_client.sanitize_for_serialization(k.pod)
self.expected_pod['spec']['containers'][0]['args'] = args
- self.expected_pod['spec']['containers'][0]['volumeMounts'] = [{
- 'name': 'test-volume',
- 'mountPath': '/tmp/test_volume',
- 'readOnly': False
- }]
- self.expected_pod['spec']['volumes'] = [{
- 'name': 'test-volume',
- 'persistentVolumeClaim': {
- 'claimName': 'test-volume'
- }
- }]
+ self.expected_pod['spec']['containers'][0]['volumeMounts'] = [
+ {'name': 'test-volume', 'mountPath': '/tmp/test_volume', 'readOnly': False}
+ ]
+ self.expected_pod['spec']['volumes'] = [
+ {'name': 'test-volume', 'persistentVolumeClaim': {'claimName': 'test-volume'}}
+ ]
self.assertEqual(self.expected_pod, actual_pod)
def test_run_as_user_root(self):
@@ -569,8 +415,8 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
security_context=security_context,
@@ -594,8 +440,8 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
security_context=security_context,
@@ -619,8 +465,8 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-fs-group",
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
security_context=security_context,
@@ -639,8 +485,8 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
startup_timeout_seconds=5,
@@ -660,8 +506,8 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
startup_timeout_seconds=5,
@@ -685,8 +531,8 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=bad_internal_command,
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
)
@@ -699,15 +545,15 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
def test_xcom_push(self):
return_value = '{"foo": "bar"\n, "buzz": 2}'
- args = ['echo \'{}\' > /airflow/xcom/return.json'.format(return_value)]
+ args = ['echo \'' + str(return_value) + '\' > /airflow/xcom/return.json']
k = KubernetesPodOperator(
namespace='default',
image="ubuntu:16.04",
cmds=["bash", "-cx"],
arguments=args,
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=True,
)
@@ -730,7 +576,8 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
# GIVEN
from airflow.utils.state import State
- configmap = 'test-configmap'
+ configmap_name = "test-config-map"
+ env_from = [k8s.V1EnvFromSource(config_map_ref=k8s.V1ConfigMapEnvSource(name=configmap_name))]
# WHEN
k = KubernetesPodOperator(
namespace='default',
@@ -738,22 +585,17 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
- configmaps=[configmap],
+ env_from=env_from,
)
# THEN
mock_monitor.return_value = (State.SUCCESS, None)
- context = self.create_context(k)
+ context = create_context(k)
k.execute(context)
- self.assertEqual(
- mock_start.call_args[0][0].spec.containers[0].env_from,
- [k8s.V1EnvFromSource(config_map_ref=k8s.V1ConfigMapEnvSource(
- name=configmap
- ))]
- )
+ self.assertEqual(mock_start.call_args[0][0].spec.containers[0].env_from, env_from)
@mock.patch("airflow.kubernetes.pod_launcher.PodLauncher.start_pod")
@mock.patch("airflow.kubernetes.pod_launcher.PodLauncher.monitor_pod")
@@ -761,6 +603,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
def test_envs_from_secrets(self, mock_client, monitor_mock, start_mock):
# GIVEN
from airflow.utils.state import State
+
secret_ref = 'secret_name'
secrets = [Secret('env', None, secret_ref)]
# WHEN
@@ -771,34 +614,40 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
arguments=["echo 10"],
secrets=secrets,
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
)
# THEN
monitor_mock.return_value = (State.SUCCESS, None)
- context = self.create_context(k)
+ context = create_context(k)
k.execute(context)
self.assertEqual(
start_mock.call_args[0][0].spec.containers[0].env_from,
- [k8s.V1EnvFromSource(secret_ref=k8s.V1SecretEnvSource(
- name=secret_ref
- ))]
+ [k8s.V1EnvFromSource(secret_ref=k8s.V1SecretEnvSource(name=secret_ref))],
)
def test_env_vars(self):
# WHEN
+ env_vars = [
+ k8s.V1EnvVar(name="ENV1", value="val1"),
+ k8s.V1EnvVar(name="ENV2", value="val2"),
+ k8s.V1EnvVar(
+ name="ENV3",
+ value_from=k8s.V1EnvVarSource(field_ref=k8s.V1ObjectFieldSelector(field_path="status.podIP")),
+ ),
+ ]
+
k = KubernetesPodOperator(
namespace='default',
image="ubuntu:16.04",
cmds=["bash", "-cx"],
arguments=["echo 10"],
- env_vars={"ENV1": "val1", "ENV2": "val2", },
- pod_runtime_info_envs=[PodRuntimeInfoEnv("ENV3", "status.podIP")],
+ env_vars=env_vars,
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
)
@@ -811,14 +660,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
self.expected_pod['spec']['containers'][0]['env'] = [
{'name': 'ENV1', 'value': 'val1'},
{'name': 'ENV2', 'value': 'val2'},
- {
- 'name': 'ENV3',
- 'valueFrom': {
- 'fieldRef': {
- 'fieldPath': 'status.podIP'
- }
- }
- }
+ {'name': 'ENV3', 'valueFrom': {'fieldRef': {'fieldPath': 'status.podIP'}}},
]
self.assertEqual(self.expected_pod, actual_pod)
@@ -828,7 +670,7 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
task_id="task" + self.get_current_task_name(),
in_cluster=False,
pod_template_file=fixture,
- do_xcom_push=True
+ do_xcom_push=True,
)
context = create_context(k)
@@ -841,10 +683,10 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
k = KubernetesPodOperator(
task_id="task" + self.get_current_task_name(),
labels={"foo": "bar", "fizz": "buzz"},
- env_vars={"env_name": "value"},
+ env_vars=[k8s.V1EnvVar(name="env_name", value="value")],
in_cluster=False,
pod_template_file=fixture,
- do_xcom_push=True
+ do_xcom_push=True,
)
context = create_context(k)
@@ -856,20 +698,14 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
def test_init_container(self):
# GIVEN
- volume_mounts = [k8s.V1VolumeMount(
- mount_path='/etc/foo',
- name='test-volume',
- sub_path=None,
- read_only=True
- )]
-
- init_environments = [k8s.V1EnvVar(
- name='key1',
- value='value1'
- ), k8s.V1EnvVar(
- name='key2',
- value='value2'
- )]
+ volume_mounts = [
+ k8s.V1VolumeMount(mount_path='/etc/foo', name='test-volume', sub_path=None, read_only=True)
+ ]
+
+ init_environments = [
+ k8s.V1EnvVar(name='key1', value='value1'),
+ k8s.V1EnvVar(name='key2', value='value2'),
+ ]
init_container = k8s.V1Container(
name="init-container",
@@ -877,34 +713,20 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
env=init_environments,
volume_mounts=volume_mounts,
command=["bash", "-cx"],
- args=["echo 10"]
+ args=["echo 10"],
)
- volume_config = {
- 'persistentVolumeClaim':
- {
- 'claimName': 'test-volume'
- }
- }
- volume = Volume(name='test-volume', configs=volume_config)
-
+ volume = k8s.V1Volume(
+ name='test-volume',
+ persistent_volume_claim=k8s.V1PersistentVolumeClaimVolumeSource(claim_name='test-volume'),
+ )
expected_init_container = {
'name': 'init-container',
'image': 'ubuntu:16.04',
'command': ['bash', '-cx'],
'args': ['echo 10'],
- 'env': [{
- 'name': 'key1',
- 'value': 'value1'
- }, {
- 'name': 'key2',
- 'value': 'value2'
- }],
- 'volumeMounts': [{
- 'mountPath': '/etc/foo',
- 'name': 'test-volume',
- 'readOnly': True
- }],
+ 'env': [{'name': 'key1', 'value': 'value1'}, {'name': 'key2', 'value': 'value2'}],
+ 'volumeMounts': [{'mountPath': '/etc/foo', 'name': 'test-volume', 'readOnly': True}],
}
k = KubernetesPodOperator(
@@ -913,8 +735,8 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
volumes=[volume],
init_containers=[init_container],
in_cluster=False,
@@ -924,30 +746,30 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
k.execute(context)
actual_pod = self.api_client.sanitize_for_serialization(k.pod)
self.expected_pod['spec']['initContainers'] = [expected_init_container]
- self.expected_pod['spec']['volumes'] = [{
- 'name': 'test-volume',
- 'persistentVolumeClaim': {
- 'claimName': 'test-volume'
- }
- }]
+ self.expected_pod['spec']['volumes'] = [
+ {'name': 'test-volume', 'persistentVolumeClaim': {'claimName': 'test-volume'}}
+ ]
self.assertEqual(self.expected_pod, actual_pod)
@mock.patch("airflow.kubernetes.pod_launcher.PodLauncher.start_pod")
@mock.patch("airflow.kubernetes.pod_launcher.PodLauncher.monitor_pod")
@mock.patch("airflow.kubernetes.kube_client.get_kube_client")
- def test_pod_template_file(self, mock_client, monitor_mock, start_mock):
+ def test_pod_template_file(
+ self, mock_client, monitor_mock, start_mock # pylint: disable=unused-argument
+ ):
from airflow.utils.state import State
- fixture = sys.path[0] + '/tests/kubernetes/pod.yaml'
+
+ path = sys.path[0] + '/tests/kubernetes/pod.yaml'
k = KubernetesPodOperator(
- task_id='task',
- pod_template_file=fixture,
- do_xcom_push=True
+ task_id="task" + self.get_current_task_name(), pod_template_file=path, do_xcom_push=True
)
+
monitor_mock.return_value = (State.SUCCESS, None)
context = create_context(k)
with self.assertLogs(k.log, level=logging.DEBUG) as cm:
k.execute(context)
- expected_line = textwrap.dedent("""\
+ expected_line = textwrap.dedent(
+ """\
DEBUG:airflow.task.operators:Starting pod:
api_version: v1
kind: Pod
@@ -956,65 +778,57 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cluster_name: null
creation_timestamp: null
deletion_grace_period_seconds: null\
- """).strip()
+ """
+ ).strip()
self.assertTrue(any(line.startswith(expected_line) for line in cm.output))
actual_pod = self.api_client.sanitize_for_serialization(k.pod)
- expected_dict = {'apiVersion': 'v1',
- 'kind': 'Pod',
- 'metadata': {'annotations': {},
- 'labels': {},
- 'name': 'memory-demo',
- 'namespace': 'mem-example'},
- 'spec': {'affinity': {},
- 'containers': [{'args': ['--vm',
- '1',
- '--vm-bytes',
- '150M',
- '--vm-hang',
- '1'],
- 'command': ['stress'],
- 'env': [],
- 'envFrom': [],
- 'image': 'apache/airflow:stress-2020.07.10-1.0.4',
- 'imagePullPolicy': 'IfNotPresent',
- 'name': 'base',
- 'ports': [],
- 'resources': {'limits': {'memory': '200Mi'},
- 'requests': {'memory': '100Mi'}},
- 'volumeMounts': [{'mountPath': '/airflow/xcom',
- 'name': 'xcom'}]},
- {'command': ['sh',
- '-c',
- 'trap "exit 0" INT; while true; do sleep '
- '30; done;'],
- 'image': 'alpine',
- 'name': 'airflow-xcom-sidecar',
- 'resources': {'requests': {'cpu': '1m'}},
- 'volumeMounts': [{'mountPath': '/airflow/xcom',
- 'name': 'xcom'}]}],
- 'hostNetwork': False,
- 'imagePullSecrets': [],
- 'initContainers': [],
- 'nodeSelector': {},
- 'restartPolicy': 'Never',
- 'securityContext': {},
- 'serviceAccountName': 'default',
- 'tolerations': [],
- 'volumes': [{'emptyDir': {}, 'name': 'xcom'}]}}
+ expected_dict = {
+ 'apiVersion': 'v1',
+ 'kind': 'Pod',
+ 'metadata': {'annotations': {}, 'labels': {}, 'name': 'memory-demo', 'namespace': 'mem-example'},
+ 'spec': {
+ 'affinity': {},
+ 'containers': [
+ {
+ 'args': ['--vm', '1', '--vm-bytes', '150M', '--vm-hang', '1'],
+ 'command': ['stress'],
+ 'env': [],
+ 'envFrom': [],
+ 'image': 'apache/airflow:stress-2020.07.10-1.0.4',
+ 'name': 'base',
+ 'ports': [],
+ 'resources': {'limits': {'memory': '200Mi'}, 'requests': {'memory': '100Mi'}},
+ 'volumeMounts': [{'mountPath': '/airflow/xcom', 'name': 'xcom'}],
+ },
+ {
+ 'command': ['sh', '-c', 'trap "exit 0" INT; while true; do sleep 30; done;'],
+ 'image': 'alpine',
+ 'name': 'airflow-xcom-sidecar',
+ 'resources': {'requests': {'cpu': '1m'}},
+ 'volumeMounts': [{'mountPath': '/airflow/xcom', 'name': 'xcom'}],
+ },
+ ],
+ 'hostNetwork': False,
+ 'imagePullSecrets': [],
+ 'initContainers': [],
+ 'nodeSelector': {},
+ 'restartPolicy': 'Never',
+ 'securityContext': {},
+ 'serviceAccountName': 'default',
+ 'tolerations': [],
+ 'volumes': [{'emptyDir': {}, 'name': 'xcom'}],
+ },
+ }
self.assertEqual(expected_dict, actual_pod)
@mock.patch("airflow.kubernetes.pod_launcher.PodLauncher.start_pod")
@mock.patch("airflow.kubernetes.pod_launcher.PodLauncher.monitor_pod")
@mock.patch("airflow.kubernetes.kube_client.get_kube_client")
def test_pod_priority_class_name(
- self,
- mock_client,
- monitor_mock,
- start_mock): # pylint: disable=unused-argument
- """Test ability to assign priorityClassName to pod
-
- """
+ self, mock_client, monitor_mock, start_mock # pylint: disable=unused-argument
+ ):
+ """Test ability to assign priorityClassName to pod"""
from airflow.utils.state import State
priority_class_name = "medium-test"
@@ -1024,15 +838,15 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
cmds=["bash", "-cx"],
arguments=["echo 10"],
labels={"foo": "bar"},
- name="test",
- task_id="task",
+ name="test-" + str(random.randint(0, 1000000)),
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
priority_class_name=priority_class_name,
)
monitor_mock.return_value = (State.SUCCESS, None)
- context = self.create_context(k)
+ context = create_context(k)
k.execute(context)
actual_pod = self.api_client.sanitize_for_serialization(k.pod)
self.expected_pod['spec']['priorityClassName'] = priority_class_name
@@ -1048,15 +862,15 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
arguments=["echo 10"],
labels={"foo": "bar"},
name=pod_name_too_long,
- task_id="task",
+ task_id="task" + self.get_current_task_name(),
in_cluster=False,
do_xcom_push=False,
)
@mock.patch("airflow.kubernetes.pod_launcher.PodLauncher.monitor_pod")
- def test_on_kill(self,
- monitor_mock): # pylint: disable=unused-argument
+ def test_on_kill(self, monitor_mock): # pylint: disable=unused-argument
from airflow.utils.state import State
+
client = kube_client.get_kube_client(in_cluster=False)
name = "test"
namespace = "default"
@@ -1082,4 +896,47 @@ class TestKubernetesPodOperatorSystem(unittest.TestCase):
with self.assertRaises(ApiException):
pod = client.read_namespaced_pod(name=name, namespace=namespace)
+ def test_reattach_failing_pod_once(self):
+ from airflow.utils.state import State
+
+ client = kube_client.get_kube_client(in_cluster=False)
+ name = "test"
+ namespace = "default"
+ k = KubernetesPodOperator(
+ namespace='default',
+ image="ubuntu:16.04",
+ cmds=["bash", "-cx"],
+ arguments=["exit 1"],
+ labels={"foo": "bar"},
+ name="test",
+ task_id=name,
+ in_cluster=False,
+ do_xcom_push=False,
+ is_delete_operator_pod=False,
+ termination_grace_period=0,
+ )
+
+ context = create_context(k)
+
+ with mock.patch("airflow.kubernetes.pod_launcher.PodLauncher.monitor_pod") as monitor_mock:
+ monitor_mock.return_value = (State.SUCCESS, None)
+ k.execute(context)
+ name = k.pod.metadata.name
+ pod = client.read_namespaced_pod(name=name, namespace=namespace)
+ while pod.status.phase != "Failed":
+ pod = client.read_namespaced_pod(name=name, namespace=namespace)
+ with self.assertRaises(AirflowException):
+ k.execute(context)
+ pod = client.read_namespaced_pod(name=name, namespace=namespace)
+ self.assertEqual(pod.metadata.labels["already_checked"], "True")
+ with mock.patch(
+ "airflow.providers.cncf.kubernetes"
+ ".operators.kubernetes_pod.KubernetesPodOperator"
+ ".create_new_pod_for_operator"
+ ) as create_mock:
+ create_mock.return_value = ("success", {}, {})
+ k.execute(context)
+ create_mock.assert_called_once()
+
+
# pylint: enable=unused-argument
diff --git a/scripts/ci/kubernetes/ci_run_kubernetes_tests.sh b/scripts/ci/kubernetes/ci_run_kubernetes_tests.sh
index 4f13335..efc5eca 100755
--- a/scripts/ci/kubernetes/ci_run_kubernetes_tests.sh
+++ b/scripts/ci/kubernetes/ci_run_kubernetes_tests.sh
@@ -21,7 +21,6 @@
kind::make_sure_kubernetes_tools_are_installed
kind::get_kind_cluster_name
-traps::add_trap kind::stop_kubectl EXIT HUP INT TERM
traps::add_trap kind::dump_kind_logs EXIT HUP INT TERM
interactive="false"
@@ -105,7 +104,6 @@ if [[ ${interactive} == "true" ]]; then
echo
echo "You are entering the virtualenv now. Type exit to exit back to the original shell"
echo
- kubectl config set-context --current --namespace=airflow
exec "${SHELL}"
else
pytest "${pytest_args[@]}" "${tests_to_run[@]}"
diff --git a/scripts/ci/kubernetes/ci_deploy_app_to_kubernetes.sh b/scripts/ci/kubernetes/ci_setup_cluster_and_deploy_airflow_to_kubernetes.sh
similarity index 94%
copy from scripts/ci/kubernetes/ci_deploy_app_to_kubernetes.sh
copy to scripts/ci/kubernetes/ci_setup_cluster_and_deploy_airflow_to_kubernetes.sh
index 2a7455a..e12f809 100755
--- a/scripts/ci/kubernetes/ci_deploy_app_to_kubernetes.sh
+++ b/scripts/ci/kubernetes/ci_setup_cluster_and_deploy_airflow_to_kubernetes.sh
@@ -23,10 +23,11 @@ traps::add_trap "kind::dump_kind_logs" EXIT HUP INT TERM
kind::make_sure_kubernetes_tools_are_installed
kind::get_kind_cluster_name
+kind::perform_kind_cluster_operation "start"
build_images::prepare_prod_build
build_images::build_prod_images
kind::build_image_for_kubernetes_tests
kind::load_image_to_kind_cluster
kind::deploy_airflow_with_helm
-kind::forward_port_to_kind_webserver
kind::deploy_test_kubernetes_resources
+kind::wait_for_webserver_healthy
diff --git a/scripts/ci/kubernetes/kind-cluster-conf.yaml b/scripts/ci/kubernetes/kind-cluster-conf.yaml
index 348fb68..df60820 100644
--- a/scripts/ci/kubernetes/kind-cluster-conf.yaml
+++ b/scripts/ci/kubernetes/kind-cluster-conf.yaml
@@ -23,6 +23,11 @@ networking:
nodes:
- role: control-plane
- role: worker
+ extraPortMappings:
+ - containerPort: 30007
+ hostPort: 8080
+ listenAddress: "0.0.0.0"
+ protocol: TCP
kubeadmConfigPatchesJson6902:
- group: kubeadm.k8s.io
version: v1beta2
diff --git a/scripts/ci/kubernetes/kind-cluster-conf.yaml b/scripts/ci/kubernetes/nodeport.yaml
similarity index 68%
copy from scripts/ci/kubernetes/kind-cluster-conf.yaml
copy to scripts/ci/kubernetes/nodeport.yaml
index 348fb68..8438281 100644
--- a/scripts/ci/kubernetes/kind-cluster-conf.yaml
+++ b/scripts/ci/kubernetes/nodeport.yaml
@@ -15,19 +15,17 @@
# specific language governing permissions and limitations
# under the License.
---
-kind: Cluster
-apiVersion: kind.sigs.k8s.io/v1alpha3
-networking:
- apiServerAddress: 0.0.0.0
- apiServerPort: 19090
-nodes:
- - role: control-plane
- - role: worker
-kubeadmConfigPatchesJson6902:
- - group: kubeadm.k8s.io
- version: v1beta2
- kind: ClusterConfiguration
- patch: |
- - op: add
- path: /apiServer/certSANs/-
- value: docker
+apiVersion: v1
+kind: Service
+metadata:
+ name: airflow-webserver-node-port
+spec:
+ type: NodePort
+ selector:
+ component: webserver
+ release: airflow
+ tier: airflow
+ ports:
+ - port: 8080
+ targetPort: 8080
+ nodePort: 30007
diff --git a/scripts/ci/kubernetes/ci_deploy_app_to_kubernetes.sh b/scripts/ci/kubernetes/redeploy_airflow.sh
similarity index 86%
rename from scripts/ci/kubernetes/ci_deploy_app_to_kubernetes.sh
rename to scripts/ci/kubernetes/redeploy_airflow.sh
index 2a7455a..7803d7c 100755
--- a/scripts/ci/kubernetes/ci_deploy_app_to_kubernetes.sh
+++ b/scripts/ci/kubernetes/redeploy_airflow.sh
@@ -23,10 +23,6 @@ traps::add_trap "kind::dump_kind_logs" EXIT HUP INT TERM
kind::make_sure_kubernetes_tools_are_installed
kind::get_kind_cluster_name
-build_images::prepare_prod_build
-build_images::build_prod_images
-kind::build_image_for_kubernetes_tests
-kind::load_image_to_kind_cluster
kind::deploy_airflow_with_helm
-kind::forward_port_to_kind_webserver
kind::deploy_test_kubernetes_resources
+kind::wait_for_webserver_healthy
diff --git a/scripts/ci/libraries/_kind.sh b/scripts/ci/libraries/_kind.sh
index 6194742..defa4de 100644
--- a/scripts/ci/libraries/_kind.sh
+++ b/scripts/ci/libraries/_kind.sh
@@ -16,14 +16,16 @@
# specific language governing permissions and limitations
# under the License.
-
-function kind::get_kind_cluster_name(){
+function kind::get_kind_cluster_name() {
# Name of the KinD cluster to connect to
export KIND_CLUSTER_NAME=${KIND_CLUSTER_NAME:="airflow-python-${PYTHON_MAJOR_MINOR_VERSION}-${KUBERNETES_VERSION}"}
readonly KIND_CLUSTER_NAME
# Name of the KinD cluster to connect to when referred to via kubectl
export KUBECTL_CLUSTER_NAME=kind-${KIND_CLUSTER_NAME}
readonly KUBECTL_CLUSTER_NAME
+ export KUBECONFIG="${BUILD_CACHE_DIR}/.kube/config"
+ mkdir -pv "${BUILD_CACHE_DIR}/.kube/"
+ touch "${KUBECONFIG}"
}
function kind::dump_kind_logs() {
@@ -40,7 +42,7 @@ function kind::dump_kind_logs() {
}
function kind::make_sure_kubernetes_tools_are_installed() {
- SYSTEM=$(uname -s| tr '[:upper:]' '[:lower:]')
+ SYSTEM=$(uname -s | tr '[:upper:]' '[:lower:]')
KIND_URL="https://github.com/kubernetes-sigs/kind/releases/download/${KIND_VERSION}/kind-${SYSTEM}-amd64"
mkdir -pv "${BUILD_CACHE_DIR}/bin"
@@ -48,7 +50,7 @@ function kind::make_sure_kubernetes_tools_are_installed() {
DOWNLOADED_KIND_VERSION=v"$(${KIND_BINARY_PATH} --version | awk '{ print $3 }')"
echo "Currently downloaded kind version = ${DOWNLOADED_KIND_VERSION}"
fi
- if [[ ! -f "${KIND_BINARY_PATH}" || ${DOWNLOADED_KIND_VERSION} != "${KIND_VERSION}" ]]; then
+ if [[ ! -f "${KIND_BINARY_PATH}" || ${DOWNLOADED_KIND_VERSION} != "${KIND_VERSION}" ]]; then
echo
echo "Downloading Kind version ${KIND_VERSION}"
repeats::run_with_retry 4 \
@@ -94,24 +96,10 @@ function kind::make_sure_kubernetes_tools_are_installed() {
}
function kind::create_cluster() {
- if [[ "${TRAVIS:="false"}" == "true" ]]; then
- # Travis CI does not handle the nice output of Kind well, so we need to capture it
- # And display only if kind fails to start
- start_output_heartbeat "Creating kubernetes cluster" 10
- set +e
- if ! OUTPUT=$(kind create cluster \
- --name "${KIND_CLUSTER_NAME}" \
- --config "${AIRFLOW_SOURCES}/scripts/ci/kubernetes/kind-cluster-conf.yaml" \
- --image "kindest/node:${KUBERNETES_VERSION}" 2>&1); then
- echo "${OUTPUT}"
- fi
- stop_output_heartbeat
- else
- kind create cluster \
- --name "${KIND_CLUSTER_NAME}" \
- --config "${AIRFLOW_SOURCES}/scripts/ci/kubernetes/kind-cluster-conf.yaml" \
- --image "kindest/node:${KUBERNETES_VERSION}"
- fi
+ kind create cluster \
+ --name "${KIND_CLUSTER_NAME}" \
+ --config "${AIRFLOW_SOURCES}/scripts/ci/kubernetes/kind-cluster-conf.yaml" \
+ --image "kindest/node:${KUBERNETES_VERSION}"
echo
echo "Created cluster ${KIND_CLUSTER_NAME}"
echo
@@ -125,9 +113,12 @@ function kind::delete_cluster() {
rm -rf "${HOME}/.kube/*"
}
-function kind::perform_kind_cluster_operation() {
- ALLOWED_KIND_OPERATIONS="[ start restart stop deploy test shell recreate ]"
+function kind::set_current_context() {
+ kubectl config set-context --current --namespace=airflow
+}
+function kind::perform_kind_cluster_operation() {
+ ALLOWED_KIND_OPERATIONS="[ start restart stop deploy test shell recreate k9s]"
set +u
if [[ -z "${1=}" ]]; then
echo >&2
@@ -170,6 +161,7 @@ function kind::perform_kind_cluster_operation() {
echo
kind::delete_cluster
kind::create_cluster
+ kind::set_current_context
elif [[ ${OPERATION} == "stop" ]]; then
echo
echo "Deleting cluster"
@@ -181,20 +173,35 @@ function kind::perform_kind_cluster_operation() {
echo "Deploying Airflow to KinD"
echo
kind::build_image_for_kubernetes_tests
+ kind::get_kind_cluster_name
kind::load_image_to_kind_cluster
kind::deploy_airflow_with_helm
- kind::forward_port_to_kind_webserver
kind::deploy_test_kubernetes_resources
+ kind::wait_for_webserver_healthy
elif [[ ${OPERATION} == "test" ]]; then
echo
echo "Testing with KinD"
echo
+ kind::set_current_context
"${AIRFLOW_SOURCES}/scripts/ci/kubernetes/ci_run_kubernetes_tests.sh"
elif [[ ${OPERATION} == "shell" ]]; then
echo
echo "Entering an interactive shell for kubernetes testing"
echo
+ kind::set_current_context
"${AIRFLOW_SOURCES}/scripts/ci/kubernetes/ci_run_kubernetes_tests.sh" "-i"
+ elif [[ ${OPERATION} == "k9s" ]]; then
+ echo
+ echo "Starting k9s CLI"
+ echo
+ export TERM=xterm-256color
+ export EDITOR=vim
+ export K9S_EDITOR=vim
+ kind::set_current_context
+ exec docker run --rm -it --network host \
+ -e COLUMNS="$(tput cols)" -e LINES="$(tput lines)" \
+ -e EDITOR -e K9S_EDITOR \
+ -v "${KUBECONFIG}:/root/.kube/config" quay.io/derailed/k9s
else
echo >&2
echo >&2 "Wrong cluster operation: ${OPERATION}. Should be one of: ${ALLOWED_KIND_OPERATIONS}"
@@ -213,8 +220,7 @@ function kind::perform_kind_cluster_operation() {
echo "Creating cluster"
echo
kind::create_cluster
- elif [[ ${OPERATION} == "stop" || ${OPERATION} == "deploy" || \
- ${OPERATION} == "test" || ${OPERATION} == "shell" ]]; then
+ elif [[ ${OPERATION} == "stop" || ${OPERATION} == "deploy" || ${OPERATION} == "test" || ${OPERATION} == "shell" ]]; then
echo >&2
echo >&2 "Cluster ${KIND_CLUSTER_NAME} does not exist. It should exist for ${OPERATION} operation"
echo >&2
@@ -245,7 +251,6 @@ function kind::check_cluster_ready_for_airflow() {
kubectl create namespace test-namespace --cluster "${KUBECTL_CLUSTER_NAME}"
}
-
function kind::build_image_for_kubernetes_tests() {
cd "${AIRFLOW_SOURCES}" || exit 1
docker build --tag "${AIRFLOW_PROD_IMAGE_KUBERNETES}" . -f - <<EOF
@@ -268,61 +273,37 @@ function kind::load_image_to_kind_cluster() {
kind load docker-image --name "${KIND_CLUSTER_NAME}" "${AIRFLOW_PROD_IMAGE_KUBERNETES}"
}
-MAX_NUM_TRIES_FOR_PORT_FORWARD=12
-readonly MAX_NUM_TRIES_FOR_PORT_FORWARD
+MAX_NUM_TRIES_FOR_HEALTH_CHECK=12
+readonly MAX_NUM_TRIES_FOR_HEALTH_CHECK
-SLEEP_TIME_FOR_PORT_FORWARD=10
-readonly SLEEP_TIME_FOR_PORT_FORWARD
-
-forwarded_port_number=8080
-
-function kind::start_kubectl_forward() {
- echo
- echo "Trying to forward port ${forwarded_port_number} to 8080 on server"
- echo
- kubectl port-forward svc/airflow-webserver "${forwarded_port_number}:8080" --namespace airflow >/dev/null &
-}
+SLEEP_TIME_FOR_HEALTH_CHECK=10
+readonly SLEEP_TIME_FOR_HEALTH_CHECK
-function kind::stop_kubectl() {
- echo
- echo "Stops all kubectl instances"
- echo
- killall kubectl || true
- sleep 10
- killall -s KILL kubectl || true
+FORWARDED_PORT_NUMBER=8080
+readonly FORWARDED_PORT_NUMBER
-}
-function kind::forward_port_to_kind_webserver() {
+function kind::wait_for_webserver_healthy() {
num_tries=0
set +e
- kind::start_kubectl_forward
- sleep "${SLEEP_TIME_FOR_PORT_FORWARD}"
- while ! curl "http://localhost:${forwarded_port_number}/health" -s | grep -q healthy; do
+ sleep "${SLEEP_TIME_FOR_HEALTH_CHECK}"
+ while ! curl "http://localhost:${FORWARDED_PORT_NUMBER}/health" -s | grep -q healthy; do
echo
- echo "Trying to establish port forwarding to 'airflow webserver'"
+ echo "Sleeping ${SLEEP_TIME_FOR_HEALTH_CHECK} while waiting for webserver being ready"
echo
- if [[ ${INCREASE_PORT_NUMBER_FOR_KUBERNETES=} == "true" ]] ; then
- forwarded_port_number=$(( forwarded_port_number + 1 ))
+ sleep "${SLEEP_TIME_FOR_HEALTH_CHECK}"
+ num_tries=$((num_tries + 1))
+ if [[ ${num_tries} == "${MAX_NUM_TRIES_FOR_HEALTH_CHECK}" ]]; then
+ >&2 echo
+ >&2 echo "Timeout while waiting for the webserver health check"
+ >&2 echo
fi
- if [[ ${num_tries} == "${MAX_NUM_TRIES_FOR_PORT_FORWARD}" ]]; then
- echo >&2
- echo >&2 "ERROR! Could not setup a forward port to Airflow's webserver after ${num_tries}! Exiting."
- echo >&2
- exit 1
- fi
- echo
- echo "Trying to establish port forwarding to 'airflow webserver'"
- echo
- kind::start_kubectl_forward
- sleep "${SLEEP_TIME_FOR_PORT_FORWARD}"
- num_tries=$(( num_tries + 1))
done
echo
- echo "Connection to 'airflow webserver' established on port ${forwarded_port_number}"
+ echo "Connection to 'airflow webserver' established on port ${FORWARDED_PORT_NUMBER}"
echo
- initialization::ga_env CLUSTER_FORWARDED_PORT "${forwarded_port_number}"
- export CLUSTER_FORWARDED_PORT="${forwarded_port_number}"
+ initialization::ga_env CLUSTER_FORWARDED_PORT "${FORWARDED_PORT_NUMBER}"
+ export CLUSTER_FORWARDED_PORT="${FORWARDED_PORT_NUMBER}"
set -e
}
@@ -348,16 +329,15 @@ function kind::deploy_airflow_with_helm() {
popd || exit 1
}
-
function kind::deploy_test_kubernetes_resources() {
echo
echo "Deploying Custom kubernetes resources"
echo
kubectl apply -f "scripts/ci/kubernetes/volumes.yaml" --namespace default
kubectl apply -f "scripts/ci/kubernetes/secrets.yaml" --namespace default
+ kubectl apply -f "scripts/ci/kubernetes/nodeport.yaml" --namespace airflow
}
-
function kind::dump_kubernetes_logs() {
POD=$(kubectl get pods -o go-template --template '{{range .items}}{{.metadata.name}}{{"\n"}}{{end}}' \
--cluster "${KUBECTL_CLUSTER_NAME}" | grep airflow | head -1)
[airflow] 04/05: For v1-10-test PRs and pushes,
use target branch scripts for images (#12339)
Posted by po...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v1-10-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit a90131c0ae098c2a063cd94fdc1f36ace818b606
Author: Jarek Potiuk <ja...@polidea.com>
AuthorDate: Fri Nov 13 14:28:36 2020 +0100
For v1-10-test PRs and pushes, use target branch scripts for images (#12339)
Previously, always master scripts were used to build images
for workflow_run, because workflow_run always runs from master
branch. However that causes some surprising effects becuase the
sripts from master had to support both master and 1.10.
This change utilises a new feature in the "get-workflow-origin"
action - to get the target branch of PR and uses ci scripts from that
target branch.
This is perfectly secure, because both v1-10-test, v1-10-stable
and future 2-0 branches can only be updated by committers,
either by direct push or by merge.
(cherry picked from commit 7c4fe19e41ae02a1df1c0a217501cae2e0e84819)
---
.github/workflows/build-images-workflow-run.yml | 12 +++++--
.github/workflows/ci.yml | 38 +++++++---------------
.../workflows/label_when_reviewed_workflow_run.yml | 2 +-
3 files changed, 22 insertions(+), 30 deletions(-)
diff --git a/.github/workflows/build-images-workflow-run.yml b/.github/workflows/build-images-workflow-run.yml
index a3277a7..af71710 100644
--- a/.github/workflows/build-images-workflow-run.yml
+++ b/.github/workflows/build-images-workflow-run.yml
@@ -53,13 +53,14 @@ jobs:
targetCommitSha: ${{ steps.source-run-info.outputs.targetCommitSha }}
pullRequestNumber: ${{ steps.source-run-info.outputs.pullRequestNumber }}
pullRequestLabels: ${{ steps.source-run-info.outputs.pullRequestLabels }}
+ targetBranch: ${{ steps.source-run-info.outputs.targetBranch }}
sourceEvent: ${{ steps.source-run-info.outputs.sourceEvent }}
cacheDirective: ${{ steps.cache-directive.outputs.docker-cache }}
buildImages: ${{ steps.build-images.outputs.buildImages }}
upgradeToLatestConstraints: ${{ steps.upgrade-constraints.outputs.upgradeToLatestConstraints }}
steps:
- name: "Get information about the original trigger of the run"
- uses: potiuk/get-workflow-origin@2ef0b065db6b688a2231f8a7f464df1aac254328 # v1_2
+ uses: potiuk/get-workflow-origin@588cc14f9f1cdf1b8be3db816855e96422204fec # v1_3
id: source-run-info
with:
token: ${{ secrets.GITHUB_TOKEN }}
@@ -326,17 +327,22 @@ jobs:
[Image Build](https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }})
for details" }
if: steps.defaults.outputs.proceed == 'true'
- - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} ) to 'main-airflow' to use main scripts"
+ - name: >
+ Checkout "${{ needs.cancel-workflow-runs.outputs.targetBranch }}" branch to 'main-airflow' folder
+ to use ci/scripts from there.
uses: actions/checkout@v2
with:
path: "main-airflow"
+ ref: "${{ needs.cancel-workflow-runs.outputs.targetBranch }}"
if: steps.defaults.outputs.proceed == 'true'
- name: "Setup python"
uses: actions/setup-python@v2
with:
python-version: ${{ needs.build-info.outputs.defaultPythonVersion }}
if: steps.defaults.outputs.proceed == 'true'
- - name: "Override 'scripts/ci' with the ${{ github.ref }} version so that the PR cannot override it."
+ - name: >
+ Override "scripts/ci" with the "${{ needs.cancel-workflow-runs.outputs.targetBranch }}" branch
+ so that the PR does not override it
# We should not override those scripts which become part of the image as they will not be
# changed in the image built - we should only override those that are executed to build
# the image.
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 81890a7..df5f53e 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -95,27 +95,15 @@ jobs:
image-build: ${{ steps.selective-checks.outputs.image-build }}
docs-build: ${{ steps.selective-checks.outputs.docs-build }}
needs-helm-tests: ${{ steps.selective-checks.outputs.needs-helm-tests }}
- needs-api-tests: ${{ steps.selective-checks.outputs.needs-api-tests }}
pullRequestNumber: ${{ steps.source-run-info.outputs.pullRequestNumber }}
pullRequestLabels: ${{ steps.source-run-info.outputs.pullRequestLabels }}
steps:
- name: "Get information about the PR"
- uses: potiuk/get-workflow-origin@2ef0b065db6b688a2231f8a7f464df1aac254328 # v1_2
+ uses: potiuk/get-workflow-origin@588cc14f9f1cdf1b8be3db816855e96422204fec # v1_3
id: source-run-info
with:
token: ${{ secrets.GITHUB_TOKEN }}
- name: "Checkout ${{ github.ref }} ( ${{ github.sha }} )"
- # First fetch the sha of merge commit in case it is pull request so that we can
- # Run selective tests
- - name: >
- Fetch merge commit ${{ github.ref }} ( ${{ github.sha }}:
- merge_commit ${{ github.event.pull_request.merge_commit_sha }} )
- uses: actions/checkout@v2
- with:
- ref: ${{ github.event.pull_request.merge_commit_sha }}
- fetch-depth: 2
- if: github.event_name == 'pull_request'
- - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} : merge commit ${{ github.merge_commit_sha }} )"
uses: actions/checkout@v2
- name: >
Event: ${{ github.event_name }}
@@ -123,7 +111,6 @@ jobs:
Branch: ${{ github.head_ref }}
Run id: ${{ github.run_id }}
Sha: ${{ github.sha }}
- Merge commit sha: ${{ github.merge_commit_sha }}
Ref: ${{ github.ref }}
run: printenv
- name: Set wait for image
@@ -164,7 +151,7 @@ jobs:
env:
BACKEND: sqlite
steps:
- - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} : merge commit ${{ github.merge_commit_sha }} )"
+ - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} )"
uses: actions/checkout@v2
if: needs.build-info.outputs.waitForImage == 'true'
- name: "Setup python"
@@ -199,7 +186,7 @@ jobs:
PYTHON_MAJOR_MINOR_VERSION: ${{needs.build-info.outputs.defaultPythonVersion}}
if: needs.build-info.outputs.basic-checks-only == 'false'
steps:
- - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} : merge commit ${{ github.merge_commit_sha }} )"
+ - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} )"
uses: actions/checkout@v2
- name: "Setup python"
uses: actions/setup-python@v2
@@ -267,7 +254,7 @@ jobs:
needs: [build-info, ci-images]
if: needs.build-info.outputs.docs-build == 'true'
steps:
- - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} : merge commit ${{ github.merge_commit_sha }})"
+ - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} )"
uses: actions/checkout@v2
- name: "Prepare CI image ${{env.PYTHON_MAJOR_MINOR_VERSION}}:${{ env.GITHUB_REGISTRY_PULL_IMAGE_TAG }}"
run: ./scripts/ci/images/ci_prepare_ci_image_on_ci.sh
@@ -348,7 +335,7 @@ jobs:
TEST_TYPE: ""
if: needs.build-info.outputs.run-tests == 'true'
steps:
- - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} : merge commit ${{ github.merge_commit_sha }})"
+ - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} )"
uses: actions/checkout@v2
- name: "Setup python"
uses: actions/setup-python@v2
@@ -400,7 +387,7 @@ jobs:
TEST_TYPE: ""
if: needs.build-info.outputs.run-tests == 'true'
steps:
- - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} : merge commit ${{ github.merge_commit_sha }})"
+ - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} )"
uses: actions/checkout@v2
- name: "Setup python"
uses: actions/setup-python@v2
@@ -449,7 +436,7 @@ jobs:
TEST_TYPE: ""
if: needs.build-info.outputs.run-tests == 'true'
steps:
- - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} : merge commit ${{ github.merge_commit_sha }})"
+ - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} )"
uses: actions/checkout@v2
- name: "Setup python"
uses: actions/setup-python@v2
@@ -503,7 +490,7 @@ jobs:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
if: needs.build-info.outputs.run-tests == 'true'
steps:
- - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} : merge commit ${{ github.merge_commit_sha }})"
+ - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} )"
uses: actions/checkout@v2
- name: "Setup python"
uses: actions/setup-python@v2
@@ -628,7 +615,7 @@ jobs:
HELM_VERSION: "${{ needs.build-info.outputs.defaultHelmVersion }}"
if: needs.build-info.outputs.run-kubernetes-tests == 'true'
steps:
- - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} : merge commit ${{ github.merge_commit_sha }})"
+ - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} )"
uses: actions/checkout@v2
- name: "Setup python"
uses: actions/setup-python@v2
@@ -736,7 +723,7 @@ jobs:
PYTHON_MAJOR_MINOR_VERSION: ${{ matrix.python-version }}
GITHUB_REGISTRY_PUSH_IMAGE_TAG: "latest"
steps:
- - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} : merge commit ${{ github.merge_commit_sha }})"
+ - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} )"
uses: actions/checkout@v2
- name: "Setup python"
uses: actions/setup-python@v2
@@ -764,7 +751,7 @@ jobs:
PYTHON_MAJOR_MINOR_VERSION: ${{ matrix.python-version }}
if: github.ref == 'refs/heads/master' || github.ref == 'refs/heads/v1-10-test'
steps:
- - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} : merge commit ${{ github.merge_commit_sha }})"
+ - name: "Checkout ${{ github.ref }} ( ${{ github.sha }} )"
uses: actions/checkout@v2
- name: "Setup python"
uses: actions/setup-python@v2
@@ -790,7 +777,6 @@ jobs:
- build-info
- constraints
- static-checks
- - static-checks-pylint
- tests-sqlite
- tests-mysql
- tests-postgres
@@ -814,7 +800,7 @@ jobs:
- name: "Commit changed constraint files for ${{needs.build-info.outputs.pythonVersions}}"
run: ./scripts/ci/constraints/ci_commit_constraints.sh
- name: "Push changes"
- uses: ad-m/github-push-action@master
+ uses: ad-m/github-push-action@40bf560936a8022e68a3c00e7d2abefaf01305a6 # v0.6.0
with:
github_token: ${{ secrets.GITHUB_TOKEN }}
branch: ${{ steps.constraints-branch.outputs.branch }}
diff --git a/.github/workflows/label_when_reviewed_workflow_run.yml b/.github/workflows/label_when_reviewed_workflow_run.yml
index c2359c8..f943609 100644
--- a/.github/workflows/label_when_reviewed_workflow_run.yml
+++ b/.github/workflows/label_when_reviewed_workflow_run.yml
@@ -30,7 +30,7 @@ jobs:
labelSet: ${{ steps.label-when-reviewed.outputs.labelSet }}
steps:
- name: "Get information about the original trigger of the run"
- uses: potiuk/get-workflow-origin@2ef0b065db6b688a2231f8a7f464df1aac254328 # v1_2
+ uses: potiuk/get-workflow-origin@588cc14f9f1cdf1b8be3db816855e96422204fec # v1_3
id: source-run-info
with:
token: ${{ secrets.GITHUB_TOKEN }}