Как прикрепить объем воздушного потока рабочего к воздушному потоку kubernetes pod operator?

Question

Я пытаюсь использовать оператор модуля kubernetes в воздушном потоке, и есть каталог, которым я хочу поделиться с модулем kubernetes на моем работнике воздушного потока, есть ли способ смонтировать каталог рабочего модуля воздушного потока в модуль kubernetes?

Я попытался использовать приведенный ниже код, но похоже, что volumn не смонтирован.

import datetime
import unittest
from unittest import TestCase
from airflow.operators.kubernetes_pod_operator import KubernetesPodOperator
from airflow.kubernetes.volume import Volume
from airflow.kubernetes.volume_mount import VolumeMount


class TestMailAlarm(TestCase):
    def setUp(self):
        self.namespace = "test-namespace"
        self.image = "ubuntu:16.04"
        self.name = "default"

        self.cluster_context = "default"

        self.dag_id = "test_dag"
        self.task_id = "root_test_dag"
        self.execution_date = datetime.datetime.now()

        self.context = {"dag_id": self.dag_id,
                        "task_id": self.task_id,
                        "execution_date": self.execution_date}

        self.cmds = ["sleep"]
        self.arguments = ["100"]

        self.volume_mount = VolumeMount('test',
                                        mount_path='/tmp',
                                        sub_path=None,
                                        read_only=False)

        volume_config = {
            'persistentVolumeClaim':
                {
                    'claimName': 'test'
                }
        }
        self.volume = Volume(name='test', configs=volume_config)

        self.operator = KubernetesPodOperator(
            namespace=self.namespace, image=self.image, name=self.name,
            cmds=self.cmds,
            arguments=self.arguments,
            startup_timeout_seconds=600,
            is_delete_operator_pod=True,
            # the operator could run successfully but the directory /tmp is not mounted to kubernetes operator
            volume=[self.volume],
            volume_mount=[self.volume_mount],
            **self.context)

    def test_execute(self):
        self.operator.execute(self.context)

Answer 1

Пример в документации очень похож на ваш код, только параметры во множественном числе volume_mounts а также volumes. Для вашего кода это будет выглядеть так:

self.operator = KubernetesPodOperator(
            namespace=self.namespace, image=self.image, name=self.name,
            cmds=self.cmds,
            arguments=self.arguments,
            startup_timeout_seconds=600,
            is_delete_operator_pod=True,
            # the operator could run successfully but the directory /tmp is not mounted to kubernetes operator
            volumes=[self.volume],
            volume_mounts=[self.volume_mount],
            **self.context)