KubernetesCeleryExecutor.yaml

apiVersion: v1
kind: Namespace
metadata:
  name: airflow

---

apiVersion: v1
kind: Service
metadata:
  name: mysql
  namespace: airflow
spec:
  type: NodePort
  ports:
  - name: mysql
    port: 3306
    targetPort: 3306
    protocol: TCP
  selector:
    app: mysql

---

apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: mysql
  namespace: airflow
spec:
  serviceName: "mysql"
  replicas: 1
  selector:
    matchLabels:
      app: mysql
  template:
    metadata:
      labels:
        app: mysql
    spec:
      containers:
      - name: mysql
        image: email2liyang/mysql-for-airflow:5.7.24
        volumeMounts:
        - name: datadir
          mountPath: /var/lib/mysql
        env:
        - name: MYSQL_ROOT_PASSWORD
          value: airflow
        - name: MYSQL_USER
          value: airflow
        - name: MYSQL_PASSWORD
          value: airflow
        - name: MYSQL_DATABASE
          value: airflow
        ports:
        - containerPort: 3306
      # No pre-stop hook is required, a SIGTERM plus some time is all that's
      # needed for graceful shutdown of a node.
      terminationGracePeriodSeconds: 60
  volumeClaimTemplates:
  - metadata:
      name: datadir
    spec:
      accessModes:
      - "ReadWriteOnce"
      resources:
        requests:
          storage: 1Gi

---

# from https://github.com/rabbitmq/rabbitmq-peer-discovery-k8s/blob/master/examples/k8s_statefulsets
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: rabbitmq 
  namespace: airflow
---
kind: Role
apiVersion: rbac.authorization.k8s.io/v1beta1
metadata:
  name: endpoint-reader
  namespace: airflow
rules:
- apiGroups: [""]
  resources: ["endpoints"]
  verbs: ["get"]
---
kind: RoleBinding
apiVersion: rbac.authorization.k8s.io/v1beta1
metadata:
  name: endpoint-reader
  namespace: airflow
subjects:
- kind: ServiceAccount
  name: rabbitmq
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: Role
  name: endpoint-reader
---
kind: Service
apiVersion: v1
metadata:
  namespace: airflow
  name: rabbitmq
  labels:
    app: rabbitmq
    type: LoadBalancer  
spec:
  type: NodePort
  ports:
   - name: http
     protocol: TCP
     port: 15672
     targetPort: 15672
     nodePort: 31672
   - name: amqp
     protocol: TCP
     port: 5672
     targetPort: 5672
     nodePort: 30672
  selector:
    app: rabbitmq
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: rabbitmq-config
  namespace: airflow
data:
  enabled_plugins: |
      [rabbitmq_management,rabbitmq_peer_discovery_k8s].
  rabbitmq.conf: |
      ## Cluster formation. See https://www.rabbitmq.com/cluster-formation.html to learn more.
      cluster_formation.peer_discovery_backend  = rabbit_peer_discovery_k8s
      cluster_formation.k8s.host = kubernetes.default.svc.cluster.local
      ## Should RabbitMQ node name be computed from the pod's hostname or IP address?
      ## IP addresses are not stable, so using [stable] hostnames is recommended when possible.
      ## Set to "hostname" to use pod hostnames.
      ## When this value is changed, so should the variable used to set the RABBITMQ_NODENAME
      ## environment variable.
      cluster_formation.k8s.address_type = hostname
      ## How often should node cleanup checks run?
      cluster_formation.node_cleanup.interval = 30
      ## Set to false if automatic removal of unknown/absent nodes
      ## is desired. This can be dangerous, see
      ##  * https://www.rabbitmq.com/cluster-formation.html#node-health-checks-and-cleanup
      ##  * https://groups.google.com/forum/#!msg/rabbitmq-users/wuOfzEywHXo/k8z_HWIkBgAJ
      cluster_formation.node_cleanup.only_log_warning = true
      cluster_partition_handling = autoheal
      ## See https://www.rabbitmq.com/ha.html#master-migration-data-locality
      queue_master_locator=min-masters
      ## See https://www.rabbitmq.com/access-control.html#loopback-users
      loopback_users.guest = false
   
---
apiVersion: apps/v1beta1
kind: StatefulSet
metadata:
  name: rabbitmq
  namespace: airflow
