Cloud Composer 2.0.24 - Apache Airflow 2.2.5

Question

I am new to Cloud Composer, Apache airflow. I have a question with SFTPToGCSOperator.

User case:

I am trying to fetch a csv file from remote host to GCS bucket.

I did my best to configure as per the apache airflow documentation. Every time I run the dag, I am getting timed out error.

I have attached a Airflow configuration screenshot along with the complete code snippet and error message below,

I would really appreciate your help on this please.

Actual error:

{base.py:70} INFO - Using connection to: id: sftp_conn. Host: 10.173.115.67 , Port: 22, Schema: , Login: ott-admin, Password: None, extra: {'key_file': '/home/airflow/gcs/data/sftp_key.txt', 'conn_timeout': '30', 'look_for_keys': 'false'}

{sftp_to_gcs.py:150} INFO - Executing copy of /home/ott/work/data/reports/output_file-001.csv to gs://apj01_ott_mig_archive/output_file-001.csv

{ssh.py:287} WARNING - No Host Key Verification. This won't protect against Man-In-The-Middle attacks

INFO - Failed to connect. Sleeping before retry attempt 1

INFO - Failed to connect. Sleeping before retry attempt 2

socket.timeout: timed out

ERROR - Failed to execute job 495 for task fetch_file_from_remote_server (timed out; 3168)

Code:

import airflow
from airflow import DAG
from datetime import datetime, timedelta
from airflow.operators.dummy import DummyOperator
from airflow.providers.google.cloud.transfers.sftp_to_gcs import SFTPToGCSOperator

default_args = {
    'owner':'sftp',
    'retries': 1,
    'retry_delay': timedelta(minutes=1),
    'email_on_failure': False,
    'email_on_retry': False
}

with DAG(
    dag_id='sftp_test',
    default_args=default_args,
    description='sftp',
    start_date=datetime(2022,8,16,2),
    schedule_interval='@weekly'
) as dag:

    start = DummyOperator(
        task_id ='start'
    )
  
    fetch_file = SFTPToGCSOperator(
        task_id='fetch_file_from_remote_server',
        source_path='/home/ott/work/data/reports/output_file-001.csv',
        destination_bucket='apj01_ott_mig_archive',
        sftp_conn_id='sftp_conn'
    )
    end = DummyOperator(task_id='end')

    start >> fetch_file >> end

Screenshot airflow sftp configuration screenshot

user_9560594 · Answer

The error socket.timeout: timed out means that the sftp server is not reachable from your composer VMs.

To solve the problem you need to check:

if the cloud composer IP is in the IP whitelist of your sftp server
if your sftp server is accessible for public or it's only accessible in a VPC, if it's the case, you need to run composer in the same VPC or create a VPC Network Peering between the two VPCs

Cloud Composer 2.0.24 - Apache Airflow 2.2.5

1 Answers