exec: fix file handle leak with container.exec_* APIs (#2320)

Requests with stream=True MUST be closed or else the connection will
never be returned to the connection pool. Both ContainerApiMixin.attach
and ExecApiMixin.exec_start were leaking in the stream=False case.
exec_start was modified to follow attach for the stream=True case as
that allows the caller to close the stream when done (untested).

Tested with:

    # Test exec_run (stream=False) - observe one less leak
    make integration-test-py3 file=models_containers_test.py' -k test_exec_run_success -vs -W error::ResourceWarning'
    # Test exec_start (stream=True, fully reads from CancellableStream)
    make integration-test-py3 file=api_exec_test.py' -k test_execute_command -vs -W error::ResourceWarning'

After this change, one resource leak is removed, the remaining resource
leaks occur because none of the tests call client.close().

Fixes #1293
(Regression from #1130)

Signed-off-by: Peter Wu <[email protected]>
Co-authored-by: Milas Bowman <[email protected]>

docker/api/client.py

@@ -406,6 +406,10 @@ def _stream_raw_result(self, response, chunk_size=1, decode=True):
         yield from response.iter_content(chunk_size, decode)
 
     def _read_from_socket(self, response, stream, tty=True, demux=False):
+        """Consume all data from the socket, close the response and return the
+        data. If stream=True, then a generator is returned instead and the
+        caller is responsible for closing the response.
+        """
         socket = self._get_raw_response_socket(response)
 
         gen = frames_iter(socket, tty)
@@ -420,8 +424,11 @@ def _read_from_socket(self, response, stream, tty=True, demux=False):
         if stream:
             return gen
         else:
-            # Wait for all the frames, concatenate them, and return the result
-            return consume_socket_output(gen, demux=demux)
+            try:
+                # Wait for all frames, concatenate them, and return the result
+                return consume_socket_output(gen, demux=demux)
+            finally:
+                response.close()
 
     def _disable_socket_timeout(self, socket):
         """ Depending on the combination of python version and whether we're

docker/api/exec_api.py

@@ -1,5 +1,6 @@
 from .. import errors
 from .. import utils
+from ..types import CancellableStream
 
 
 class ExecApiMixin:
@@ -125,9 +126,10 @@ def exec_start(self, exec_id, detach=False, tty=False, stream=False,
             detach (bool): If true, detach from the exec command.
                 Default: False
             tty (bool): Allocate a pseudo-TTY. Default: False
-            stream (bool): Stream response data. Default: False
+            stream (bool): Return response data progressively as an iterator
+                of strings, rather than a single string.
             socket (bool): Return the connection socket to allow custom
-                read/write operations.
+                read/write operations. Must be closed by the caller when done.
             demux (bool): Return stdout and stderr separately
 
         Returns:
@@ -161,7 +163,15 @@ def exec_start(self, exec_id, detach=False, tty=False, stream=False,
             stream=True
         )
         if detach:
-            return self._result(res)
+            try:
+                return self._result(res)
+            finally:
+                res.close()
         if socket:
             return self._get_raw_response_socket(res)
-        return self._read_from_socket(res, stream, tty=tty, demux=demux)
+
+        output = self._read_from_socket(res, stream, tty=tty, demux=demux)
+        if stream:
+            return CancellableStream(output, res)
+        else:
+            return output