在Airflow DAG内调用REST API失败的原因排查

我在Docker中运行本地Airflow实例（v2.6.1版本），在Airflow外部的本地文件中调用Airflow REST API创建连接可正常执行：

import requests

def create_connection():
    response = requests.post(
        url='http://localhost:8080/api/v1/connections',
        headers={
            'Content-type': 'application/json',
            'Accept': 'application/json'
        },
        auth=(
            'my_airflow_username',
            'my_airflow_password'
        ),
        json={
            'connection_id': 'new_connection',
            'conn_type': 'generic',
            'description': 'this is the description'
        }
    )
    return response

print(create_connection().status_code)

返回状态码200并按预期创建连接。但将相同代码放入DAG中执行时：

from datetime import datetime
from airflow.decorators import dag
from airflow.operators.python import PythonOperator
import requests

@dag(
    dag_id=f'test_dag',
    start_date=datetime(2023,5,11),
    schedule_interval=None,
    max_active_runs=1
)

def execute():

    def create_connection():
        response = requests.post(
            url='http://localhost:8080/api/v1/connections',
            headers={
                'Content-type': 'application/json',
                'Accept': 'application/json'
            },
            auth=(
                'my_airflow_username',
                'my_airflow_password'
            ),
            json={
                'connection_id': 'new_connection',
                'conn_type': 'generic',
                'description': 'this is the description'
            }
        )
        return response

    create_connection = PythonOperator(
        task_id='create_connection',
        python_callable=create_connection
    )

    create_connection
    
execute()

出现错误并返回如下栈追踪：

Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/urllib3/connection.py", line 175, in _new_conn
    (self._dns_host, self.port), self.timeout, **extra_kw
  File "/home/airflow/.local/lib/python3.7/site-packages/urllib3/util/connection.py", line 95, in create_connection
    raise err
  File "/home/airflow/.local/lib/python3.7/site-packages/urllib3/util/connection.py", line 85, in create_connection
    sock.connect(sa)
ConnectionRefusedError: [Errno 111] Connection refused

During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/urllib3/connectionpool.py", line 710, in urlopen
    chunked=chunked,
  File "/home/airflow/.local/lib/python3.7/site-packages/urllib3/connectionpool.py", line 398, in _make_request
    conn.request(method, url, **httplib_request_kw)
  File "/home/airflow/.local/lib/python3.7/site-packages/urllib3/connection.py", line 244, in request
    super(HTTPConnection, self).request(method, url, body=body, headers=headers)
  File "/usr/local/lib/python3.7/http/client.py", line 1281, in request
    self._send_request(method, url, body, headers, encode_chunked)
  File "/usr/local/lib/python3.7/http/client.py", line 1327, in _send_request
    self.endheaders(body, encode_chunked=encode_chunked)
  File "/usr/local/lib/python3.7/http/client.py", line 1276, in endheaders
    self._send_output(message_body, encode_chunked=encode_chunked)
  File "/usr/local/lib/python3.7/http/client.py", line 1036, in _send_output
    self.send(msg)
  File "/usr/local/lib/python3.7/http/client.py", line 976, in send
    self.connect()
  File "/home/airflow/.local/lib/python3.7/site-packages/urllib3/connection.py", line 205, in connect
    conn = self._new_conn()
  File "/home/airflow/.local/lib/python3.7/site-packages/urllib3/connection.py", line 187, in _new_conn
    self, "Failed to establish a new connection: %s" % e
urllib3.exceptions.NewConnectionError: <urllib3.connection.HTTPConnection object at 0xffff8bb14090>: Failed to establish a new connection: [Errno 111] Connection refused

During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/requests/adapters.py", line 497, in send
    chunked=chunked,
  File "/home/airflow/.local/lib/python3.7/site-packages/urllib3/connectionpool.py", line 788, in urlopen
    method, url, error=e, _pool=self, _stacktrace=sys.exc_info()[2]
  File "/home/airflow/.local/lib/python3.7/site-packages/urllib3/util/retry.py", line 592, in increment
    raise MaxRetryError(_pool, url, error or ResponseError(cause))
urllib3.exceptions.MaxRetryError: HTTPConnectionPool(host='localhost', port=8080): Max retries exceeded with url: /api/v1/connections (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0xffff8bb14090>: Failed to establish a new connection: [Errno 111] Connection refused'))

During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/operators/python.py", line 181, in execute
    return_value = self.execute_callable()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/operators/python.py", line 198, in execute_callable
    return self.python_callable(*self.op_args, **self.op_kwargs)
  File "/opt/airflow/dags/crmf/development/credential_testing/dag2.py", line 31, in create_connection
    'description': 'this is the description'
  File "/home/airflow/.local/lib/python3.7/site-packages/requests/api.py", line 115, in post
    return request("post", url, data=data, json=json, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/requests/api.py", line 59, in request
    return session.request(method=method, url=url, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/requests/sessions.py", line 587, in request
    resp = self.send(prep, **send_kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/requests/sessions.py", line 701, in send
    r = adapter.send(request, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/requests/adapters.py", line 519, in send
    raise ConnectionError(e, request=request)
requests.exceptions.ConnectionError: HTTPConnectionPool(host='localhost', port=8080): Max retries exceeded with url: /api/v1/connections (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0xffff8bb14090>: Failed to establish a new connection: [Errno 111] Connection refused'))

我期望能像调用其他API一样在DAG内调用Airflow API，请问为何Airflow API连接失败？

核心原因 #

Docker容器内的localhost指向容器自身，而非宿主机。本地外部调用http://localhost:8080时，请求指向宿主机暴露的Airflow Webserver端口；但DAG任务运行在Airflow Worker（或Scheduler）容器中，此时localhost指代Worker容器本身，而Worker容器并未在8080端口启动Webserver服务，因此出现连接拒绝错误。

解决方法 #

方法1：使用Webserver容器名称作为主机名 #

Docker Compose部署的Airflow中，Webserver容器默认名称为airflow-webserver。将API请求URL修改为：

url='http://airflow-webserver:8080/api/v1/connections'

容器名称可在docker-compose.yaml中确认，确保Worker与Webserver容器处于同一Docker网络（Compose默认会创建统一网络）。

方法2：替换为宿主机局域网IP #

找到宿主机的局域网IP（如192.168.1.100），修改URL：

url='http://192.168.1.100:8080/api/v1/connections'

注意：此方式依赖宿主机IP稳定，IP变动需同步修改。

方法3：直接使用Airflow内部API创建连接（推荐） #

无需通过REST API绕路，Airflow提供内部方法直接操作连接，更高效且避免网络问题。修改DAG中的create_connection函数：

from airflow.models.connection import Connection
from airflow.utils.session import create_session

def create_connection():
    conn = Connection(
        conn_id='new_connection',
        conn_type='generic',
        description='this is the description'
    )
    with create_session() as session:
        if not session.query(Connection).filter(Connection.conn_id == conn.conn_id).first():
            session.add(conn)
            session.commit()
            return "Connection created successfully"
        else:
            return "Connection already exists"

该方式直接操作Airflow元数据库，无需网络请求，是更规范的实现方式。

内容的提问来源于stack exchange，提问作者gbeaven