Airflow on Kubernetes

def as_pod(self): if self.kube_config.pod_template_file: return PodGenerator(pod_template_file=self.kube_config.pod_template_file).gen_pod() pod = PodGenerator( image=self.kube_config.kube_image, image_pull_policy=self.kube_config.kube_image_pull_policy or 'IfNotPresent', image_pull_secrets=self.kube_config.image_pull_secrets, volumes=self._get_volumes(), volume_mounts=self._get_volume_mounts(), init_containers=self._get_init_containers(), labels=self.kube_config.kube_labels, annotations=self.kube_config.kube_annotations, affinity=self.kube_config.kube_affinity, tolerations=self.kube_config.kube_tolerations, envs=self._get_environment(), node_selectors=self.kube_config.kube_node_selectors, service_account_name=self.kube_config.worker_service_account_name or 'default', restart_policy='Never' ).gen_pod() pod.spec.containers[0].env_from = pod.spec.containers[0].env_from or [] pod.spec.containers[0].env_from.extend(self._get_env_from()) pod.spec.security_context = self._get_security_context() return append_to_pod(pod, self._get_secrets())

minikube start 😄 Darwin 13.4 (arm64) 의 minikube v1.30.1 🎉 minikube 1.31.2 이 사용가능합니다! 다음 경로에서 다운받으세요: https://github.com/kubernetes/minikube/releases/tag/v1.31.2 💡 해당 알림을 비활성화하려면 다음 명령어를 실행하세요. 'minikube config set WantUpdateNotification false' ✨ 자동적으로 docker 드라이버가 선택되었습니다 📌 Using Docker Desktop driver with root privileges 👍 minikube 클러스터의 minikube 컨트롤 플레인 노드를 시작하는 중 🚜 베이스 이미지를 다운받는 중 ... 🔥 Creating docker container (CPUs=2, Memory=7903MB) ... 🐳 쿠버네티스 v1.26.3 을 Docker 23.0.2 런타임으로 설치하는 중 ▪ 인증서 및 키를 생성하는 중 ... ▪ 컨트롤 플레인이 부팅... ▪ RBAC 규칙을 구성하는 중 ... 🔗 Configuring bridge CNI (Container Networking Interface) ... ▪ Using image gcr.io/k8s-minikube/storage-provisioner:v5 🔎 Kubernetes 구성 요소를 확인... 🌟 애드온 활성화 : storage-provisioner, default-storageclass 🏄 끝났습니다! kubectl이 "minikube" 클러스터와 "default" 네임스페이스를 기본적으로 사용하도록 구성되었습니다.

(root user) curl -fsSL -o get_helm.sh https://raw.githubusercontent.com/helm/helm/main/scripts/get-helm-3 chmod 700 get_helm.sh ./get_helm.sh which helm /usr/local/bin/helm helm의 위치 찾기: 먼저, root 사용자로 로그인하여 helm의 정확한 위치를 찾습니다. sudo su which helm 이 명령은 helm의 경로를 출력합니다. 예를 들어, 출력 결과가 /usr/local/bin/helm이라면, 이 경로를 기록해 두세요. PATH 환경 변수 수정: 이제 사용자로 돌아가서 PATH 환경 변수에 helm의 위치를 추가합니다. 사용자의 셸 설정 파일 (예: .bashrc 또는 .zshrc)을 편집해야 합니다. exit # root 사용자에서 나오기 (user) nano ~/.bashrc # 또는 사용하는 셸에 맞는 설정 파일을 엽니다 파일의 마지막에 다음 라인을 추가합니다 (여기서는 helm이 /usr/local/bin/helm에 위치한다고 가정합니다):] export PATH=$PATH:/usr/local/bin 변경사항을 저장하고 편집기를 종료합니다. 변경사항 적용: 변경사항을 적용하기 위해 셸 설정 파일을 다시 로드합니다. source ~/.bashrc # 또는 사용하는 셸에 맞는 설정 파일 helm 실행 확인: 이제 helm 명령어를 실행하여 변경사항이 올바르게 적용되었는지 확인합니다. helm version

helm repo add apache-airflow https://airflow.apache.org

# values.yaml airflowVersion: "2.6.2"

helm upgrade --install airflow apache-airflow/airflow --namespace airflow --create-namespace -f my-values.yaml

kubectl port-forward svc/airflow-webserver 8080:8080 -n airflow kubectl port-forward svc/airflow-webserver 20000:8080 -n airflow

kubectl get pods -n airflow NAME READY STATUS RESTARTS AGE airflow-postgresql-0 1/1 Running 0 2m12s airflow-redis-0 1/1 Running 0 2m12s airflow-scheduler-68d6b6cdd4-z2ngc 2/2 Running 0 2m12s airflow-statsd-77685bcd45-t6pfj 1/1 Running 0 2m12s airflow-triggerer-0 2/2 Running 0 2m12s airflow-webserver-bf8bbc9cb-qgd85 1/1 Running 0 2m12s airflow-worker-0 2/2 Running 0 2m12s

