Repository: nameko/opentelemetry-instrumentation-nameko (language: python)

Issue:
add tests for amqp module
Add final tests for the AMQP module


Discussion / hints:

Fix patch:
diff --git a/.coveragerc b/.coveragerc
new file mode 100644
index 0000000..8f80d31
--- /dev/null
+++ b/.coveragerc
@@ -0,0 +1,9 @@
+[run]
+concurrency = eventlet
+branch = 1
+source =
+    nameko_opentelemetry
+    tests
+[report]
+show_missing = true
+fail_under = 100
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
new file mode 100644
index 0000000..8ae9fc1
--- /dev/null
+++ b/.github/workflows/ci.yaml
@@ -0,0 +1,90 @@
+on:
+  push:
+    branches:
+    - "**"
+  pull_request:
+    branches:
+    - master
+
+  workflow_dispatch:
+
+jobs:
+  static:
+    runs-on: ubuntu-latest
+    steps:
+    - name: checkout
+      uses: actions/checkout@v2
+
+    - name: setup python
+      uses: actions/setup-python@v2
+      with:
+        python-version: 3.9
+
+    - name: install test dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install tox
+
+    - name: run tox
+      run: tox
+      env:
+        TOXENV: static
+
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version:
+        - 3.6
+        - 3.7
+        - 3.8
+        - 3.9
+
+    steps:
+    - name: checkout
+      uses: actions/checkout@v2
+
+    - name: setup python
+      uses: actions/setup-python@v2
+      with:
+        python-version: ${{ matrix.python-version }}
+
+    - name: start rabbitmq
+      run: |
+        docker run -d --rm -p 15672:15672 -p 5672:5672 -p 5671:5671 --name nameko-rabbitmq ghcr.io/nameko/nameko-rabbitmq:3.6.6
+
+    - name: install test dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install tox
+
+    - name: run tox
+      run: tox
+      env:
+        TOXENV: py${{ matrix.python-version }}-test
+
+  distribute:
+    runs-on: ubuntu-latest
+    needs:
+    - static
+    - test
+    if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags')
+
+    steps:
+    - name: checkout
+      uses: actions/checkout@v2
+
+    - name: setup python
+      uses: actions/setup-python@v2
+      with:
+        python-version: 3.9
+
+    - name: generate distributables
+      run: |
+        python setup.py sdist bdist_wheel
+
+    - name: publish to pypi
+      uses: pypa/gh-action-pypi-publish@release/v1
+      with:
+        user: __token__
+        password: ${{ secrets.PYPI_API_TOKEN }}
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..a6455d9
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,5 @@
+__pycache__
+*.egg-info
+.tox
+.coverage
+.pytest_cache
\ No newline at end of file
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
new file mode 100644
index 0000000..2d67b56
--- /dev/null
+++ b/.pre-commit-config.yaml
@@ -0,0 +1,22 @@
+repos:
+- repo: git://github.com/pre-commit/pre-commit-hooks
+  rev: v1.3.0
+  hooks:
+  - id: trailing-whitespace
+  - id: check-ast
+  - id: check-merge-conflict
+  - id: fix-encoding-pragma
+  - id: flake8
+- repo: https://github.com/pycqa/isort
+  rev: 5.6.3
+  hooks:
+  - id: isort
+- repo: https://github.com/ambv/black
+  rev: 19.10b0
+  hooks:
+  - id: black
+- repo: https://github.com/adrienverge/yamllint.git
+  rev: v1.25.0
+  hooks:
+  - id: yamllint
+    args: ["-d", "relaxed"]
diff --git a/LICENSE b/LICENSE
index 261eeb9..66a27ec 100644
--- a/LICENSE
+++ b/LICENSE
@@ -175,27 +175,3 @@
 
    END OF TERMS AND CONDITIONS
 
