`container.attach(stream=True)` causing "ResourceWarning: unclosed <socket.socket …>"

Attempting to use container.attach(stream=True) to stream logs from a container appears to leak unclosed sockets.

The following test case:

import contextlib
import unittest

import docker


class TestDocker(unittest.TestCase):

   def test(self):
      for count in range(10):
         with contextlib.closing(docker.from_env()) as client:
            container = client.containers.run('alpine',
               auto_remove=True,
               command=('/bin/sh', '-c', 'echo Hello; sleep 1'),
               detach=True,
               init=True,
               tty=True
            )
            with contextlib.closing(container.attach(
               logs=True,
               stdout=True,
               stream=True
            )) as logs:
               next(logs, b'').decode()
            container.stop()

yields the following ouput:

test (container.test_docker.TestDocker) ... /usr/lib/python3.8/email/feedparser.py:158: ResourceWarning: unclosed <socket.socket [closed] fd=7, family=AddressFamily.AF_UNIX, type=SocketKind.SOCK_STREAM, proto=0>
  _factory(policy=self.policy)
ResourceWarning: Enable tracemalloc to get the object allocation traceback
/usr/lib/python3.8/email/feedparser.py:158: ResourceWarning: unclosed <socket.socket [closed] fd=8, family=AddressFamily.AF_UNIX, type=SocketKind.SOCK_STREAM, proto=0>
  _factory(policy=self.policy)
ResourceWarning: Enable tracemalloc to get the object allocation traceback
/usr/lib/python3.8/email/feedparser.py:158: ResourceWarning: unclosed <socket.socket [closed] fd=9, family=AddressFamily.AF_UNIX, type=SocketKind.SOCK_STREAM, proto=0>
  _factory(policy=self.policy)
ResourceWarning: Enable tracemalloc to get the object allocation traceback
/usr/lib/python3.8/email/feedparser.py:158: ResourceWarning: unclosed <socket.socket [closed] fd=10, family=AddressFamily.AF_UNIX, type=SocketKind.SOCK_STREAM, proto=0>
  _factory(policy=self.policy)
ResourceWarning: Enable tracemalloc to get the object allocation traceback
ok

----------------------------------------------------------------------
Ran 1 test in 4.980s

OK

The root cause appears to relate to a combination of APIClient._read_from_socket(…) and CancellableStream: https://github.com/docker/docker-py/blob/a3652028b1ead708bd9191efb286f909ba6c2a49/docker/api/container.py#L61-L65

The documentation for APIClient._read_from_socket(…) states:

If stream=True, then a generator is returned instead and the caller is responsible for closing the response.

and if stream is not True then the implementation calls response.close(): https://github.com/docker/docker-py/blob/a3652028b1ead708bd9191efb286f909ba6c2a49/docker/api/client.py#L443-L447

however, the current implementation of CancellableStream.close() does not call self._response.close().

Modifying the test case to be:

import contextlib
import unittest

import docker


class TestDocker(unittest.TestCase):

   def test(self):
      for count in range(10):
         with contextlib.closing(docker.from_env()) as client:
            container = client.containers.run('alpine',
               auto_remove=True,
               command=('/bin/sh', '-c', 'echo Hello; sleep 1'),
               detach=True,
               init=True,
               tty=True
            )
            logs = container.attach(
               logs=True,
               stdout=True,
               stream=True
            )
            next(logs, b'').decode()
            logs._response.close()
            container.stop()

(i.e. not bothering to call CancellableStream.close(), but manually calling CancellableStream._response.close() appears to resolve the "ResourceWarning: unclosed <socket.socket …>" warnings; which appears to suggest that CancellableStream.close() should call self._response.close() and perhaps the rest of the current implementation that appears to be seeking a socket in order to close it is unnecessary?

This issue may, or may not, be that same as that reported in #3268.

Reproduce the ResourceWarning with the provided unittest case, then read docker/api/container.py, docker/api/client.py, and docker/types/daemon.py, focusing on _read_from_socket and CancellableStream.close. Verify how the response is owned and closed, and add regression coverage showing that closing the stream leaves no unclosed sockets or warnings.