气流 SSH 操作错误：[Errno 2] 没有这样的文件或目录：

Question

气流 1.10.10
minikube 1.22.0
亚马逊emr

我在 kubernetes(minikube) 上运行气流。 Dags 从 github 同步。 Amazon EMR 上的 spark-submit 作为 CLI 模式。

为了做到这一点，我附上了 EMR pem 密钥。 因此，我从 AWS S3 获取 pem 密钥，而 ExtraInitContainer 正在获取图像 awscli 并将卷安装在 airlfow/sshpem

当我从气流 WebUI 建立连接时报告错误 “con_type”：“ssh” "key_file": "/opt/sshepm/emr.pem"

SSH operator error: [Errno 2] No such file or directory: '/opt/airflow/sshpem/emr.pem'

它就在那里。我认为这与一些 PATH 或权限问题有关，因为我在 ExtraInitContainer 上获得了 emr.pem 并且它的权限是 root。虽然我暂时将用户更改为 1000:1000，但气流 WebUI 在获取密钥时无法获取此目录存在一些问题。

完整日志如下

> Traceback (most recent call last):   File
    > "/home/airflow/.local/lib/python3.6/site-packages/airflow/contrib/operators/ssh_operator.py",
    > line 108, in execute
    >     with self.ssh_hook.get_conn() as ssh_client:   File "/home/airflow/.local/lib/python3.6/site-packages/airflow/contrib/hooks/ssh_hook.py",
    > line 194, in get_conn
    >     client.connect(**connect_kwargs)   File "/home/airflow/.local/lib/python3.6/site-packages/paramiko/client.py",
    > line 446, in connect
    >     passphrase,   File "/home/airflow/.local/lib/python3.6/site-packages/paramiko/client.py",
    > line 677, in _auth
    >     key_filename, pkey_class, passphrase   File "/home/airflow/.local/lib/python3.6/site-packages/paramiko/client.py",
    > line 586, in _key_from_filepath
    >     key = klass.from_private_key_file(key_path, password)   File "/home/airflow/.local/lib/python3.6/site-packages/paramiko/pkey.py",
    > line 235, in from_private_key_file
    >     key = cls(filename=filename, password=password)   File "/home/airflow/.local/lib/python3.6/site-packages/paramiko/rsakey.py",
    > line 55, in __init__
    >     self._from_private_key_file(filename, password)   File "/home/airflow/.local/lib/python3.6/site-packages/paramiko/rsakey.py",
    > line 175, in _from_private_key_file
    >     data = self._read_private_key_file("RSA", filename, password)   File
    > "/home/airflow/.local/lib/python3.6/site-packages/paramiko/pkey.py",
    > line 307, in _read_private_key_file
    >     with open(filename, "r") as f: FileNotFoundError: [Errno 2] No such file or directory: '/opt/airflow/sshpem/emr-pa.pem'
    > 
    > During handling of the above exception, another exception occurred:
    > 
    > Traceback (most recent call last):   File
    > "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py",
    > line 979, in _run_raw_task
    >     result = task_copy.execute(context=context)   File "/opt/airflow/class101-airflow/plugins/operators/emr_ssh_operator.py",
    > line 107, in execute
    >     super().execute(context)   File "/home/airflow/.local/lib/python3.6/site-packages/airflow/contrib/operators/ssh_operator.py",
    > line 177, in execute
    >     raise AirflowException("SSH operator error: {0}".format(str(e))) airflow.exceptions.AirflowException: SSH operator error: [Errno 2] No
    > such file or directory: '/opt/airflow/sshpem/emr-pa.pem' [2021-07-14
    > 05:40:31,624] Marking task as UP_FOR_RETRY. dag_id=test_staging,
    > task_id=extract_categories_from_mongo, execution_date=20210712T190000,
    > start_date=20210714T054031, end_date=20210714T054031 [2021-07-14
    > 05:40:36,303] Task exited with return code 1

气流主页：/opt/airflow
dags：/opt/airflow//dags
pemkey : /opt/sshpem/
气流.cfg：/opt/airflow
气流环境： 出口 PATH="/home/airflow/.local/bin:$PATH"

我的 yaml 文件

airflow: 
   image: 
     repository: airflow
   executor: KubernetesExecutor
   extraVolumeMounts:
    - name: sshpem
      mountPath: /opt/airflow/sshpem
   extraVolumes:
    - name: sshpem
      emptyDir: {}
scheduler:
    extraInitContainers:
    - name: emr-key-file-download
      image: amazon/aws-cli
      command: [
        "sh",
        "-c",
        "aws s3 cp s3://mykeyfile/path.my.pem&& \
        chown -R 1000:1000 /opt/airflow/sshpem/"
      volumeMounts:
        - mountPath: /opt/airflow/sshpem
          name: sshpem

Answer 1

你使用的是 KubernetesExecutor 还是 CeleryExecutor？

如果是前者，那么您必须确保将额外的 init 容器添加到您正在使用的 pod_template（KubernetesExecutor 中的任务）作为单独的 POD 运行。

如果是后者，您应该确保也为工作人员添加了额外的 init 容器，而不仅仅是为调度程序添加）。

顺便说一句。 Airflow 1.10 已于 2021 年 6 月 17 日终止使用，它甚至不会收到关键的安全修复程序。您可以观看我们在最近的 Airflow 峰会“保持您的 Airflow 安全”-https://airflowsummit.org/sessions/2021/panel-airflow-security/ 上的演讲，了解升级到 Airflow 2 的重要性。