Docker容器无法与Apache Airflow中的LocalStack通信怎么解决？

问题背景 #

我用Apache Airflow的DockerOperator启动业务容器，想通过Testcontainers测试容器与LocalStack的交互，配置如下：

• Airflow DAG通过DockerOperator运行业务容器
• 业务容器需连接LocalStack创建S3存储桶
• LocalStack的端点URL由Testcontainers动态分配

但反复遇到连接错误：

EndpointConnectionError: Could not connect to the endpoint URL: "http://localhost:<random_port>/"

已尝试的排查操作 #

• 确认Airflow和测试脚本中的AWS_ENDPOINT_URL均指向Testcontainers分配的动态LocalStack URL
• 尝试通过环境变量将正确端点传递给Docker容器
• 试过把DockerOperator的network_mode设为"host"，并将URL改为http://localhost:4566，问题依旧

补充：根据相关讨论，这是testcontainers-python的已知bug。

临时解决办法 #

1. 共享自定义Docker网络 #

让Testcontainers启动的LocalStack和Airflow DockerOperator使用同一个自定义网络，业务容器通过LocalStack的容器名称访问，避免localhost的端口映射问题：

• 在测试脚本中创建自定义网络：

  from testcontainers.core.network import Network
  
  network = Network().create()
  

• 启动LocalStack时关联该网络：

  from testcontainers.localstack import LocalStackContainer
  
  with LocalStackContainer("localstack/localstack:latest", network=network.get_name()) as localstack:
      # 业务容器直接用容器名访问LocalStack
      localstack_endpoint = "http://localstack:4566"
      # 将该端点传递给Airflow（比如通过Variable）
  

• 在DockerOperator中配置相同网络，并使用容器名作为端点：

  DockerOperator(
      task_id="run_container",
      image="your-business-image",
      environment={
          "AWS_ENDPOINT_URL": "http://localstack:4566",
          "AWS_ACCESS_KEY_ID": "test",
          "AWS_SECRET_ACCESS_KEY": "test"
      },
      network_mode=network.get_name(),
      command="aws s3 mb s3://test-bucket --endpoint-url $AWS_ENDPOINT_URL"
  )
  

2. 固定端口+主机IP适配 #

如果使用host网络，强制Testcontainers固定映射LocalStack的4566端口，同时业务容器用主机可访问的地址：

with LocalStackContainer("localstack/localstack:latest") as localstack:
    # 绑定本地4566端口到容器的4566端口
    localstack.with_bind_ports(4566, 4566)
    # Docker Desktop用host.docker.internal，Linux用主机实际IP
    localstack_endpoint = "http://host.docker.internal:4566"

代码示例参考 #

dag.py #

from airflow import DAG
from airflow.providers.docker.operators.docker import DockerOperator
from datetime import datetime

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 1, 1),
}

with DAG('localstack_s3_test', default_args=default_args, schedule_interval=None) as dag:
    create_s3_bucket = DockerOperator(
        task_id='create_s3_bucket',
        image='s3-client-image',
        environment={
            'AWS_ENDPOINT_URL': '{{ var.value.localstack_endpoint }}',
            'AWS_ACCESS_KEY_ID': 'test',
            'AWS_SECRET_ACCESS_KEY': 'test'
        },
        command='aws s3 mb s3://test-bucket --endpoint-url $AWS_ENDPOINT_URL',
        network_mode='test-network'
    )

test_dag.py #

from airflow.models import Variable
from testcontainers.localstack import LocalStackContainer
from testcontainers.core.network import Network
from airflow.utils.state import State
from airflow.models import DagRun
from datetime import datetime

def test_dag_with_localstack():
    # 创建自定义网络
    network = Network().create()
    network_name = network.get_name()

    with LocalStackContainer("localstack/localstack:latest", network=network_name) as localstack:
        # 设置Airflow变量传递端点
        Variable.set("localstack_endpoint", "http://localstack:4566")
        
        # 触发DAG运行
        dag_run = DagRun.create(
            dag_id='localstack_s3_test',
            execution_date=datetime.now(),
            state=State.RUNNING
        )
        dag_run.run()
        
        # 验证任务执行成功
        assert dag_run.state == State.SUCCESS

requirements.txt #

apache-airflow>=2.5.0
apache-airflow-providers-docker>=3.0.0
testcontainers>=3.7.0
boto3>=1.26.0

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