spec:
  serviceName: rabbitmq
  replicas: 1
  template:
    metadata:
      labels:
        app: rabbitmq
    spec:
      serviceAccountName: rabbitmq
      terminationGracePeriodSeconds: 10
      containers:        
      - name: rabbitmq-k8s
        image: rabbitmq:3.7
        volumeMounts:
          - name: config-volume
            mountPath: /etc/rabbitmq
          - name: datadir
            mountPath: /var/lib/rabbitmq  
        ports:
          - name: http
            protocol: TCP
            containerPort: 15672
          - name: amqp
            protocol: TCP
            containerPort: 5672
        livenessProbe:
          exec:
            command: ["rabbitmqctl", "status"]
          initialDelaySeconds: 60
          # See https://www.rabbitmq.com/monitoring.html for monitoring frequency recommendations.
          periodSeconds: 60
          timeoutSeconds: 15
        readinessProbe:
          exec:
            command: ["rabbitmqctl", "status"]
          initialDelaySeconds: 20
          periodSeconds: 60
          timeoutSeconds: 10
        imagePullPolicy: Always
        env:
          - name: MY_POD_IP
            valueFrom:
              fieldRef:
                fieldPath: status.podIP
          - name: MY_POD_NAME
            valueFrom:
              fieldRef:
                fieldPath: metadata.name           
          - name: RABBITMQ_USE_LONGNAME
            value: "true"
          # See a note on cluster_formation.k8s.address_type in the config file section
          - name: RABBITMQ_NODENAME
            value: "rabbit@$(MY_POD_NAME).rabbitmq.airflow.svc.cluster.local"
          - name: K8S_SERVICE_NAME
            value: "rabbitmq"
          - name: K8S_HOSTNAME_SUFFIX
            value: '.rabbitmq.airflow.svc.cluster.local'
          - name: RABBITMQ_ERLANG_COOKIE
            value: "mycookie" 
      volumes:
        - name: config-volume
          configMap:
            name: rabbitmq-config
            items:
            - key: rabbitmq.conf
              path: rabbitmq.conf
            - key: enabled_plugins
              path: enabled_plugins
  volumeClaimTemplates:
  - metadata:
      name: datadir
    spec:
      accessModes:
      - "ReadWriteOnce"
      resources:
        requests:
          storage: 5Gi            
---

apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: airflow-celery-webserver
  namespace: airflow
spec:
  replicas: 1
  selector:
    matchLabels:
      app: airflow-celery-webserver
  template:
    metadata:
      labels:
        name: airflow-celery-webserver
        app: airflow-celery-webserver
    spec:
      containers:
      - name: airflow-celery-webserver
        image: docker.io/email2liyang/docker-airflow:1.10.2
        imagePullPolicy: Always
        env:
        - name: LOAD_EX
          value: "y"
        - name: EXECUTOR
          value: Celery
        - name: AIRFLOW_MYSQL_DB_HOST
          value: mysql
        - name: AIRFLOW_MYSQL_DB_PORT
          value: "3306"
        - name: FERNET_KEY
          value: 46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho=
        - name: AIRFLOW__CELERY__BROKER_URL
          value: amqp://guest:guest@rabbitmq:5672
        - name: AIRFLOW__CELERY__RESULT_BACKEND
          value: db+mysql://airflow:airflow@$(AIRFLOW_MYSQL_DB_HOST):$(AIRFLOW_MYSQL_DB_PORT)/airflow
        - name: AIRFLOW__CORE__SQL_ALCHEMY_CONN
          value: mysql://airflow:airflow@$(AIRFLOW_MYSQL_DB_HOST):$(AIRFLOW_MYSQL_DB_PORT)/airflow
        - name: AIRFLOW__CORE__EXECUTOR
          value: CeleryExecutor
        readinessProbe:
          httpGet:
            path: /admin/
            port: 8080
          initialDelaySeconds: 8
          timeoutSeconds: 10
        livenessProbe:
          httpGet:
            path: /admin/
            port: 8080
          initialDelaySeconds: 8
          timeoutSeconds: 10
        ports:
        - name: webserver
          containerPort: 8080
        command: ["/entrypoint.sh"]
        args: ["webserver"]
---

apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: airflow-celery-flower
  namespace: airflow
spec:
  replicas: 1
  selector:
    matchLabels:
      app: airflow-celery-flower
  template:
    metadata:
      labels:
        name: airflow-celery-flower
        app: airflow-celery-flower
    spec:
      containers:
      - name: airflow-celery-flower
        image: docker.io/email2liyang/docker-airflow:1.10.2
        imagePullPolicy: Always
        env:
        - name: LOAD_EX
          value: "y"
        - name: EXECUTOR
          value: Celery
        - name: AIRFLOW_MYSQL_DB_HOST
          value: mysql
        - name: AIRFLOW_MYSQL_DB_PORT
          value: "3306"
        - name: FERNET_KEY
          value: 46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho=
        - name: AIRFLOW__CELERY__BROKER_URL
          value: amqp://guest:guest@rabbitmq:5672
        - name: AIRFLOW__CELERY__RESULT_BACKEND
          value: db+mysql://airflow:airflow@$(AIRFLOW_MYSQL_DB_HOST):$(AIRFLOW_MYSQL_DB_PORT)/airflow
        - name: AIRFLOW__CORE__SQL_ALCHEMY_CONN
          value: mysql://airflow:airflow@$(AIRFLOW_MYSQL_DB_HOST):$(AIRFLOW_MYSQL_DB_PORT)/airflow
        - name: AIRFLOW__CORE__EXECUTOR
          value: CeleryExecutor
        readinessProbe:
          httpGet:
            path: /
            port: 5555
          initialDelaySeconds: 8
          timeoutSeconds: 10
        livenessProbe:
          httpGet:
            path: /
            port: 5555
          initialDelaySeconds: 8
          timeoutSeconds: 10
        ports:
        - name: flower
          containerPort: 5555
        command: ["/entrypoint.sh"]
        args: ["flower"]