-   APPENDIX: How to apply the Apache License to your work.
-
-      To apply the Apache License to your work, attach the following
-      boilerplate notice, with the fields enclosed by brackets "[]"
-      replaced with your own identifying information. (Don't include
-      the brackets!)  The text should be enclosed in the appropriate
-      comment syntax for the file format. We also recommend that a
-      file or class name and description of purpose be included on the
-      same "printed page" as the copyright notice for easier
-      identification within third-party archives.
-
-   Copyright [yyyy] [name of copyright owner]
-
-   Licensed under the Apache License, Version 2.0 (the "License");
-   you may not use this file except in compliance with the License.
-   You may obtain a copy of the License at
-
-       http://www.apache.org/licenses/LICENSE-2.0
-
-   Unless required by applicable law or agreed to in writing, software
-   distributed under the License is distributed on an "AS IS" BASIS,
-   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-   See the License for the specific language governing permissions and
-   limitations under the License.
diff --git a/Makefile b/Makefile
new file mode 100644
index 0000000..460188f
--- /dev/null
+++ b/Makefile
@@ -0,0 +1,9 @@
+static:
+	pre-commit run --all-files
+
+test:
+	py.test tests -v
+
+coverage:
+	coverage run -m pytest tests -v
+	coverage report
\ No newline at end of file
diff --git a/README.md b/README.md
index 568763a..031ec3b 100644
--- a/README.md
+++ b/README.md
@@ -1,1 +1,90 @@
-# nameko-opentelemetry
\ No newline at end of file
+# nameko-opentelemetry
+
+OpenTelemetry instrumentation for Nameko microservices.
+
+## Usage
+
+Inside your codebase, before the service is started, apply the instrumentation:
+
+``` python
+# Instrument nameko
+from nameko_opentelemetry import NamekoInstrumentor
+NamekoInstrumentor().instrument(
+    # optionally pass any customised entrypoint adapters (see below for details)
+    entrypoint_adapters={
+        "my.custom.EntrypointType": "my.custom.EntrypointAdapter"
+    },
+    # turn on logging of request headers and context data
+    send_headers=True,
+    # turn on logging of request payloads and entrypoint arguments
+    send_request_payloads=True,
+    # turn on logging of response payloads and entrypoint results
+    send_response_payloads=True,
+    # change the default length at which headers and payloads are truncated
+    max_truncate_length=1000,
+)
+
+# You will also want to instrument any other libraries you're using, e.g.
+from opentelemetry.instrumentation.requests import RequestsInstrumentor
+RequestsInstrumentor().instrument()
+
+# All entrypoints, and all built-in dependency providers and clients are now instrumented.
+# To use the instrumentation somehow, attach a span processor, e.g.
+from opentelemetry import trace
+from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor
+trace.get_tracer_provider().add_span_processor(
+    SimpleSpanProcessor(ConsoleSpanExporter())
+)
+```
+
+## Approach
+
+This library uses monkey-patching, as many other third-party libraries do when adding instrumentation. This has the advantage of keeping the main library "clean" of instrumentation code, but at the expense of some brittleness. It's probably fine for now.
+
+## Applied Instrumentation
+
+Server spans are generated for every entrypoint execution, with helpful default span names and attributes. To customise these for a third-party entrypoint, you can define a new `EntrypointAdapter`. See below.
+
+Client spans are generated by all the built-in dependency providers and standalone clients, again with helpful defaults for span names and attributes. To instrument third-party dependency providers or clients, you may need to create standalone instrumentation -- or rely on pre-existing instrumentation of an underlying library (e.g. requests, sqlalchemy, etc.)
+
+
+### Entrypoint Adapters
+
+This instrumentation generates a default span name and attributes for every entrypoint that executes. It assumes that the `context_data` for the call includes any propagation headers from whichever transport was used. If you want to customise the span name or attributes, or if the entrypoint _does not_ populate the worker's `context_data` with headers from the transport, you can implement an entrypoint adapter. As an example, parts of the adapter for the built-in `@http` entrypoint:
+
+``` python
+class HttpEntrypointAdapter(EntrypointAdapter):
+    """ Implemented according to https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/semantic_conventions/http.md
+    """
+
+    def get_metadata(self):
+        # http headers are not automatically added to the worker's context data,
+        # so we get them from the `request` argument
+        return self.worker_ctx.args[0].headers
+
+    def get_span_name(self):
+        # http entrypoints use the url path as a span name, as per
+        # https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/semantic_conventions/http.md
+        return self.worker_ctx.entrypoint.url
+
+    def get_attributes(self):
+        """ Arbitrary additional attributes to include
+        """
+        default_attributes = super(). get_attributes()
+        default_attributes.update({
+            "appropriately.namespaced.key": "value"
+        })
+
+    def get_result_attributes(self, result):
+        """ Return attributes derived from the `result` returned by the entrypoint method.
+        """
+
+    def get_exception_attributes(self, exc_info):
+        """ Return attributes derived from any exception thrown by the entrypoint method.
+        """
+```
+
+# TODO
+
+- [ ] Refactor RPC patching to remove some brittleness
+- [ ] Add support for scrubbing sensitive data
\ No newline at end of file
diff --git a/demo.py b/demo.py
new file mode 100644
index 0000000..e275dfc
--- /dev/null
+++ b/demo.py
@@ -0,0 +1,106 @@
+# -*- coding: utf-8 -*-
+"""
+Demo service that exercises much of the instrumentation in this package.
+
+Supply a honeycomb API key to send the spans there for visual exploration.
+
+Usage:
+
+HONEYCOMB_API_KEY=secret nameko run demo \
+    --define AMQP_URI=pyamqp://guest:guest@localhost/ \
+    --define WEB_SERVER_ADDRESS=0.0.0.0:8000
+
+"""
+import logging
+import os
+import random
+import time
+
+import nameko
+import opentelemetry.instrumentation.requests
+import requests
+from kombu.messaging import Exchange, Queue
+from nameko.events import EventDispatcher, event_handler
+from nameko.messaging import Publisher, consume
+from nameko.rpc import ServiceRpc, rpc
+from nameko.timer import timer
+from nameko.web.handlers import http
+from opentelemetry import trace
+from opentelemetry.ext.honeycomb import HoneycombSpanExporter
+from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+
+from nameko_opentelemetry import NamekoInstrumentor
+
+
+logging.basicConfig()
+
+logger = logging.getLogger(__name__)
+
+
+provider = TracerProvider(resource=Resource.create({"blah.blah": "foobar"}))
+trace.set_tracer_provider(provider)
+
+# instrument requests
+opentelemetry.instrumentation.requests.RequestsInstrumentor().instrument(
+    tracer_provider=provider, entrypoint_adapters={}
+)
+# instrument nameko
+NamekoInstrumentor().instrument(tracer_provider=provider)
+
+# export spans to honeycomb
+exporter = HoneycombSpanExporter(
+    service_name="demo-service",
+    writekey=os.environ.get("HONEYCOMB_API_KEY"),
+    dataset="test",
+)
+trace.get_tracer_provider().add_span_processor(BatchSpanProcessor(exporter))
+
+
+exchange = Exchange(name="barfoo")
+queue = Queue(name="barfoo", exchange=exchange)
+
+
+class Service:
+    name = "demo"
+
+    dispatch = EventDispatcher()
+    publish = Publisher()
+    demo_rpc = ServiceRpc("demo")
+
+    @consume(queue)
+    def handle_barfoo(self, body):
+        time.sleep(random.random())
+        return body
+
+    @event_handler("demo", "foobar")
+    def handle_foobar(self, payload):
+        time.sleep(random.random())
+        self.publish("barfoo", routing_key="barfoo")
+        return "ok"
+
+    @http("GET", "/matt")
+    def matt(self, request):
+        time.sleep(random.random())
+        self.dispatch("foobar", "yo")
+        time.sleep(random.random())
+        return "ok"
+
+    @rpc
+    def upper(self, string):
+        time.sleep(random.random())
+        requests.get(f"http://{nameko.config['WEB_SERVER_ADDRESS']}/matt")
+        time.sleep(random.random())
+        return string.upper()
+
+    @http("GET", "/hello")
+    def hello(self, request):
+        time.sleep(random.random())
+        requests.get("http://google.com")
+        time.sleep(random.random())
+        return 200, self.demo_rpc.upper("matt")
+
+    @timer(interval=1)
+    def tick(self):
+        requests.get(f"http://{nameko.config['WEB_SERVER_ADDRESS']}/hello")
diff --git a/nameko_opentelemetry/__init__.py b/nameko_opentelemetry/__init__.py
new file mode 100644
index 0000000..bed2a05
--- /dev/null
+++ b/nameko_opentelemetry/__init__.py
@@ -0,0 +1,42 @@
+# -*- coding: utf-8 -*-
+from opentelemetry import trace
+from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
+
+from nameko_opentelemetry import amqp, entrypoints, events, http, messaging, rpc
+from nameko_opentelemetry.package import _instruments
+from nameko_opentelemetry.version import __version__
+
+
+def active_tracer():
+    provider = trace.get_tracer_provider()
+    return trace.get_tracer(__name__, __version__, provider)
+
+
+class NamekoInstrumentor(BaseInstrumentor):
+    def instrumentation_dependencies(self):
+        return _instruments
+
+    def _instrument(self, **config):
+        """
+        ...
+        """
+        tracer = active_tracer()
+
+        # client_request_hook = kwargs.get("client_request_hook", None)
+        # client_response_hook = kwargs.get("client_response_hook", None)
+        # server_request_hook = kwargs.get("server_request_hook", None)
+
+        entrypoints.instrument(tracer, config)
+        http.instrument(tracer, config)
+        amqp.instrument(tracer, config)
+        rpc.instrument(tracer, config)
+        events.instrument(tracer, config)
+        messaging.instrument(tracer, config)
+
+    def _uninstrument(self, **kwargs):
+        entrypoints.uninstrument()
+        http.uninstrument()
+        amqp.uninstrument()
+        rpc.uninstrument()
+        events.uninstrument()
+        messaging.uninstrument()
diff --git a/nameko_opentelemetry/amqp.py b/nameko_opentelemetry/amqp.py
new file mode 100644
index 0000000..0ca1d59
--- /dev/null
+++ b/nameko_opentelemetry/amqp.py
@@ -0,0 +1,105 @@
+# -*- coding: utf-8 -*-
+""" Utility functions that extract relevant attributes from AMQP
+publishers and consumers.
+"""
+from functools import partial
+
+import nameko.amqp.publish
+from opentelemetry import trace
+from opentelemetry.instrumentation.utils import unwrap
+from wrapt import wrap_function_wrapper
+
+from nameko_opentelemetry.scrubbers import scrub
+from nameko_opentelemetry.utils import serialise_to_string
+
+
+PREFIX = "nameko.amqp"
+
+
+def get_routing_key(publisher, kwargs):
+    """
+    Extract routing key from combined publisher options, so we can report it as
+    an attribute in its own right.
+    """
+    options = publisher.publish_kwargs.copy()
+    options.update(kwargs)
+    return serialise_to_string(options.get("routing_key"))
+
+
+def get_headers(publisher, kwargs, config):
+    """
+    Extract final headers included in the published message. Must be extracted
+    from several combined sources.
+    """
+    headers = publisher.publish_kwargs.get("headers", {})
+    headers.update(kwargs.get("headers", {}))
+    headers.update(kwargs.get("extra_headers", {}))
+    return serialise_to_string(scrub(headers, config))
+
+
+def amqp_publisher_attributes(publisher, kwargs, config):
+    """
+    Extract attributes relevant to AMQP message publishers.
+
+    The publisher instance and the kwargs passed to it as publish-time are
+    combined before the attribute value is extracted.
+    """
+
+    def generic_getter(attribute):
+        return serialise_to_string(
+            kwargs.get(attribute, getattr(publisher, attribute, None))
+        )
+
+    attributes = {
+        f"{PREFIX}.amqp_uri": generic_getter("amqp_uri"),
+        f"{PREFIX}.ssl": generic_getter("ssl"),
+        f"{PREFIX}.use_confirms": generic_getter("use_confirms"),
+        f"{PREFIX}.delivery_mode": generic_getter("delivery_mode"),
+        f"{PREFIX}.mandatory": generic_getter("mandatory"),
+        f"{PREFIX}.priority": generic_getter("priority"),
+        f"{PREFIX}.expiration": generic_getter("expiration"),
+        f"{PREFIX}.serializer": generic_getter("serializer"),
+        f"{PREFIX}.compression": generic_getter("compression"),
+        f"{PREFIX}.retry": generic_getter("retry"),
+        f"{PREFIX}.retry_policy": generic_getter("retry_policy"),
+        f"{PREFIX}.declarations": generic_getter("declare"),
+        f"{PREFIX}.transport_options": generic_getter("transport_options"),
+        f"{PREFIX}.publish_kwargs": generic_getter("publish_kwargs"),
+        f"{PREFIX}.routing_key": get_routing_key(publisher, kwargs),
+    }
+
+    if config.get("send_headers"):  # pragma: no cover (temporary; module needs tests)
+        attributes.update({f"{PREFIX}.headers": get_headers(publisher, kwargs, config)})
+
+    return attributes
+
+
+def amqp_consumer_attributes(consumer):
+    """
+    Extract attributes relevant to AMQP message consumers.
+    """
+    return {
+        f"{PREFIX}.amqp_uri": serialise_to_string(consumer.amqp_uri),
+        f"{PREFIX}.ssl": serialise_to_string(consumer.ssl),
+        f"{PREFIX}.prefetch_count": serialise_to_string(consumer.prefetch_count),
+        f"{PREFIX}.heartbeat": serialise_to_string(consumer.heartbeat),
+        f"{PREFIX}.accept": serialise_to_string(consumer.accept),
+        f"{PREFIX}.queues": serialise_to_string(consumer.queues),
+        f"{PREFIX}.consumer_options": serialise_to_string(consumer.consumer_options),
+    }
+
+
+def publish(tracer, config, wrapped, instance, args, kwargs):
+    current_span = trace.get_current_span()
+    current_span.set_attributes(amqp_publisher_attributes(instance, kwargs, config))
+    return wrapped(*args, **kwargs)
+
+
+def instrument(tracer, config):
+    wrap_function_wrapper(
+        "nameko.amqp.publish", "Publisher.publish", partial(publish, tracer, config),
+    )
+
+
+def uninstrument():
+    unwrap(nameko.amqp.publish.Publisher, "publish")
diff --git a/nameko_opentelemetry/entrypoints.py b/nameko_opentelemetry/entrypoints.py
new file mode 100644
index 0000000..ca8f3b3
--- /dev/null
+++ b/nameko_opentelemetry/entrypoints.py
@@ -0,0 +1,293 @@
+# -*- coding: utf-8 -*-
+"""
+This module patches the Nameko ServiceContainer so that every entrypoint that fires
+generates a span with helpful defaults.
+
+The kind, name, attributes and status of the span are determined by the
+EntrypointAdapter class. More specialised versions can be provided by passing an
+appropriate dictionary as `entrypoint_adapters` when invoking
+`NamekoInstrumentor.instrument()`.
+
+For example:
+
+    entrypoint_to_adapter_map = {
+        "my.custom.EntrypointType": "my.custom.EntrypointAdapter"
+    }
+
+    instrumentor = NamekoInstrumentor()
+    instrumentor.instrument(entrypoint_adapters=entrypoint_to_adapter_map)
+
+"""
+import inspect
+import socket
+import warnings
+from collections import defaultdict
+from functools import partial
+from traceback import format_exception
+from weakref import WeakKeyDictionary
+
+import nameko.containers
+from nameko.utils import get_redacted_args
+from opentelemetry import context, trace
+from opentelemetry.instrumentation.utils import unwrap
+from opentelemetry.propagate import extract
+from opentelemetry.trace.status import Status, StatusCode
+from opentelemetry.util._time import _time_ns
+from wrapt import wrap_function_wrapper
+
+from nameko_opentelemetry import utils
+from nameko_opentelemetry.scrubbers import scrub
+
+
+DEFAULT_ADAPTERS = {
+    "nameko.rpc.Rpc": ("nameko_opentelemetry.rpc.RpcEntrypointAdapter"),
+    "nameko.web.handlers.HttpRequestHandler": (
+        "nameko_opentelemetry.http.HttpEntrypointAdapter"
+    ),
+    "nameko.events.EventHandler": (
+        "nameko_opentelemetry.events.EventHandlerEntrypointAdapter"
+    ),
+    "nameko.messaging.Consumer": (
+        "nameko_opentelemetry.messaging.ConsumerEntrypointAdapter"
+    ),
+    "nameko.timer.Timer": ("nameko_opentelemetry.timer.TimerEntrypointAdapter"),
+}
+
+active_spans = WeakKeyDictionary()
+adapter_types = defaultdict(lambda: EntrypointAdapter)
+
+
+class EntrypointAdapter:
+    """ Default entrypoint adapter. This implementation is used unless there's
+    a more specific adapter set for the firing entrypoint's type.
+    """
+
+    span_kind = trace.SpanKind.SERVER
+
+    def __init__(self, worker_ctx, config):
+        self.worker_ctx = worker_ctx
+        self.config = config
+
+    def get_span_name(self):
+        return (
+            f"{self.worker_ctx.service_name}.{self.worker_ctx.entrypoint.method_name}"
+        )
+
+    def get_metadata(self):
+        return self.worker_ctx.context_data
+
+    def exception_was_expected(self, exc):
+        expected_exceptions = getattr(
+            self.worker_ctx.entrypoint, "expected_exceptions", None
+        )
+        expected_exceptions = expected_exceptions or tuple()
+        return isinstance(exc, expected_exceptions)
+
+    def start_span(self, span):
+        if span.is_recording():
+            span.set_attributes(self.get_attributes())
+
+    def end_span(self, span, result, exc_info):
+        if span.is_recording():
+
+            if exc_info:
+                span.record_exception(
+                    exc_info[1],
+                    escaped=True,
+                    attributes=self.get_exception_attributes(exc_info),
+                )
+            else:
+                span.set_attributes(self.get_result_attributes(result))
+
+            status = self.get_status(result, exc_info)
+            span.set_status(status)
+
+    def get_attributes(self):
+        """ Common attributes for most entrypoints.
+        """
+        entrypoint = self.worker_ctx.entrypoint
+
+        attributes = {
+            "service_name": self.worker_ctx.service_name,
+            "entrypoint_type": type(entrypoint).__name__,
+            "method_name": entrypoint.method_name,
+            "active_workers": self.worker_ctx.container._worker_pool.running(),
+            "available_workers": self.worker_ctx.container._worker_pool.free(),
+        }
+
+        if self.config.get("send_headers"):
+            attributes.update(
+                {
+                    "context_data": utils.serialise_to_string(
+                        scrub(self.worker_ctx.data, self.config)
+                    )
+                }
+            )
+
+        if self.config.get("send_request_payloads"):
+
+            if getattr(entrypoint, "sensitive_arguments", None):
+                call_args = get_redacted_args(
+                    entrypoint, *self.worker_ctx.args, **self.worker_ctx.kwargs
+                )
+                redacted = True
+            else:
+                method = getattr(
+                    entrypoint.container.service_cls, entrypoint.method_name
+                )
+                call_args = inspect.getcallargs(
+                    method, None, *self.worker_ctx.args, **self.worker_ctx.kwargs
+                )
+                del call_args["self"]
+                redacted = False
+
+            call_args, truncated = utils.truncate(
+                utils.serialise_to_string(scrub(call_args, self.config)),
+                max_len=self.config.get("truncate_max_length"),
+            )
+
+            attributes.update(
+                {
+                    "call_args": call_args,
+                    "call_args_redacted": str(redacted),
+                    "call_args_truncated": str(truncated),
+                }
+            )
+
+        return attributes
+
+    def get_exception_attributes(self, exc_info):
+        """ Additional attributes to save alongside a worker exception.
+        """
+        exc_type, exc, _ = exc_info
+
+        try:
+            stacktrace = "\n".join(format_exception(*exc_info))
+        except Exception:
+            stacktrace = "Exception occurred on stacktrace formatting"
+
+        return {
+            "exception.stacktrace": stacktrace,
+            "exception.expected": str(self.exception_was_expected(exc)),
+        }
+
+    def get_result_attributes(self, result):
+        """ Attributes describing the entrypoint method result.
+        """
+        if self.config.get("send_response_payloads"):
+
+            attributes = {
+                "result": utils.serialise_to_string(scrub(result or "", self.config))
+            }
+        else:
+            attributes = {}
+        return attributes
+
+    def get_status(self, result, exc_info):
+        """ Span status for this worker.
+        """
+        if exc_info:
+            exc_type, exc, _ = exc_info
+
+            if not self.exception_was_expected(exc):
+                return Status(
+                    StatusCode.ERROR,
+                    description="{}: {}".format(type(exc).__name__, exc),
+                )
+
+        return Status(StatusCode.OK)
+
+
+def adapter_factory(worker_ctx, config):
+    adapter_class = adapter_types[type(worker_ctx.entrypoint)]
+    return adapter_class(worker_ctx, config)
+
+
+def worker_setup(tracer, config, wrapped, instance, args, kwargs):
+    """ Wrap nameko.containers.ServiceContainer._worker_setup.
+
+    Creates a new span for each entrypoint that fires. The name of the
+    span and its attributes are determined by the entrypoint "adapter"
+    that is configured for that entrypoint, or the default implementation.
+    """
+    (worker_ctx,) = args
+
+    adapter = adapter_factory(worker_ctx, config)
+    ctx = extract(adapter.get_metadata())
+    token = context.attach(ctx)
+
+    span = tracer.start_span(
+        adapter.get_span_name(),
+        kind=adapter.span_kind,
+        attributes={"hostname": socket.gethostname()},
+        start_time=_time_ns(),
+    )
+    # don't automatically record the exception or set status, because
+    # we do that in the entrypoint adapter's `end_span` method
+    activation = trace.use_span(
+        span, record_exception=False, set_status_on_exception=False
+    )
+    activation.__enter__()
+    active_spans[worker_ctx] = (activation, span, token, adapter)
+
+    adapter.start_span(span)
+
+    wrapped(*args, **kwargs)
+
+
+def worker_result(tracer, config, wrapped, instance, args, kwargs):
+    """ Wrap nameko.containers.ServiceContainer._worker_result.
+
+    Finds the existing span for this worker and closes it. Additional
+    attributes and status are set by the configured entrypoint adapter.
+    """
+    (worker_ctx, result, exc_info) = args
+
+    activated = active_spans.get(worker_ctx)
+    if not activated:
+        # something went wrong when starting the span; nothing more to do
+        warnings.warn("worker result when no active span")
+        return
+
+    activation, span, token, adapter = activated
+
+    adapter.end_span(span, result, exc_info)
+
+    if exc_info is None:
+        activation.__exit__(None, None, None)
+    else:
+        activation.__exit__(*exc_info)
+
+    span.end(_time_ns())
+    context.detach(token)
+
+    wrapped(*args, **kwargs)
+
+
+def instrument(tracer, config):
+
+    # set up entrypoint adapters
+    adapter_config = DEFAULT_ADAPTERS.copy()
+    adapter_config.update(config.get("entrypoint_adapters", {}))
+
+    for entrypoint_path, adapter_path in adapter_config.items():
+        entrypoint_class = utils.import_by_path(entrypoint_path)
+        adapter_class = utils.import_by_path(adapter_path)
+        adapter_types[entrypoint_class] = adapter_class
+
+    # apply patches
+    wrap_function_wrapper(
+        "nameko.containers",
+        "ServiceContainer._worker_setup",
+        partial(worker_setup, tracer, config),
+    )
+    wrap_function_wrapper(
+        "nameko.containers",
+        "ServiceContainer._worker_result",
+        partial(worker_result, tracer, config),
+    )
+
+
+def uninstrument():
+    unwrap(nameko.containers.ServiceContainer, "_worker_setup")
+    unwrap(nameko.containers.ServiceContainer, "_worker_result")
diff --git a/nameko_opentelemetry/events.py b/nameko_opentelemetry/events.py
new file mode 100644
index 0000000..85a6a49
--- /dev/null
+++ b/nameko_opentelemetry/events.py
@@ -0,0 +1,159 @@
+# -*- coding: utf-8 -*-
+""" This modules applies patches to capture spans when events are dispatched, by both
+the dependency provider and the standalone client.
+
+The entrypoint adapter for event handler entrypoint is defined here too.
+"""
+from functools import partial
+
+import nameko.events
+import nameko.standalone.events
+from nameko.standalone.events import get_event_exchange
+from opentelemetry import trace
+from opentelemetry.instrumentation.utils import unwrap
+from opentelemetry.propagate import inject
+from wrapt import FunctionWrapper, wrap_function_wrapper
+
+from nameko_opentelemetry.amqp import amqp_consumer_attributes
+from nameko_opentelemetry.entrypoints import EntrypointAdapter
+from nameko_opentelemetry.scrubbers import scrub
+from nameko_opentelemetry.utils import (
+    call_function_get_frame,
+    serialise_to_string,
+    truncate,
+)
+
+
+class EventHandlerEntrypointAdapter(EntrypointAdapter):
+    """ Adapter customisation for EventHandler entrypoints.
+    """
+
+    span_kind = trace.SpanKind.CONSUMER
+
+    def get_attributes(self):
+        """ Include configuration of the entrypoint, and AMQP consumer attributes.
+        """
+        attrs = super().get_attributes()
+
+        entrypoint = self.worker_ctx.entrypoint
+
+        attrs.update(
+            {
+                "nameko.events.handler_type": entrypoint.handler_type,
+                "nameko.events.reliable_delivery": str(entrypoint.reliable_delivery),
+                "nameko.events.requeue_on_error": str(entrypoint.requeue_on_error),
+            }
+        )
+
+        consumer = self.worker_ctx.entrypoint.consumer
+        attrs.update(amqp_consumer_attributes(consumer))
+        return attrs
+
+
+def collect_client_attributes(
+    config, exchange_name, event_type, event_data, publisher, kwargs
+):
+    attributes = {
+        "nameko.events.exchange": exchange_name,
+        "nameko.events.event_type": event_type,
+    }
+    if config.get("send_request_payloads"):
+        data, truncated = truncate(
+            serialise_to_string(scrub(event_data, config)),
+            max_len=config.get("truncate_max_length"),
+        )
+        attributes.update(
+            {
+                "nameko.events.event_data": data,
+                "nameko.events.event_data_truncated": str(truncated),
+            }
+        )
+    return attributes
+
+
+def get_dependency(tracer, config, wrapped, instance, args, kwargs):
+    """ Wrap nameko.events.EventDispatcher.get_dependency.
+
+    Creates a PRODUCER span around the dispatch of the message, including all the
+    AMQP publisher attributes.
+    """
+    dispatcher = instance
+    (worker_ctx,) = args
+
+    def wrapped_dispatch(wrapped, instance, args, kwargs):
+        event_type, event_data = args
+
+        attributes = collect_client_attributes(
+            config,
+            dispatcher.exchange.name,
+            event_type,
+            event_data,
+            dispatcher.publisher,
+            kwargs,
+        )
+
+        with tracer.start_as_current_span(
+            f"Dispatch event {worker_ctx.service_name}.{event_type}",
+            attributes=attributes,
+            kind=trace.SpanKind.PRODUCER,
+        ):
+            inject(worker_ctx.context_data)
+            return wrapped(*args, **kwargs)
+
+    dispatch = wrapped(*args, **kwargs)
+    return FunctionWrapper(dispatch, wrapped_dispatch)
+
+
+def event_dispatcher(
+    tracer, config, wrapped, instance, args, kwargs
+):  # pragma: no cover -- call_function_get_frame messes up coverage collection
+    """ Wrap nameko.standalone.events.event_dispatcher.
+
+    Creates a PRODUCER span around the dispatch of the message, including all the
+    AMQP publisher attributes.
+    """
+    headers = kwargs.get("headers", {})
+    kwargs["headers"] = headers
+    frame, dispatch = call_function_get_frame(wrapped, *args, **kwargs)
+
+    # egregious hack: publisher is instantiated inside event_dispatcher function
+    # and only available in its locals
+    publisher = frame.f_locals["publisher"]
+
+    def wrapped_dispatch(wrapped, instance, args, kwargs):
+        service_name, event_type, event_data = args
+
+        exchange = get_event_exchange(service_name)
+
+        attributes = collect_client_attributes(
+            config, exchange.name, event_type, event_data, publisher, kwargs,
+        )
+
+        with tracer.start_as_current_span(
+            f"Dispatch event {service_name}.{event_type}",
+            attributes=attributes,
+            kind=trace.SpanKind.PRODUCER,
+        ):
+            inject(headers)
+            return wrapped(*args, **kwargs)
+
+    return FunctionWrapper(dispatch, wrapped_dispatch)
+
+
+def instrument(tracer, config):
+    wrap_function_wrapper(
+        "nameko.events",
+        "EventDispatcher.get_dependency",
+        partial(get_dependency, tracer, config),
+    )
+
+    wrap_function_wrapper(
+        "nameko.standalone.events",
+        "event_dispatcher",
+        partial(event_dispatcher, tracer, config),
+    )
+
+
+def uninstrument():
+    unwrap(nameko.events.EventDispatcher, "get_dependency")
+    unwrap(nameko.standalone.events, "event_dispatcher")
diff --git a/nameko_opentelemetry/http.py b/nameko_opentelemetry/http.py
new file mode 100644
index 0000000..475d3a0
--- /dev/null
+++ b/nameko_opentelemetry/http.py
@@ -0,0 +1,158 @@
+# -*- coding: utf-8 -*-
+import inspect
+from functools import partial
+
+import nameko.web.server
+from nameko.web.handlers import Response
+from nameko.web.server import HTTPException
+from opentelemetry import trace
+from opentelemetry.instrumentation import wsgi
+from opentelemetry.instrumentation.utils import http_status_to_status_code, unwrap
+from opentelemetry.semconv.trace import SpanAttributes
+from opentelemetry.trace.status import Status, StatusCode
+from wrapt import wrap_function_wrapper
+
+from nameko_opentelemetry import utils
+from nameko_opentelemetry.entrypoints import EntrypointAdapter
+from nameko_opentelemetry.scrubbers import scrub
+
+
+class HttpEntrypointAdapter(EntrypointAdapter):
+    """ Implemented according to
+    https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/semantic_conventions/http.md
+    """
+
+    def get_metadata(self):
+        # TODO: why doesn't http entrypoint populate context data?
+        # alternatively, why do AMQP (and gRPC?) extensions feel they can turn
+        # _all_ amqp headers into context data?
+        return self.worker_ctx.args[0].headers
+
+    def get_span_name(self):
+        return self.worker_ctx.entrypoint.url
+
+    @property
+    def request(self):
+        worker_ctx = self.worker_ctx
+        entrypoint = worker_ctx.entrypoint
+
+        method = getattr(entrypoint.container.service_cls, entrypoint.method_name)
+        call_args = inspect.getcallargs(
+            method, None, *worker_ctx.args, **worker_ctx.kwargs
+        )
+
+        return call_args.get("request")
+
+    def get_attributes(self):
+
+        attributes = super().get_attributes()
+        attributes.pop("call_args", None)
+        attributes.pop("call_args_redacted", None)
+        attributes.pop("call_args_truncated", None)
+
+        request = self.request
+        data = request.data or request.form
+
+        attributes.update(wsgi.collect_request_attributes(request.environ))
+
+        if self.config.get("send_headers"):
+            headers = {}
+            for key, value in request.environ.items():
+                key = str(key)
+                if key.startswith("HTTP_") and key not in (
+                    "HTTP_CONTENT_TYPE",
+                    "HTTP_CONTENT_LENGTH",
+                ):
+                    headers[key[5:].lower()] = str(value)
+                elif key in ("CONTENT_TYPE", "CONTENT_LENGTH"):
+                    headers[key.lower()] = str(value)
+
+            attributes.update(
+                {
+                    "request.headers": utils.serialise_to_string(
+                        scrub(headers, self.config)
+                    )
+                }
+            )
+
+        if self.config.get("send_request_payloads"):
+            attributes.update(
+                {"request.data": utils.serialise_to_string(scrub(data, self.config))}
+            )
+
+        return attributes
+
+    def get_result_attributes(self, result):
+        """ Return serialisable result data
+        """
+        if not isinstance(result, Response):
+            if isinstance(result, tuple):
+                if len(result) == 3:
+                    status, headers, payload = result
+                else:
+                    status, payload = result
+                    headers = {}
+            else:
+                payload = result
+                status = 200
+                headers = {}
+
+            result = Response(payload, headers=headers.items(), status=status)
+
+        attributes = {
+            "response.content_type": result.content_type,
+            SpanAttributes.HTTP_RESPONSE_CONTENT_LENGTH: result.content_length,
+            SpanAttributes.HTTP_STATUS_CODE: result.status_code,
+        }
+
+        if self.config.get("send_response_payloads"):
+
+            response, truncated = utils.truncate(
+                scrub(result.get_data(), self.config),
+                max_len=self.config.get("truncate_max_length"),
+            )
+            attributes.update(
+                {"response.data": response, "response.data_truncated": str(truncated)}
+            )
+
+        return attributes
+
+    def get_status(self, result, exc_info):
+        if exc_info:
+            exc_type, exc, _ = exc_info
+
+            return Status(
+                StatusCode.ERROR, description="{}: {}".format(type(exc).__name__, exc),
+            )
+
+        result_attributes = self.get_result_attributes(result)
+        return Status(
+            http_status_to_status_code(
+                result_attributes[SpanAttributes.HTTP_STATUS_CODE]
+            )
+        )
+
+
+def wsgi_app_call(tracer, config, wrapped, instance, args, kwargs):
+    environ, start_response = args
+    try:
+        instance.url_map.bind_to_environ(environ).match()
+    except HTTPException as exc:
+        span = tracer.start_span(
+            wsgi.get_default_span_name(environ), kind=trace.SpanKind.SERVER
+        )
+        span.set_status(Status(StatusCode.ERROR, description=str(exc)))
+        with trace.use_span(span, end_on_exit=True):
+            return wrapped(*args, **kwargs)
+    else:
+        return wrapped(*args, **kwargs)
+
+
+def instrument(tracer, config):
+    wrap_function_wrapper(
+        "nameko.web.server", "WsgiApp.__call__", partial(wsgi_app_call, tracer, config),
+    )
+
+
+def uninstrument():
+    unwrap(nameko.web.server.WsgiApp, "__call__")
diff --git a/nameko_opentelemetry/messaging.py b/nameko_opentelemetry/messaging.py
new file mode 100644
index 0000000..4e498dd
--- /dev/null
+++ b/nameko_opentelemetry/messaging.py
@@ -0,0 +1,91 @@
+# -*- coding: utf-8 -*-
+""" This modules applies patches to capture spans when messages are published by
+the Publisher dependency provider.
+
+The entrypoint adapter for consumer entrypoint is defined here too.
+"""
+from functools import partial
+
+import nameko.messaging
+from opentelemetry import trace
+from opentelemetry.instrumentation.utils import unwrap
+from opentelemetry.propagate import inject
+from wrapt import FunctionWrapper, wrap_function_wrapper
+
+from nameko_opentelemetry.amqp import amqp_consumer_attributes
+from nameko_opentelemetry.entrypoints import EntrypointAdapter
+from nameko_opentelemetry.scrubbers import scrub
+from nameko_opentelemetry.utils import serialise_to_string, truncate
+
+
+class ConsumerEntrypointAdapter(EntrypointAdapter):
+    """ Adapter customisation for Consumer entrypoints.
+    """
+
+    span_kind = trace.SpanKind.CONSUMER
+
+    def get_attributes(self):
+        """ Include configuration of the entrypoint, and AMQP consumer attributes.
+        """
+        attrs = super().get_attributes()
+
+        entrypoint = self.worker_ctx.entrypoint
+
+        attrs.update(
+            {"nameko.messaging.requeue_on_error": str(entrypoint.requeue_on_error)}
+        )
+
+        consumer = self.worker_ctx.entrypoint.consumer
+        attrs.update(amqp_consumer_attributes(consumer))
+        return attrs
+
+
+def get_dependency(tracer, config, wrapped, instance, args, kwargs):
+    """ Wrap nameko.events.Consumer.get_dependency.
+
+    Creates a PRODUCER span around the publish of the message, including all the
+    AMQP publisher attributes.
+    """
+
+    (worker_ctx,) = args
+    exchange = instance.exchange
+
+    def wrapped_publish(wrapped, instance, args, kwargs):
+        (msg,) = args
+
+        target = exchange and exchange.name or "default-exchange"
+
+        attributes = {"nameko.messaging.exchange": target}
+        if config.get("send_request_payloads"):
+
+            data, truncated = truncate(
+                serialise_to_string(scrub(msg, config)),
+                max_len=config.get("truncate_max_length"),
+            )
+            attributes.update(
+                {
+                    "nameko.messaging.payload": data,
+                    "nameko.messaging.payload_truncated": str(truncated),
+                }
+            )
+
+        with tracer.start_as_current_span(
+            f"Publish to {target}", attributes=attributes, kind=trace.SpanKind.PRODUCER,
+        ):
+            inject(worker_ctx.context_data)
+            return wrapped(*args, **kwargs)
+
+    publish = wrapped(*args, **kwargs)
+    return FunctionWrapper(publish, wrapped_publish)
+
+
+def instrument(tracer, config):
+    wrap_function_wrapper(
+        "nameko.messaging",
+        "Publisher.get_dependency",
+        partial(get_dependency, tracer, config),
+    )
+
+
+def uninstrument():
+    unwrap(nameko.messaging.Publisher, "get_dependency")
diff --git a/nameko_opentelemetry/package.py b/nameko_opentelemetry/package.py
new file mode 100644
index 0000000..6186f31
--- /dev/null
+++ b/nameko_opentelemetry/package.py
@@ -0,0 +1,2 @@
+# -*- coding: utf-8 -*-
+_instruments = ("nameko >= 3.0.0rc1",)
diff --git a/nameko_opentelemetry/rpc.py b/nameko_opentelemetry/rpc.py
new file mode 100644
index 0000000..71ab1d0
--- /dev/null
+++ b/nameko_opentelemetry/rpc.py
@@ -0,0 +1,157 @@
+# -*- coding: utf-8 -*-
+""" This modules applies patches to capture spans when RPC messages are sent by the
+clients (both the dependency provider and the standalone client).
+
+There are also patches that handle cases where RPC messages are received, but no
+entrypoint fires, and therefore the normal entrypoint instrumentation won't apply.
+
+The entrypoint adapter for RPC entrypoints is defined here too.
+"""
+from functools import partial
+from weakref import WeakKeyDictionary
+
+import nameko.rpc
+from nameko.exceptions import IncorrectSignature, MethodNotFound
+from opentelemetry import trace
+from opentelemetry.instrumentation.utils import unwrap
+from opentelemetry.propagate import inject
+from opentelemetry.trace.status import Status, StatusCode
+from opentelemetry.util._time import _time_ns
+from wrapt import wrap_function_wrapper
+
+from nameko_opentelemetry.amqp import amqp_consumer_attributes
+from nameko_opentelemetry.entrypoints import EntrypointAdapter
+
+
+publishers = {}
+active_spans = WeakKeyDictionary()
+
+
+class RpcEntrypointAdapter(EntrypointAdapter):
+    """ Adapter customisation for RPC entrypoints.
+    """
+
+    def get_attributes(self):
+        """ Include AMQP consumer attributes
+        """
+        attributes = super().get_attributes()
+
+        consumer = self.worker_ctx.entrypoint.rpc_consumer.consumer
+        attributes.update(amqp_consumer_attributes(consumer))
+        return attributes
+
+
+def initiate_call(tracer, config, wrapped, instance, args, kwargs):
+    """ Wrap nameko.rpc.Client._call, so that we can start a span when an RPC call
+    is initiated. This code path is active in the nameko.rpc.ClusterRpc and
+    nameko.rpc.ServiceRpc dependency providers, as well as the standalone client
+    defined in nameko.rpc.standalone.
+    """
+    client = instance
+
+    attributes = {
+        "nameko.rpc.target_service": client.service_name,
+        "nameko.rpc.target_method": client.method_name,
+        # XXX send payload? probably should, for consistency w/ everything else
+    }
+
+    span = tracer.start_span(
+        kind=trace.SpanKind.CLIENT,
+        name=f"RPC to {instance.identifier}",
+        attributes=attributes,
+        start_time=_time_ns(),
+    )
+    activation = trace.use_span(span)
+    activation.__enter__()
+
+    inject(instance.context_data)
+
+    rpc_call = wrapped(*args, **kwargs)
+
+    span.set_attributes({"nameko.rpc.correlation_id": rpc_call.correlation_id})
+
+    active_spans[rpc_call] = (activation, span)
+
+    return rpc_call
+
+
+def get_response(tracer, config, wrapped, instance, args, kwargs):
+    """ Wrap nameko.rpc.Client._call, so that we can terminate the active span.
+
+    This code path is active in the nameko.rpc.ClusterRpc and nameko.rpc.ServiceRpc
+    dependency providers, as well as the standalone client defined in
+    nameko.rpc.standalone.
+    """
+    resp = wrapped(*args, **kwargs)
+    activation, span = active_spans[instance]
+    activation.__exit__(None, None, None)
+    span.end(_time_ns())
+    return resp
+
+
+def consumer_handle_message(tracer, config, wrapped, instance, args, kwargs):
+    """ Wrap nameko.rpc.RpcConsumer.handle_message.
+
+    In the case where an RPC message is received for a method that doesn't exist,
+    no entrypoint will fire and so the entrypoint instrumentation won't
+    generate a span. We generate one here instead.
+
+    TODO make sure to add server attributes.
+    """
+    body, message = args
+    routing_key = message.delivery_info["routing_key"]
+    try:
+        instance.get_provider_for_method(routing_key)
+    except MethodNotFound as exc:
+        span = tracer.start_span(routing_key, kind=trace.SpanKind.SERVER)
+        span.set_status(Status(StatusCode.ERROR, description=f"MethodNotFound: {exc}"))
+        with trace.use_span(span, end_on_exit=True):
+            return wrapped(*args, **kwargs)
+    else:
+        return wrapped(*args, **kwargs)
+
+
+def entrypoint_handle_message(tracer, config, wrapped, instance, args, kwargs):
+    """ Wrap nameko.rpc.Rpc.handle_message.
+
+    In the case where an RPC message is received for a valid method, but with an
+    invalid signature, no entrypoint will fire and so the entrypoint instrumentation
+    won't generate a span. We generate one here instead.
+    """
+    body, message = args
+    try:
+        instance.check_signature(body.get("args"), body.get("kwargs"))
+    except IncorrectSignature as exc:
+        name = f"{instance.container.service_name}.{instance.method_name}"
+        span = tracer.start_span(name, kind=trace.SpanKind.SERVER)
+        span.set_status(Status(StatusCode.ERROR, description=f"MethodNotFound: {exc}"))
+        with trace.use_span(span, end_on_exit=True):
+            return wrapped(*args, **kwargs)
+    else:
+        return wrapped(*args, **kwargs)
+
+
+def instrument(tracer, config):
+    wrap_function_wrapper(
+        "nameko.rpc", "Client._call", partial(initiate_call, tracer, config)
+    )
+    wrap_function_wrapper(
+        "nameko.rpc", "RpcCall.get_response", partial(get_response, tracer, config)
+    )
+    wrap_function_wrapper(
+        "nameko.rpc",
+        "RpcConsumer.handle_message",
+        partial(consumer_handle_message, tracer, config),
+    )
+    wrap_function_wrapper(
+        "nameko.rpc",
+        "Rpc.handle_message",
+        partial(entrypoint_handle_message, tracer, config),
+    )
+
+
+def uninstrument():
+    unwrap(nameko.rpc.Client, "_call")
+    unwrap(nameko.rpc.RpcCall, "get_response")
+    unwrap(nameko.rpc.RpcConsumer, "handle_message")
+    unwrap(nameko.rpc.Rpc, "handle_message")
diff --git a/nameko_opentelemetry/scrubbers.py b/nameko_opentelemetry/scrubbers.py
new file mode 100644
index 0000000..5940f36
--- /dev/null
+++ b/nameko_opentelemetry/scrubbers.py
@@ -0,0 +1,113 @@
+# -*- coding: utf-8 -*-
+import collections
+import re
+
+from nameko_opentelemetry.utils import import_by_path
+
+
+SCRUBBED = "scrubbed"
+
+
+def scrubbers(config):
+    for scrubber_path in config.get(
+        "scrubbers", ["nameko_opentelemetry.scrubbers.DefaultScrubber"]
+    ):
+        scrubber_cls = import_by_path(scrubber_path)
+        yield scrubber_cls(config)
+
+
+def scrub(data, config):
+    for scrubber in scrubbers(config):
+        data = scrubber.scrub(data)
+    return data
+
+
+class DefaultScrubber:
+
+    SENSITIVE_KEYS = (
+        "token",
+        "password",
+        "jwt",
+        "auth",
+        "password",
+        "secret",
+        "passwd",
+        "api_key",
+        "apikey",
+        "access_token",
+        "auth",
+        "credentials",
+        "mysql_pwd",
+        "stripeToken",
+    )
+
+    COMMON_KEY_PREFIXES = ("nameko.", "x-")
+
+    SENSITIVE_VALUES = (
+        re.compile(r"[a-z0-9._%\+\-—|]+@[a-z0-9.\-—|]+\.[a-z|]{2,6}"),  # email address
+    )
+
+    REPLACEMENT = SCRUBBED
+
+    def __init__(self, config):
+        self.config = config
+
+    def sensitive_key(self, key):
+        for prefix in self.COMMON_KEY_PREFIXES:
+            if key.startswith(prefix):
+                chop = len(prefix)
+                key = key[chop:]
+        return key in self.SENSITIVE_KEYS
+
+    def sensitive_value(self, value):
+        for regex in self.SENSITIVE_VALUES:  # pragma: no cover (branch to exit)
+            return regex.match(value)
+
+    def scrub(self, data):
+        """ `data` can be a dict, an iterable, or a scalar value.
+
+        Returns a scubbed version and leaves original unchanged.
+        """
+        if isinstance(data, dict):
+            data = data.copy()
+            replace_keys = {}
+            for key, value in data.items():
+                if self.sensitive_key(key):
+                    value = self.REPLACEMENT
+                else:
+                    value = self.scrub(value)
+
+                if self.sensitive_value(key):
+                    replace_keys[key] = value
+                else:
+                    data[key] = value
+
+            for key, value in replace_keys.items():
+                del data[key]
+                data[self.REPLACEMENT] = value
+
+            return data
+
+        if isinstance(data, collections.abc.Iterable) and not isinstance(
+            data, (str, bytes)
+        ):
+            scrubbed = list(map(self.scrub, data))
+            try:
+                return type(data)(scrubbed)
+            except TypeError:
+                return list(scrubbed)
+
+        if isinstance(data, bytes):
+            try:
+                decoded = data.decode("utf-8")
+            except UnicodeDecodeError:
+                pass
+            else:
+                if self.sensitive_value(decoded):
+                    return self.REPLACEMENT.encode("utf-8")
+
+        if isinstance(data, str):
+            if self.sensitive_value(str(data)):
+                return self.REPLACEMENT
+
+        return data
diff --git a/nameko_opentelemetry/timer.py b/nameko_opentelemetry/timer.py
new file mode 100644
index 0000000..87e2cc6
--- /dev/null
+++ b/nameko_opentelemetry/timer.py
@@ -0,0 +1,18 @@
+# -*- coding: utf-8 -*-
+from nameko_opentelemetry.entrypoints import EntrypointAdapter
+
+
+class TimerEntrypointAdapter(EntrypointAdapter):
+    """ Adapter customisation for Timer entrypoints.
+    """
+
+    def get_attributes(self):
+        attrs = super().get_attributes()
+
+        attrs.update(
+            {
+                "nameko.timer.interval": self.worker_ctx.entrypoint.interval,
+                "nameko.timer.eager": self.worker_ctx.entrypoint.eager,
+            }
+        )
+        return attrs
diff --git a/nameko_opentelemetry/utils.py b/nameko_opentelemetry/utils.py
new file mode 100644
index 0000000..750559b
--- /dev/null
+++ b/nameko_opentelemetry/utils.py
@@ -0,0 +1,106 @@
+# -*- coding: utf-8 -*-
+import collections.abc
+import json
+import logging
+import sys
+from importlib import import_module
+
+import six
+
+
+TRUNCATE_MAX_LENGTH = 200
+
+logger = logging.getLogger(__name__)
+
+
+def serialise_to_json(value):
+    """ Safely serialise ``value`` to JSON formatted string
+    """
+    return json.dumps(safe_for_serialisation(value))
+
+
+def serialise_to_string(value):
+    """ Safely serialise ``value`` to string representation
+    """
+    return str(safe_for_serialisation(value))
+
+
+def safe_for_serialisation(value):
+    no_op_types = six.string_types + six.integer_types + (float, type(None))
+    if isinstance(value, no_op_types):
+        return value
+    if isinstance(value, bytes):
+        return value.decode("utf-8", "ignore")
+    if isinstance(value, dict):
+        return {
+            safe_for_serialisation(key): safe_for_serialisation(val)
+            for key, val in six.iteritems(value)
+        }
+    if isinstance(value, collections.abc.Iterable):
+        return list(map(safe_for_serialisation, value))
+    try:
+        return six.text_type(value)
+    except Exception:
+        logger.warning("failed to get string representation", exc_info=True)
+        return "failed to get string representation"
+
+
+def import_by_path(dotted_path):
+    """
+    Import a dotted module path and return the attribute/class designated by
+    the last name in the path. Raise ImportError if the import failed.
+
+    Borrowed from ``django.utils.module_loading.import_string``.
+
+    """
+    try:
+        module_path, class_name = dotted_path.rsplit(".", 1)
+    except ValueError:
+        raise ImportError("{} doesn't look like a module path".format(dotted_path))
+
+    module = import_module(module_path)
+
+    try:
+        return getattr(module, class_name)
+    except AttributeError:
+        raise ImportError(
+            'Module "{}" does not define a "{}" attribute/class'.format(
+                module_path, class_name
+            )
+        )
+
+
+def truncate(value, max_len=None):
+    if max_len is None:
+        max_len = TRUNCATE_MAX_LENGTH
+    if len(value) > max_len:
+        return value[:max_len], True
+    return value, False
+
+
+def call_function_get_frame(
+    func, *args, **kwargs
+):  # pragma: no cover -- cannot be captured by coverage
+    """
+    Calls the function `func` with the specified arguments and keyword
+    arguments and snatches its local frame before it actually executes.
+
+    Borrowed from https://stackoverflow.com/questions/4214936
+    """
+
+    frame = None
+    trace = sys.gettrace()
+
+    def snatch_locals(_frame, name, arg):
+        nonlocal frame
+        if frame is None and name == "call":
+            frame = _frame
+            sys.settrace(trace)
+        return trace
+
+    sys.settrace(snatch_locals)
+    try:
+        result = func(*args, **kwargs)
+    finally:
+        sys.settrace(trace)
+    return frame, result
diff --git a/nameko_opentelemetry/version.py b/nameko_opentelemetry/version.py
new file mode 100644
index 0000000..4742c9c
--- /dev/null
+++ b/nameko_opentelemetry/version.py
@@ -0,0 +1,2 @@
+# -*- coding: utf-8 -*-
+__version__ = "0.1.0"
diff --git a/setup.cfg b/setup.cfg
new file mode 100644
index 0000000..11ce6ed
--- /dev/null
+++ b/setup.cfg
@@ -0,0 +1,26 @@
+[bdist_wheel]
+universal = 1
+
+[flake8]
+exclude =
+    .git,
+    .tox,
+    __pycache__,
+max-line-length = 88
+[isort]
+line_length=88
+known_first_party=nameko_opentelemetry
+known_standard_library=
+indent='    '
+multi_line_output=3
+include_trailing_comma=true
+forced_separate=
+default_section=THIRDPARTY
+lines_after_imports=2
+skip=.tox,.git
+[tool:pytest]
+filterwarnings =
+    ignore::DeprecationWarning:kombu.utils.compat
+    ignore::DeprecationWarning:dns.namedict
+    ignore::DeprecationWarning:dns.hash
+    ignore::DeprecationWarning:nameko.messaging
\ No newline at end of file
diff --git a/setup.py b/setup.py
new file mode 100644
index 0000000..37e1bde
--- /dev/null
+++ b/setup.py
@@ -0,0 +1,46 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+import os
+
+from setuptools import find_packages, setup
+
+
+BASE_DIR = os.path.dirname(__file__)
+PACKAGE_INFO = {}
+
+VERSION_FILENAME = os.path.join(BASE_DIR, "nameko_opentelemetry", "version.py")
+with open(VERSION_FILENAME) as f:
+    exec(f.read(), PACKAGE_INFO)
+
+PACKAGE_FILENAME = os.path.join(BASE_DIR, "nameko_opentelemetry", "package.py")
+with open(PACKAGE_FILENAME) as f:
+    exec(f.read(), PACKAGE_INFO)
+
+
+setup(
+    name="opentelemetry-instrumentation-nameko",
+    description="Nameko extension producing opentelemetry data",
+    version=PACKAGE_INFO["__version__"],
+    author="Nameko Authors",
+    url="https://github.com/nameko/opentelemetry-instrumentation-nameko",
+    packages=find_packages(exclude=["test", "test.*"]),
+    install_requires=[
+        "nameko==3.0.0rc9",
+        "opentelemetry-api",
+        "opentelemetry-instrumentation",
+        "opentelemetry-instrumentation-wsgi",
+    ],
+    extras_require={
+        "dev": list(PACKAGE_INFO["_instruments"])
+        + [
+            "coverage",
+            "pytest",
+            "opentelemetry-sdk",
+            "opentelemetry-instrumentation-requests",
+        ],
+        "instruments": PACKAGE_INFO["_instruments"],
+    },
+    dependency_links=[],
+    zip_safe=True,
+    license="Apache License, Version 2.0",
+)
diff --git a/tox.ini b/tox.ini
new file mode 100644
index 0000000..82771bb
--- /dev/null
+++ b/tox.ini
@@ -0,0 +1,14 @@
+[tox]
+envlist = static, {py3.6,py3.7,py3.8,py3.9}-test
+skipsdist = True
+
+[testenv]
+allowlist_externals = make
+
+commands =
+    static: pip install pre-commit
+    static: make static
+
+    test: pip install --pre nameko
+    test: pip install --editable .[dev]
+    test: make coverage


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..da6723b
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,64 @@
+# -*- coding: utf-8 -*-
+import pytest
+from opentelemetry import trace
+from opentelemetry.instrumentation.utils import unwrap
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+from wrapt import wrap_function_wrapper
+
+from nameko_opentelemetry import NamekoInstrumentor
+
+
+@pytest.fixture(scope="session")
+def filter_spans():
+    def filter_rabbit(span):
+        if "api/vhosts/nameko_test" not in span.attributes.get(
+            "http.url", ""
+        ):  # pragma: no cover (don't cover false branch)
+            return True
+
+    def get_finished_spans(wrapped, instance, args, kwargs):
+        spans = wrapped(*args, **kwargs)
+        return list(filter(filter_rabbit, spans))
+
+    wrap_function_wrapper(
+        "opentelemetry.sdk.trace.export.in_memory_span_exporter",
+        "InMemorySpanExporter.get_finished_spans",
+        get_finished_spans,
+    )
+    yield
+    unwrap(InMemorySpanExporter, "get_finished_spans")
+
+
+@pytest.fixture(scope="session")
+def memory_exporter(filter_spans):
+    return InMemorySpanExporter()
+
+
+@pytest.fixture(scope="session")
+def trace_provider(memory_exporter):
+    provider = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(memory_exporter))
+    trace.set_tracer_provider(provider)
+
+
+@pytest.fixture
+def config():
+    return {
+        "send_headers": True,
+        "send_request_payloads": True,
+        "send_response_payloads": True,
+        "send_context_data": True,  # XXX?
+        "truncate_max_length": 200,
+    }
+
+
+@pytest.fixture(autouse=True)
+def instrument(trace_provider, memory_exporter, config):
+    instrumentor = NamekoInstrumentor()
+
+    instrumentor.instrument(**config)
+    yield
+    memory_exporter.clear()
+    instrumentor.uninstrument()
diff --git a/tests/test_amqp.py b/tests/test_amqp.py
new file mode 100644
index 0000000..e2b3734
--- /dev/null
+++ b/tests/test_amqp.py
@@ -0,0 +1,212 @@
+# -*- coding: utf-8 -*-
+# test add to active span
+# test no active span
+
+import ast
+
+import nameko
+import pytest
+from nameko.amqp.consume import Consumer
+from nameko.amqp.publish import Publisher
+
+from nameko_opentelemetry import active_tracer
+from nameko_opentelemetry.amqp import amqp_consumer_attributes
+
+
+class TestAddToActiveSpan:
+    @pytest.fixture
+    def publisher(self, rabbit_config):
+        return Publisher(amqp_uri=nameko.config["AMQP_URI"], use_confirms=True)
+
+    def test_no_active_span(self, publisher, memory_exporter):
+        publisher.publish("msg")
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 0
+
+    def test_active_span(self, publisher, rabbit_config, memory_exporter):
+        with active_tracer().start_as_current_span("foobar", attributes={"foo": "bar"}):
+            publisher.publish("msg")
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        assert spans[0].name == "foobar"
+        assert spans[0].attributes["nameko.amqp.amqp_uri"] == nameko.config["AMQP_URI"]
+
+
+class TestPublisherAttributes:
+    @pytest.fixture
+    def publisher(self, rabbit_config):
+        return Publisher(
+            amqp_uri=nameko.config["AMQP_URI"],
+            routing_key="foo",
+            headers={
+                "header-1": "value-1",
+                "header-2": "value-2",
+                "header-3": "value-3",
+            },
+        )
+
+    def test_defaults(self, publisher, memory_exporter):
+        with active_tracer().start_as_current_span("publish"):
+            publisher.publish("msg")
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+        assert attributes["nameko.amqp.amqp_uri"] == nameko.config["AMQP_URI"]
+        assert attributes["nameko.amqp.use_confirms"] == "True"
+
+        # just check the other keys all exist
+        for attribute in (
+            "nameko.amqp.amqp_uri",
+            "nameko.amqp.ssl",
+            "nameko.amqp.use_confirms",
+            "nameko.amqp.delivery_mode",
+            "nameko.amqp.mandatory",
+            "nameko.amqp.priority",
+            "nameko.amqp.expiration",
+            "nameko.amqp.serializer",
+            "nameko.amqp.compression",
+            "nameko.amqp.retry",
+            "nameko.amqp.retry_policy",
+            "nameko.amqp.declarations",
+            "nameko.amqp.transport_options",
+            "nameko.amqp.publish_kwargs",
+        ):
+            assert attribute in attributes
+
+    def test_publish_time_overrides(self, publisher, memory_exporter):
+        with active_tracer().start_as_current_span("publish"):
+            publisher.publish("msg", use_confirms=False)
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+        assert attributes["nameko.amqp.use_confirms"] == "False"
+
+    def test_routing_key_from_publisher(self, publisher, memory_exporter):
+        with active_tracer().start_as_current_span("publish"):
+            publisher.publish("msg")
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+        assert attributes["nameko.amqp.routing_key"] == "foo"
+
+    def test_routing_key_from_publish_time(self, publisher, memory_exporter):
+        with active_tracer().start_as_current_span("publish"):
+            publisher.publish("msg", routing_key="bar")
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+        assert attributes["nameko.amqp.routing_key"] == "bar"
+
+    def test_headers_from_publisher(self, publisher, memory_exporter):
+        with active_tracer().start_as_current_span("publish"):
+            publisher.publish("msg")
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+        headers = ast.literal_eval(attributes["nameko.amqp.headers"])
+        assert headers == {
+            "header-1": "value-1",
+            "header-2": "value-2",
+            "header-3": "value-3",
+        }
+
+    def test_headers_from_publish_time(self, publisher, memory_exporter):
+        with active_tracer().start_as_current_span("publish"):
+            publisher.publish(
+                "msg", headers={"header-3": "REPLACED", "header-4": "value-4"}
+            )
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+        headers = ast.literal_eval(attributes["nameko.amqp.headers"])
+        assert headers == {
+            "header-1": "value-1",
+            "header-2": "value-2",
+            "header-3": "REPLACED",
+            "header-4": "value-4",
+        }
+
+    def test_extra_headers_from_publish_time(self, publisher, memory_exporter):
+        with active_tracer().start_as_current_span("publish"):
+            publisher.publish(
+                "msg",
+                headers={"header-4": "value-4"},
+                extra_headers={"header-3": "REPLACED", "header-4": "REPLACED"},
+            )
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+        headers = ast.literal_eval(attributes["nameko.amqp.headers"])
+        assert headers == {
+            "header-1": "value-1",
+            "header-2": "value-2",
+            "header-3": "REPLACED",
+            "header-4": "REPLACED",
+        }
+
+
+class TestSendHeaders:
+    @pytest.fixture
+    def publisher(self, rabbit_config):
+        return Publisher(amqp_uri=nameko.config["AMQP_URI"])
+
+    @pytest.fixture(params=[True, False], ids=["send_headers", "no_send_headers"])
+    def send_headers(self, request):
+        return request.param
+
+    @pytest.fixture
+    def config(self, config, send_headers):
+        config["send_headers"] = send_headers
+        return config
+
+    def test_send_headers_toggle(self, send_headers, publisher, memory_exporter):
+        with active_tracer().start_as_current_span("publish"):
+            publisher.publish("msg")
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+        if send_headers:
+            assert "nameko.amqp.headers" in attributes
+        else:
+            assert "nameko.amqp.headers" not in attributes
+
+
+class TestConsumerAttributes:
+    @pytest.fixture
+    def consumer(self, rabbit_config):
+        return Consumer(nameko.config["AMQP_URI"])
+
+    def test_defaults(self, consumer):
+
+        attributes = amqp_consumer_attributes(consumer)
+        assert attributes["nameko.amqp.amqp_uri"] == nameko.config["AMQP_URI"]
+
+        # just check the other keys all exist
+        for attribute in (
+            "nameko.amqp.ssl",
+            "nameko.amqp.prefetch_count",
+            "nameko.amqp.heartbeat",
+            "nameko.amqp.accept",
+            "nameko.amqp.queues",
+            "nameko.amqp.consumer_options",
+        ):
+            assert attribute in attributes
diff --git a/tests/test_entrypoints.py b/tests/test_entrypoints.py
new file mode 100644
index 0000000..f743167
--- /dev/null
+++ b/tests/test_entrypoints.py
@@ -0,0 +1,646 @@
+# -*- coding: utf-8 -*-
+import socket
+import uuid
+from unittest.mock import Mock, patch
+
+import pytest
+from nameko.extensions import DependencyProvider
+from nameko.rpc import ServiceRpc, rpc
+from nameko.standalone.rpc import ServiceRpcClient
+from nameko.testing.services import dummy, entrypoint_hook, entrypoint_waiter
+from nameko.testing.utils import get_extension
+from nameko.utils import REDACTED
+from opentelemetry import trace
+from opentelemetry.trace import SpanKind
+from opentelemetry.trace.status import StatusCode
+
+from nameko_opentelemetry.entrypoints import EntrypointAdapter
+
+
+class TestWrappedMethods:
+    @pytest.fixture
+    def track_worker_setup(self):
+        return Mock()
+
+    @pytest.fixture
+    def track_worker_result(self):
+        return Mock()
+
+    @pytest.fixture
+    def container(self, container_factory, track_worker_setup, track_worker_result):
+        class Tracker(DependencyProvider):
+            def worker_setup(self, worker_ctx):
+                track_worker_setup(worker_ctx)
+
+            def worker_result(self, worker_ctx, result, exc_info):
+                track_worker_result(worker_ctx)
+
+        class Service:
+            name = "service"
+
+            tracker = Tracker()
+
+            @rpc
+            def method(self, arg, kwarg=None):
+                return "OK"
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    def test_wrapped_methods_called(
+        self, container, memory_exporter, track_worker_setup, track_worker_result
+    ):
+
+        with entrypoint_hook(container, "method") as hook:
+            assert hook("arg", kwarg="kwarg") == "OK"
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        assert len(track_worker_setup.call_args_list) == 1
+        assert len(track_worker_result.call_args_list) == 1
+
+
+class TestSpanAttributes:
+    @pytest.fixture(params=[True, False], ids=["send_headers", "no_send_headers"])
+    def send_headers(self, request):
+        return request.param
+
+    @pytest.fixture
+    def config(self, config, send_headers):
+        # disable headers based on param
+        config["send_headers"] = send_headers
+        return config
+
+    @pytest.fixture
+    def container(self, container_factory):
+        class Service:
+            name = "service"
+
+            @rpc
+            def method(self, arg, kwarg=None):
+                return "OK"
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    def test_common(self, container, memory_exporter, send_headers):
+
+        with entrypoint_hook(container, "method") as hook:
+            assert hook("arg", kwarg="kwarg") == "OK"
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        assert spans[0].name == "service.method"
+
+        attributes = spans[0].attributes
+        assert attributes["hostname"] == socket.gethostname()
+        assert attributes["service_name"] == "service"
+        assert attributes["entrypoint_type"] == "Rpc"
+        assert attributes["method_name"] == "method"
+        assert attributes["active_workers"] == 1
+        assert attributes["available_workers"] == 9
+
+        if send_headers:
+            assert "call_id_stack" in attributes["context_data"]
+        else:
+            assert "call_id_stack" not in attributes
+
+
+class TestNoTracer:
+    @pytest.fixture
+    def container(self, container_factory):
+        class Service:
+            name = "service"
+
+            @rpc
+            def method(self, arg, kwarg=None):
+                return "OK"
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @pytest.fixture
+    def trace_provider(self):
+        """ Temporarily replace the configured trace provider with the default
+        provider that would be used if no SDK was in use.
+        """
+        with patch("nameko_opentelemetry.trace") as patched:
+            patched.get_tracer.return_value = trace.get_tracer(
+                __name__, "", trace._DefaultTracerProvider()
+            )
+            yield
+
+    def test_not_recording(self, trace_provider, container, memory_exporter):
+        with entrypoint_hook(container, "method") as hook:
+            assert hook("arg", kwarg="kwarg") == "OK"
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 0
+
+
+class CustomAdapter(EntrypointAdapter):
+    def get_span_name(self):
+        return "custom_span_name"
+
+
+class TestCustomAdapter:
+    @pytest.fixture
+    def container(self, container_factory):
+        class Service:
+            name = "service"
+
+            @rpc
+            def method(self, arg, kwarg=None):
+                return "OK"
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @pytest.fixture
+    def config(self, config):
+        config["entrypoint_adapters"] = {
+            "nameko.rpc.Rpc": "test_entrypoints.CustomAdapter"
+        }
+        return config
+
+    def test_custom_adapter(self, container, memory_exporter):
+
+        with entrypoint_hook(container, "method") as hook:
+            assert hook("arg", kwarg="kwarg") == "OK"
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        assert spans[0].name == "custom_span_name"
+
+
+class TestResultAttributes:
+    @pytest.fixture(
+        params=[True, False], ids=["send_response_payloads", "no_response_payloads"]
+    )
+    def send_response_payloads(self, request):
+        return request.param
+
+    @pytest.fixture
+    def config(self, config, send_response_payloads):
+        # disable headers based on param
+        config["send_response_payloads"] = send_response_payloads
+        return config
+
+    @pytest.fixture
+    def unserializable_object(self):
+        return object()
+
+    @pytest.fixture
+    def container(self, container_factory, unserializable_object):
+        class Service:
+            name = "service"
+
+            class Error(Exception):
+                pass
+
+            @rpc
+            def method(self, arg, kwarg=None):
+                return "OK"
+
+            @dummy
+            def unserializable(self, arg, kwarg=None):
+                return unserializable_object
+
+            @rpc
+            def raises(self, arg, kwarg=None):
+                raise self.Error("boom")
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    def test_simple(self, container, memory_exporter, send_response_payloads):
+
+        with entrypoint_hook(container, "method") as hook:
+            assert hook("arg", kwarg="kwarg") == "OK"
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+        assert attributes["method_name"] == "method"
+
+        if send_response_payloads:
+            assert attributes["result"] == "OK"
+        else:
+            assert "result" not in attributes
+
+    def test_exception(self, container, memory_exporter, send_response_payloads):
+
+        with entrypoint_hook(container, "raises") as hook:
+            with pytest.raises(Exception):
+                hook("arg", kwarg="kwarg")
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+        assert attributes["method_name"] == "raises"
+
+        if send_response_payloads:
+            assert attributes.get("result") is None
+        else:
+            assert "result" not in attributes
+
+    def test_unserializable_result(
+        self, container, memory_exporter, unserializable_object, send_response_payloads
+    ):
+
+        with entrypoint_hook(container, "unserializable") as hook:
+            hook("arg", kwarg="kwarg")
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+
+        if send_response_payloads:
+            assert attributes["method_name"] == "unserializable"
+            assert attributes["result"] == str(unserializable_object)
+        else:
+            assert "result" not in attributes
+
+
+class TestExceptions:
+    @pytest.fixture
+    def container(self, container_factory):
+        class Service:
+            name = "service"
+
+            class Error(Exception):
+                pass
+
+            @rpc
+            def method(self, arg, kwarg=None):
+                return "OK"
+
+            @rpc
+            def raises(self, arg, kwarg=None):
+                raise self.Error("boom")
+
+            @rpc(expected_exceptions=Error)
+            def raises_expected(self, arg, kwarg=None):
+                raise self.Error("boom")
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    def test_success(self, container, memory_exporter):
+
+        with entrypoint_hook(container, "method") as hook:
+            assert hook("arg", kwarg="kwarg") == "OK"
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        span = spans[0]
+        assert len(span.events) == 0
+
+    def test_exception(self, container, memory_exporter):
+
+        with entrypoint_hook(container, "raises") as hook:
+            with pytest.raises(Exception):
+                hook("arg", kwarg="kwarg")
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        span = spans[0]
+        assert len(span.events) == 1
+
+        event = span.events[0]
+        assert event.name == "exception"
+        assert event.attributes["exception.type"] == "Error"
+        assert event.attributes["exception.message"] == "boom"
+        assert 'raise self.Error("boom")' in event.attributes["exception.stacktrace"]
+        assert event.attributes["exception.escaped"] == "True"
+        # extra attributes
+        assert event.attributes["exception.expected"] == "False"
+
+    def test_expected_exception(self, container, memory_exporter):
+
+        with entrypoint_hook(container, "raises_expected") as hook:
+            with pytest.raises(Exception):
+                hook("arg", kwarg="kwarg")
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        span = spans[0]
+        assert len(span.events) == 1
+
+        event = span.events[0]
+        assert event.name == "exception"
+        assert event.attributes["exception.type"] == "Error"
+        assert event.attributes["exception.message"] == "boom"
+        assert 'raise self.Error("boom")' in event.attributes["exception.stacktrace"]
+        assert event.attributes["exception.escaped"] == "True"
+        # extra attributes
+        assert event.attributes["exception.expected"] == "True"
+
+    @patch(
+        "nameko_opentelemetry.entrypoints.format_exception",
+        side_effect=Exception("boom"),
+    )
+    def test_unserializable_exception(
+        self, format_exccepion, container, memory_exporter
+    ):
+
+        with entrypoint_hook(container, "raises_expected") as hook:
+            with pytest.raises(Exception):
+                hook("arg", kwarg="kwarg")
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        span = spans[0]
+        assert len(span.events) == 1
+
+        event = span.events[0]
+        assert event.name == "exception"
+        assert event.attributes["exception.type"] == "Error"
+        assert event.attributes["exception.message"] == "boom"
+        assert (
+            "Exception occurred on stacktrace formatting"
+            in event.attributes["exception.stacktrace"]
+        )
+        assert event.attributes["exception.escaped"] == "True"
+        # extra attributes
+        assert event.attributes["exception.expected"] == "True"
+
+
+class TestCallArgs:
+    @pytest.fixture(
+        params=[True, False], ids=["send_request_payloads", "no_request_payloads"]
+    )
+    def send_request_payloads(self, request):
+        return request.param
+
+    @pytest.fixture
+    def config(self, config, send_request_payloads):
+        # override default truncation length
+        config["truncate_max_length"] = 300
+        # disable call args based on param
+        config["send_request_payloads"] = send_request_payloads
+        return config
+
+    @pytest.fixture
+    def container(self, container_factory):
+        class Service:
+            name = "service"
+
+            @rpc(sensitive_arguments="kwarg.foo")
+            def method(self, arg, kwarg=None):
+                return "OK"
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    def test_call_args(self, container, memory_exporter, send_request_payloads):
+
+        with entrypoint_hook(container, "method") as hook:
+            assert hook("arg", kwarg="kwarg") == "OK"
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+
+        if send_request_payloads:
+            assert attributes["call_args"] == "{'arg': 'arg', 'kwarg': 'kwarg'}"
+        else:
+            assert "call_args" not in attributes
+
+    def test_call_args_truncation(
+        self, container, memory_exporter, send_request_payloads
+    ):
+
+        with entrypoint_hook(container, "method") as hook:
+            assert hook("arg" * 1000) == "OK"
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+
+        if send_request_payloads:
+            assert len(attributes["call_args"]) == 300
+            assert attributes["call_args_truncated"] == "True"
+        else:
+            assert "call_args_truncated" not in attributes
+
+    def test_call_args_redaction(
+        self, container, memory_exporter, send_request_payloads
+    ):
+
+        with entrypoint_hook(container, "method") as hook:
+            assert hook("arg", kwarg={"foo": "FOO", "bar": "BAR"}) == "OK"
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+
+        if send_request_payloads:
+            assert attributes["call_args"] == (
+                f"{{'arg': 'arg', 'kwarg': {{'foo': '{REDACTED}', 'bar': 'BAR'}}}}"
+            )
+            assert attributes["call_args_redacted"] == "True"
+        else:
+            assert "call_args_redacted" not in attributes
+
+
+class TestStatus:
+    @pytest.fixture
+    def container(self, container_factory):
+        class Service:
+            name = "service"
+
+            class Error(Exception):
+                pass
+
+            @rpc
+            def method(self, arg, kwarg=None):
+                return "OK"
+
+            @rpc
+            def raises(self, arg, kwarg=None):
+                raise self.Error("boom")
+
+            @rpc(expected_exceptions=Error)
+            def raises_expected(self, arg, kwarg=None):
+                raise self.Error("boom")
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    def test_success(self, container, memory_exporter):
+
+        with entrypoint_hook(container, "method") as hook:
+            assert hook("arg", kwarg="kwarg") == "OK"
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        span = spans[0]
+        assert span.status.is_ok
+        assert span.status.status_code == StatusCode.OK
+
+    def test_exception(self, container, memory_exporter):
+
+        with entrypoint_hook(container, "raises") as hook:
+            with pytest.raises(Exception):
+                hook("arg", kwarg="kwarg")
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        span = spans[0]
+        assert not span.status.is_ok
+        assert span.status.status_code == StatusCode.ERROR
+        assert span.status.description == "Error: boom"
+
+    def test_expected_exception(self, container, memory_exporter):
+
+        with entrypoint_hook(container, "raises_expected") as hook:
+            with pytest.raises(Exception):
+                hook("arg", kwarg="kwarg")
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        span = spans[0]
+        assert span.status.is_ok
+        assert span.status.status_code == StatusCode.OK
+
+
+class TestPartialSpan:
+    @pytest.fixture
+    def container(self, container_factory):
+        class Service:
+            name = "service"
+
+            @rpc
+            def method(self, arg, kwarg=None):
+                return "OK"
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @patch("nameko_opentelemetry.entrypoints.active_spans")
+    def test_span_not_started(self, active_spans, container, memory_exporter):
+
+        # fake a missing span
+        active_spans.get.return_value = None
+
+        with pytest.warns(UserWarning) as warnings:
+            with entrypoint_hook(container, "method") as hook:
+                assert hook("arg", kwarg="kwarg") == "OK"
+
+        assert "no active span" in str(warnings[0].message)
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 0
+
+
+class TestScrubbing:
+    @pytest.fixture
+    def container(self, container_factory, rabbit_config):
+        class Service:
+            name = "service"
+
+            self_rpc = ServiceRpc("service")
+
+            @rpc
+            def method(self, arg, password=None):
+                return {"token": "should-be-secret"}
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @pytest.fixture(params=["standalone", "dependency_provider"])
+    def client(self, rabbit_config, request, container):
+
+        context_data = {
+            "call_id": f"service.method.{uuid.uuid4()}",
+            "token": "should-be-secret",
+        }
+
+        if request.param == "standalone":
+            with ServiceRpcClient("service", context_data=context_data) as client:
+                yield client
+        if request.param == "dependency_provider":
+            dp = get_extension(container, ServiceRpc)
+            yield dp.get_dependency(Mock(context_data=context_data))
+
+    def test_response_scrubber(self, container, client, memory_exporter):
+
+        with entrypoint_waiter(container, "method"):
+            assert client.method("arg", password="password") == {
+                "token": "should-be-secret"
+            }
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        server_span = list(filter(lambda span: span.kind == SpanKind.SERVER, spans))[0]
+
+        assert server_span.attributes["result"] == "{'token': 'scrubbed'}"
+
+    def test_call_args_scrubber(self, container, client, memory_exporter):
+
+        with entrypoint_waiter(container, "method"):
+            assert client.method("arg", password="password") == {
+                "token": "should-be-secret"
+            }
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        server_span = list(filter(lambda span: span.kind == SpanKind.SERVER, spans))[0]
+
+        assert (
+            server_span.attributes["call_args"]
+            == "{'arg': 'arg', 'password': 'scrubbed'}"
+        )
+
+    def test_context_data_scrubber(self, container, client, memory_exporter):
+
+        with entrypoint_waiter(container, "method"):
+            assert client.method("arg", password="password") == {
+                "token": "should-be-secret"
+            }
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        server_span = list(filter(lambda span: span.kind == SpanKind.SERVER, spans))[0]
+
+        assert "'token': 'scrubbed'" in server_span.attributes["context_data"]
diff --git a/tests/test_events.py b/tests/test_events.py
new file mode 100644
index 0000000..c84d629
--- /dev/null
+++ b/tests/test_events.py
@@ -0,0 +1,333 @@
+# -*- coding: utf-8 -*-
+from unittest.mock import Mock
+
+import nameko.standalone.events
+import pytest
+from nameko.events import EventDispatcher, event_handler
+from nameko.testing.services import entrypoint_waiter
+from nameko.testing.utils import get_extension
+from opentelemetry.trace import SpanKind
+
+from nameko_opentelemetry import active_tracer
+from nameko_opentelemetry.scrubbers import SCRUBBED
+
+
+class TestCaptureIncomingContext:
+    @pytest.fixture
+    def container(self, container_factory, rabbit_config):
+        class Service:
+            name = "service"
+
+            dispatch = EventDispatcher()
+
+            @event_handler("service", "example")
+            def handle(self, payload):
+                return payload
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @pytest.fixture(params=["standalone", "dependency_provider"])
+    def dispatch(self, rabbit_config, request, container):
+        if request.param == "standalone":
+            dispatch = nameko.standalone.events.event_dispatcher()
+            yield lambda event_type, payload: dispatch("service", event_type, payload)
+
+        if request.param == "dependency_provider":
+            dp = get_extension(container, EventDispatcher)
+            yield dp.get_dependency(Mock(context_data={}))
+
+    def test_incoming_context(self, container, dispatch, memory_exporter):
+
+        payload = "payload"
+        with entrypoint_waiter(container, "handle") as result:
+            dispatch("example", payload)
+        assert result.get() == payload
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        client_span = list(filter(lambda span: span.kind == SpanKind.PRODUCER, spans))[
+            0
+        ]
+        server_span = list(filter(lambda span: span.kind == SpanKind.CONSUMER, spans))[
+            0
+        ]
+
+        assert client_span.parent is None
+        assert server_span.parent.span_id == client_span.get_span_context().span_id
+
+
+class TestServerAttributes:
+    @pytest.fixture
+    def container(self, container_factory, rabbit_config):
+        class Service:
+            name = "service"
+
+            @event_handler("service", "example", requeue_on_error=True)
+            def handle(self, payload):
+                return payload
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @pytest.fixture
+    def dispatch(self, rabbit_config):
+        dispatch = nameko.standalone.events.event_dispatcher()
+        yield lambda event_type, payload: dispatch("service", event_type, payload)
+
+    def test_event_attributes(self, container, dispatch, memory_exporter):
+
+        payload = "payload"
+        with entrypoint_waiter(container, "handle") as result:
+            dispatch("example", payload)
+        assert result.get() == payload
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        server_span = list(filter(lambda span: span.kind == SpanKind.CONSUMER, spans))[
+            0
+        ]
+
+        attributes = server_span.attributes
+        assert attributes["nameko.events.handler_type"] == "service_pool"
+        assert attributes["nameko.events.reliable_delivery"] == "True"
+        assert attributes["nameko.events.requeue_on_error"] == "True"
+
+    def test_consumer_attributes(self, container, dispatch, memory_exporter):
+
+        payload = "payload"
+        with entrypoint_waiter(container, "handle") as result:
+            dispatch("example", payload)
+        assert result.get() == payload
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        server_span = list(filter(lambda span: span.kind == SpanKind.CONSUMER, spans))[
+            0
+        ]
+
+        attributes = server_span.attributes
+        assert attributes["nameko.amqp.prefetch_count"] == "10"
+        assert attributes["nameko.amqp.heartbeat"] == "60"
+        # no need to test all
+
+
+class TestClientAttributes:
+    @pytest.fixture(
+        params=[True, False], ids=["send_request_payloads", "no_send_request_payloads"]
+    )
+    def send_request_payloads(self, request):
+        return request.param
+
+    @pytest.fixture
+    def config(self, config, send_request_payloads):
+        # disable request payloads based on param
+        config["send_request_payloads"] = send_request_payloads
+        return config
+
+    @pytest.fixture
+    def container(self, container_factory, rabbit_config):
+        class Service:
+            name = "service"
+
+            dispatch = EventDispatcher(expiration=10)
+
+            @event_handler("service", "example")
+            def handle(self, payload):
+                return payload
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @pytest.fixture(params=["standalone", "dependency_provider"])
+    def dispatch(self, rabbit_config, request, container):
+        if request.param == "standalone":
+            dispatch = nameko.standalone.events.event_dispatcher(expiration=10)
+            yield lambda event_type, payload: dispatch("service", event_type, payload)
+
+        if request.param == "dependency_provider":
+            dp = get_extension(container, EventDispatcher)
+            yield dp.get_dependency(Mock(context_data={}))
+
+    def test_event_attributes(
+        self, container, dispatch, memory_exporter, send_request_payloads
+    ):
+
+        payload = "payload"
+        with entrypoint_waiter(container, "handle") as result:
+            dispatch("example", payload)
+        assert result.get() == payload
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        client_span = list(filter(lambda span: span.kind == SpanKind.PRODUCER, spans))[
+            0
+        ]
+
+        attributes = client_span.attributes
+        assert attributes["nameko.events.exchange"] == "service.events"
+        assert attributes["nameko.events.event_type"] == "example"
+
+        if send_request_payloads:
+            assert attributes["nameko.events.event_data"] == "payload"
+            assert attributes["nameko.events.event_data_truncated"] == "False"
+        else:
+            assert "nameko.events.event_data" not in attributes
+            assert "nameko.events.event_data_truncated" not in attributes
+
+    def test_publisher_attributes(self, container, dispatch, memory_exporter):
+
+        payload = "payload"
+        with entrypoint_waiter(container, "handle") as result:
+            dispatch("example", payload)
+        assert result.get() == payload
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        client_span = list(filter(lambda span: span.kind == SpanKind.PRODUCER, spans))[
+            0
+        ]
+
+        attributes = client_span.attributes
+        assert attributes["nameko.amqp.mandatory"] == "False"
+        assert attributes["nameko.amqp.expiration"] == "10"
+        # no need to test all
+
+
+class TestAdditionalSpans:
+    @pytest.fixture
+    def container(self, container_factory, rabbit_config):
+        class Service:
+            name = "service"
+
+            @event_handler("service", "example")
+            def handle(self, payload):
+                with active_tracer().start_as_current_span(
+                    "foobar", attributes={"foo": "bar"}
+                ):
+                    return payload
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @pytest.fixture
+    def dispatch(self, rabbit_config):
+        dispatch = nameko.standalone.events.event_dispatcher()
+        yield lambda event_type, payload: dispatch("service", event_type, payload)
+
+    def test_internal_span(self, container, dispatch, memory_exporter):
+
+        payload = "payload"
+        with entrypoint_waiter(container, "handle") as result:
+            dispatch("example", payload)
+        assert result.get() == payload
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 3
+
+        internal_span = list(
+            filter(lambda span: span.kind == SpanKind.INTERNAL, spans)
+        )[0]
+
+        assert internal_span.name == "foobar"
+
+
+class TestScrubbing:
+    @pytest.fixture
+    def container(self, container_factory, rabbit_config):
+        class Service:
+            name = "service"
+
+            dispatch = EventDispatcher(expiration=10, headers={"password": "secret"})
+
+            @event_handler("service", "example")
+            def handle(self, payload):
+                return payload
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @pytest.fixture(params=["standalone", "dependency_provider"])
+    def dispatch(self, rabbit_config, request, container):
+        if request.param == "standalone":
+            dispatch = nameko.standalone.events.event_dispatcher(
+                expiration=10, headers={"password": "secret"}
+            )
+            yield lambda event_type, payload: dispatch("service", event_type, payload)
+
+        if request.param == "dependency_provider":
+            dp = get_extension(container, EventDispatcher)
+            yield dp.get_dependency(Mock(context_data={}))
+
+    def test_payload_scrubber(self, container, dispatch, memory_exporter):
+
+        payload = {"auth": "token"}
+        with entrypoint_waiter(container, "handle") as result:
+            dispatch("example", payload)
+        assert result.get() == payload
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        client_span = list(filter(lambda span: span.kind == SpanKind.PRODUCER, spans))[
+            0
+        ]
+
+        attributes = client_span.attributes
+        assert attributes["nameko.events.event_data"] == f"{{'auth': '{SCRUBBED}'}}"
+
+    def test_call_args_scrubber(self, container, dispatch, memory_exporter):
+
+        payload = {"auth": "token"}
+        with entrypoint_waiter(container, "handle") as result:
+            dispatch("example", payload)
+        assert result.get() == payload
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        server_span = list(filter(lambda span: span.kind == SpanKind.CONSUMER, spans))[
+            0
+        ]
+
+        attributes = server_span.attributes
+        assert attributes["call_args"] == f"{{'payload': {{'auth': '{SCRUBBED}'}}}}"
+
+    def test_header_scrubber(self, container, dispatch, memory_exporter):
+
+        payload = {"auth": "token"}
+        with entrypoint_waiter(container, "handle") as result:
+            dispatch("example", payload)
+        assert result.get() == payload
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        client_span = list(filter(lambda span: span.kind == SpanKind.PRODUCER, spans))[
+            0
+        ]
+        server_span = list(filter(lambda span: span.kind == SpanKind.CONSUMER, spans))[
+            0
+        ]
+
+        # headers scrubbed at client
+        assert (
+            f"'password': '{SCRUBBED}'" in client_span.attributes["nameko.amqp.headers"]
+        )
+        # context data scrubbed at server
+        assert f"'password': '{SCRUBBED}'" in server_span.attributes["context_data"]
diff --git a/tests/test_http.py b/tests/test_http.py
new file mode 100644
index 0000000..8b9dc59
--- /dev/null
+++ b/tests/test_http.py
@@ -0,0 +1,663 @@
+# -*- coding: utf-8 -*-
+import socket
+
+import pytest
+from nameko.testing.services import entrypoint_waiter
+from nameko.web.handlers import Response, http
+from opentelemetry.instrumentation.requests import RequestsInstrumentor
+from opentelemetry.semconv.trace import SpanAttributes
+from opentelemetry.trace import SpanKind
+from opentelemetry.trace.status import StatusCode
+
+from nameko_opentelemetry.scrubbers import SCRUBBED
+
+
+class TestCaptureIncomingContext:
+    @pytest.fixture
+    def container(self, container_factory, web_config):
+        class Service:
+            name = "service"
+
+            @http("GET", "/resource")
+            def get_resource(self, request):
+                return "OK"
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @pytest.fixture
+    def instrument_requests(self):
+        instrumentor = RequestsInstrumentor()
+        instrumentor.instrument()
+        yield
+        instrumentor.uninstrument()
+
+    def test_incoming_context(
+        self, container, web_session, memory_exporter, instrument_requests
+    ):
+        with entrypoint_waiter(container, "get_resource"):
+            resp = web_session.get("/resource")
+
+        assert resp.status_code == 200
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        client_span = list(filter(lambda span: span.kind == SpanKind.CLIENT, spans))[0]
+        server_span = list(filter(lambda span: span.kind == SpanKind.SERVER, spans))[0]
+
+        assert client_span.parent is None
+        assert server_span.parent.span_id == client_span.get_span_context().span_id
+
+
+class TestSpanName:
+    @pytest.fixture
+    def container(self, container_factory, web_config):
+        class Service:
+            name = "service"
+
+            @http("GET", "/resource/<int:value>")
+            def get_resource(self, request, value):
+                return f"OK {value}"
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    def test_match(self, container, web_session, memory_exporter):
+
+        with entrypoint_waiter(container, "get_resource"):
+            resp = web_session.get("/resource/1")
+
+        assert resp.status_code == 200
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        assert spans[0].name == "/resource/<int:value>"
+
+    def test_no_match(self, container, web_session, memory_exporter):
+
+        resp = web_session.get("/missing")
+        assert resp.status_code == 404
+
+        container.stop()
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        assert spans[0].name == "HTTP GET"
+
+
+class TestNoEntrypointFired:
+    """ Test cases where the request is aborted before finding an entrypoint
+    """
+
+    @pytest.fixture
+    def container(self, container_factory, web_config):
+        class Service:
+            name = "service"
+
+            @http("GET", "/resource")
+            def get_resource(self, request):
+                return "OK"  # pragma: no cover
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    def test_method_not_found(self, container, web_session, memory_exporter):
+
+        resp = web_session.get("/missing")
+        assert resp.status_code == 404
+
+        container.stop()
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        span = spans[0]
+
+        assert spans[0].name == "HTTP GET"
+
+        assert not span.status.is_ok
+        assert span.status.status_code == StatusCode.ERROR
+        assert "404 Not Found" in span.status.description
+
+    def test_method_not_allowed(self, container, web_session, memory_exporter):
+
+        resp = web_session.post("/resource")
+        assert resp.status_code == 405
+
+        container.stop()
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        span = spans[0]
+
+        assert spans[0].name == "HTTP POST"
+
+        assert not span.status.is_ok
+        assert span.status.status_code == StatusCode.ERROR
+        assert "405 Method Not Allowed" in span.status.description
+
+
+class TestSpanAttributes:
+    @pytest.fixture
+    def container(self, container_factory, web_config):
+        class Service:
+            name = "service"
+
+            @http("GET", "/resource")
+            def get_resource(self, request):
+                return "OK"
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    def test_common(self, container, web_session, memory_exporter):
+
+        with entrypoint_waiter(container, "get_resource"):
+            resp = web_session.get("/resource")
+
+        assert resp.status_code == 200
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+        assert attributes["hostname"] == socket.gethostname()
+        assert attributes["service_name"] == "service"
+        assert attributes["entrypoint_type"] == "HttpRequestHandler"
+        assert attributes["method_name"] == "get_resource"
+
+
+class TestCallArgs:
+    @pytest.fixture(params=[True, False], ids=["send_headers", "no_send_headers"])
+    def send_headers(self, request):
+        return request.param
+
+    @pytest.fixture(
+        params=[True, False], ids=["send_request_payloads", "no_send_request_payloads"]
+    )
+    def send_request_payloads(self, request):
+        return request.param
+
+    @pytest.fixture
+    def config(self, config, send_headers, send_request_payloads):
+        # disable headers based on param
+        config["send_headers"] = send_headers
+        # disable request payload based on param
+        config["send_request_payloads"] = send_request_payloads
+        return config
+
+    @pytest.fixture
+    def container(self, container_factory, web_config):
+        class Service:
+            name = "service"
+
+            @http("GET", "/resource")
+            def get_resource(self, request):
+                return "OK"
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    def test_wsgi_common(self, container, web_session, memory_exporter):
+        """ These are determined by the Opentelemetry WSGI middleware module
+        """
+        with entrypoint_waiter(container, "get_resource"):
+            resp = web_session.get("/resource")
+
+        assert resp.status_code == 200
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+        assert attributes[SpanAttributes.HTTP_METHOD] == "GET"
+        assert attributes[SpanAttributes.HTTP_SCHEME] == "http"
+        # no need to test them exhaustively
+
+    def test_request_data(
+        self, container, web_session, memory_exporter, send_request_payloads
+    ):
+        with entrypoint_waiter(container, "get_resource"):
+            resp = web_session.get("/resource", data="foobar")
+
+        assert resp.status_code == 200
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+
+        if send_request_payloads:
+            assert attributes["request.data"] == "foobar"
+        else:
+            assert "request.data" not in attributes
+
+    def test_request_headers(
+        self, container, web_session, memory_exporter, send_headers
+    ):
+
+        with entrypoint_waiter(container, "get_resource"):
+            resp = web_session.get("/resource", headers={"name": "matt"})
+
+        assert resp.status_code == 200
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+        if send_headers:
+            assert "'name': 'matt'" in attributes["request.headers"]
+        else:
+            assert "request.headers" not in attributes
+
+
+class TestExceptions:
+    @pytest.fixture
+    def container(self, container_factory, web_config):
+        class Service:
+            name = "service"
+
+            class Error(Exception):
+                pass
+
+            @http("POST", "/resource")
+            def raises(self, request):
+                raise self.Error("boom")
+
+            @http("DELETE", "/resource", expected_exceptions=Error)
+            def raises_expected(self, request):
+                raise self.Error("boom")
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    def test_exception(self, container, web_session, memory_exporter):
+
+        with entrypoint_waiter(container, "raises"):
+            resp = web_session.post("/resource")
+
+        assert resp.status_code == 500
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        span = spans[0]
+        assert len(span.events) == 1
+
+        event = span.events[0]
+        assert event.name == "exception"
+        assert event.attributes["exception.type"] == "Error"
+        assert event.attributes["exception.message"] == "boom"
+        assert 'raise self.Error("boom")' in event.attributes["exception.stacktrace"]
+        assert event.attributes["exception.escaped"] == "True"
+        # extra attributes
+        assert event.attributes["exception.expected"] == "False"
+
+    def test_expected_exception(self, container, web_session, memory_exporter):
+
+        with entrypoint_waiter(container, "raises_expected"):
+            resp = web_session.delete("/resource")
+
+        assert resp.status_code == 400
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        span = spans[0]
+        assert len(span.events) == 1
+
+        event = span.events[0]
+        assert event.name == "exception"
+        assert event.attributes["exception.type"] == "Error"
+        assert event.attributes["exception.message"] == "boom"
+        assert 'raise self.Error("boom")' in event.attributes["exception.stacktrace"]
+        assert event.attributes["exception.escaped"] == "True"
+        # extra attributes
+        assert event.attributes["exception.expected"] == "True"
+
+
+class TestResult:
+    @pytest.fixture(
+        params=[True, False],
+        ids=["send_response_payloads", "no_send_response_payloads"],
+    )
+    def send_response_payloads(self, request):
+        return request.param
+
+    @pytest.fixture
+    def config(self, config, send_response_payloads):
+        # override default truncation length
+        config["truncate_max_length"] = 300
+        # disable response payload based on param
+        config["send_response_payloads"] = send_response_payloads
+        return config
+
+    @pytest.fixture
+    def container(self, container_factory, web_config):
+        class Service:
+            name = "service"
+
+            @http("GET", "/simple")
+            def simple_result(self, request):
+                return "OK"
+
+            @http("GET", "/tuple")
+            def tuple_result(self, request):
+                return (
+                    401,
+                    {"Content-Type": "application/json"},
+                    '{"authorized": false}',
+                )
+
+            @http("GET", "/response")
+            def response_result(self, request):
+                return Response(
+                    "Permission denied", status=403, content_type="text/plain"
+                )
+
+            @http("GET", "/big")
+            def truncate_result(self, request):
+                return "x" * 1000
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    def test_simple_result(
+        self, container, web_session, memory_exporter, send_response_payloads
+    ):
+
+        with entrypoint_waiter(container, "simple_result"):
+            resp = web_session.get("/simple")
+
+        assert resp.status_code == 200
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+        assert attributes["response.content_type"] == "text/plain; charset=utf-8"
+        assert attributes[SpanAttributes.HTTP_RESPONSE_CONTENT_LENGTH] == 2
+        assert attributes[SpanAttributes.HTTP_STATUS_CODE] == 200
+
+        if send_response_payloads:
+            assert attributes["response.data"] == "OK"
+            assert attributes["response.data_truncated"] == "False"
+        else:
+            assert "response.data" not in attributes
+            assert "response.data_truncated" not in attributes
+
+    def test_tuple_result(
+        self, container, web_session, memory_exporter, send_response_payloads
+    ):
+
+        with entrypoint_waiter(container, "tuple_result"):
+            resp = web_session.get("/tuple")
+
+        assert resp.status_code == 401
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+        assert attributes["response.content_type"] == "application/json"
+        assert attributes[SpanAttributes.HTTP_RESPONSE_CONTENT_LENGTH] == 21
+        assert attributes[SpanAttributes.HTTP_STATUS_CODE] == 401
+
+        if send_response_payloads:
+            assert attributes["response.data"] == '{"authorized": false}'
+            assert attributes["response.data_truncated"] == "False"
+        else:
+            assert "response.data" not in attributes
+            assert "response.data_truncated" not in attributes
+
+    def test_response_result(
+        self, container, web_session, memory_exporter, send_response_payloads
+    ):
+
+        with entrypoint_waiter(container, "response_result"):
+            resp = web_session.get("/response")
+
+        assert resp.status_code == 403
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+        assert attributes["response.content_type"] == "text/plain"
+        assert attributes[SpanAttributes.HTTP_RESPONSE_CONTENT_LENGTH] == 17
+        assert attributes[SpanAttributes.HTTP_STATUS_CODE] == 403
+
+        if send_response_payloads:
+            assert attributes["response.data"] == "Permission denied"
+            assert attributes["response.data_truncated"] == "False"
+        else:
+            assert "response.data" not in attributes
+            assert "response.data_truncated" not in attributes
+
+    def test_truncated_result(
+        self, container, web_session, memory_exporter, send_response_payloads
+    ):
+
+        with entrypoint_waiter(container, "truncate_result"):
+            resp = web_session.get("/big")
+
+        assert resp.status_code == 200
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+        assert attributes["response.content_type"] == "text/plain; charset=utf-8"
+        assert attributes[SpanAttributes.HTTP_RESPONSE_CONTENT_LENGTH] == 1000
+        assert attributes[SpanAttributes.HTTP_STATUS_CODE] == 200
+
+        if send_response_payloads:
+            assert attributes["response.data"] == "x" * 300
+            assert attributes["response.data_truncated"] == "True"
+        else:
+            assert "response.data" not in attributes
+            assert "response.data_truncated" not in attributes
+
+
+class TestStatus:
+    """
+    Spec says status code MUST be set for 4xx and 5xx errors, and MUST NOT be set for
+    1xx, 2xx and 3xx responses. See
+    https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/semantic_conventions/http.md#status
+    """
+
+    @pytest.fixture
+    def container(self, container_factory, web_config):
+        class Service:
+            name = "service"
+
+            class Error(Exception):
+                pass
+
+            @http("GET", "/resource")
+            def get_resource(self, request):
+                return "OK"
+
+            @http("GET", "/redirect")
+            def get_redirect(self, request):
+                return 302, "https://example.org"
+
+            @http("GET", "/unauthorized")
+            def get_status_code(self, request):
+                return 401, "Unauthorized"
+
+            @http("GET", "/response")
+            def get_response(self, request):
+                return Response("Permission denied", status=403)
+
+            @http("POST", "/resource")
+            def raises(self, request):
+                raise self.Error("boom")
+
+            @http("DELETE", "/resource", expected_exceptions=Error)
+            def raises_expected(self, request):
+                raise self.Error("boom")
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    def test_success(self, container, web_session, memory_exporter):
+
+        with entrypoint_waiter(container, "get_resource"):
+            resp = web_session.get("/resource")
+
+        assert resp.status_code == 200
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        span = spans[0]
+        assert span.status.is_ok
+        assert span.status.status_code == StatusCode.UNSET
+
+    def test_redirect(self, container, web_session, memory_exporter):
+
+        with entrypoint_waiter(container, "get_redirect"):
+            resp = web_session.get("/redirect")
+
+        assert resp.status_code == 302
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        span = spans[0]
+        assert span.status.is_ok
+        assert span.status.status_code == StatusCode.UNSET
+
+    def test_status_code(self, container, web_session, memory_exporter):
+
+        with entrypoint_waiter(container, "get_status_code"):
+            resp = web_session.get("/unauthorized")
+
+        assert resp.status_code == 401
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        span = spans[0]
+        assert not span.status.is_ok
+        assert span.status.status_code == StatusCode.ERROR
+
+    def test_response_object(self, container, web_session, memory_exporter):
+
+        with entrypoint_waiter(container, "get_response"):
+            resp = web_session.get("/response")
+
+        assert resp.status_code == 403
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        span = spans[0]
+        assert not span.status.is_ok
+        assert span.status.status_code == StatusCode.ERROR
+
+    def test_exception(self, container, web_session, memory_exporter):
+
+        with entrypoint_waiter(container, "raises"):
+            resp = web_session.post("/resource")
+
+        assert resp.status_code == 500
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        span = spans[0]
+        assert not span.status.is_ok
+        assert span.status.status_code == StatusCode.ERROR
+        assert span.status.description == "Error: boom"
+
+    def test_expected_exception(self, container, web_session, memory_exporter):
+
+        with entrypoint_waiter(container, "raises_expected"):
+            resp = web_session.delete("/resource")
+
+        assert resp.status_code == 400
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        span = spans[0]
+        assert not span.status.is_ok
+        assert span.status.status_code == StatusCode.ERROR
+        assert span.status.description == "Error: boom"
+
+
+class TestScrubbing:
+    @pytest.fixture
+    def container(self, container_factory, web_config):
+        class Service:
+            name = "service"
+
+            @http("GET", "/resource")
+            def get_resource(self, request):
+                return (request.data or b"OK").decode("utf-8")
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    def test_scrub_headers(self, container, web_session, memory_exporter):
+
+        with entrypoint_waiter(container, "get_resource"):
+            resp = web_session.get("/resource", headers={"token": "secret"})
+
+        assert resp.status_code == 200
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        server_span = spans[0]
+        assert f"'token': '{SCRUBBED}'" in server_span.attributes["request.headers"]
+
+    def test_scrub_request_payload(self, container, web_session, memory_exporter):
+
+        with entrypoint_waiter(container, "get_resource"):
+            resp = web_session.get("/resource", data={"token": "secret"})
+
+        assert resp.status_code == 200
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        server_span = spans[0]
+        assert server_span.attributes["request.data"] == f"{{'token': '{SCRUBBED}'}}"
+
+    def test_scrub_response(self, container, web_session, memory_exporter):
+
+        with entrypoint_waiter(container, "get_resource"):
+            resp = web_session.get("/resource", data="matt@pacerevenue.com")
+
+        assert resp.status_code == 200
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        server_span = spans[0]
+        assert server_span.attributes["response.data"] == SCRUBBED
diff --git a/tests/test_messaging.py b/tests/test_messaging.py
new file mode 100644
index 0000000..31b02a5
--- /dev/null
+++ b/tests/test_messaging.py
@@ -0,0 +1,322 @@
+# -*- coding: utf-8 -*-
+from unittest.mock import Mock
+
+import pytest
+from kombu.messaging import Exchange, Queue
+from nameko.messaging import Publisher, consume
+from nameko.testing.services import entrypoint_waiter
+from nameko.testing.utils import get_extension
+from opentelemetry.trace import SpanKind
+
+from nameko_opentelemetry import active_tracer
+from nameko_opentelemetry.scrubbers import SCRUBBED
+
+
+exchange = Exchange(name="test")
+queue = Queue(name="test", exchange=exchange)
+
+
+class TestCaptureIncomingContext:
+    @pytest.fixture
+    def container(self, container_factory, rabbit_config):
+        class Service:
+            name = "service"
+
+            publish = Publisher(exchange)
+
+            @consume(queue)
+            def handle(self, payload):
+                return payload
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @pytest.fixture
+    def publish(self, rabbit_config, container):
+        dp = get_extension(container, Publisher)
+        return dp.get_dependency(Mock(context_data={}))
+
+    def test_incoming_context(self, container, publish, memory_exporter):
+
+        payload = "payload"
+        with entrypoint_waiter(container, "handle") as result:
+            publish(payload)
+        assert result.get() == payload
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        client_span = list(filter(lambda span: span.kind == SpanKind.PRODUCER, spans))[
+            0
+        ]
+        server_span = list(filter(lambda span: span.kind == SpanKind.CONSUMER, spans))[
+            0
+        ]
+
+        assert client_span.parent is None
+        assert server_span.parent.span_id == client_span.get_span_context().span_id
+
+
+class TestServerAttributes:
+    @pytest.fixture
+    def container(self, container_factory, rabbit_config):
+        class Service:
+            name = "service"
+
+            publish = Publisher(exchange)
+
+            @consume(queue, requeue_on_error=True)
+            def handle(self, payload):
+                return payload
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @pytest.fixture
+    def publish(self, rabbit_config, container):
+        dp = get_extension(container, Publisher)
+        return dp.get_dependency(Mock(context_data={}))
+
+    def test_messaging_attributes(self, container, publish, memory_exporter):
+
+        payload = "payload"
+        with entrypoint_waiter(container, "handle") as result:
+            publish(payload)
+        assert result.get() == payload
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        server_span = list(filter(lambda span: span.kind == SpanKind.CONSUMER, spans))[
+            0
+        ]
+
+        attributes = server_span.attributes
+        assert attributes["nameko.messaging.requeue_on_error"] == "True"
+
+    def test_consumer_attributes(self, container, publish, memory_exporter):
+
+        payload = "payload"
+        with entrypoint_waiter(container, "handle") as result:
+            publish(payload)
+        assert result.get() == payload
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        server_span = list(filter(lambda span: span.kind == SpanKind.CONSUMER, spans))[
+            0
+        ]
+
+        attributes = server_span.attributes
+        assert attributes["nameko.amqp.prefetch_count"] == "10"
+        assert attributes["nameko.amqp.heartbeat"] == "60"
+        # no need to test all
+
+
+class TestClientAttributes:
+    @pytest.fixture(
+        params=[True, False], ids=["send_request_payloads", "no_send_request_payloads"]
+    )
+    def send_request_payloads(self, request):
+        return request.param
+
+    @pytest.fixture
+    def config(self, config, send_request_payloads):
+        # disable request payloads based on param
+        config["send_request_payloads"] = send_request_payloads
+        return config
+
+    @pytest.fixture
+    def container(self, container_factory, rabbit_config):
+        class Service:
+            name = "service"
+
+            publish = Publisher(exchange)
+
+            @consume(queue)
+            def handle(self, payload):
+                return payload
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @pytest.fixture
+    def publish(self, rabbit_config, container):
+        dp = get_extension(container, Publisher)
+        return dp.get_dependency(Mock(context_data={}))
+
+    def test_event_attributes(
+        self, container, publish, memory_exporter, send_request_payloads
+    ):
+
+        payload = "payload"
+        with entrypoint_waiter(container, "handle") as result:
+            publish(payload)
+        assert result.get() == payload
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        client_span = list(filter(lambda span: span.kind == SpanKind.PRODUCER, spans))[
+            0
+        ]
+
+        attributes = client_span.attributes
+        assert attributes["nameko.messaging.exchange"] == exchange.name
+
+        if send_request_payloads:
+            assert attributes["nameko.messaging.payload"] == "payload"
+            assert attributes["nameko.messaging.payload_truncated"] == "False"
+        else:
+            assert "nameko.messaging.payload" not in attributes
+            assert "nameko.messaging.payload_truncated" not in attributes
+
+    def test_publisher_attributes(self, container, publish, memory_exporter):
+
+        payload = "payload"
+        with entrypoint_waiter(container, "handle") as result:
+            publish(payload, expiration=10)
+        assert result.get() == payload
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        client_span = list(filter(lambda span: span.kind == SpanKind.PRODUCER, spans))[
+            0
+        ]
+
+        attributes = client_span.attributes
+        assert attributes["nameko.amqp.mandatory"] == "False"
+        assert attributes["nameko.amqp.expiration"] == "10"
+        # no need to test all
+
+
+class TestAdditionalSpans:
+    @pytest.fixture
+    def container(self, container_factory, rabbit_config):
+        class Service:
+            name = "service"
+
+            publish = Publisher(exchange)
+
+            @consume(queue)
+            def handle(self, payload):
+                with active_tracer().start_as_current_span(
+                    "foobar", attributes={"foo": "bar"}
+                ):
+                    return payload
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @pytest.fixture
+    def publish(self, rabbit_config, container):
+        dp = get_extension(container, Publisher)
+        return dp.get_dependency(Mock(context_data={}))
+
+    def test_internal_span(self, container, publish, memory_exporter):
+
+        payload = "payload"
+        with entrypoint_waiter(container, "handle") as result:
+            publish(payload)
+        assert result.get() == payload
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 3
+
+        internal_span = list(
+            filter(lambda span: span.kind == SpanKind.INTERNAL, spans)
+        )[0]
+
+        assert internal_span.name == "foobar"
+
+
+class TestScrubbing:
+    @pytest.fixture
+    def container(self, container_factory, rabbit_config):
+        class Service:
+            name = "service"
+
+            publish = Publisher(exchange)
+
+            @consume(queue)
+            def handle(self, payload):
+                return payload
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @pytest.fixture
+    def publish(self, rabbit_config, container):
+        dp = get_extension(container, Publisher)
+        return dp.get_dependency(Mock(context_data={}))
+
+    def test_payload_scrubber(self, container, publish, memory_exporter):
+
+        payload = {"auth": "token"}
+        with entrypoint_waiter(container, "handle") as result:
+            publish(payload)
+        assert result.get() == payload
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        client_span = list(filter(lambda span: span.kind == SpanKind.PRODUCER, spans))[
+            0
+        ]
+
+        attributes = client_span.attributes
+        assert attributes["nameko.messaging.payload"] == f"{{'auth': '{SCRUBBED}'}}"
+
+    def test_call_args_scrubber(self, container, publish, memory_exporter):
+
+        payload = {"auth": "token"}
+        with entrypoint_waiter(container, "handle") as result:
+            publish(payload)
+        assert result.get() == payload
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        server_span = list(filter(lambda span: span.kind == SpanKind.CONSUMER, spans))[
+            0
+        ]
+
+        attributes = server_span.attributes
+        assert attributes["call_args"] == f"{{'payload': {{'auth': '{SCRUBBED}'}}}}"
+
+    def test_header_scrubber(self, container, publish, memory_exporter):
+
+        payload = {"auth": "token"}
+        with entrypoint_waiter(container, "handle") as result:
+            publish(payload, headers={"password": "secret"})
+        assert result.get() == payload
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        client_span = list(filter(lambda span: span.kind == SpanKind.PRODUCER, spans))[
+            0
+        ]
+        server_span = list(filter(lambda span: span.kind == SpanKind.CONSUMER, spans))[
+            0
+        ]
+
+        # headers scrubbed at client
+        assert (
+            f"'password': '{SCRUBBED}'" in client_span.attributes["nameko.amqp.headers"]
+        )
+
+        # context data scrubbed at server
+        assert f"'password': '{SCRUBBED}'" in server_span.attributes["context_data"]
diff --git a/tests/test_rpc.py b/tests/test_rpc.py
new file mode 100644
index 0000000..8549961
--- /dev/null
+++ b/tests/test_rpc.py
@@ -0,0 +1,287 @@
+# -*- coding: utf-8 -*-
+import uuid
+from unittest.mock import Mock
+
+import pytest
+from nameko.exceptions import IncorrectSignature, MethodNotFound
+from nameko.rpc import ServiceRpc, rpc
+from nameko.standalone.rpc import ServiceRpcClient
+from nameko.testing.services import entrypoint_waiter
+from nameko.testing.utils import get_extension
+from opentelemetry.trace import SpanKind
+from opentelemetry.trace.status import StatusCode
+
+from nameko_opentelemetry import active_tracer
+from nameko_opentelemetry.scrubbers import SCRUBBED
+
+
+class TestCaptureIncomingContext:
+    @pytest.fixture
+    def container(self, container_factory, rabbit_config):
+        class Service:
+            name = "service"
+
+            self_rpc = ServiceRpc("service")
+
+            @rpc
+            def method(self, arg, kwarg=None):
+                return "OK"
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @pytest.fixture(params=["standalone", "dependency_provider"])
+    def client(self, rabbit_config, request, container):
+        if request.param == "standalone":
+            with ServiceRpcClient("service") as client:
+                yield client
+        if request.param == "dependency_provider":
+            dp = get_extension(container, ServiceRpc)
+            yield dp.get_dependency(
+                Mock(context_data={"call_id": f"service.method.{uuid.uuid4()}"})
+            )
+
+    def test_incoming_context(self, container, client, memory_exporter):
+
+        with entrypoint_waiter(container, "method"):
+            assert client.method("arg", kwarg="kwarg") == "OK"
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        client_span = list(filter(lambda span: span.kind == SpanKind.CLIENT, spans))[0]
+        server_span = list(filter(lambda span: span.kind == SpanKind.SERVER, spans))[0]
+
+        assert client_span.parent is None
+        assert server_span.parent.span_id == client_span.get_span_context().span_id
+
+
+class TestNoEntrypointFired:
+    """ Test cases where the request is aborted before finding an entrypoint
+    """
+
+    @pytest.fixture
+    def container(self, container_factory, rabbit_config):
+        class Service:
+            name = "service"
+
+            @rpc
+            def method(self, arg, kwarg=None):
+                return "OK"  # pragma: no cover
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @pytest.fixture
+    def client(self, rabbit_config):
+        with ServiceRpcClient("service") as client:
+            yield client
+
+    def test_method_not_found(self, container, client, memory_exporter):
+
+        with pytest.raises(MethodNotFound):
+            client.not_a_method()
+
+        container.stop()
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        server_span = list(filter(lambda span: span.kind == SpanKind.SERVER, spans))[0]
+
+        assert not server_span.status.is_ok
+        assert server_span.status.status_code == StatusCode.ERROR
+        assert server_span.status.description == "MethodNotFound: not_a_method"
+
+    def test_incorrect_signature(self, container, client, memory_exporter):
+        with pytest.raises(IncorrectSignature):
+            client.method("foo", "bar", "baz")
+
+        container.stop()
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        server_span = list(filter(lambda span: span.kind == SpanKind.SERVER, spans))[0]
+
+        assert not server_span.status.is_ok
+        assert server_span.status.status_code == StatusCode.ERROR
+        assert "IncorrectSignature" in server_span.status.description
+
+
+class TestServerAttributes:
+    @pytest.fixture
+    def container(self, container_factory, rabbit_config):
+        class Service:
+            name = "service"
+
+            @rpc
+            def method(self, arg, kwarg=None):
+                return "OK"
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @pytest.fixture
+    def client(self, rabbit_config):
+        with ServiceRpcClient("service") as client:
+            yield client
+
+    def test_consumer_attributes(self, container, client, memory_exporter):
+
+        with entrypoint_waiter(container, "method"):
+            assert client.method("arg", kwarg="kwarg") == "OK"
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        server_span = list(filter(lambda span: span.kind == SpanKind.SERVER, spans))[0]
+
+        attributes = server_span.attributes
+        assert attributes["nameko.amqp.prefetch_count"] == "10"
+        assert attributes["nameko.amqp.heartbeat"] == "60"
+        # no need to test all
+
+
+class TestClientAttributes:
+    @pytest.fixture
+    def container(self, container_factory, rabbit_config):
+        class Service:
+            name = "service"
+
+            self_rpc = ServiceRpc("service")
+
+            @rpc
+            def method(self, arg, kwarg=None):
+                return "OK"
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @pytest.fixture(params=["standalone", "dependency_provider"])
+    def client(self, rabbit_config, request, container):
+        if request.param == "standalone":
+            with ServiceRpcClient("service") as client:
+                yield client
+        if request.param == "dependency_provider":
+            dp = get_extension(container, ServiceRpc)
+            yield dp.get_dependency(
+                Mock(context_data={"call_id": f"service.method.{uuid.uuid4()}"})
+            )
+
+    def test_rpc_attributes(self, container, client, memory_exporter):
+
+        with entrypoint_waiter(container, "method"):
+            assert client.method("arg", kwarg="kwarg") == "OK"
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        client_span = list(filter(lambda span: span.kind == SpanKind.CLIENT, spans))[0]
+
+        attributes = client_span.attributes
+        assert attributes["nameko.rpc.target_service"] == "service"
+        assert attributes["nameko.rpc.target_method"] == "method"
+
+    def test_publisher_attributes(self, container, client, memory_exporter):
+
+        with entrypoint_waiter(container, "method"):
+            assert client.method("arg", kwarg="kwarg") == "OK"
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        client_span = list(filter(lambda span: span.kind == SpanKind.CLIENT, spans))[0]
+
+        attributes = client_span.attributes
+        assert attributes["nameko.amqp.mandatory"] == "True"
+        assert attributes["nameko.amqp.retry"] == "True"
+        # no need to test all
+
+
+class TestAdditionalSpans:
+    @pytest.fixture
+    def container(self, container_factory, rabbit_config):
+        class Service:
+            name = "service"
+
+            @rpc
+            def method(self, arg, kwarg=None):
+                with active_tracer().start_as_current_span(
+                    "foobar", attributes={"foo": "bar"}
+                ):
+                    return "OK"
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @pytest.fixture
+    def client(self, rabbit_config):
+        with ServiceRpcClient("service") as client:
+            yield client
+
+    def test_internal_span(self, container, client, memory_exporter):
+
+        with entrypoint_waiter(container, "method"):
+            assert client.method("arg", kwarg="kwarg") == "OK"
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 3
+
+        internal_span = list(
+            filter(lambda span: span.kind == SpanKind.INTERNAL, spans)
+        )[0]
+
+        assert internal_span.name == "foobar"
+
+
+class TestScrubbing:
+    @pytest.fixture
+    def container(self, container_factory, rabbit_config):
+        class Service:
+            name = "service"
+
+            self_rpc = ServiceRpc("service")
+
+            @rpc
+            def method(self, arg, kwarg=None):
+                return "OK"
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    @pytest.fixture(params=["standalone", "dependency_provider"])
+    def client(self, rabbit_config, request, container):
+        if request.param == "standalone":
+            with ServiceRpcClient("service", context_data={"auth": "token"}) as client:
+                yield client
+        if request.param == "dependency_provider":
+            dp = get_extension(container, ServiceRpc)
+            yield dp.get_dependency(Mock(context_data={"auth": "token"}))
+
+    def test_header_scrubber(self, container, client, memory_exporter):
+
+        with entrypoint_waiter(container, "method"):
+            assert client.method("arg", kwarg="kwarg") == "OK"
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 2
+
+        client_span = list(filter(lambda span: span.kind == SpanKind.CLIENT, spans))[0]
+
+        assert (
+            f"'nameko.auth': '{SCRUBBED}'"
+            in client_span.attributes["nameko.amqp.headers"]
+        )
diff --git a/tests/test_scrubbers.py b/tests/test_scrubbers.py
new file mode 100644
index 0000000..54b05ab
--- /dev/null
+++ b/tests/test_scrubbers.py
@@ -0,0 +1,134 @@
+# -*- coding: utf-8 -*-
+import pytest
+
+from nameko_opentelemetry.scrubbers import SCRUBBED, DefaultScrubber, scrub
+
+
+class TestDefaultScubber:
+    @pytest.fixture
+    def scrubber(self):
+        return DefaultScrubber(config={})
+
+    @pytest.mark.parametrize(
+        "value",
+        [
+            "value",
+            b"value",
+            b"\x90",
+            1,
+            1.0,
+            True,
+            [1, 2, 3],
+            dict(foo="bar"),
+            (1, 2, 3),
+            object(),
+        ],
+    )
+    def test_innocuous(self, value, scrubber):
+        assert scrubber.scrub(value) == value
+
+    def test_sensitive_string(self, scrubber):
+        assert scrubber.scrub("matt@pacerevenue.com") == SCRUBBED
+
+    def test_sensitive_bytes(self, scrubber):
+        assert scrubber.scrub(b"matt@pacerevenue.com") == SCRUBBED.encode("utf-8")
+
+    def test_list_with_sensitive_item(self, scrubber):
+        data = ["foo", "matt@pacerevenue.com", "bar"]
+        assert scrubber.scrub(data) == ["foo", SCRUBBED, "bar"]
+
+        # source data not modified
+        assert data == ["foo", "matt@pacerevenue.com", "bar"]
+
+    def test_tuple_with_sensitive_item(self, scrubber):
+        data = ("foo", "matt@pacerevenue.com", "bar")
+        assert scrubber.scrub(data) == ("foo", SCRUBBED, "bar")
+
+        # source data not modified
+        assert data == ("foo", "matt@pacerevenue.com", "bar")
+
+    def test_generator_with_sensitive_item(self, scrubber):
+        def items():
+            yield "foo"
+            yield "matt@pacerevenue.com"
+            yield "bar"
+
+        data = items()
+        assert scrubber.scrub(data) == ["foo", SCRUBBED, "bar"]
+
+    def test_dict_with_sensitive_key(self, scrubber):
+        data = {"password": "foobar"}
+        assert scrubber.scrub(data) == {"password": SCRUBBED}
+
+        # source data not modified
+        assert data == {"password": "foobar"}
+
+    def test_dict_with_sensitive_value(self, scrubber):
+        data = {"email": "matt@pacerevenue.com"}
+        assert scrubber.scrub(data) == {"email": SCRUBBED}
+
+        # source data not modified
+        assert data == {"email": "matt@pacerevenue.com"}
+
+    def test_dict_with_sensitive_value_for_key(self, scrubber):
+        data = {"matt@pacerevenue.com": "me"}
+        assert scrubber.scrub(data) == {SCRUBBED: "me"}
+
+        # source data not modified
+        assert data == {"matt@pacerevenue.com": "me"}
+
+    def test_dict_with_sensitive_sub_key(self, scrubber):
+        data = {"innocuous": {"email": "matt@pacerevenue.com"}}
+        assert scrubber.scrub(data) == {"innocuous": {"email": SCRUBBED}}
+
+        # source data not modified
+        assert data == {"innocuous": {"email": "matt@pacerevenue.com"}}
+
+    def test_dict_with_nested_list_containing_sensitive_value(self, scrubber):
+        data = {"innocuous": ["foo", "matt@pacerevenue.com", "bar"]}
+        assert scrubber.scrub(data) == {"innocuous": ["foo", SCRUBBED, "bar"]}
+
+        # source data not modified
+        assert data == {"innocuous": ["foo", "matt@pacerevenue.com", "bar"]}
+
+    def test_list_with_nested_dict_containing_sensitive_data(self, scrubber):
+        data = ["foo", {"email": "matt@pacerevenue.com", "secret": "shh"}, "bar"]
+        assert scrubber.scrub(data) == [
+            "foo",
+            {"email": SCRUBBED, "secret": SCRUBBED},
+            "bar",
+        ]
+
+        # source data not modified
+        assert data == [
+            "foo",
+            {"email": "matt@pacerevenue.com", "secret": "shh"},
+            "bar",
+        ]
+
+    def test_common_key_prefixes(self, scrubber):
+        data = {"x-token": "secret", "nameko.token": "secret"}
+        assert scrubber.scrub(data) == {"x-token": SCRUBBED, "nameko.token": SCRUBBED}
+
+        # source data not modified
+        assert data == {"x-token": "secret", "nameko.token": "secret"}
+
+
+class CustomScrubber(DefaultScrubber):
+    SENSITIVE_KEYS = ("name",)
+    REPLACEMENT = "***"
+
+
+class TestScrub:
+    @pytest.fixture
+    def config(self):
+        return {
+            "scrubbers": (
+                "test_scrubbers.CustomScrubber",
+                "nameko_opentelemetry.scrubbers.DefaultScrubber",
+            )
+        }
+
+    def test_register_new_scrubber(self, config):
+        data = {"name": "Matt", "email": "matt@pacerevenue.com"}
+        assert scrub(data, config) == {"name": "***", "email": "***"}
diff --git a/tests/test_timer.py b/tests/test_timer.py
new file mode 100644
index 0000000..4cce39a
--- /dev/null
+++ b/tests/test_timer.py
@@ -0,0 +1,32 @@
+# -*- coding: utf-8 -*-
+import pytest
+from nameko.testing.services import entrypoint_waiter
+from nameko.timer import timer
+
+
+class TestServerAttributes:
+    @pytest.fixture
+    def container(self, container_factory, rabbit_config):
+        class Service:
+            name = "service"
+
+            @timer(interval=0.001, eager=True)
+            def method(self):
+                return "OK"
+
+        container = container_factory(Service)
+        container.start()
+
+        return container
+
+    def test_attributes(self, container, memory_exporter):
+
+        with entrypoint_waiter(container, "method"):
+            pass
+
+        spans = memory_exporter.get_finished_spans()
+        assert len(spans) == 1
+
+        attributes = spans[0].attributes
+        assert attributes["nameko.timer.interval"] == 0.001
+        assert attributes["nameko.timer.eager"] is True
diff --git a/tests/test_utils.py b/tests/test_utils.py
new file mode 100644
index 0000000..bea9045
--- /dev/null
+++ b/tests/test_utils.py
@@ -0,0 +1,94 @@
+# -*- coding: utf-8 -*-
+import pytest
+
+from nameko_opentelemetry import utils
+
+
+@pytest.mark.parametrize(
+    ("input_", "expected_output"),
+    (
+        ("string", "string"),
+        ([1, 0.5, "string"], [1, 0.5, "string"]),
+        ((1, 0.5, "string"), [1, 0.5, "string"]),
+        ({1: 2, 0.5: 0.5}, {1: 2, 0.5: 0.5}),
+        ({"string": "string"}, {"string": "string"}),
+        ({1}, [1]),
+        (object, repr(object)),
+        (None, None),
+        (
+            ((1, 0.5, "string", object), [], {}, set()),
+            [[1, 0.5, "string", repr(object)], [], {}, []],
+        ),
+        (
+            ((), [1, 0.5, "string", object], {}),
+            [[], [1, 0.5, "string", repr(object)], {}],
+        ),
+        (
+            ((), [], {1: {0.5: ("string", object)}}),
+            [[], [], {1: {0.5: ["string", repr(object)]}}],
+        ),
+        (b"bytes to decode", "bytes to decode"),
+    ),
+)
+def test_safe_for_serialisation(input_, expected_output):
+    assert utils.safe_for_serialisation(input_) == expected_output
+
+
+def test_safe_for_serialisation_repr_failure():
+    class CannotRepr(object):
+        def __repr__(self):
+            raise Exception("boom")
+
+    assert utils.safe_for_serialisation([1, CannotRepr(), 2]) == [
+        1,
+        "failed to get string representation",
+        2,
+    ]
+
+
+def test_serialise_to_json():
+    assert (
+        utils.serialise_to_json(((), [], {1: {0.5: ("string", object)}}))
+        == '[[], [], {"1": {"0.5": ["string", "<class \'object\'>"]}}]'
+    )
+
+
+def test_serialise_to_string():
+    assert (
+        utils.serialise_to_string(((), [], {1: {0.5: ("string", object)}}))
+        == "[[], [], {1: {0.5: ['string', \"<class 'object'>\"]}}]"
+    )
+
+
+class SomeClass:
+    pass
+
+
+def test_import_by_path():
+    cls = utils.import_by_path("test_utils.SomeClass")
+    assert cls == SomeClass
+
+
+def test_import_by_path_error_not_a_path():
+    with pytest.raises(ImportError) as exc:
+        utils.import_by_path("no_dots_in_path")
+    assert str(exc.value) == ("no_dots_in_path doesn't look like a module path")
+
+
+def test_import_by_path_not_found_error():
+    with pytest.raises(ImportError) as exc:
+        utils.import_by_path("test_utils.Nothing")
+    assert str(exc.value) == (
+        'Module "test_utils" does not define a "Nothing" attribute/class'
+    )
+
+
+class TestTruncate:
+    def test_dont_truncate(self):
+        assert utils.truncate("short") == ("short", False)
+
+    def test_truncate_default_length(self):
+        assert utils.truncate("x" * 1000) == ("x" * utils.TRUNCATE_MAX_LENGTH, True)
+
+    def test_truncate_custom_length(self):
+        assert utils.truncate("x" * 1000, max_len=500) == ("x" * 500, True)