We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Leaking sockets and connections
It looks like ddtrace writer is not closing the http connections it opens. 

```
  File "../lib/python3.7/threading.py", line 885, in _bootstrap
    self._bootstrap_inner()
  File "../lib/python3.7/threading.py", line 917, in _bootstrap_inner
    self.run()
  File "../lib/python3.7/threading.py", line 865, in run
    self._target(*self._args, **self._kwargs)
  File "../lib/python3.7/site-packages/ddtrace/writer.py", line 168, in _target
    result_services = None
  File "../lib/python3.7/http/client.py", line 408, in close
    self._close_conn()
  File "../lib/python3.7/http/client.py", line 401, in _close_conn
    fp.close()
  File "../lib/python3.7/socket.py", line 660, in close
    self._sock = None
  File "../lib/python3.7/warnings.py", line 99, in _showwarnmsg
    msg.file, msg.line)
  File "/app/core.py", line 30, in warn_with_traceback
    traceback.print_stack(file=log)
../lib/python3.7/socket.py:660: ResourceWarning: unclosed <socket.socket fd=14, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('0.0.0.0', 54954), raddr=('0.0.0.0', 8126)>
  self._sock = None
```


Looking at the code, the issue is in the `_put` method of the `API` object. It creates an `HTTPConnection` but doesn't close it. 

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `ddtrace/api.py`
Content:
```
1 # stdlib
2 import logging
3 import time
4 import ddtrace
5 from json import loads
6 
7 # project
8 from .encoding import get_encoder, JSONEncoder
9 from .compat import httplib, PYTHON_VERSION, PYTHON_INTERPRETER, get_connection_response
10 
11 
12 log = logging.getLogger(__name__)
13 
14 TRACE_COUNT_HEADER = 'X-Datadog-Trace-Count'
15 
16 _VERSIONS = {'v0.4': {'traces': '/v0.4/traces',
17                       'services': '/v0.4/services',
18                       'compatibility_mode': False,
19                       'fallback': 'v0.3'},
20              'v0.3': {'traces': '/v0.3/traces',
21                      'services': '/v0.3/services',
22                       'compatibility_mode': False,
23                       'fallback': 'v0.2'},
24              'v0.2': {'traces': '/v0.2/traces',
25                      'services': '/v0.2/services',
26                       'compatibility_mode': True,
27                       'fallback': None}}
28 
29 def _parse_response_json(response):
30     """
31     Parse the content of a response object, and return the right type,
32     can be a string if the output was plain text, or a dictionnary if
33     the output was a JSON.
34     """
35     if hasattr(response, 'read'):
36         body = response.read()
37         try:
38             if not isinstance(body, str) and hasattr(body, 'decode'):
39                 body = body.decode('utf-8')
40             if hasattr(body, 'startswith') and body.startswith('OK'):
41                 # This typically happens when using a priority-sampling enabled
42                 # library with an outdated agent. It still works, but priority sampling
43                 # will probably send too many traces, so the next step is to upgrade agent.
44                 log.debug("'OK' is not a valid JSON, please make sure trace-agent is up to date")
45                 return
46             content = loads(body)
47             return content
48         except (ValueError, TypeError) as err:
49             log.debug("unable to load JSON '%s': %s" % (body, err))
50 
51 class API(object):
52     """
53     Send data to the trace agent using the HTTP protocol and JSON format
54     """
55     def __init__(self, hostname, port, headers=None, encoder=None, priority_sampling=False):
56         self.hostname = hostname
57         self.port = port
58 
59         self._headers = headers or {}
60         self._version = None
61 
62         if priority_sampling:
63             self._set_version('v0.4', encoder=encoder)
64         else:
65             self._set_version('v0.3', encoder=encoder)
66 
67         self._headers.update({
68             'Datadog-Meta-Lang': 'python',
69             'Datadog-Meta-Lang-Version': PYTHON_VERSION,
70             'Datadog-Meta-Lang-Interpreter': PYTHON_INTERPRETER,
71             'Datadog-Meta-Tracer-Version': ddtrace.__version__,
72         })
73 
74     def _set_version(self, version, encoder=None):
75         if version not in _VERSIONS:
76             version = 'v0.2'
77         if version == self._version:
78             return
79         self._version = version
80         self._traces = _VERSIONS[version]['traces']
81         self._services = _VERSIONS[version]['services']
82         self._fallback = _VERSIONS[version]['fallback']
83         self._compatibility_mode = _VERSIONS[version]['compatibility_mode']
84         if self._compatibility_mode:
85             self._encoder = JSONEncoder()
86         else:
87             self._encoder = encoder or get_encoder()
88         # overwrite the Content-type with the one chosen in the Encoder
89         self._headers.update({'Content-Type': self._encoder.content_type})
90 
91     def _downgrade(self):
92         """
93         Downgrades the used encoder and API level. This method must fallback to a safe
94         encoder and API, so that it will success despite users' configurations. This action
95         ensures that the compatibility mode is activated so that the downgrade will be
96         executed only once.
97         """
98         self._set_version(self._fallback)
99 
100     def send_traces(self, traces):
101         if not traces:
102             return
103         start = time.time()
104         data = self._encoder.encode_traces(traces)
105         response = self._put(self._traces, data, len(traces))
106 
107         # the API endpoint is not available so we should downgrade the connection and re-try the call
108         if response.status in [404, 415] and self._fallback:
109             log.debug('calling endpoint "%s" but received %s; downgrading API', self._traces, response.status)
110             self._downgrade()
111             return self.send_traces(traces)
112 
113         log.debug("reported %d traces in %.5fs", len(traces), time.time() - start)
114         return response
115 
116     def send_services(self, services):
117         if not services:
118             return
119         s = {}
120         for service in services:
121             s.update(service)
122         data = self._encoder.encode_services(s)
123         response = self._put(self._services, data)
124 
125         # the API endpoint is not available so we should downgrade the connection and re-try the call
126         if response.status in [404, 415] and self._fallback:
127             log.debug('calling endpoint "%s" but received %s; downgrading API', self._services, response.status)
128             self._downgrade()
129             return self.send_services(services)
130 
131         log.debug("reported %d services", len(services))
132         return response
133 
134     def _put(self, endpoint, data, count=0):
135         conn = httplib.HTTPConnection(self.hostname, self.port)
136 
137         headers = self._headers
138         if count:
139             headers = dict(self._headers)
140             headers[TRACE_COUNT_HEADER] = str(count)
141 
142         conn.request("PUT", endpoint, data, headers)
143         return get_connection_response(conn)
144
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/ddtrace/api.py b/ddtrace/api.py
--- a/ddtrace/api.py
+++ b/ddtrace/api.py
@@ -133,11 +133,13 @@
 
     def _put(self, endpoint, data, count=0):
         conn = httplib.HTTPConnection(self.hostname, self.port)
-
-        headers = self._headers
-        if count:
-            headers = dict(self._headers)
-            headers[TRACE_COUNT_HEADER] = str(count)
-
-        conn.request("PUT", endpoint, data, headers)
-        return get_connection_response(conn)
+        try:
+            headers = self._headers
+            if count:
+                headers = dict(self._headers)
+                headers[TRACE_COUNT_HEADER] = str(count)
+
+            conn.request("PUT", endpoint, data, headers)
+            return get_connection_response(conn)
+        finally:
+            conn.close()

{"golden_diff": "diff --git a/ddtrace/api.py b/ddtrace/api.py\n--- a/ddtrace/api.py\n+++ b/ddtrace/api.py\n@@ -133,11 +133,13 @@\n \n     def _put(self, endpoint, data, count=0):\n         conn = httplib.HTTPConnection(self.hostname, self.port)\n-\n-        headers = self._headers\n-        if count:\n-            headers = dict(self._headers)\n-            headers[TRACE_COUNT_HEADER] = str(count)\n-\n-        conn.request(\"PUT\", endpoint, data, headers)\n-        return get_connection_response(conn)\n+        try:\n+            headers = self._headers\n+            if count:\n+                headers = dict(self._headers)\n+                headers[TRACE_COUNT_HEADER] = str(count)\n+\n+            conn.request(\"PUT\", endpoint, data, headers)\n+            return get_connection_response(conn)\n+        finally:\n+            conn.close()\n", "issue": "Leaking sockets and connections\nIt looks like ddtrace writer is not closing the http connections it opens. \r\n\r\n```\r\n  File \"../lib/python3.7/threading.py\", line 885, in _bootstrap\r\n    self._bootstrap_inner()\r\n  File \"../lib/python3.7/threading.py\", line 917, in _bootstrap_inner\r\n    self.run()\r\n  File \"../lib/python3.7/threading.py\", line 865, in run\r\n    self._target(*self._args, **self._kwargs)\r\n  File \"../lib/python3.7/site-packages/ddtrace/writer.py\", line 168, in _target\r\n    result_services = None\r\n  File \"../lib/python3.7/http/client.py\", line 408, in close\r\n    self._close_conn()\r\n  File \"../lib/python3.7/http/client.py\", line 401, in _close_conn\r\n    fp.close()\r\n  File \"../lib/python3.7/socket.py\", line 660, in close\r\n    self._sock = None\r\n  File \"../lib/python3.7/warnings.py\", line 99, in _showwarnmsg\r\n    msg.file, msg.line)\r\n  File \"/app/core.py\", line 30, in warn_with_traceback\r\n    traceback.print_stack(file=log)\r\n../lib/python3.7/socket.py:660: ResourceWarning: unclosed <socket.socket fd=14, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('0.0.0.0', 54954), raddr=('0.0.0.0', 8126)>\r\n  self._sock = None\r\n```\r\n\r\n\r\nLooking at the code, the issue is in the `_put` method of the `API` object. It creates an `HTTPConnection` but doesn't close it. \n", "before_files": [{"content": "# stdlib\nimport logging\nimport time\nimport ddtrace\nfrom json import loads\n\n# project\nfrom .encoding import get_encoder, JSONEncoder\nfrom .compat import httplib, PYTHON_VERSION, PYTHON_INTERPRETER, get_connection_response\n\n\nlog = logging.getLogger(__name__)\n\nTRACE_COUNT_HEADER = 'X-Datadog-Trace-Count'\n\n_VERSIONS = {'v0.4': {'traces': '/v0.4/traces',\n                      'services': '/v0.4/services',\n                      'compatibility_mode': False,\n                      'fallback': 'v0.3'},\n             'v0.3': {'traces': '/v0.3/traces',\n                     'services': '/v0.3/services',\n                      'compatibility_mode': False,\n                      'fallback': 'v0.2'},\n             'v0.2': {'traces': '/v0.2/traces',\n                     'services': '/v0.2/services',\n                      'compatibility_mode': True,\n                      'fallback': None}}\n\ndef _parse_response_json(response):\n    \"\"\"\n    Parse the content of a response object, and return the right type,\n    can be a string if the output was plain text, or a dictionnary if\n    the output was a JSON.\n    \"\"\"\n    if hasattr(response, 'read'):\n        body = response.read()\n        try:\n            if not isinstance(body, str) and hasattr(body, 'decode'):\n                body = body.decode('utf-8')\n            if hasattr(body, 'startswith') and body.startswith('OK'):\n                # This typically happens when using a priority-sampling enabled\n                # library with an outdated agent. It still works, but priority sampling\n                # will probably send too many traces, so the next step is to upgrade agent.\n                log.debug(\"'OK' is not a valid JSON, please make sure trace-agent is up to date\")\n                return\n            content = loads(body)\n            return content\n        except (ValueError, TypeError) as err:\n            log.debug(\"unable to load JSON '%s': %s\" % (body, err))\n\nclass API(object):\n    \"\"\"\n    Send data to the trace agent using the HTTP protocol and JSON format\n    \"\"\"\n    def __init__(self, hostname, port, headers=None, encoder=None, priority_sampling=False):\n        self.hostname = hostname\n        self.port = port\n\n        self._headers = headers or {}\n        self._version = None\n\n        if priority_sampling:\n            self._set_version('v0.4', encoder=encoder)\n        else:\n            self._set_version('v0.3', encoder=encoder)\n\n        self._headers.update({\n            'Datadog-Meta-Lang': 'python',\n            'Datadog-Meta-Lang-Version': PYTHON_VERSION,\n            'Datadog-Meta-Lang-Interpreter': PYTHON_INTERPRETER,\n            'Datadog-Meta-Tracer-Version': ddtrace.__version__,\n        })\n\n    def _set_version(self, version, encoder=None):\n        if version not in _VERSIONS:\n            version = 'v0.2'\n        if version == self._version:\n            return\n        self._version = version\n        self._traces = _VERSIONS[version]['traces']\n        self._services = _VERSIONS[version]['services']\n        self._fallback = _VERSIONS[version]['fallback']\n        self._compatibility_mode = _VERSIONS[version]['compatibility_mode']\n        if self._compatibility_mode:\n            self._encoder = JSONEncoder()\n        else:\n            self._encoder = encoder or get_encoder()\n        # overwrite the Content-type with the one chosen in the Encoder\n        self._headers.update({'Content-Type': self._encoder.content_type})\n\n    def _downgrade(self):\n        \"\"\"\n        Downgrades the used encoder and API level. This method must fallback to a safe\n        encoder and API, so that it will success despite users' configurations. This action\n        ensures that the compatibility mode is activated so that the downgrade will be\n        executed only once.\n        \"\"\"\n        self._set_version(self._fallback)\n\n    def send_traces(self, traces):\n        if not traces:\n            return\n        start = time.time()\n        data = self._encoder.encode_traces(traces)\n        response = self._put(self._traces, data, len(traces))\n\n        # the API endpoint is not available so we should downgrade the connection and re-try the call\n        if response.status in [404, 415] and self._fallback:\n            log.debug('calling endpoint \"%s\" but received %s; downgrading API', self._traces, response.status)\n            self._downgrade()\n            return self.send_traces(traces)\n\n        log.debug(\"reported %d traces in %.5fs\", len(traces), time.time() - start)\n        return response\n\n    def send_services(self, services):\n        if not services:\n            return\n        s = {}\n        for service in services:\n            s.update(service)\n        data = self._encoder.encode_services(s)\n        response = self._put(self._services, data)\n\n        # the API endpoint is not available so we should downgrade the connection and re-try the call\n        if response.status in [404, 415] and self._fallback:\n            log.debug('calling endpoint \"%s\" but received %s; downgrading API', self._services, response.status)\n            self._downgrade()\n            return self.send_services(services)\n\n        log.debug(\"reported %d services\", len(services))\n        return response\n\n    def _put(self, endpoint, data, count=0):\n        conn = httplib.HTTPConnection(self.hostname, self.port)\n\n        headers = self._headers\n        if count:\n            headers = dict(self._headers)\n            headers[TRACE_COUNT_HEADER] = str(count)\n\n        conn.request(\"PUT\", endpoint, data, headers)\n        return get_connection_response(conn)\n", "path": "ddtrace/api.py"}], "after_files": [{"content": "# stdlib\nimport logging\nimport time\nimport ddtrace\nfrom json import loads\n\n# project\nfrom .encoding import get_encoder, JSONEncoder\nfrom .compat import httplib, PYTHON_VERSION, PYTHON_INTERPRETER, get_connection_response\n\n\nlog = logging.getLogger(__name__)\n\nTRACE_COUNT_HEADER = 'X-Datadog-Trace-Count'\n\n_VERSIONS = {'v0.4': {'traces': '/v0.4/traces',\n                      'services': '/v0.4/services',\n                      'compatibility_mode': False,\n                      'fallback': 'v0.3'},\n             'v0.3': {'traces': '/v0.3/traces',\n                     'services': '/v0.3/services',\n                      'compatibility_mode': False,\n                      'fallback': 'v0.2'},\n             'v0.2': {'traces': '/v0.2/traces',\n                     'services': '/v0.2/services',\n                      'compatibility_mode': True,\n                      'fallback': None}}\n\ndef _parse_response_json(response):\n    \"\"\"\n    Parse the content of a response object, and return the right type,\n    can be a string if the output was plain text, or a dictionnary if\n    the output was a JSON.\n    \"\"\"\n    if hasattr(response, 'read'):\n        body = response.read()\n        try:\n            if not isinstance(body, str) and hasattr(body, 'decode'):\n                body = body.decode('utf-8')\n            if hasattr(body, 'startswith') and body.startswith('OK'):\n                # This typically happens when using a priority-sampling enabled\n                # library with an outdated agent. It still works, but priority sampling\n                # will probably send too many traces, so the next step is to upgrade agent.\n                log.debug(\"'OK' is not a valid JSON, please make sure trace-agent is up to date\")\n                return\n            content = loads(body)\n            return content\n        except (ValueError, TypeError) as err:\n            log.debug(\"unable to load JSON '%s': %s\" % (body, err))\n\nclass API(object):\n    \"\"\"\n    Send data to the trace agent using the HTTP protocol and JSON format\n    \"\"\"\n    def __init__(self, hostname, port, headers=None, encoder=None, priority_sampling=False):\n        self.hostname = hostname\n        self.port = port\n\n        self._headers = headers or {}\n        self._version = None\n\n        if priority_sampling:\n            self._set_version('v0.4', encoder=encoder)\n        else:\n            self._set_version('v0.3', encoder=encoder)\n\n        self._headers.update({\n            'Datadog-Meta-Lang': 'python',\n            'Datadog-Meta-Lang-Version': PYTHON_VERSION,\n            'Datadog-Meta-Lang-Interpreter': PYTHON_INTERPRETER,\n            'Datadog-Meta-Tracer-Version': ddtrace.__version__,\n        })\n\n    def _set_version(self, version, encoder=None):\n        if version not in _VERSIONS:\n            version = 'v0.2'\n        if version == self._version:\n            return\n        self._version = version\n        self._traces = _VERSIONS[version]['traces']\n        self._services = _VERSIONS[version]['services']\n        self._fallback = _VERSIONS[version]['fallback']\n        self._compatibility_mode = _VERSIONS[version]['compatibility_mode']\n        if self._compatibility_mode:\n            self._encoder = JSONEncoder()\n        else:\n            self._encoder = encoder or get_encoder()\n        # overwrite the Content-type with the one chosen in the Encoder\n        self._headers.update({'Content-Type': self._encoder.content_type})\n\n    def _downgrade(self):\n        \"\"\"\n        Downgrades the used encoder and API level. This method must fallback to a safe\n        encoder and API, so that it will success despite users' configurations. This action\n        ensures that the compatibility mode is activated so that the downgrade will be\n        executed only once.\n        \"\"\"\n        self._set_version(self._fallback)\n\n    def send_traces(self, traces):\n        if not traces:\n            return\n        start = time.time()\n        data = self._encoder.encode_traces(traces)\n        response = self._put(self._traces, data, len(traces))\n\n        # the API endpoint is not available so we should downgrade the connection and re-try the call\n        if response.status in [404, 415] and self._fallback:\n            log.debug('calling endpoint \"%s\" but received %s; downgrading API', self._traces, response.status)\n            self._downgrade()\n            return self.send_traces(traces)\n\n        log.debug(\"reported %d traces in %.5fs\", len(traces), time.time() - start)\n        return response\n\n    def send_services(self, services):\n        if not services:\n            return\n        s = {}\n        for service in services:\n            s.update(service)\n        data = self._encoder.encode_services(s)\n        response = self._put(self._services, data)\n\n        # the API endpoint is not available so we should downgrade the connection and re-try the call\n        if response.status in [404, 415] and self._fallback:\n            log.debug('calling endpoint \"%s\" but received %s; downgrading API', self._services, response.status)\n            self._downgrade()\n            return self.send_services(services)\n\n        log.debug(\"reported %d services\", len(services))\n        return response\n\n    def _put(self, endpoint, data, count=0):\n        conn = httplib.HTTPConnection(self.hostname, self.port)\n        try:\n            headers = self._headers\n            if count:\n                headers = dict(self._headers)\n                headers[TRACE_COUNT_HEADER] = str(count)\n\n            conn.request(\"PUT\", endpoint, data, headers)\n            return get_connection_response(conn)\n        finally:\n            conn.close()\n", "path": "ddtrace/api.py"}]}