---  

apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: airflow-celery-scheduler
  namespace: airflow
spec:
  replicas: 1
  selector:
    matchLabels:
      app: airflow-celery-scheduler
  template:
    metadata:
      labels:
        name: airflow-celery-scheduler
        app: airflow-celery-scheduler
    spec:
      containers:
      - name: airflow-celery-scheduler
        image: docker.io/email2liyang/docker-airflow:1.10.2
        imagePullPolicy: Always
        env:
        - name: LOAD_EX
          value: "y"
        - name: EXECUTOR
          value: Celery
        - name: AIRFLOW_MYSQL_DB_HOST
          value: mysql
        - name: AIRFLOW_MYSQL_DB_PORT
          value: "3306"
        - name: FERNET_KEY
          value: 46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho=
        - name: AIRFLOW__CELERY__BROKER_URL
          value: amqp://guest:guest@rabbitmq:5672
        - name: AIRFLOW__CELERY__RESULT_BACKEND
          value: db+mysql://airflow:airflow@$(AIRFLOW_MYSQL_DB_HOST):$(AIRFLOW_MYSQL_DB_PORT)/airflow
        - name: AIRFLOW__CORE__SQL_ALCHEMY_CONN
          value: mysql://airflow:airflow@$(AIRFLOW_MYSQL_DB_HOST):$(AIRFLOW_MYSQL_DB_PORT)/airflow
        - name: AIRFLOW__CORE__EXECUTOR
          value: CeleryExecutor  
        readinessProbe:
          exec:
            command:
            - /bin/bash
            - -c
            - ps -ef | grep scheduler | grep -v "grep"
          initialDelaySeconds: 8
          timeoutSeconds: 10
        livenessProbe:
          exec:
            command:
            - /bin/bash
            - -c
            - ps -ef | grep scheduler | grep -v "grep"
          initialDelaySeconds: 8
          timeoutSeconds: 10
        command: ["/entrypoint.sh"]
        args: ["scheduler"]
---

apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: airflow-celery-worker
  namespace: airflow
spec:
  replicas: 2
  selector:
    matchLabels:
      app: airflow-celery-worker
  template:
    metadata:
      labels:
        name: airflow-celery-worker
        app: airflow-celery-worker
    spec:
      containers:
      - name: airflow-celery-worker
        image: docker.io/email2liyang/docker-airflow:1.10.2
        imagePullPolicy: Always
        env:
        - name: LOAD_EX
          value: "y"
        - name: EXECUTOR
          value: Celery
        - name: AIRFLOW_MYSQL_DB_HOST
          value: mysql
        - name: AIRFLOW_MYSQL_DB_PORT
          value: "3306"
        - name: FERNET_KEY
          value: 46BKJoQYlPPOexq0OhDZnIlNepKFf87WFwLbfzqDDho=
        - name: AIRFLOW__CELERY__BROKER_URL
          value: amqp://guest:guest@rabbitmq:5672
        - name: AIRFLOW__CELERY__RESULT_BACKEND
          value: db+mysql://airflow:airflow@$(AIRFLOW_MYSQL_DB_HOST):$(AIRFLOW_MYSQL_DB_PORT)/airflow
        - name: AIRFLOW__CORE__SQL_ALCHEMY_CONN
          value: mysql://airflow:airflow@$(AIRFLOW_MYSQL_DB_HOST):$(AIRFLOW_MYSQL_DB_PORT)/airflow
        - name: AIRFLOW__CORE__EXECUTOR
          value: CeleryExecutor  
        readinessProbe:
          exec:
            command:
            - /bin/bash
            - -c
            - ps -ef | grep celeryd | grep -v "grep"
          initialDelaySeconds: 8
          timeoutSeconds: 10
        livenessProbe:
          exec:
            command:
            - /bin/bash
            - -c
            - ps -ef | grep celeryd | grep -v "grep"
          initialDelaySeconds: 8
          timeoutSeconds: 10
        command: ["/entrypoint.sh"]
        args: ["worker"]
---
apiVersion: v1
kind: Service
metadata:
  name: airflow-celery-webserver
  namespace: airflow
spec:
  type: NodePort
  ports:
  - name: webserver
    port: 8080
    targetPort: webserver
  selector:
    app: airflow-celery-webserver
---
apiVersion: v1
kind: Service
metadata:
  name: airflow-celery-flower
  namespace: airflow
spec:
  type: NodePort
  ports:
  - name: flower
    port: 5555
    targetPort: flower
  selector:
    app: airflow-celery-flower