ssh-keygen -t rsa -b 4096 -C "your_email@example.com"

pbcopy < airflow_ssh_key.pub

kubectl create secret generic airflow-git-ssh-secret \ --from-file=gitSshKey=/Users/xxxx/.ssh/airflow_ssh_key \ --from-file=id_ed25519.pub=/Users/xxxx/.ssh/airflow_ssh_key.pub \ -n airflow

cd ~/.ssh vi values.yaml

/# Git sync airflowVersion: "2.6.2" extraEnv: | - name: "AIRFLOW__CORE__PLUGINS_FOLDER" value: "/opt/airflow/dags/repo/plugins" # Git sync dags: persistence: # Enable persistent volume for storing dags enabled: true #변경 1 # Volume size for dags size: 10Gi # If using a custom storageClass, pass name here storageClassName: # access mode of the persistent volume accessMode: ReadWriteOnce ## the name of an existing PVC to use existingClaim: gitSync: enabled: true #변경 2 repo: git@github.com:xxx/xxxx.git #변경 3 branch: main #변경 4 rev: HEAD depth: 1 # the number of consecutive failures allowed before aborting maxFailures: 0 # subpath within the repo where dags are located # should be "" if dags are at repo root subPath: "" credentialsSecret: git-credentials sshKeySecret: airflow-git-ssh-secret #변경 5 wait: 10 containerName: git-sync uid: 65533 # When not set, the values defined in the global securityContext will be used securityContext: {} # runAsUser: 65533 # runAsGroup: 0 extraVolumeMounts: [] env: [] resources: {} # limits: # cpu: 100m # memory: 128Mi # requests: # cpu: 100m # memory: 128Mi

helm upgrade --install airflow apache-airflow/airflow -n airflow -f my-values.yaml

Error: UPGRADE FAILED: cannot patch "airflow-scheduler" with kind Deployment: Deployment.apps "airflow-scheduler" is invalid: [spec.template.spec.containers[1].volumeMounts[1].name: Not found: "git-sync-ssh-key", spec.template.spec.initContainers[1].volumeMounts[1].name: Not found: "git-sync-ssh-key"]

vi values.yaml /# Airflow scheduler settings

scheduler: extraVolumes: - name: git-sync-ssh-key secret: secretName: airflow-git-ssh-secret

helm upgrade --install airflow apache-airflow/airflow -n airflow -f values.yaml

