diff --git a/elasticapm/conf/__init__.py b/elasticapm/conf/__init__.py
index d787491ab..6d19eb96c 100644
--- a/elasticapm/conf/__init__.py
+++ b/elasticapm/conf/__init__.py
@@ -883,6 +883,15 @@ def setup_logging(handler):
 
     For a typical Python install:
 
+    >>> from elasticapm.handlers.logging import LoggingHandler
+    >>> client = ElasticAPM(...)
+    >>> setup_logging(LoggingHandler(client))
+
+    Within Django:
+
+    >>> from elasticapm.contrib.django.handlers import LoggingHandler
+    >>> setup_logging(LoggingHandler())
+
     Returns a boolean based on if logging was configured or not.
     """
     # TODO We should probably revisit this. Does it make more sense as
diff --git a/elasticapm/contrib/django/handlers.py b/elasticapm/contrib/django/handlers.py
index 550cfae87..c980acc4f 100644
--- a/elasticapm/contrib/django/handlers.py
+++ b/elasticapm/contrib/django/handlers.py
@@ -31,11 +31,44 @@
 
 from __future__ import absolute_import
 
+import logging
 import sys
 import warnings
 
 from django.conf import settings as django_settings
 
+from elasticapm import get_client
+from elasticapm.handlers.logging import LoggingHandler as BaseLoggingHandler
+from elasticapm.utils.logging import get_logger
+
+logger = get_logger("elasticapm.logging")
+
+
+class LoggingHandler(BaseLoggingHandler):
+    def __init__(self, level=logging.NOTSET) -> None:
+        warnings.warn(
+            "The LoggingHandler is deprecated and will be removed in v7.0 of the agent. "
+            "Please use `log_ecs_reformatting` and ship the logs with Elastic "
+            "Agent or Filebeat instead. "
+            "https://www.elastic.co/guide/en/apm/agent/python/current/logs.html",
+            DeprecationWarning,
+        )
+        # skip initialization of BaseLoggingHandler
+        logging.Handler.__init__(self, level=level)
+
+    @property
+    def client(self):
+        return get_client()
+
+    def _emit(self, record, **kwargs):
+        from elasticapm.contrib.django.middleware import LogMiddleware
+
+        # Fetch the request from a threadlocal variable, if available
+        request = getattr(LogMiddleware.thread, "request", None)
+        request = getattr(record, "request", request)
+
+        return super(LoggingHandler, self)._emit(record, request=request, **kwargs)
+
 
 def exception_handler(client, request=None, **kwargs):
     def actually_do_stuff(request=None, **kwargs) -> None:
diff --git a/elasticapm/contrib/flask/__init__.py b/elasticapm/contrib/flask/__init__.py
index 4be9fe7ae..fdb6906dd 100644
--- a/elasticapm/contrib/flask/__init__.py
+++ b/elasticapm/contrib/flask/__init__.py
@@ -31,6 +31,9 @@
 
 from __future__ import absolute_import
 
+import logging
+import warnings
+
 import flask
 from flask import request, signals
 
@@ -38,8 +41,9 @@
 import elasticapm.instrumentation.control
 from elasticapm import get_client
 from elasticapm.base import Client
-from elasticapm.conf import constants
+from elasticapm.conf import constants, setup_logging
 from elasticapm.contrib.flask.utils import get_data_from_request, get_data_from_response
+from elasticapm.handlers.logging import LoggingHandler
 from elasticapm.traces import execution_context
 from elasticapm.utils import build_name_with_http_method_prefix
 from elasticapm.utils.disttracing import TraceParent
@@ -77,14 +81,17 @@ class ElasticAPM(object):
     >>> elasticapm.capture_message('hello, world!')
     """
 
-    def __init__(self, app=None, client=None, client_cls=Client, **defaults) -> None:
+    def __init__(self, app=None, client=None, client_cls=Client, logging=False, **defaults) -> None:
         self.app = app
+        self.logging = logging
+        if self.logging:
+            warnings.warn(
+                "Flask log shipping is deprecated. See the Flask docs for more info and alternatives.",
+                DeprecationWarning,
+            )
         self.client = client or get_client()
         self.client_cls = client_cls
 
