This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 33ecca1 Use jobs check command for liveness probe check in airflow 2
(#22143)
33ecca1 is described below
commit 33ecca1b9ab99d9d15006df77757825c81c24f84
Author: Ephraim Anierobi <[email protected]>
AuthorDate: Fri Mar 11 08:03:08 2022 +0100
Use jobs check command for liveness probe check in airflow 2 (#22143)
This PR removes the current use of python code for liveness probe check
commands in the Scheduler & Triggerer deployments
Co-authored-by: Jed Cunningham
<[email protected]>
---
chart/templates/_helpers.yaml | 35 +++++++++++++++++++
.../templates/scheduler/scheduler-deployment.yaml | 8 +++--
.../templates/triggerer/triggerer-deployment.yaml | 8 +++--
chart/values.schema.json | 20 ++++++++---
chart/values.yaml | 39 ++--------------------
5 files changed, 63 insertions(+), 47 deletions(-)
diff --git a/chart/templates/_helpers.yaml b/chart/templates/_helpers.yaml
index 74a5bf9..80102ef 100644
--- a/chart/templates/_helpers.yaml
+++ b/chart/templates/_helpers.yaml
@@ -607,6 +607,41 @@ Create the name of the cleanup service account to use
{{- end }}
{{- end }}
+{{define "scheduler_liveness_check_command"}}
+
+ {{- if semverCompare ">=2.0.0" .Values.airflowVersion }}
+ - sh
+ - -c
+ - |
+ CONNECTION_CHECK_MAX_COUNT=0 AIRFLOW__LOGGING__LOGGING_LEVEL=ERROR exec
/entrypoint \
+ airflow jobs check --job-type SchedulerJob --hostname $(hostname)
+ {{- else }}
+ - sh
+ - -c
+ - |
+ CONNECTION_CHECK_MAX_COUNT=0 exec /entrypoint python -Wignore -c "
+ import os
+ os.environ['AIRFLOW__CORE__LOGGING_LEVEL'] = 'ERROR'
+ os.environ['AIRFLOW__LOGGING__LOGGING_LEVEL'] = 'ERROR'
+ from airflow.jobs.scheduler_job import SchedulerJob
+ from airflow.utils.db import create_session
+ from airflow.utils.net import get_hostname
+ import sys
+ with create_session() as session:
+ job =
session.query(SchedulerJob).filter_by(hostname=get_hostname()).order_by(
+ SchedulerJob.latest_heartbeat.desc()).limit(1).first()
+ sys.exit(0 if job.is_alive() else 1)"
+ {{- end }}
+{{- end }}
+
+{{define "triggerer_liveness_check_command"}}
+ - sh
+ - -c
+ - |
+ CONNECTION_CHECK_MAX_COUNT=0 AIRFLOW__LOGGING__LOGGING_LEVEL=ERROR exec
/entrypoint \
+ airflow jobs check --job-type TriggererJob --hostname $(hostname)
+{{- end }}
+
{{ define "registry_docker_config" -}}
{{- $host := .Values.registry.connection.host }}
{{- $email := .Values.registry.connection.email }}
diff --git a/chart/templates/scheduler/scheduler-deployment.yaml
b/chart/templates/scheduler/scheduler-deployment.yaml
index d144bb0..c2cece0 100644
--- a/chart/templates/scheduler/scheduler-deployment.yaml
+++ b/chart/templates/scheduler/scheduler-deployment.yaml
@@ -165,9 +165,11 @@ spec:
periodSeconds: {{ .Values.scheduler.livenessProbe.periodSeconds }}
exec:
command:
-{{- with .Values.scheduler.livenessProbe.command }}
-{{ toYaml . | indent 16 }}
-{{- end }}
+ {{- if .Values.scheduler.livenessProbe.command }}
+ {{ toYaml .Values.scheduler.livenessProbe.command | nindent
16 }}
+ {{- else}}
+ {{- include "scheduler_liveness_check_command" . | nindent
16 }}
+ {{- end }}
{{- if and $local (not $elasticsearch) }}
# Serve logs if we're in local mode and we don't have elasticsearch
enabled.
ports:
diff --git a/chart/templates/triggerer/triggerer-deployment.yaml
b/chart/templates/triggerer/triggerer-deployment.yaml
index 9a8f328..4147a2d 100644
--- a/chart/templates/triggerer/triggerer-deployment.yaml
+++ b/chart/templates/triggerer/triggerer-deployment.yaml
@@ -168,9 +168,11 @@ spec:
periodSeconds: {{ .Values.triggerer.livenessProbe.periodSeconds }}
exec:
command:
-{{- with .Values.triggerer.livenessProbe.command }}
-{{ toYaml . | indent 16 }}
-{{- end }}
+ {{- if .Values.triggerer.livenessProbe.command }}
+ {{ toYaml .Values.triggerer.livenessProbe.command | nindent
16 }}
+ {{- else }}
+ {{- include "triggerer_liveness_check_command" . | nindent
16 }}
+ {{- end }}
{{- if and (.Values.dags.gitSync.enabled) (not
.Values.dags.persistence.enabled) }}
{{- include "git_sync_container" . | indent 8 }}
{{- end }}
diff --git a/chart/values.schema.json b/chart/values.schema.json
index 39c49a4..70aab49 100644
--- a/chart/values.schema.json
+++ b/chart/values.schema.json
@@ -1507,9 +1507,15 @@
},
"command": {
"description": "Command for livenessProbe",
- "type": "array",
+ "type": [
+ "array",
+ "null"
+ ],
"items": {
- "type": "string"
+ "type": [
+ "string",
+ "null"
+ ]
}
}
}
@@ -1821,9 +1827,15 @@
},
"command": {
"description": "Command for livenessProbe",
- "type": "array",
+ "type": [
+ "array",
+ "null"
+ ],
"items": {
- "type": "string"
+ "type": [
+ "string",
+ "null"
+ ]
}
}
}
diff --git a/chart/values.yaml b/chart/values.yaml
index 810c1e0..87a0def 100644
--- a/chart/values.yaml
+++ b/chart/values.yaml
@@ -563,23 +563,7 @@ scheduler:
timeoutSeconds: 20
failureThreshold: 5
periodSeconds: 60
- command:
- - sh
- - -c
- - |
- CONNECTION_CHECK_MAX_COUNT=0 exec /entrypoint python -Wignore -c "
- import os
- os.environ['AIRFLOW__CORE__LOGGING_LEVEL'] = 'ERROR'
- os.environ['AIRFLOW__LOGGING__LOGGING_LEVEL'] = 'ERROR'
- from airflow.jobs.scheduler_job import SchedulerJob
- from airflow.utils.db import create_session
- from airflow.utils.net import get_hostname
- import sys
- with create_session() as session:
- job =
session.query(SchedulerJob).filter_by(hostname=get_hostname()).order_by(
- SchedulerJob.latest_heartbeat.desc()).limit(1).first()
- sys.exit(0 if job.is_alive() else 1)"
-
+ command: ~
# Airflow 2.0 allows users to run multiple schedulers,
# However this feature is only recommended for MySQL 8+ and Postgres
replicas: 1
@@ -968,26 +952,7 @@ triggerer:
timeoutSeconds: 20
failureThreshold: 5
periodSeconds: 60
- command:
- - sh
- - -c
- - |
- CONNECTION_CHECK_MAX_COUNT=0 exec /entrypoint python -Wignore -c "
- import os
- os.environ['AIRFLOW__CORE__LOGGING_LEVEL'] = 'ERROR'
- os.environ['AIRFLOW__LOGGING__LOGGING_LEVEL'] = 'ERROR'
-
- from airflow.jobs.triggerer_job import TriggererJob
- from airflow.utils.db import create_session
- from airflow.utils.net import get_hostname
- import sys
-
- with create_session() as session:
- job =
session.query(TriggererJob).filter_by(hostname=get_hostname()).order_by(
- TriggererJob.latest_heartbeat.desc()).limit(1).first()
-
- sys.exit(0 if job.is_alive() else 1)
- "
+ command: ~
# Create ServiceAccount
serviceAccount:
