Docker Python SDK删除容器后出现未关闭Socket资源警告问题排查

问题 #

在MacBook上安装了Docker Desktop，使用Docker Python SDK 6.0.1版本（5.0.3版本也存在相同问题）。测试用例中创建本地容器并存储在container变量，删除容器时执行：

container.remove(force=True)
container = None

执行最后一行时偶尔收到警告：

$HOME/git/es-client/test/local_es_decorator.py:197: ResourceWarning: unclosed <socket.socket fd=5, family=AddressFamily.AF_UNIX, type=SocketKind.SOCK_STREAM, proto=0, raddr=$HOME/.docker/run/docker.sock>
  self._local_es_container = None
Object allocated at (most recent call last):
  File "$HOME/.pyenv/versions/es-client/lib/python3.8/site-packages/docker/transport/unixconn.py", lineno 28
    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)

在两行代码间加5秒sleep无法解决，且仅在某个虚拟环境中出现该警告，不清楚环境差异。

以下是创建和删除容器的装饰器代码：

import asyncio
import tracemalloc
from functools import wraps
from os import getenv
from pprint import pprint
from typing import Optional
from unittest import TestCase

import requests
from docker import types, from_env
from docker.errors import NotFound
from docker.models.containers import Container
from elasticsearch_dsl import connections
from time import sleep

from ESClient.connection import set_endpoint
from ESClient.utils import measure_time

ELASTICSEARCH_TEST_PORT = 9200
TEST_ELASTICSEARCH_ENDPOINT = 'http://localhost'
TEST_ELASTICSEARCH_CONNECTION = connections.create_connection(hosts=['localhost'], port=ELASTICSEARCH_TEST_PORT,
                                                              timeout=10, max_retries=2)
ELASTICSEARCH_TEST_URL = f'{TEST_ELASTICSEARCH_ENDPOINT}:{ELASTICSEARCH_TEST_PORT}'
CONTAINER_NAME = 'elasticsearch_test'


# noinspection PyPep8Naming
class use_local_elasticsearch(TestContextDecorator):
    """
    Class decorator for unittest.TestCase to make use of a Docker ElasticSearch container for that test case

    This introduces a dependency to the Docker SDK
    You also need to have Docker installed, see https://docs.docker.com/get-docker/

    Usage example:

    If we have a TestCase class called FindAutoRestockTestCase that tests integration with ES (therefore we don't want
    to just mock those interactions but to actually have our code integrate to a real ES cluster), just decorate the
    class with this decorator as  follows:

    @use_local_elasticsearch()
    class FindAutoRestockTestCase(TestCase):
        ...

    This will spin up a local ES docker container, create all required indices there, set the endpoint in ESClient
    (if needed) and override settings ELASTICSEARCH_ENDPOINT and ELASTICSEARCH_CONNECTION before running the tests in
    the class, so no other change is needed. After every test method in the class is executed, the container will be
    removed (unless destroy_container = False), and settings are restored to their previous values.

    Note that the cluster is spin up when you initialize the class. This means that at the time that
    setUpClass or setUpTestData is called, the cluster is not available yet.
    """

    def __init__(self, destroy_container: bool = True):
        super().__init__()

        tracemalloc.start()

        self._destroy_container = destroy_container
        self._local_es_container: Optional[Container] = None

        self._decorated_class = None
        self._enabled = False

    def enable(self):
        """
        Call method in charge of creating ES container with indices, set endpoint of ESClient if present,
        and enable override_settings

        If the environment variable USE_EXISTING_ES is set, we don't spin up a Docker container from Python.
        This is used by CircleCI. Rationale: The tests in CircleCI are running from a Docker container with Python.
        This particular container does not have access to Docker itself, and therefore cannot create containers
        itself. We create an existing Docker container in the CircleCI script and reuse it here. In your local
        environment however, we do have access to Docker, and as long as the app is running, we can (and will if
        necessary) create a Docker container whenever it is necessary

        Often, a test class wants to add test data to ElasticSearch one time for all test methods. However, we
        cannot create this data in setupTestData(), since this method is called before the class is decorated
        and this method enable() is called, and hence no ElasticSearch is running at that moment in time.

        This method enable() is called for every test case method inside a TestClass, so if we setup the cluster
        in this method and reset all the indices, we have to do this for every test method. This is time consuming
        if there are many test cases inside a test class.
        In order to only one time setup the cluster per test class and create the test data in ElasticSearch only
        once after the cluster is initialized, we do the following:
        - Run the classmethod on_elasticsearch_loaded() on the decorated test class. Individual classes can
          implement this method as they see fit to generate test data. If the test class doesn't implement
          this method, no test data is generated at this point.
        - Keep track of whether we have initialized ElasticSearch already in the variable self._enabled. If the
          cluster has been initialized, we don't set it up again.

        If destroy_container=True, we setup a fresh container for every test method, and the classmethod
        on_elasticsearch_loaded() is called for every test method. It is therefore important that this method is
        idempotent if it has effects other than in ElasticSearch, e.g. creating data in SQL.
        """

        if not self._local_es_container or not self._enabled:
            if not getenv('USE_EXISTING_ES'):
                self._spin_up_es_container()
            self._wait_until_es_runs()
            set_endpoint(TEST_ELASTICSEARCH_ENDPOINT)

            if hasattr(self._decorated_class, 'on_elasticsearch_loaded'):
                self._decorated_class.on_elasticsearch_loaded()
            self._enabled = True

    def disable(self):
        """
        Remove ES container
        """

        if self._destroy_container and self._local_es_container:
            self._local_es_container.remove(force=True)
            pprint(tracemalloc.get_traced_memory())
            # tracemalloc.stop()
            traceback = tracemalloc.get_object_traceback(self._local_es_container)
            print(traceback)
            print('-' * 80)
            self._local_es_container = None

    def decorate_class(self, cls):
        """
        Call decorate_class of both parent and "grandparent" classes,
        as both are needed but parent class does not call its parent
        """

        result_cls = super().decorate_class(cls)
        # Keep track of the decorated class in order to call on_elasticsearch_loaded()
        # on it, see docblock of enable() for more details
        self._decorated_class = result_cls
        return result_cls

    def _spin_up_es_container(self):
        """
        Spin up a Docker ES container if there is not yet one running

        For CircleCI, we set up a Docker container in the .circleci configuration, and reuse that one
        In that case, the environment variable USE_EXISTING_ES=1 and this function is not called at all.
        It would fail, because we could not create a Docker container with ElasticSearch from the
        Python docker container in which the tests are running.
        """

        # Spin up ElasticSearch docker container
        es_environment = {'cluster.name': 'docker-cluster', 'discovery.type': 'single-node',
                          'bootstrap.memory_lock': 'true', 'xpack.security.enabled': 'false',
                          'ES_JAVA_OPTS': '-Xms128m -Xmx128m'}

        client = from_env()
        ulimit = types.Ulimit(name='memlock', soft=-1, hard=-1)

        # Check if there is already a running test ES container locally. If so, use it
        try:
            container = client.containers.get(CONTAINER_NAME)
            if container.status != 'running':
                container.start()
        except NotFound:
            # If there is not yet a running test ES container, run it
            container = client.containers.run('elasticsearch:7.12.0', name=CONTAINER_NAME, detach=True,
                                              environment=es_environment, ulimits=[ulimit],
                                              ports={str(ELASTICSEARCH_TEST_PORT): ELASTICSEARCH_TEST_PORT})

        # container_starting, _ = container.exec_run(f'curl {ELASTICSEARCH_TEST_URL}')

        # Store a reference to the container, so we can remove it on tear down
        self._local_es_container = container

    def _wait_until_es_runs(self):
        # Wait until ElasticSearch is running before we try to create indices
        container_starting = True
        with measure_time() as t:
            while container_starting:
                try:
                    requests.get(ELASTICSEARCH_TEST_URL)
                    container_starting = False
                except requests.exceptions.ConnectionError as e:
                    if t.seconds < 5 * 60:
                        sleep(5)
                    else:
                        msg = 'ElasticSearch container is not running after 5 minutes'
                        raise requests.exceptions.ConnectionError(msg) from e

原因分析与解决方案 #

核心原因 #

• Docker SDK的socket资源未及时关闭：Container对象内部持有与Docker daemon通信的Unix socket连接。调用container.remove()后，SDK不会立即关闭socket，而是依赖Python垃圾回收机制处理。当你把container设为None时，对象被标记为可回收，但垃圾回收时机不确定，导致socket未及时关闭触发警告。
• 虚拟环境差异的本质：不同虚拟环境的Python版本、垃圾回收阈值、依赖库（如urllib3）版本不同，会影响垃圾回收触发时机。部分环境中垃圾回收可能在警告输出前就处理了socket，因此无警告。
• tracemalloc的干扰：装饰器中启动的tracemalloc会追踪内存分配，可能延迟垃圾回收执行，使得socket未关闭的状态被检测到，进而触发警告。

解决办法 #

1. 显式关闭容器关联的客户端连接
   每个Container对象关联一个client实例，显式关闭该client的连接池可确保socket被释放：

def disable(self):
    if self._destroy_container and self._local_es_container:
        self._local_es_container.remove(force=True)
        # 显式关闭客户端连接
        self._local_es_container.client.close()
        self._local_es_container = None

2. 管理Docker客户端生命周期
   在_spin_up_es_container中保留客户端实例，在disable方法中统一关闭：

def _spin_up_es_container(self):
    # ... 其他代码 ...
    client = from_env()
    # ... 创建/启动容器逻辑 ...
    self._docker_client = client
    self._local_es_container = container

def disable(self):
    if self._destroy_container and self._local_es_container:
        self._local_es_container.remove(force=True)
        if hasattr(self, '_docker_client'):
            self._docker_client.close()
        self._local_es_container = None

3. 调整tracemalloc使用时机
   在disable方法中关闭tracemalloc，避免干扰垃圾回收：

def disable(self):
    if self._destroy_container and self._local_es_container:
        self._local_es_container.remove(force=True)
        pprint(tracemalloc.get_traced_memory())
        tracemalloc.stop()  # 关闭tracemalloc
        traceback = tracemalloc.get_object_traceback(self._local_es_container)
        print(traceback)
        print('-' * 80)
        self._local_es_container = None

4. 手动触发垃圾回收
   设置container = None后手动调用垃圾回收，确保socket被及时回收：

import gc

def disable(self):
    if self._destroy_container and self._local_es_container:
        self._local_es_container.remove(force=True)
        self._local_es_container = None
        gc.collect()  # 手动触发垃圾回收

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