-        if "logging" in defaults:
-            raise ValueError("Flask log shipping has been removed, drop the ElasticAPM logging parameter")
-
         if app:
             self.init_app(app, **defaults)
 
@@ -120,6 +127,14 @@ def init_app(self, app, **defaults) -> None:
 
             self.client = self.client_cls(config, **defaults)
 
+        # 0 is a valid log level (NOTSET), so we need to check explicitly for it
+        if self.logging or self.logging is logging.NOTSET:
+            if self.logging is not True:
+                kwargs = {"level": self.logging}
+            else:
+                kwargs = {}
+            setup_logging(LoggingHandler(self.client, **kwargs))
+
         signals.got_request_exception.connect(self.handle_exception, sender=app, weak=False)
 
         try:
diff --git a/elasticapm/handlers/logging.py b/elasticapm/handlers/logging.py
index bcdd15bb0..96718d2db 100644
--- a/elasticapm/handlers/logging.py
+++ b/elasticapm/handlers/logging.py
@@ -32,11 +32,181 @@
 from __future__ import absolute_import
 
 import logging
+import sys
+import traceback
+import warnings
 
 import wrapt
 
 from elasticapm import get_client
+from elasticapm.base import Client
 from elasticapm.traces import execution_context
+from elasticapm.utils.stacks import iter_stack_frames
+
+
+class LoggingHandler(logging.Handler):
+    def __init__(self, *args, **kwargs) -> None:
+        warnings.warn(
+            "The LoggingHandler is deprecated and will be removed in v7.0 of "
+            "the agent. Please use `log_ecs_reformatting` and ship the logs "
+            "with Elastic Agent or Filebeat instead. "
+            "https://www.elastic.co/guide/en/apm/agent/python/current/logs.html",
+            DeprecationWarning,
+        )
+        self.client = None
+        if "client" in kwargs:
+            self.client = kwargs.pop("client")
+        elif len(args) > 0:
+            arg = args[0]
+            if isinstance(arg, Client):
+                self.client = arg
+
+        if not self.client:
+            client_cls = kwargs.pop("client_cls", None)
+            if client_cls:
+                self.client = client_cls(*args, **kwargs)
+            else:
+                warnings.warn(
+                    "LoggingHandler requires a Client instance. No Client was received.",
+                    DeprecationWarning,
+                )
+                self.client = Client(*args, **kwargs)
+        logging.Handler.__init__(self, level=kwargs.get("level", logging.NOTSET))
+
+    def emit(self, record):
+        self.format(record)
+
+        # Avoid typical config issues by overriding loggers behavior
+        if record.name.startswith(("elasticapm.errors",)):
+            sys.stderr.write(record.getMessage() + "\n")
+            return
+
+        try:
+            return self._emit(record)
+        except Exception:
+            sys.stderr.write("Top level ElasticAPM exception caught - failed creating log record.\n")
+            sys.stderr.write(record.getMessage() + "\n")
+            sys.stderr.write(traceback.format_exc() + "\n")
+
+            try:
+                self.client.capture("Exception")
+            except Exception:
+                pass
+
+    def _emit(self, record, **kwargs):
+        data = {}
+
+        for k, v in record.__dict__.items():
+            if "." not in k and k not in ("culprit",):
+                continue
+            data[k] = v
+
+        stack = getattr(record, "stack", None)
+        if stack is True:
+            stack = iter_stack_frames(config=self.client.config)
+
+        if stack:
+            frames = []
+            started = False
+            last_mod = ""
+            for item in stack:
+                if isinstance(item, (list, tuple)):
+                    frame, lineno = item
+                else:
+                    frame, lineno = item, item.f_lineno
+
+                if not started:
+                    f_globals = getattr(frame, "f_globals", {})
+                    module_name = f_globals.get("__name__", "")
+                    if last_mod.startswith("logging") and not module_name.startswith("logging"):
+                        started = True
+                    else:
+                        last_mod = module_name
+                        continue
+                frames.append((frame, lineno))
+            stack = frames
+
+        custom = getattr(record, "data", {})
+        # Add in all of the data from the record that we aren't already capturing
+        for k in record.__dict__.keys():
+            if k in (
+                "stack",
+                "name",
+                "args",
+                "msg",
+                "levelno",
+                "exc_text",
+                "exc_info",
+                "data",
+                "created",
+                "levelname",
+                "msecs",
+                "relativeCreated",
+            ):
+                continue
+            if k.startswith("_"):
+                continue
+            custom[k] = record.__dict__[k]
+
+        # If there's no exception being processed,
+        # exc_info may be a 3-tuple of None
+        # http://docs.python.org/library/sys.html#sys.exc_info
+        if record.exc_info and all(record.exc_info):
+            handler = self.client.get_handler("elasticapm.events.Exception")
+            exception = handler.capture(self.client, exc_info=record.exc_info)
+        else:
+            exception = None
+
+        return self.client.capture(
+            "Message",
+            param_message={"message": str(record.msg), "params": record.args},
+            stack=stack,
+            custom=custom,
+            exception=exception,
+            level=record.levelno,
+            logger_name=record.name,
+            **kwargs,
+        )
+
+
+class LoggingFilter(logging.Filter):
+    """
+    This filter doesn't actually do any "filtering" -- rather, it just adds
+    three new attributes to any "filtered" LogRecord objects:
+
+    * elasticapm_transaction_id
+    * elasticapm_trace_id
+    * elasticapm_span_id
+    * elasticapm_service_name
+
+    These attributes can then be incorporated into your handlers and formatters,
+    so that you can tie log messages to transactions in elasticsearch.
+
+    This filter also adds these fields to a dictionary attribute,
+    `elasticapm_labels`, using the official tracing fields names as documented
+    here: https://www.elastic.co/guide/en/ecs/current/ecs-tracing.html
+
+    Note that if you're using Python 3.2+, by default we will add a
+    LogRecordFactory to your root logger which will add these attributes
+    automatically.
+    """
+
+    def __init__(self, name=""):
+        super().__init__(name=name)
+        warnings.warn(
+            "The LoggingFilter is deprecated and will be removed in v7.0 of "
+            "the agent. On Python 3.2+, by default we add a LogRecordFactory to "
+            "your root logger automatically"
+            "https://www.elastic.co/guide/en/apm/agent/python/current/logs.html",
+            DeprecationWarning,
+        )
+
+    def filter(self, record):
+        """
+        Add elasticapm attributes to `record`.
+        """
+        _add_attributes_to_log_record(record)
+        return True
 
 
 @wrapt.decorator