from datetime import datetime, timedelta from textwrap import dedent from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.python_operator import PythonOperator from datetime import datetime def test(): print(datetime.now()) with DAG( 'tutorial', # These args will get passed on to each operator # You can override them on a per-task basis during operator initialization default_args={ 'depends_on_past': False, 'email': ['airflow@example.com'], 'email_on_failure': False, 'email_on_retry': False, 'retries': 1, 'retry_delay': timedelta(minutes=5), }, description='A simple tutorial DAG', schedule_interval='1 * * * *', start_date=datetime(2021, 1, 1), catchup=False, tags=['example'], ) as dag: # t1, t2 and t3 are examples of tasks created by instantiating operators t1 = PythonOperator( task_id='print_date', python_callable=test, ) t2 = BashOperator( task_id='sleep', depends_on_past=False, bash_command='sleep 5', retries=3, ) t1.doc_md = dedent( """\ #### Task Documentation You can document your task using the attributes `doc_md` (markdown), `doc` (plain text), `doc_rst`, `doc_json`, `doc_yaml` which gets rendered in the UI's Task Instance Details page.  """ ) dag.doc_md = __doc__ # providing that you have a docstring at the beginning of the DAG dag.doc_md = """ This is a documentation placed anywhere """ # otherwise, type it like this templated_command = dedent( """ {% for i in range(5) %} echo "{{ ds }}" echo "{{ macros.ds_add(ds, 7)}}" {% endfor %} """ ) t3 = BashOperator( task_id='templated', depends_on_past=False, bash_command=templated_command, ) t1 >> [t2, t3]

vi values.yaml

airflowVersion: "2.6.2" extraEnv: | - name: "AIRFLOW__CORE__PLUGINS_FOLDER" value: "/opt/airflow/dags/repo/airflow/plugins" executor: "KubernetesExecutor" dags: persistence: # Enable persistent volume for storing dags enabled: true #변경 1 # Volume size for dags size: 10Gi # If using a custom storageClass, pass name here storageClassName: # access mode of the persistent volume accessMode: ReadWriteOnce ## the name of an existing PVC to use existingClaim: gitSync: enabled: true repo: git@github.com:xxx/xxxx.git branch: main rev: HEAD depth: 1 # the number of consecutive failures allowed before aborting maxFailures: 0 # subpath within the repo where dags are located # should be "" if dags are at repo root subPath: "airflow/dags" credentialsSecret: git-credentials sshKeySecret: airflow-git-ssh-secret #변경 5 wait: 10 containerName: git-sync uid: 65533 # When not set, the values defined in the global securityContext will be used securityContext: {} # runAsUser: 65533 # runAsGroup: 0 extraVolumeMounts: [] env: [] resources: {} # limits: # cpu: 100m # memory: 128Mi # requests: # cpu: 100m # memory: 128Mi scheduler: extraVolumes: - name: git-sync-ssh-key secret: secretName: airflow-git-ssh-secret

python3 -c 'import secrets; print(secrets.token_hex(16))' f83bab1deefae2199d27d2621762c30b

kubectl create secret generic \ airflow-webserver-secret \ --from-literal="webserver-secret-key=f83bab1deefae2199d27d2621762c30b" \ -n airflow secret/airflow-webserver-secret created

airflowVersion: "2.6.2" extraEnv: | - name: "AIRFLOW__CORE__PLUGINS_FOLDER" value: "/opt/airflow/dags/repo/airflow/plugins" # 추가 webserverSecretKey: webserver-secret-key webserverSecretKeySecretName: airflow-webserver-secret executor: "KubernetesExecutor" dags: persistence: # Enable persistent volume for storing dags enabled: true #변경 1 # Volume size for dags size: 10Gi # If using a custom storageClass, pass name here storageClassName: # access mode of the persistent volume accessMode: ReadWriteOnce ## the name of an existing PVC to use existingClaim: gitSync: enabled: true repo: https://github.com/xxx/xxx.git branch: main rev: HEAD depth: 1 # the number of consecutive failures allowed before aborting maxFailures: 0 # subpath within the repo where dags are located # should be "" if dags are at repo root subPath: "airflow/dags" # credentialsSecret: git-credentials # sshKeySecret: airflow-git-ssh-secret #변경 5 wait: 10 containerName: git-sync uid: 65533 # When not set, the values defined in the global securityContext will be used securityContext: {} # runAsUser: 65533 # runAsGroup: 0 extraVolumeMounts: [] env: [] resources: {} # limits: # cpu: 100m # memory: 128Mi # requests: # cpu: 100m # memory: 128Mi # 추가 webserver: service: type: NodePort ## service annotations ports: - name: airflow-ui port: "{{ .Values.ports.airflowUI }}" targetPort: "{{ .Values.ports.airflowUI }}" nodePort: 31151

helm upgrade --install airflow apache-airflow/airflow -n airflow -f my-values.yaml

$ minikube ip 192.168.49.2 $ kubectl get svc -n airflow NAME TYPE CLUSTER-IP EXTERNAL-IP PORT(S) AGE airflow-postgresql ClusterIP 10.100.22.149 <none> 5432/TCP 30m airflow-postgresql-hl ClusterIP None <none> 5432/TCP 30m airflow-statsd ClusterIP 10.107.91.48 <none> 9125/UDP,9102/TCP 30m airflow-triggerer ClusterIP None <none> 8794/TCP 30m airflow-webserver NodePort 10.100.214.92 <none> 8080:31151/TCP 30m # 아래의 주소로 접근 http://192.168.49.2:31151

$ kubectl logs airflow-webserver-78866bff9b-cbrx4 -n airflow Traceback (most recent call last): File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/log/file_task_handler.py", line 505, in _read_from_logs_server response = _fetch_logs_from_service(url, rel_path) File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/log/file_task_handler.py", line 93, in _fetch_logs_from_service headers={"Authorization": signer.generate_signed_token({"filename": log_relative_path})}, File "/home/airflow/.local/lib/python3.7/site-packages/httpx/_api.py", line 201, in get trust_env=trust_env, File "/home/airflow/.local/lib/python3.7/site-packages/httpx/_api.py", line 110, in request follow_redirects=follow_redirects, File "/home/airflow/.local/lib/python3.7/site-packages/httpx/_client.py", line 821, in request return self.send(request, auth=auth, follow_redirects=follow_redirects) File "/home/airflow/.local/lib/python3.7/site-packages/httpx/_client.py", line 912, in send history=[], File "/home/airflow/.local/lib/python3.7/site-packages/httpx/_client.py", line 939, in _send_handling_auth history=history, File "/home/airflow/.local/lib/python3.7/site-packages/httpx/_client.py", line 973, in _send_handling_redirects response = self._send_single_request(request) File "/home/airflow/.local/lib/python3.7/site-packages/httpx/_client.py", line 1009, in _send_single_request response = transport.handle_request(request) File "/home/airflow/.local/lib/python3.7/site-packages/httpx/_transports/default.py", line 218, in handle_request resp = self._pool.handle_request(req) File "/usr/local/lib/python3.7/contextlib.py", line 130, in __exit__ self.gen.throw(type, value, traceback) File "/home/airflow/.local/lib/python3.7/site-packages/httpx/_transports/default.py", line 77, in map_httpcore_exceptions raise mapped_exc(message) from exc httpx.ConnectError: [Errno -2] Name or service not known

logs: persistence: enabled: true