We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
celery 4.1 worker tasks aren't being traced with ddtrace-py 0.11.1
```
ddtrace==0.11.1
celery==4.1.0
```

I'm manually patching celery due to #423 

I'm getting traces for 'defaultdb' (sqlite), 'redis', and 'postgres' when I run the celery worker

However, I'm not receiving any traces whatsoever for the task runs, and no 'celery' service appears in datadog.

Here's a gist with my worker startup in debug logging mode:
https://gist.github.com/eedwards-sk/f924548c043859901db6918aec95dada

I'm sending traces to a local running agent and I know that's working because I'm seeing the flush records in the logs, and I get traces for the above mentioned services when I'm running it.

(We have another application that is using celery 3.x and their traces show up for tasks, which we instrumented in more or less the same way)

Here's a gist showing how we're instantiating celery and the tracer:
https://gist.github.com/eedwards-sk/fe81f62406e7a245b7c26a9ada19c658

--- 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/contrib/celery/task.py`
Content:
```
1 # Third party
2 import wrapt
3 import inspect
4 import celery
5 
6 # Project
7 from ddtrace import Pin
8 from ddtrace.ext import AppTypes
9 from ...ext import errors
10 from .util import APP, PRODUCER_SERVICE, WORKER_SERVICE, meta_from_context, require_pin
11 
12 PRODUCER_ROOT_SPAN = 'celery.apply'
13 WORKER_ROOT_SPAN = 'celery.run'
14 # Task operations
15 TASK_TAG_KEY = 'celery.action'
16 TASK_APPLY = 'apply'
17 TASK_APPLY_ASYNC = 'apply_async'
18 TASK_RUN = 'run'
19 
20 
21 def patch_task(task, pin=None):
22     """ patch_task will add tracing to a celery task """
23     # The service set here is actually ignored, because it's not possible to
24     # be certain whether this process is being used as a worker, a producer,
25     # or both. So the service as recorded in traces is set based on the actual
26     # work being done (ie. apply/apply_async vs run).
27     pin = pin or Pin(service=WORKER_SERVICE, app=APP, app_type=AppTypes.worker)
28 
29     patch_methods = [
30         ('__init__', _task_init),
31         ('run', _task_run),
32         ('apply', _task_apply),
33         ('apply_async', _task_apply_async),
34     ]
35     for method_name, wrapper in patch_methods:
36         # Get original method
37         method = getattr(task, method_name, None)
38         if method is None:
39             continue
40 
41         # Do not patch if method is already patched
42         if isinstance(method, wrapt.ObjectProxy):
43             continue
44 
45         # If the function as been applied as a decorator for v1 Celery tasks, then a different patching is needed
46         if inspect.isclass(task) and issubclass(task, celery.task.Task):
47             wrapped = wrapt.FunctionWrapper(method, wrapper)
48             setattr(task, method_name, wrapped)
49             continue
50         # Patch method
51         # DEV: Using `BoundFunctionWrapper` ensures our `task` wrapper parameter is properly set
52         setattr(task, method_name, wrapt.BoundFunctionWrapper(method, task, wrapper))
53 
54     # Attach our pin to the app
55     pin.onto(task)
56     return task
57 
58 def unpatch_task(task):
59     """ unpatch_task will remove tracing from a celery task """
60     patched_methods = [
61         '__init__',
62         'run',
63         'apply',
64         'apply_async',
65     ]
66     for method_name in patched_methods:
67         # Get wrapped method
68         wrapper = getattr(task, method_name, None)
69         if wrapper is None:
70             continue
71 
72         # Only unpatch if wrapper is an `ObjectProxy`
73         if not isinstance(wrapper, wrapt.ObjectProxy):
74             continue
75 
76         # Restore original method
77         setattr(task, method_name, wrapper.__wrapped__)
78 
79     return task
80 
81 
82 def _task_init(func, task, args, kwargs):
83     func(*args, **kwargs)
84 
85     # Patch this task if our pin is enabled
86     pin = Pin.get_from(task)
87     if pin and pin.enabled():
88         patch_task(task, pin=pin)
89 
90 
91 @require_pin
92 def _task_run(pin, func, task, args, kwargs):
93     with pin.tracer.trace(WORKER_ROOT_SPAN, service=WORKER_SERVICE, resource=task.name) as span:
94         # Set meta data from task request
95         span.set_metas(meta_from_context(task.request))
96         span.set_meta(TASK_TAG_KEY, TASK_RUN)
97 
98         # Call original `run` function
99         return func(*args, **kwargs)
100 
101 
102 @require_pin
103 def _task_apply(pin, func, task, args, kwargs):
104     with pin.tracer.trace(PRODUCER_ROOT_SPAN, service=PRODUCER_SERVICE, resource=task.name) as span:
105         # Call the original `apply` function
106         res = func(*args, **kwargs)
107 
108         # Set meta data from response
109         span.set_meta('id', res.id)
110         span.set_meta('state', res.state)
111         span.set_meta(TASK_TAG_KEY, TASK_APPLY)
112         if res.traceback:
113             span.error = 1
114             span.set_meta(errors.STACK, res.traceback)
115         return res
116 
117 
118 @require_pin
119 def _task_apply_async(pin, func, task, args, kwargs):
120     with pin.tracer.trace(PRODUCER_ROOT_SPAN, service=PRODUCER_SERVICE, resource=task.name) as span:
121         # Extract meta data from `kwargs`
122         meta_keys = (
123             'compression', 'countdown', 'eta', 'exchange', 'expires',
124             'priority', 'routing_key', 'serializer', 'queue',
125         )
126         for name in meta_keys:
127             if name in kwargs:
128                 span.set_meta(name, kwargs[name])
129         span.set_meta(TASK_TAG_KEY, TASK_APPLY_ASYNC)
130 
131         # Call the original `apply_async` function
132         res = func(*args, **kwargs)
133 
134         # Set meta data from response
135         # DEV: Calling `res.traceback` or `res.state` will make an
136         #   API call to the backend for the properties
137         span.set_meta('id', res.id)
138         return res
139
```

Path: `ddtrace/contrib/celery/patch.py`
Content:
```
1 import celery
2 
3 from wrapt import wrap_function_wrapper as _w
4 
5 from .app import patch_app, unpatch_app
6 from .registry import _wrap_register
7 from ...utils.wrappers import unwrap as _u
8 
9 
10 def patch():
11     """Instrument Celery base application and the `TaskRegistry` so
12     that any new registered task is automatically instrumented
13     """
14     setattr(celery, 'Celery', patch_app(celery.Celery))
15     _w('celery.app.registry', 'TaskRegistry.register', _wrap_register)
16 
17 
18 def unpatch():
19     """Removes instrumentation from Celery"""
20     setattr(celery, 'Celery', unpatch_app(celery.Celery))
21     _u(celery.app.registry.TaskRegistry, 'register')
22
```
--- 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/contrib/celery/patch.py b/ddtrace/contrib/celery/patch.py
--- a/ddtrace/contrib/celery/patch.py
+++ b/ddtrace/contrib/celery/patch.py
@@ -3,19 +3,24 @@
 from wrapt import wrap_function_wrapper as _w
 
 from .app import patch_app, unpatch_app
+from .task import _wrap_shared_task
 from .registry import _wrap_register
 from ...utils.wrappers import unwrap as _u
 
 
 def patch():
     """Instrument Celery base application and the `TaskRegistry` so
-    that any new registered task is automatically instrumented
+    that any new registered task is automatically instrumented. In the
+    case of Django-Celery integration, also the `@shared_task` decorator
+    must be instrumented because Django doesn't use the Celery registry.
     """
     setattr(celery, 'Celery', patch_app(celery.Celery))
     _w('celery.app.registry', 'TaskRegistry.register', _wrap_register)
+    _w('celery', 'shared_task', _wrap_shared_task)
 
 
 def unpatch():
     """Removes instrumentation from Celery"""
     setattr(celery, 'Celery', unpatch_app(celery.Celery))
     _u(celery.app.registry.TaskRegistry, 'register')
+    _u(celery, 'shared_task')
diff --git a/ddtrace/contrib/celery/task.py b/ddtrace/contrib/celery/task.py
--- a/ddtrace/contrib/celery/task.py
+++ b/ddtrace/contrib/celery/task.py
@@ -79,6 +79,14 @@
     return task
 
 
+def _wrap_shared_task(decorator, instance, args, kwargs):
+    """Wrapper for Django-Celery shared tasks. `shared_task` is a decorator
+    that returns a `Task` from the given function.
+    """
+    task = decorator(*args, **kwargs)
+    return patch_task(task)
+
+
 def _task_init(func, task, args, kwargs):
     func(*args, **kwargs)

{"golden_diff": "diff --git a/ddtrace/contrib/celery/patch.py b/ddtrace/contrib/celery/patch.py\n--- a/ddtrace/contrib/celery/patch.py\n+++ b/ddtrace/contrib/celery/patch.py\n@@ -3,19 +3,24 @@\n from wrapt import wrap_function_wrapper as _w\n \n from .app import patch_app, unpatch_app\n+from .task import _wrap_shared_task\n from .registry import _wrap_register\n from ...utils.wrappers import unwrap as _u\n \n \n def patch():\n     \"\"\"Instrument Celery base application and the `TaskRegistry` so\n-    that any new registered task is automatically instrumented\n+    that any new registered task is automatically instrumented. In the\n+    case of Django-Celery integration, also the `@shared_task` decorator\n+    must be instrumented because Django doesn't use the Celery registry.\n     \"\"\"\n     setattr(celery, 'Celery', patch_app(celery.Celery))\n     _w('celery.app.registry', 'TaskRegistry.register', _wrap_register)\n+    _w('celery', 'shared_task', _wrap_shared_task)\n \n \n def unpatch():\n     \"\"\"Removes instrumentation from Celery\"\"\"\n     setattr(celery, 'Celery', unpatch_app(celery.Celery))\n     _u(celery.app.registry.TaskRegistry, 'register')\n+    _u(celery, 'shared_task')\ndiff --git a/ddtrace/contrib/celery/task.py b/ddtrace/contrib/celery/task.py\n--- a/ddtrace/contrib/celery/task.py\n+++ b/ddtrace/contrib/celery/task.py\n@@ -79,6 +79,14 @@\n     return task\n \n \n+def _wrap_shared_task(decorator, instance, args, kwargs):\n+    \"\"\"Wrapper for Django-Celery shared tasks. `shared_task` is a decorator\n+    that returns a `Task` from the given function.\n+    \"\"\"\n+    task = decorator(*args, **kwargs)\n+    return patch_task(task)\n+\n+\n def _task_init(func, task, args, kwargs):\n     func(*args, **kwargs)\n", "issue": "celery 4.1 worker tasks aren't being traced with ddtrace-py 0.11.1\n```\r\nddtrace==0.11.1\r\ncelery==4.1.0\r\n```\r\n\r\nI'm manually patching celery due to #423 \r\n\r\nI'm getting traces for 'defaultdb' (sqlite), 'redis', and 'postgres' when I run the celery worker\r\n\r\nHowever, I'm not receiving any traces whatsoever for the task runs, and no 'celery' service appears in datadog.\r\n\r\nHere's a gist with my worker startup in debug logging mode:\r\nhttps://gist.github.com/eedwards-sk/f924548c043859901db6918aec95dada\r\n\r\nI'm sending traces to a local running agent and I know that's working because I'm seeing the flush records in the logs, and I get traces for the above mentioned services when I'm running it.\r\n\r\n(We have another application that is using celery 3.x and their traces show up for tasks, which we instrumented in more or less the same way)\r\n\r\nHere's a gist showing how we're instantiating celery and the tracer:\r\nhttps://gist.github.com/eedwards-sk/fe81f62406e7a245b7c26a9ada19c658\n", "before_files": [{"content": "# Third party\nimport wrapt\nimport inspect\nimport celery\n\n# Project\nfrom ddtrace import Pin\nfrom ddtrace.ext import AppTypes\nfrom ...ext import errors\nfrom .util import APP, PRODUCER_SERVICE, WORKER_SERVICE, meta_from_context, require_pin\n\nPRODUCER_ROOT_SPAN = 'celery.apply'\nWORKER_ROOT_SPAN = 'celery.run'\n# Task operations\nTASK_TAG_KEY = 'celery.action'\nTASK_APPLY = 'apply'\nTASK_APPLY_ASYNC = 'apply_async'\nTASK_RUN = 'run'\n\n\ndef patch_task(task, pin=None):\n    \"\"\" patch_task will add tracing to a celery task \"\"\"\n    # The service set here is actually ignored, because it's not possible to\n    # be certain whether this process is being used as a worker, a producer,\n    # or both. So the service as recorded in traces is set based on the actual\n    # work being done (ie. apply/apply_async vs run).\n    pin = pin or Pin(service=WORKER_SERVICE, app=APP, app_type=AppTypes.worker)\n\n    patch_methods = [\n        ('__init__', _task_init),\n        ('run', _task_run),\n        ('apply', _task_apply),\n        ('apply_async', _task_apply_async),\n    ]\n    for method_name, wrapper in patch_methods:\n        # Get original method\n        method = getattr(task, method_name, None)\n        if method is None:\n            continue\n\n        # Do not patch if method is already patched\n        if isinstance(method, wrapt.ObjectProxy):\n            continue\n\n        # If the function as been applied as a decorator for v1 Celery tasks, then a different patching is needed\n        if inspect.isclass(task) and issubclass(task, celery.task.Task):\n            wrapped = wrapt.FunctionWrapper(method, wrapper)\n            setattr(task, method_name, wrapped)\n            continue\n        # Patch method\n        # DEV: Using `BoundFunctionWrapper` ensures our `task` wrapper parameter is properly set\n        setattr(task, method_name, wrapt.BoundFunctionWrapper(method, task, wrapper))\n\n    # Attach our pin to the app\n    pin.onto(task)\n    return task\n\ndef unpatch_task(task):\n    \"\"\" unpatch_task will remove tracing from a celery task \"\"\"\n    patched_methods = [\n        '__init__',\n        'run',\n        'apply',\n        'apply_async',\n    ]\n    for method_name in patched_methods:\n        # Get wrapped method\n        wrapper = getattr(task, method_name, None)\n        if wrapper is None:\n            continue\n\n        # Only unpatch if wrapper is an `ObjectProxy`\n        if not isinstance(wrapper, wrapt.ObjectProxy):\n            continue\n\n        # Restore original method\n        setattr(task, method_name, wrapper.__wrapped__)\n\n    return task\n\n\ndef _task_init(func, task, args, kwargs):\n    func(*args, **kwargs)\n\n    # Patch this task if our pin is enabled\n    pin = Pin.get_from(task)\n    if pin and pin.enabled():\n        patch_task(task, pin=pin)\n\n\n@require_pin\ndef _task_run(pin, func, task, args, kwargs):\n    with pin.tracer.trace(WORKER_ROOT_SPAN, service=WORKER_SERVICE, resource=task.name) as span:\n        # Set meta data from task request\n        span.set_metas(meta_from_context(task.request))\n        span.set_meta(TASK_TAG_KEY, TASK_RUN)\n\n        # Call original `run` function\n        return func(*args, **kwargs)\n\n\n@require_pin\ndef _task_apply(pin, func, task, args, kwargs):\n    with pin.tracer.trace(PRODUCER_ROOT_SPAN, service=PRODUCER_SERVICE, resource=task.name) as span:\n        # Call the original `apply` function\n        res = func(*args, **kwargs)\n\n        # Set meta data from response\n        span.set_meta('id', res.id)\n        span.set_meta('state', res.state)\n        span.set_meta(TASK_TAG_KEY, TASK_APPLY)\n        if res.traceback:\n            span.error = 1\n            span.set_meta(errors.STACK, res.traceback)\n        return res\n\n\n@require_pin\ndef _task_apply_async(pin, func, task, args, kwargs):\n    with pin.tracer.trace(PRODUCER_ROOT_SPAN, service=PRODUCER_SERVICE, resource=task.name) as span:\n        # Extract meta data from `kwargs`\n        meta_keys = (\n            'compression', 'countdown', 'eta', 'exchange', 'expires',\n            'priority', 'routing_key', 'serializer', 'queue',\n        )\n        for name in meta_keys:\n            if name in kwargs:\n                span.set_meta(name, kwargs[name])\n        span.set_meta(TASK_TAG_KEY, TASK_APPLY_ASYNC)\n\n        # Call the original `apply_async` function\n        res = func(*args, **kwargs)\n\n        # Set meta data from response\n        # DEV: Calling `res.traceback` or `res.state` will make an\n        #   API call to the backend for the properties\n        span.set_meta('id', res.id)\n        return res\n", "path": "ddtrace/contrib/celery/task.py"}, {"content": "import celery\n\nfrom wrapt import wrap_function_wrapper as _w\n\nfrom .app import patch_app, unpatch_app\nfrom .registry import _wrap_register\nfrom ...utils.wrappers import unwrap as _u\n\n\ndef patch():\n    \"\"\"Instrument Celery base application and the `TaskRegistry` so\n    that any new registered task is automatically instrumented\n    \"\"\"\n    setattr(celery, 'Celery', patch_app(celery.Celery))\n    _w('celery.app.registry', 'TaskRegistry.register', _wrap_register)\n\n\ndef unpatch():\n    \"\"\"Removes instrumentation from Celery\"\"\"\n    setattr(celery, 'Celery', unpatch_app(celery.Celery))\n    _u(celery.app.registry.TaskRegistry, 'register')\n", "path": "ddtrace/contrib/celery/patch.py"}], "after_files": [{"content": "# Third party\nimport wrapt\nimport inspect\nimport celery\n\n# Project\nfrom ddtrace import Pin\nfrom ddtrace.ext import AppTypes\nfrom ...ext import errors\nfrom .util import APP, PRODUCER_SERVICE, WORKER_SERVICE, meta_from_context, require_pin\n\nPRODUCER_ROOT_SPAN = 'celery.apply'\nWORKER_ROOT_SPAN = 'celery.run'\n# Task operations\nTASK_TAG_KEY = 'celery.action'\nTASK_APPLY = 'apply'\nTASK_APPLY_ASYNC = 'apply_async'\nTASK_RUN = 'run'\n\n\ndef patch_task(task, pin=None):\n    \"\"\" patch_task will add tracing to a celery task \"\"\"\n    # The service set here is actually ignored, because it's not possible to\n    # be certain whether this process is being used as a worker, a producer,\n    # or both. So the service as recorded in traces is set based on the actual\n    # work being done (ie. apply/apply_async vs run).\n    pin = pin or Pin(service=WORKER_SERVICE, app=APP, app_type=AppTypes.worker)\n\n    patch_methods = [\n        ('__init__', _task_init),\n        ('run', _task_run),\n        ('apply', _task_apply),\n        ('apply_async', _task_apply_async),\n    ]\n    for method_name, wrapper in patch_methods:\n        # Get original method\n        method = getattr(task, method_name, None)\n        if method is None:\n            continue\n\n        # Do not patch if method is already patched\n        if isinstance(method, wrapt.ObjectProxy):\n            continue\n\n        # If the function as been applied as a decorator for v1 Celery tasks, then a different patching is needed\n        if inspect.isclass(task) and issubclass(task, celery.task.Task):\n            wrapped = wrapt.FunctionWrapper(method, wrapper)\n            setattr(task, method_name, wrapped)\n            continue\n        # Patch method\n        # DEV: Using `BoundFunctionWrapper` ensures our `task` wrapper parameter is properly set\n        setattr(task, method_name, wrapt.BoundFunctionWrapper(method, task, wrapper))\n\n    # Attach our pin to the app\n    pin.onto(task)\n    return task\n\ndef unpatch_task(task):\n    \"\"\" unpatch_task will remove tracing from a celery task \"\"\"\n    patched_methods = [\n        '__init__',\n        'run',\n        'apply',\n        'apply_async',\n    ]\n    for method_name in patched_methods:\n        # Get wrapped method\n        wrapper = getattr(task, method_name, None)\n        if wrapper is None:\n            continue\n\n        # Only unpatch if wrapper is an `ObjectProxy`\n        if not isinstance(wrapper, wrapt.ObjectProxy):\n            continue\n\n        # Restore original method\n        setattr(task, method_name, wrapper.__wrapped__)\n\n    return task\n\n\ndef _wrap_shared_task(decorator, instance, args, kwargs):\n    \"\"\"Wrapper for Django-Celery shared tasks. `shared_task` is a decorator\n    that returns a `Task` from the given function.\n    \"\"\"\n    task = decorator(*args, **kwargs)\n    return patch_task(task)\n\n\ndef _task_init(func, task, args, kwargs):\n    func(*args, **kwargs)\n\n    # Patch this task if our pin is enabled\n    pin = Pin.get_from(task)\n    if pin and pin.enabled():\n        patch_task(task, pin=pin)\n\n\n@require_pin\ndef _task_run(pin, func, task, args, kwargs):\n    with pin.tracer.trace(WORKER_ROOT_SPAN, service=WORKER_SERVICE, resource=task.name) as span:\n        # Set meta data from task request\n        span.set_metas(meta_from_context(task.request))\n        span.set_meta(TASK_TAG_KEY, TASK_RUN)\n\n        # Call original `run` function\n        return func(*args, **kwargs)\n\n\n@require_pin\ndef _task_apply(pin, func, task, args, kwargs):\n    with pin.tracer.trace(PRODUCER_ROOT_SPAN, service=PRODUCER_SERVICE, resource=task.name) as span:\n        # Call the original `apply` function\n        res = func(*args, **kwargs)\n\n        # Set meta data from response\n        span.set_meta('id', res.id)\n        span.set_meta('state', res.state)\n        span.set_meta(TASK_TAG_KEY, TASK_APPLY)\n        if res.traceback:\n            span.error = 1\n            span.set_meta(errors.STACK, res.traceback)\n        return res\n\n\n@require_pin\ndef _task_apply_async(pin, func, task, args, kwargs):\n    with pin.tracer.trace(PRODUCER_ROOT_SPAN, service=PRODUCER_SERVICE, resource=task.name) as span:\n        # Extract meta data from `kwargs`\n        meta_keys = (\n            'compression', 'countdown', 'eta', 'exchange', 'expires',\n            'priority', 'routing_key', 'serializer', 'queue',\n        )\n        for name in meta_keys:\n            if name in kwargs:\n                span.set_meta(name, kwargs[name])\n        span.set_meta(TASK_TAG_KEY, TASK_APPLY_ASYNC)\n\n        # Call the original `apply_async` function\n        res = func(*args, **kwargs)\n\n        # Set meta data from response\n        # DEV: Calling `res.traceback` or `res.state` will make an\n        #   API call to the backend for the properties\n        span.set_meta('id', res.id)\n        return res\n", "path": "ddtrace/contrib/celery/task.py"}, {"content": "import celery\n\nfrom wrapt import wrap_function_wrapper as _w\n\nfrom .app import patch_app, unpatch_app\nfrom .task import _wrap_shared_task\nfrom .registry import _wrap_register\nfrom ...utils.wrappers import unwrap as _u\n\n\ndef patch():\n    \"\"\"Instrument Celery base application and the `TaskRegistry` so\n    that any new registered task is automatically instrumented. In the\n    case of Django-Celery integration, also the `@shared_task` decorator\n    must be instrumented because Django doesn't use the Celery registry.\n    \"\"\"\n    setattr(celery, 'Celery', patch_app(celery.Celery))\n    _w('celery.app.registry', 'TaskRegistry.register', _wrap_register)\n    _w('celery', 'shared_task', _wrap_shared_task)\n\n\ndef unpatch():\n    \"\"\"Removes instrumentation from Celery\"\"\"\n    setattr(celery, 'Celery', unpatch_app(celery.Celery))\n    _u(celery.app.registry.TaskRegistry, 'register')\n    _u(celery, 'shared_task')\n", "path": "ddtrace/contrib/celery/patch.py"}]}