We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
aiopg instrumentation creating nested traces
**Describe your environment**
Python 3.6.12
Macos 11.1
opentelemetry-python-contrib: Tried released version, as well as `main`.
aiohttp==3.7.3
aiopg==1.1.0

 Describe any aspect of your environment relevant to the problem, including your Python version, [platform](https://docs.python.org/3/library/platform.html), version numbers of installed dependencies, information about your cloud hosting provider, etc. If you're reporting a problem with a specific version of a library in this repo, please check whether the problem has been fixed on main.

**Steps to reproduce**
I am trying the `aiopg` instrumentation. I am using it along aiohttp. This may be an usage issue, but I have created a minimal example in [here](https://github.com/cdvv7788/aiopg-instrumentor-bug) (Just install requirements, run, and visit `localhost:4000`...you will need a postgres database named test and jaeger running somewhere).
The trace will keep growing with every request, nesting new traces every time:
![image](https://user-images.githubusercontent.com/5531776/108002651-4494bd80-6fbe-11eb-8ebf-0067905e6ab6.png)


**What is the expected behavior?**
Every connection has a single trace (no nesting)

**What is the actual behavior?**
Every time a request is made, a new trace is nested and sends the whole stack to jaeger

**Notes**
This may be a bad usage issue. If that is the case, please let me know what may be causing it and how to use this properly.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/aiopg_integration.py`
Content:
```
1 import typing
2 
3 import wrapt
4 from aiopg.utils import _ContextManager, _PoolAcquireContextManager
5 
6 from opentelemetry.instrumentation.dbapi import (
7     CursorTracer,
8     DatabaseApiIntegration,
9 )
10 from opentelemetry.trace import SpanKind
11 
12 
13 # pylint: disable=abstract-method
14 class AsyncProxyObject(wrapt.ObjectProxy):
15     def __aiter__(self):
16         return self.__wrapped__.__aiter__()
17 
18     async def __anext__(self):
19         result = await self.__wrapped__.__anext__()
20         return result
21 
22     async def __aenter__(self):
23         return await self.__wrapped__.__aenter__()
24 
25     async def __aexit__(self, exc_type, exc_val, exc_tb):
26         return await self.__wrapped__.__aexit__(exc_type, exc_val, exc_tb)
27 
28     def __await__(self):
29         return self.__wrapped__.__await__()
30 
31 
32 class AiopgIntegration(DatabaseApiIntegration):
33     async def wrapped_connection(
34         self,
35         connect_method: typing.Callable[..., typing.Any],
36         args: typing.Tuple[typing.Any, typing.Any],
37         kwargs: typing.Dict[typing.Any, typing.Any],
38     ):
39         """Add object proxy to connection object."""
40         connection = await connect_method(*args, **kwargs)
41         # pylint: disable=protected-access
42         self.get_connection_attributes(connection._conn)
43         return get_traced_connection_proxy(connection, self)
44 
45     async def wrapped_pool(self, create_pool_method, args, kwargs):
46         pool = await create_pool_method(*args, **kwargs)
47         async with pool.acquire() as connection:
48             # pylint: disable=protected-access
49             self.get_connection_attributes(connection._conn)
50         return get_traced_pool_proxy(pool, self)
51 
52 
53 def get_traced_connection_proxy(
54     connection, db_api_integration, *args, **kwargs
55 ):
56     # pylint: disable=abstract-method
57     class TracedConnectionProxy(AsyncProxyObject):
58         # pylint: disable=unused-argument
59         def __init__(self, connection, *args, **kwargs):
60             super().__init__(connection)
61 
62         def cursor(self, *args, **kwargs):
63             coro = self._cursor(*args, **kwargs)
64             return _ContextManager(coro)
65 
66         async def _cursor(self, *args, **kwargs):
67             # pylint: disable=protected-access
68             cursor = await self.__wrapped__._cursor(*args, **kwargs)
69             return get_traced_cursor_proxy(cursor, db_api_integration)
70 
71     return TracedConnectionProxy(connection, *args, **kwargs)
72 
73 
74 def get_traced_pool_proxy(pool, db_api_integration, *args, **kwargs):
75     # pylint: disable=abstract-method
76     class TracedPoolProxy(AsyncProxyObject):
77         # pylint: disable=unused-argument
78         def __init__(self, pool, *args, **kwargs):
79             super().__init__(pool)
80 
81         def acquire(self):
82             """Acquire free connection from the pool."""
83             coro = self._acquire()
84             return _PoolAcquireContextManager(coro, self)
85 
86         async def _acquire(self):
87             # pylint: disable=protected-access
88             connection = await self.__wrapped__._acquire()
89             return get_traced_connection_proxy(
90                 connection, db_api_integration, *args, **kwargs
91             )
92 
93     return TracedPoolProxy(pool, *args, **kwargs)
94 
95 
96 class AsyncCursorTracer(CursorTracer):
97     async def traced_execution(
98         self,
99         cursor,
100         query_method: typing.Callable[..., typing.Any],
101         *args: typing.Tuple[typing.Any, typing.Any],
102         **kwargs: typing.Dict[typing.Any, typing.Any]
103     ):
104         name = ""
105         if args:
106             name = self.get_operation_name(cursor, args)
107 
108         if not name:
109             name = (
110                 self._db_api_integration.database
111                 if self._db_api_integration.database
112                 else self._db_api_integration.name
113             )
114 
115         with self._db_api_integration.get_tracer().start_as_current_span(
116             name, kind=SpanKind.CLIENT
117         ) as span:
118             self._populate_span(span, cursor, *args)
119             return await query_method(*args, **kwargs)
120 
121 
122 def get_traced_cursor_proxy(cursor, db_api_integration, *args, **kwargs):
123     _traced_cursor = AsyncCursorTracer(db_api_integration)
124 
125     # pylint: disable=abstract-method
126     class AsyncCursorTracerProxy(AsyncProxyObject):
127 
128         # pylint: disable=unused-argument
129         def __init__(self, cursor, *args, **kwargs):
130             super().__init__(cursor)
131 
132         async def execute(self, *args, **kwargs):
133             result = await _traced_cursor.traced_execution(
134                 self, self.__wrapped__.execute, *args, **kwargs
135             )
136             return result
137 
138         async def executemany(self, *args, **kwargs):
139             result = await _traced_cursor.traced_execution(
140                 self, self.__wrapped__.executemany, *args, **kwargs
141             )
142             return result
143 
144         async def callproc(self, *args, **kwargs):
145             result = await _traced_cursor.traced_execution(
146                 self, self.__wrapped__.callproc, *args, **kwargs
147             )
148             return result
149 
150     return AsyncCursorTracerProxy(cursor, *args, **kwargs)
151
```
--- 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/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/aiopg_integration.py b/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/aiopg_integration.py
--- a/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/aiopg_integration.py
+++ b/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/aiopg_integration.py
@@ -86,9 +86,11 @@
         async def _acquire(self):
             # pylint: disable=protected-access
             connection = await self.__wrapped__._acquire()
-            return get_traced_connection_proxy(
-                connection, db_api_integration, *args, **kwargs
-            )
+            if not isinstance(connection, AsyncProxyObject):
+                connection = get_traced_connection_proxy(
+                    connection, db_api_integration, *args, **kwargs
+                )
+            return connection
 
     return TracedPoolProxy(pool, *args, **kwargs)

{"golden_diff": "diff --git a/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/aiopg_integration.py b/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/aiopg_integration.py\n--- a/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/aiopg_integration.py\n+++ b/instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/aiopg_integration.py\n@@ -86,9 +86,11 @@\n         async def _acquire(self):\n             # pylint: disable=protected-access\n             connection = await self.__wrapped__._acquire()\n-            return get_traced_connection_proxy(\n-                connection, db_api_integration, *args, **kwargs\n-            )\n+            if not isinstance(connection, AsyncProxyObject):\n+                connection = get_traced_connection_proxy(\n+                    connection, db_api_integration, *args, **kwargs\n+                )\n+            return connection\n \n     return TracedPoolProxy(pool, *args, **kwargs)\n", "issue": "aiopg instrumentation creating nested traces\n**Describe your environment**\r\nPython 3.6.12\r\nMacos 11.1\r\nopentelemetry-python-contrib: Tried released version, as well as `main`.\r\naiohttp==3.7.3\r\naiopg==1.1.0\r\n\r\n Describe any aspect of your environment relevant to the problem, including your Python version, [platform](https://docs.python.org/3/library/platform.html), version numbers of installed dependencies, information about your cloud hosting provider, etc. If you're reporting a problem with a specific version of a library in this repo, please check whether the problem has been fixed on main.\r\n\r\n**Steps to reproduce**\r\nI am trying the `aiopg` instrumentation. I am using it along aiohttp. This may be an usage issue, but I have created a minimal example in [here](https://github.com/cdvv7788/aiopg-instrumentor-bug) (Just install requirements, run, and visit `localhost:4000`...you will need a postgres database named test and jaeger running somewhere).\r\nThe trace will keep growing with every request, nesting new traces every time:\r\n![image](https://user-images.githubusercontent.com/5531776/108002651-4494bd80-6fbe-11eb-8ebf-0067905e6ab6.png)\r\n\r\n\r\n**What is the expected behavior?**\r\nEvery connection has a single trace (no nesting)\r\n\r\n**What is the actual behavior?**\r\nEvery time a request is made, a new trace is nested and sends the whole stack to jaeger\r\n\r\n**Notes**\r\nThis may be a bad usage issue. If that is the case, please let me know what may be causing it and how to use this properly.\r\n\n", "before_files": [{"content": "import typing\n\nimport wrapt\nfrom aiopg.utils import _ContextManager, _PoolAcquireContextManager\n\nfrom opentelemetry.instrumentation.dbapi import (\n    CursorTracer,\n    DatabaseApiIntegration,\n)\nfrom opentelemetry.trace import SpanKind\n\n\n# pylint: disable=abstract-method\nclass AsyncProxyObject(wrapt.ObjectProxy):\n    def __aiter__(self):\n        return self.__wrapped__.__aiter__()\n\n    async def __anext__(self):\n        result = await self.__wrapped__.__anext__()\n        return result\n\n    async def __aenter__(self):\n        return await self.__wrapped__.__aenter__()\n\n    async def __aexit__(self, exc_type, exc_val, exc_tb):\n        return await self.__wrapped__.__aexit__(exc_type, exc_val, exc_tb)\n\n    def __await__(self):\n        return self.__wrapped__.__await__()\n\n\nclass AiopgIntegration(DatabaseApiIntegration):\n    async def wrapped_connection(\n        self,\n        connect_method: typing.Callable[..., typing.Any],\n        args: typing.Tuple[typing.Any, typing.Any],\n        kwargs: typing.Dict[typing.Any, typing.Any],\n    ):\n        \"\"\"Add object proxy to connection object.\"\"\"\n        connection = await connect_method(*args, **kwargs)\n        # pylint: disable=protected-access\n        self.get_connection_attributes(connection._conn)\n        return get_traced_connection_proxy(connection, self)\n\n    async def wrapped_pool(self, create_pool_method, args, kwargs):\n        pool = await create_pool_method(*args, **kwargs)\n        async with pool.acquire() as connection:\n            # pylint: disable=protected-access\n            self.get_connection_attributes(connection._conn)\n        return get_traced_pool_proxy(pool, self)\n\n\ndef get_traced_connection_proxy(\n    connection, db_api_integration, *args, **kwargs\n):\n    # pylint: disable=abstract-method\n    class TracedConnectionProxy(AsyncProxyObject):\n        # pylint: disable=unused-argument\n        def __init__(self, connection, *args, **kwargs):\n            super().__init__(connection)\n\n        def cursor(self, *args, **kwargs):\n            coro = self._cursor(*args, **kwargs)\n            return _ContextManager(coro)\n\n        async def _cursor(self, *args, **kwargs):\n            # pylint: disable=protected-access\n            cursor = await self.__wrapped__._cursor(*args, **kwargs)\n            return get_traced_cursor_proxy(cursor, db_api_integration)\n\n    return TracedConnectionProxy(connection, *args, **kwargs)\n\n\ndef get_traced_pool_proxy(pool, db_api_integration, *args, **kwargs):\n    # pylint: disable=abstract-method\n    class TracedPoolProxy(AsyncProxyObject):\n        # pylint: disable=unused-argument\n        def __init__(self, pool, *args, **kwargs):\n            super().__init__(pool)\n\n        def acquire(self):\n            \"\"\"Acquire free connection from the pool.\"\"\"\n            coro = self._acquire()\n            return _PoolAcquireContextManager(coro, self)\n\n        async def _acquire(self):\n            # pylint: disable=protected-access\n            connection = await self.__wrapped__._acquire()\n            return get_traced_connection_proxy(\n                connection, db_api_integration, *args, **kwargs\n            )\n\n    return TracedPoolProxy(pool, *args, **kwargs)\n\n\nclass AsyncCursorTracer(CursorTracer):\n    async def traced_execution(\n        self,\n        cursor,\n        query_method: typing.Callable[..., typing.Any],\n        *args: typing.Tuple[typing.Any, typing.Any],\n        **kwargs: typing.Dict[typing.Any, typing.Any]\n    ):\n        name = \"\"\n        if args:\n            name = self.get_operation_name(cursor, args)\n\n        if not name:\n            name = (\n                self._db_api_integration.database\n                if self._db_api_integration.database\n                else self._db_api_integration.name\n            )\n\n        with self._db_api_integration.get_tracer().start_as_current_span(\n            name, kind=SpanKind.CLIENT\n        ) as span:\n            self._populate_span(span, cursor, *args)\n            return await query_method(*args, **kwargs)\n\n\ndef get_traced_cursor_proxy(cursor, db_api_integration, *args, **kwargs):\n    _traced_cursor = AsyncCursorTracer(db_api_integration)\n\n    # pylint: disable=abstract-method\n    class AsyncCursorTracerProxy(AsyncProxyObject):\n\n        # pylint: disable=unused-argument\n        def __init__(self, cursor, *args, **kwargs):\n            super().__init__(cursor)\n\n        async def execute(self, *args, **kwargs):\n            result = await _traced_cursor.traced_execution(\n                self, self.__wrapped__.execute, *args, **kwargs\n            )\n            return result\n\n        async def executemany(self, *args, **kwargs):\n            result = await _traced_cursor.traced_execution(\n                self, self.__wrapped__.executemany, *args, **kwargs\n            )\n            return result\n\n        async def callproc(self, *args, **kwargs):\n            result = await _traced_cursor.traced_execution(\n                self, self.__wrapped__.callproc, *args, **kwargs\n            )\n            return result\n\n    return AsyncCursorTracerProxy(cursor, *args, **kwargs)\n", "path": "instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/aiopg_integration.py"}], "after_files": [{"content": "import typing\n\nimport wrapt\nfrom aiopg.utils import _ContextManager, _PoolAcquireContextManager\n\nfrom opentelemetry.instrumentation.dbapi import (\n    CursorTracer,\n    DatabaseApiIntegration,\n)\nfrom opentelemetry.trace import SpanKind\n\n\n# pylint: disable=abstract-method\nclass AsyncProxyObject(wrapt.ObjectProxy):\n    def __aiter__(self):\n        return self.__wrapped__.__aiter__()\n\n    async def __anext__(self):\n        result = await self.__wrapped__.__anext__()\n        return result\n\n    async def __aenter__(self):\n        return await self.__wrapped__.__aenter__()\n\n    async def __aexit__(self, exc_type, exc_val, exc_tb):\n        return await self.__wrapped__.__aexit__(exc_type, exc_val, exc_tb)\n\n    def __await__(self):\n        return self.__wrapped__.__await__()\n\n\nclass AiopgIntegration(DatabaseApiIntegration):\n    async def wrapped_connection(\n        self,\n        connect_method: typing.Callable[..., typing.Any],\n        args: typing.Tuple[typing.Any, typing.Any],\n        kwargs: typing.Dict[typing.Any, typing.Any],\n    ):\n        \"\"\"Add object proxy to connection object.\"\"\"\n        connection = await connect_method(*args, **kwargs)\n        # pylint: disable=protected-access\n        self.get_connection_attributes(connection._conn)\n        return get_traced_connection_proxy(connection, self)\n\n    async def wrapped_pool(self, create_pool_method, args, kwargs):\n        pool = await create_pool_method(*args, **kwargs)\n        async with pool.acquire() as connection:\n            # pylint: disable=protected-access\n            self.get_connection_attributes(connection._conn)\n        return get_traced_pool_proxy(pool, self)\n\n\ndef get_traced_connection_proxy(\n    connection, db_api_integration, *args, **kwargs\n):\n    # pylint: disable=abstract-method\n    class TracedConnectionProxy(AsyncProxyObject):\n        # pylint: disable=unused-argument\n        def __init__(self, connection, *args, **kwargs):\n            super().__init__(connection)\n\n        def cursor(self, *args, **kwargs):\n            coro = self._cursor(*args, **kwargs)\n            return _ContextManager(coro)\n\n        async def _cursor(self, *args, **kwargs):\n            # pylint: disable=protected-access\n            cursor = await self.__wrapped__._cursor(*args, **kwargs)\n            return get_traced_cursor_proxy(cursor, db_api_integration)\n\n    return TracedConnectionProxy(connection, *args, **kwargs)\n\n\ndef get_traced_pool_proxy(pool, db_api_integration, *args, **kwargs):\n    # pylint: disable=abstract-method\n    class TracedPoolProxy(AsyncProxyObject):\n        # pylint: disable=unused-argument\n        def __init__(self, pool, *args, **kwargs):\n            super().__init__(pool)\n\n        def acquire(self):\n            \"\"\"Acquire free connection from the pool.\"\"\"\n            coro = self._acquire()\n            return _PoolAcquireContextManager(coro, self)\n\n        async def _acquire(self):\n            # pylint: disable=protected-access\n            connection = await self.__wrapped__._acquire()\n            if not isinstance(connection, AsyncProxyObject):\n                connection = get_traced_connection_proxy(\n                    connection, db_api_integration, *args, **kwargs\n                )\n            return connection\n\n    return TracedPoolProxy(pool, *args, **kwargs)\n\n\nclass AsyncCursorTracer(CursorTracer):\n    async def traced_execution(\n        self,\n        cursor,\n        query_method: typing.Callable[..., typing.Any],\n        *args: typing.Tuple[typing.Any, typing.Any],\n        **kwargs: typing.Dict[typing.Any, typing.Any]\n    ):\n        name = \"\"\n        if args:\n            name = self.get_operation_name(cursor, args)\n\n        if not name:\n            name = (\n                self._db_api_integration.database\n                if self._db_api_integration.database\n                else self._db_api_integration.name\n            )\n\n        with self._db_api_integration.get_tracer().start_as_current_span(\n            name, kind=SpanKind.CLIENT\n        ) as span:\n            self._populate_span(span, cursor, *args)\n            return await query_method(*args, **kwargs)\n\n\ndef get_traced_cursor_proxy(cursor, db_api_integration, *args, **kwargs):\n    _traced_cursor = AsyncCursorTracer(db_api_integration)\n\n    # pylint: disable=abstract-method\n    class AsyncCursorTracerProxy(AsyncProxyObject):\n\n        # pylint: disable=unused-argument\n        def __init__(self, cursor, *args, **kwargs):\n            super().__init__(cursor)\n\n        async def execute(self, *args, **kwargs):\n            result = await _traced_cursor.traced_execution(\n                self, self.__wrapped__.execute, *args, **kwargs\n            )\n            return result\n\n        async def executemany(self, *args, **kwargs):\n            result = await _traced_cursor.traced_execution(\n                self, self.__wrapped__.executemany, *args, **kwargs\n            )\n            return result\n\n        async def callproc(self, *args, **kwargs):\n            result = await _traced_cursor.traced_execution(\n                self, self.__wrapped__.callproc, *args, **kwargs\n            )\n            return result\n\n    return AsyncCursorTracerProxy(cursor, *args, **kwargs)\n", "path": "instrumentation/opentelemetry-instrumentation-aiopg/src/opentelemetry/instrumentation/aiopg/aiopg_integration.py"}]}