diff --git a/tests/contrib/django/django_tests.py b/tests/contrib/django/django_tests.py
index 312583ac1..94843a83f 100644
--- a/tests/contrib/django/django_tests.py
+++ b/tests/contrib/django/django_tests.py
@@ -62,6 +62,7 @@
 from elasticapm.conf.constants import ERROR, SPAN, TRANSACTION
 from elasticapm.contrib.django.apps import ElasticAPMConfig
 from elasticapm.contrib.django.client import client, get_client
+from elasticapm.contrib.django.handlers import LoggingHandler
 from elasticapm.contrib.django.middleware.wsgi import ElasticAPM
 from elasticapm.utils.disttracing import TraceParent
 from tests.contrib.django.conftest import BASE_TEMPLATE_DIR
@@ -409,6 +410,25 @@ def test_ignored_exception_is_ignored(django_elasticapm_client, client):
     assert len(django_elasticapm_client.events[ERROR]) == 0
 
 
+def test_record_none_exc_info(django_elasticapm_client):
+    # sys.exc_info can return (None, None, None) if no exception is being
+    # handled anywhere on the stack. See:
+    #  http://docs.python.org/library/sys.html#sys.exc_info
+    record = logging.LogRecord(
+        "foo", logging.INFO, pathname=None, lineno=None, msg="test", args=(), exc_info=(None, None, None)
+    )
+    handler = LoggingHandler()
+    handler.emit(record)
+
+    assert len(django_elasticapm_client.events[ERROR]) == 1
+    event = django_elasticapm_client.events[ERROR][0]
+
+    assert event["log"]["param_message"] == "test"
+    assert event["log"]["logger_name"] == "foo"
+    assert event["log"]["level"] == "info"
+    assert "exception" not in event
+
+
 def test_404_middleware(django_elasticapm_client, client):
     with override_settings(
         **middleware_setting(django.VERSION, ["elasticapm.contrib.django.middleware.Catch404Middleware"])
@@ -1012,6 +1032,54 @@ def test_filter_matches_module_only(django_sending_elasticapm_client):
     assert len(django_sending_elasticapm_client.httpserver.requests) == 1
 
 
+def test_django_logging_request_kwarg(django_elasticapm_client):
+    handler = LoggingHandler()
+
+    logger = logging.getLogger(__name__)
+    logger.handlers = []
+    logger.addHandler(handler)
+
+    logger.error(
+        "This is a test error",
+        extra={
+            "request": WSGIRequest(
+                environ={
+                    "wsgi.input": io.StringIO(),
+                    "REQUEST_METHOD": "POST",
+                    "SERVER_NAME": "testserver",
+                    "SERVER_PORT": "80",
+                    "CONTENT_TYPE": "application/json",
+                    "ACCEPT": "application/json",
+                }
+            )
+        },
+    )
+
+    assert len(django_elasticapm_client.events[ERROR]) == 1
+    event = django_elasticapm_client.events[ERROR][0]
+    assert "request" in event["context"]
+    request = event["context"]["request"]
+    assert request["method"] == "POST"
+
+
+def test_django_logging_middleware(django_elasticapm_client, client):
+    handler = LoggingHandler()
+
+    logger = logging.getLogger("logmiddleware")
+    logger.handlers = []
+    logger.addHandler(handler)
+    logger.level = logging.INFO
+
+    with override_settings(
+        **middleware_setting(django.VERSION, ["elasticapm.contrib.django.middleware.LogMiddleware"])
+    ):
+        client.get(reverse("elasticapm-logging"))
+    assert len(django_elasticapm_client.events[ERROR]) == 1
+    event = django_elasticapm_client.events[ERROR][0]
+    assert "request" in event["context"]
+    assert event["context"]["request"]["url"]["pathname"] == reverse("elasticapm-logging")
+
+
 def client_get(client, url):
     return client.get(url)
 
diff --git a/tests/contrib/flask/flask_tests.py b/tests/contrib/flask/flask_tests.py
index 38657a6f6..a54cfe75a 100644
--- a/tests/contrib/flask/flask_tests.py
+++ b/tests/contrib/flask/flask_tests.py
@@ -50,11 +50,6 @@
 pytestmark = pytest.mark.flask
 
 
-def test_logging_parameter_raises_exception():
-    with pytest.raises(ValueError, match="Flask log shipping has been removed, drop the ElasticAPM logging parameter"):
-        ElasticAPM(config=None, logging=True)
-
-
 def test_error_handler(flask_apm_client):
     client = flask_apm_client.app.test_client()
     response = client.get("/an-error/")
@@ -446,6 +441,32 @@ def test_rum_tracing_context_processor(flask_apm_client):
         assert callable(context["apm"]["span_id"])
 
 
+@pytest.mark.parametrize("flask_apm_client", [{"logging": True}], indirect=True)
+def test_logging_enabled(flask_apm_client):
+    logger = logging.getLogger()
+    logger.error("test")
+    error = flask_apm_client.client.events[ERROR][0]
+    assert error["log"]["level"] == "error"
+    assert error["log"]["message"] == "test"
+
+
+@pytest.mark.parametrize("flask_apm_client", [{"logging": False}], indirect=True)
+def test_logging_disabled(flask_apm_client):
+    logger = logging.getLogger()
+    logger.error("test")
+    assert len(flask_apm_client.client.events[ERROR]) == 0
+
+
+@pytest.mark.parametrize("flask_apm_client", [{"logging": logging.ERROR}], indirect=True)
+def test_logging_by_level(flask_apm_client):
+    logger = logging.getLogger()
+    logger.warning("test")
+    logger.error("test")
+    assert len(flask_apm_client.client.events[ERROR]) == 1
+    error = flask_apm_client.client.events[ERROR][0]
+    assert error["log"]["level"] == "error"
+
+
 def test_flask_transaction_ignore_urls(flask_apm_client):
     resp = flask_apm_client.app.test_client().get("/users/")
     resp.close()
diff --git a/tests/handlers/logging/logging_tests.py b/tests/handlers/logging/logging_tests.py
index 00a1a4ab6..8cc8fc4f1 100644
--- a/tests/handlers/logging/logging_tests.py
+++ b/tests/handlers/logging/logging_tests.py
@@ -40,13 +40,238 @@
 
 from elasticapm.conf import Config
 from elasticapm.conf.constants import ERROR
-from elasticapm.handlers.logging import Formatter
+from elasticapm.handlers.logging import Formatter, LoggingFilter, LoggingHandler
 from elasticapm.handlers.structlog import structlog_processor
 from elasticapm.traces import capture_span
 from elasticapm.utils.stacks import iter_stack_frames
 from tests.fixtures import TempStoreClient
 
 
+@pytest.fixture()
+def logger(elasticapm_client):
+    elasticapm_client.config.include_paths = ["tests", "elasticapm"]
+    handler = LoggingHandler(elasticapm_client)
+    logger = logging.getLogger(__name__)
+    logger.handlers = []
+    logger.addHandler(handler)
+    logger.client = elasticapm_client
+    logger.level = logging.INFO
+    return logger
+
+
+def test_logger_basic(logger):
+    logger.error("This is a test error")
+
+    assert len(logger.client.events) == 1
+    event = logger.client.events[ERROR][0]
+    assert event["log"]["logger_name"] == __name__
+    assert event["log"]["level"] == "error"
+    assert event["log"]["message"] == "This is a test error"
+    assert "stacktrace" in event["log"]
+    assert "exception" not in event
+    assert "param_message" in event["log"]
+    assert event["log"]["param_message"] == "This is a test error"
+
+
+def test_logger_warning(logger):
+    logger.warning("This is a test warning")
+    assert len(logger.client.events) == 1
+    event = logger.client.events[ERROR][0]
+    assert event["log"]["logger_name"] == __name__
+    assert event["log"]["level"] == "warning"
+    assert "exception" not in event
+    assert "param_message" in event["log"]
+    assert event["log"]["param_message"] == "This is a test warning"
+
+
+def test_logger_extra_data(logger):
+    logger.info("This is a test info with a url", extra=dict(data=dict(url="http://example.com")))
+    assert len(logger.client.events) == 1
+    event = logger.client.events[ERROR][0]
+    assert event["context"]["custom"]["url"] == "http://example.com"
+    assert "stacktrace" in event["log"]
+    assert "exception" not in event
+    assert "param_message" in event["log"]
+    assert event["log"]["param_message"] == "This is a test info with a url"
+
+
+def test_logger_exc_info(logger):
+    try:
+        raise ValueError("This is a test ValueError")
+    except ValueError:
+        logger.info("This is a test info with an exception", exc_info=True)
+
+    assert len(logger.client.events) == 1
+    event = logger.client.events[ERROR][0]
+
+    # assert event['message'] == 'This is a test info with an exception'
+    assert "exception" in event
+    assert "stacktrace" in event["exception"]
+    exc = event["exception"]
+    assert exc["type"] == "ValueError"
+    assert exc["message"] == "ValueError: This is a test ValueError"
+    assert "param_message" in event["log"]
+    assert event["log"]["message"] == "This is a test info with an exception"
+
+
+def test_message_params(logger):
+    logger.info("This is a test of %s", "args")
+    assert len(logger.client.events) == 1
+    event = logger.client.events[ERROR][0]
+    assert "exception" not in event
+    assert "param_message" in event["log"]
+    assert event["log"]["message"] == "This is a test of args"
+    assert event["log"]["param_message"] == "This is a test of %s"
+
+
+def test_record_stack(logger):
+    logger.info("This is a test of stacks", extra={"stack": True})
+    assert len(logger.client.events) == 1
+    event = logger.client.events[ERROR][0]
+    frames = event["log"]["stacktrace"]
+    assert len(frames) != 1
+    frame = frames[0]
+    assert frame["module"] == __name__
+    assert "exception" not in event
+    assert "param_message" in event["log"]
+    assert event["log"]["param_message"] == "This is a test of stacks"
+    assert event["culprit"] == "tests.handlers.logging.logging_tests.test_record_stack"
+    assert event["log"]["message"] == "This is a test of stacks"
+
+
+def test_no_record_stack(logger):
+    logger.info("This is a test of no stacks", extra={"stack": False})
+    assert len(logger.client.events) == 1
+    event = logger.client.events[ERROR][0]
+    assert event.get("culprit") == None
+    assert event["log"]["message"] == "This is a test of no stacks"
+    assert "stacktrace" not in event["log"]
+    assert "exception" not in event
+    assert "param_message" in event["log"]
+    assert event["log"]["param_message"] == "This is a test of no stacks"
+
+
+def test_no_record_stack_via_config(logger):
+    logger.client.config.auto_log_stacks = False
+    logger.info("This is a test of no stacks")
+    assert len(logger.client.events) == 1
+    event = logger.client.events[ERROR][0]
+    assert event.get("culprit") == None
+    assert event["log"]["message"] == "This is a test of no stacks"
+    assert "stacktrace" not in event["log"]
+    assert "exception" not in event
+    assert "param_message" in event["log"]
+    assert event["log"]["param_message"] == "This is a test of no stacks"
+
+
+def test_explicit_stack(logger):
+    logger.info("This is a test of stacks", extra={"stack": iter_stack_frames()})
+    assert len(logger.client.events) == 1
+    event = logger.client.events[ERROR][0]
+    assert "culprit" in event, event
+    assert event["culprit"] == "tests.handlers.logging.logging_tests.test_explicit_stack"
+    assert "message" in event["log"], event
+    assert event["log"]["message"] == "This is a test of stacks"
+    assert "exception" not in event
+    assert "param_message" in event["log"]
+    assert event["log"]["param_message"] == "This is a test of stacks"
+    assert "stacktrace" in event["log"]
+
+
+def test_extra_culprit(logger):
+    logger.info("This is a test of stacks", extra={"culprit": "foo.bar"})
+    assert len(logger.client.events) == 1
+    event = logger.client.events[ERROR][0]
+    assert event["culprit"] == "foo.bar"
+    assert "culprit" not in event["context"]["custom"]
+
+
+def test_logger_exception(logger):
+    try:
+        raise ValueError("This is a test ValueError")
+    except ValueError:
+        logger.exception("This is a test with an exception", extra={"stack": True})
+
+    assert len(logger.client.events) == 1
+    event = logger.client.events[ERROR][0]
+
+    assert event["log"]["message"] == "This is a test with an exception"
+    assert "stacktrace" in event["log"]
+    assert "exception" in event
+    exc = event["exception"]
+    assert exc["type"] == "ValueError"
+    assert exc["message"] == "ValueError: This is a test ValueError"
+    assert "param_message" in event["log"]
+    assert event["log"]["message"] == "This is a test with an exception"
+
+
+def test_client_arg(elasticapm_client):
+    handler = LoggingHandler(elasticapm_client)
+    assert handler.client == elasticapm_client
+
+
+def test_client_kwarg(elasticapm_client):
+    handler = LoggingHandler(client=elasticapm_client)
+    assert handler.client == elasticapm_client
+
+
+def test_logger_setup():
+    handler = LoggingHandler(
+        server_url="foo", service_name="bar", secret_token="baz", metrics_interval="0ms", client_cls=TempStoreClient
+    )
+    client = handler.client
+    assert client.config.server_url == "foo"
+    assert client.config.service_name == "bar"
+    assert client.config.secret_token == "baz"
+    assert handler.level == logging.NOTSET
+
+
+def test_logging_handler_emit_error(capsys, elasticapm_client):
+    handler = LoggingHandler(elasticapm_client)
+    handler._emit = lambda: 1 / 0
+    handler.emit(LogRecord("x", 1, "/ab/c/", 10, "Oops", (), None))
+    out, err = capsys.readouterr()
+    assert "Top level ElasticAPM exception caught" in err
+    assert "Oops" in err
+
+
+def test_logging_handler_dont_emit_elasticapm(capsys, elasticapm_client):
+    handler = LoggingHandler(elasticapm_client)
+    handler.emit(LogRecord("elasticapm.errors", 1, "/ab/c/", 10, "Oops", (), None))
+    out, err = capsys.readouterr()
+    assert "Oops" in err
+
+
+def test_logging_handler_emit_error_non_str_message(capsys, elasticapm_client):
+    handler = LoggingHandler(elasticapm_client)
+    handler._emit = lambda: 1 / 0
+    handler.emit(LogRecord("x", 1, "/ab/c/", 10, ValueError("oh no"), (), None))
+    out, err = capsys.readouterr()
+    assert "Top level ElasticAPM exception caught" in err
+    assert "oh no" in err
+
+
+def test_arbitrary_object(logger):
+    logger.error(["a", "list", "of", "strings"])
+    assert len(logger.client.events) == 1
+    event = logger.client.events[ERROR][0]
+    assert "param_message" in event["log"]
+    assert event["log"]["param_message"] == "['a', 'list', 'of', 'strings']"
+
+
+def test_logging_filter_no_span(elasticapm_client):
+    transaction = elasticapm_client.begin_transaction("test")
+    f = LoggingFilter()
+    record = logging.LogRecord(__name__, logging.DEBUG, __file__, 252, "dummy_msg", [], None)
+    f.filter(record)
+    assert record.elasticapm_transaction_id == transaction.id
+    assert record.elasticapm_service_name == transaction.tracer.config.service_name
+    assert record.elasticapm_service_environment == transaction.tracer.config.environment
+    assert record.elasticapm_trace_id == transaction.trace_parent.trace_id
+    assert record.elasticapm_span_id is None
+    assert record.elasticapm_labels
+
+
 def test_structlog_processor_no_span(elasticapm_client):
     transaction = elasticapm_client.begin_transaction("test")
     event_dict = {}
@@ -56,6 +281,37 @@ def test_structlog_processor_no_span(elasticapm_client):
     assert "span.id" not in new_dict
 
 
+@pytest.mark.parametrize("elasticapm_client", [{"transaction_max_spans": 5}], indirect=True)
+def test_logging_filter_span(elasticapm_client):
+    transaction = elasticapm_client.begin_transaction("test")
+    with capture_span("test") as span:
+        f = LoggingFilter()
+        record = logging.LogRecord(__name__, logging.DEBUG, __file__, 252, "dummy_msg", [], None)
+        f.filter(record)
+        assert record.elasticapm_transaction_id == transaction.id
+        assert record.elasticapm_service_name == transaction.tracer.config.service_name
+        assert record.elasticapm_service_environment == transaction.tracer.config.environment
+        assert record.elasticapm_trace_id == transaction.trace_parent.trace_id
+        assert record.elasticapm_span_id == span.id
+        assert record.elasticapm_labels
+
+    # Capture too many spans so we start dropping
+    for i in range(10):
+        with capture_span("drop"):
+            pass
+
+    # Test logging with DroppedSpan
+    with capture_span("drop") as span:
+        record = logging.LogRecord(__name__, logging.DEBUG, __file__, 252, "dummy_msg2", [], None)
+        f.filter(record)
+        assert record.elasticapm_transaction_id == transaction.id
+        assert record.elasticapm_service_name == transaction.tracer.config.service_name
+        assert record.elasticapm_service_environment == transaction.tracer.config.environment
+        assert record.elasticapm_trace_id == transaction.trace_parent.trace_id
+        assert record.elasticapm_span_id is None
+        assert record.elasticapm_labels
+
+
 @pytest.mark.parametrize("elasticapm_client", [{"transaction_max_spans": 5}], indirect=True)
 def test_structlog_processor_span(elasticapm_client):
     transaction = elasticapm_client.begin_transaction("test")
@@ -117,6 +373,18 @@ def test_formatter():
     assert hasattr(record, "elasticapm_service_environment")
 
 
+def test_logging_handler_no_client(recwarn):
+    # In 6.0, this should be changed to expect a ValueError instead of a log
+    warnings.simplefilter("always")
+    LoggingHandler(transport_class="tests.fixtures.DummyTransport")
+    while True:
+        # If we never find our desired warning this will eventually throw an
+        # AssertionError
+        w = recwarn.pop(DeprecationWarning)
+        if "LoggingHandler requires a Client instance" in w.message.args[0]:
+            return True
+
+
 @pytest.mark.parametrize(
     "elasticapm_client,expected",
     [