I have deployed apache airflow in azure kubernetes.
Helm repository of apache airflow: https://github.com/apache/airflow/tree/master/chart
AKS version: 1.16.13
I am reading the dags in my GitHub with git-sync, for that, I have modified the values.yml file:
dags:
persistence:
# Enable persistent volume for storing dags
enabled: false
# Volume size for dags
size: 1Gi
# If using a custom storageClass, pass name here
storageClassName:
# access mode of the persistent volume
accessMode: ReadWriteMany
## the name of an existing PVC to use
existingClaim: ~
gitSync:
enabled: true
# git repo clone url
# ssh examples ssh://git@github.com/apache/airflow.git
# git@github.com:apache/airflow.git
# https example: https://github.com/apache/airflow.git
repo: https://my_github_repository.git
branch: master
rev: HEAD
root: "/git"
dest: "repo"
Once I have airflow deployed I have tested it with this dag:
from airflow import DAG
from datetime import datetime, timedelta
from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator
from airflow.operators.dummy_operator import DummyOperator
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date': datetime.utcnow(),
'email': ['airflow@example.com'],
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': timedelta(minutes=5)
}
dag = DAG(
'kubernetes_sample', default_args=default_args, schedule_interval=timedelta(minutes=10))
start = DummyOperator(task_id='run_this_first', dag=dag)
passing = KubernetesPodOperator(namespace='default',
image="python:3.8-slim-buster",
cmds=["python3","-c"],
arguments=["print('hello world')"],
labels={"foo": "bar"},
name="passing-test",
task_id="passing-task",
get_logs=True,
dag=dag
)
passing.set_upstream(start)
It works fine. Now I want to use my own images. For that, I am using azure containers, and following this guide: https://airflow.readthedocs.io/en/latest/howto/operator/kubernetes.html, I am using this code to create the secret to access to my azure registry:
kubectl create secret docker-registry testquay \
--docker-server=quay.io \
--docker-username=<Profile name> \
--docker-password=<password>
I build my image and I test locally and it is working. I upload the image to azure container registry and I write the following dag:
from airflow import DAG
from datetime import datetime, timedelta
from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator
from airflow.operators.dummy_operator import DummyOperator
from kubernetes.client import models as k8s
import logging
import os
import sys
import traceback
try:
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date': datetime.utcnow(),
'email': ['airflow@example.com'],
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': timedelta(minutes=5)
}
dag = DAG(
'test1', default_args=default_args, schedule_interval=timedelta(minutes=10))
start = DummyOperator(task_id='run_this_first', dag=dag)
quay_k8s = KubernetesPodOperator(
namespace='default',
image='<MY_SERVER_NAME>/testingairlfowdags:latest',
image_pull_secrets=[k8s.V1LocalObjectReference('azure-registry')],
name="testingairlfowdags",
is_delete_operator_pod=True,
in_cluster=True,
task_id="task-two",
get_logs=True,
log_events_on_failure = True,
dag=dag
)
start >> quay_k8s
except Exception as e:
error_message = {
"message": "An internal error ocurred"
,"error": str(e)
, "error information" : str(sys.exc_info())
, "traceback": str(traceback.format_exc())
}
logging.info(error_message)
But when the quay_k8s task starts a pod is created and suddenly killed, I cannot get any log.
The Kubernetes dashboard shows me this when the pod is initialised:

But then:
