prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  apm-agent-python and structlog - mapper_parsing_exception for `event.dataset` 
  **Describe the bug**: When using `structlog` and with `elasticapm` and the log processer `elasticapm.structlog_processor`, we have recently seen Logstash refuse our logger with the following error: 
  
  ```
  [2021-10-05T12:10:10,746][WARN ][logstash.outputs.elasticsearch][main][a2a92c7c0ddf765b1969e7e8d4a302b6deca976af4c80a2d9706ccdf2486267b] Could not index event to Elasticsearch. {:status=>400, :action=>["index", {:_id=>nil, :_index=>"logstash-2021.10.05", :routing=>nil}, {"stage_environment"=>"prod", "service.name"=>"flow", "company"=>"<PayerCompany: Hidden Company Name [Parent Company: Hidden Company Name [CRN: DKXXXXXXXX]] [CRN: DKXXXXXXXX]>", "@version"=>"1", "host"=>"167.71.1.240", "sentry"=>"skipped", "timestamp"=>"2021-10-05T12:10:00.483890Z", "logger"=>"account_service.models", "event.dataset"=>"flow", "event"=>"PayerCompany change state request", "level"=>"debug", "port"=>58652, "new"=>"APPROVED", "override"=>false, "@timestamp"=>2021-10-05T12:10:10.462Z, "old"=>"APPROVED", "modline"=>"account_service.models:159"}], :response=>{"index"=>{"_index"=>"logstash-2021.10.05", "_type"=>"_doc", "_id"=>"ST1cUHwBFM723LU2e_JV", "status"=>400, "error"=>{"type"=>"mapper_parsing_exception", "reason"=>"Could not dynamically add mapping for field [event.dataset]. Existing mapping for [event] must be of type object but found [text]."}}}}
  ```
  
  We setup our structlog like this in the Django settings file: 
  
  ```python
  processors = [
      structlog.stdlib.filter_by_level,
      structlog.stdlib.add_log_level,
      structlog.stdlib.add_logger_name,
      log_processors.add_module_and_lineno,
      log_processors.normalize_datatypes,
      log_processors.attach_environment,
      log_processors.timestamper,
      structlog_processor,  # this is the processor that seems to cause the error
      SentryJsonProcessor(level=logging.ERROR, tag_keys="__all__"),
      structlog.stdlib.PositionalArgumentsFormatter(),
      structlog.processors.StackInfoRenderer(),
      structlog.processors.format_exc_info,
      structlog.processors.UnicodeDecoder(),
      structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
  ]
  
  
  # Structlog
  structlog.configure(
      processors=processors,
      context_class=structlog.threadlocal.wrap_dict(dict),
      logger_factory=structlog.stdlib.LoggerFactory(),
      wrapper_class=structlog.stdlib.BoundLogger,
      cache_logger_on_first_use=True,
  )
  ```
  
  If we remove `structlog_processor` all loggers are received by Logstash with no problems. 
  
  If we write a small customer log processor that strips the `event.dataset` (e.g. `del event_dict["event.dataset"]` it also works fine again. 
  
  **To Reproduce**
  
  1. Setup structlog
  2. Add the elasticapm.structlog_processor to the list of processors for structlog
  3. Send a logger (e.g. `logger.info("test")`). 
  
  **Environment (please complete the following information)**
  - OS: Debian (slim)
  - Python version: 3.9.7
  - Framework and version [e.g. Django 2.1]: 3.2.8
  - APM Server version: 7.15
  - Agent version: N/A
  
  
  **Additional context**
  
  We use `python-logstash-async` for delivery of the logs to the logstash server. 
  
  requirements.txt (excerpts)
  ```pip
  elastic-apm==6.5.0
  elasticsearch==7.15.0
  django-structlog==2.1.3
  python-logstash-async==2.3.0
  structlog==21.1.0
  structlog-sentry==1.4.0
  ``` 
  
  --- END ISSUE ---
  
  Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
  --- BEGIN FILES ---
  Path: `elasticapm/handlers/structlog.py`
  Content:
  ```
  1 #  Copyright (c) 2019, Elasticsearch BV
  2 #  All rights reserved.
  3 #
  4 #  Redistribution and use in source and binary forms, with or without
  5 #  modification, are permitted provided that the following conditions are met:
  6 #
  7 #  * Redistributions of source code must retain the above copyright notice, this
  8 #    list of conditions and the following disclaimer.
  9 #
  10 #  * Redistributions in binary form must reproduce the above copyright notice,
  11 #    this list of conditions and the following disclaimer in the documentation
  12 #    and/or other materials provided with the distribution.
  13 #
  14 #  * Neither the name of the copyright holder nor the names of its
  15 #    contributors may be used to endorse or promote products derived from
  16 #    this software without specific prior written permission.
  17 #
  18 #  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  19 #  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  20 #  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  21 #  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
  22 #  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  23 #  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
  24 #  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
  25 #  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
  26 #  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  27 
  28 
  29 from __future__ import absolute_import
  30 
  31 from elasticapm import get_client
  32 from elasticapm.traces import execution_context
  33 
  34 
  35 def structlog_processor(logger, method_name, event_dict):
  36     """
  37     Add three new entries to the event_dict for any processed events:
  38 
  39     * transaction.id
  40     * trace.id
  41     * span.id
  42 
  43     Only adds non-None IDs.
  44 
  45     :param logger:
  46         Unused (logger instance in structlog)
  47     :param method_name:
  48         Unused (wrapped method_name)
  49     :param event_dict:
  50         Event dictionary for the event we're processing
  51     :return:
  52         `event_dict`, with three new entries.
  53     """
  54     transaction = execution_context.get_transaction()
  55     if transaction:
  56         event_dict["transaction.id"] = transaction.id
  57     client = get_client()
  58     if client:
  59         event_dict["service.name"] = client.config.service_name
  60         event_dict["event.dataset"] = f"{client.config.service_name}"
  61     if transaction and transaction.trace_parent:
  62         event_dict["trace.id"] = transaction.trace_parent.trace_id
  63     span = execution_context.get_span()
  64     if span and span.id:
  65         event_dict["span.id"] = span.id
  66     return event_dict
  67
  ```
  --- END FILES ---
  
  Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.
  
  Here is an example:
  
  ```diff
  diff --git a/examples/server_async.py b/examples/server_async.py
  --- a/examples/server_async.py
  +++ b/examples/server_async.py
  @@ -313,4 +313,4 @@
   
   
   if __name__ == "__main__":
  -    asyncio.run(run_async_server("."), debug=True)
  +    asyncio.run(run_async_server(), debug=True)
  diff --git a/examples/server_sync.py b/examples/server_sync.py
  --- a/examples/server_sync.py
  +++ b/examples/server_sync.py
  @@ -313,5 +313,5 @@
   
   
   if __name__ == "__main__":
  -    server = run_sync_server(".")
  +    server = run_sync_server()
       server.shutdown()
  
  ```
  
golden_diff: diff --git a/elasticapm/handlers/structlog.py b/elasticapm/handlers/structlog.py
  --- a/elasticapm/handlers/structlog.py
  +++ b/elasticapm/handlers/structlog.py
  @@ -57,7 +57,7 @@
       client = get_client()
       if client:
           event_dict["service.name"] = client.config.service_name
  -        event_dict["event.dataset"] = f"{client.config.service_name}"
  +        event_dict["event"] = {"dataset": f"{client.config.service_name}"}
       if transaction and transaction.trace_parent:
           event_dict["trace.id"] = transaction.trace_parent.trace_id
       span = execution_context.get_span()
  
verification_info: {"golden_diff": "diff --git a/elasticapm/handlers/structlog.py b/elasticapm/handlers/structlog.py\n--- a/elasticapm/handlers/structlog.py\n+++ b/elasticapm/handlers/structlog.py\n@@ -57,7 +57,7 @@\n     client = get_client()\n     if client:\n         event_dict[\"service.name\"] = client.config.service_name\n-        event_dict[\"event.dataset\"] = f\"{client.config.service_name}\"\n+        event_dict[\"event\"] = {\"dataset\": f\"{client.config.service_name}\"}\n     if transaction and transaction.trace_parent:\n         event_dict[\"trace.id\"] = transaction.trace_parent.trace_id\n     span = execution_context.get_span()\n", "issue": "apm-agent-python and structlog - mapper_parsing_exception for `event.dataset` \n**Describe the bug**: When using `structlog` and with `elasticapm` and the log processer `elasticapm.structlog_processor`, we have recently seen Logstash refuse our logger with the following error: \r\n\r\n```\r\n[2021-10-05T12:10:10,746][WARN ][logstash.outputs.elasticsearch][main][a2a92c7c0ddf765b1969e7e8d4a302b6deca976af4c80a2d9706ccdf2486267b] Could not index event to Elasticsearch. {:status=>400, :action=>[\"index\", {:_id=>nil, :_index=>\"logstash-2021.10.05\", :routing=>nil}, {\"stage_environment\"=>\"prod\", \"service.name\"=>\"flow\", \"company\"=>\"<PayerCompany: Hidden Company Name [Parent Company: Hidden Company Name [CRN: DKXXXXXXXX]] [CRN: DKXXXXXXXX]>\", \"@version\"=>\"1\", \"host\"=>\"167.71.1.240\", \"sentry\"=>\"skipped\", \"timestamp\"=>\"2021-10-05T12:10:00.483890Z\", \"logger\"=>\"account_service.models\", \"event.dataset\"=>\"flow\", \"event\"=>\"PayerCompany change state request\", \"level\"=>\"debug\", \"port\"=>58652, \"new\"=>\"APPROVED\", \"override\"=>false, \"@timestamp\"=>2021-10-05T12:10:10.462Z, \"old\"=>\"APPROVED\", \"modline\"=>\"account_service.models:159\"}], :response=>{\"index\"=>{\"_index\"=>\"logstash-2021.10.05\", \"_type\"=>\"_doc\", \"_id\"=>\"ST1cUHwBFM723LU2e_JV\", \"status\"=>400, \"error\"=>{\"type\"=>\"mapper_parsing_exception\", \"reason\"=>\"Could not dynamically add mapping for field [event.dataset]. Existing mapping for [event] must be of type object but found [text].\"}}}}\r\n```\r\n\r\nWe setup our structlog like this in the Django settings file: \r\n\r\n```python\r\nprocessors = [\r\n    structlog.stdlib.filter_by_level,\r\n    structlog.stdlib.add_log_level,\r\n    structlog.stdlib.add_logger_name,\r\n    log_processors.add_module_and_lineno,\r\n    log_processors.normalize_datatypes,\r\n    log_processors.attach_environment,\r\n    log_processors.timestamper,\r\n    structlog_processor,  # this is the processor that seems to cause the error\r\n    SentryJsonProcessor(level=logging.ERROR, tag_keys=\"__all__\"),\r\n    structlog.stdlib.PositionalArgumentsFormatter(),\r\n    structlog.processors.StackInfoRenderer(),\r\n    structlog.processors.format_exc_info,\r\n    structlog.processors.UnicodeDecoder(),\r\n    structlog.stdlib.ProcessorFormatter.wrap_for_formatter,\r\n]\r\n\r\n\r\n# Structlog\r\nstructlog.configure(\r\n    processors=processors,\r\n    context_class=structlog.threadlocal.wrap_dict(dict),\r\n    logger_factory=structlog.stdlib.LoggerFactory(),\r\n    wrapper_class=structlog.stdlib.BoundLogger,\r\n    cache_logger_on_first_use=True,\r\n)\r\n```\r\n\r\nIf we remove `structlog_processor` all loggers are received by Logstash with no problems. \r\n\r\nIf we write a small customer log processor that strips the `event.dataset` (e.g. `del event_dict[\"event.dataset\"]` it also works fine again. \r\n\r\n**To Reproduce**\r\n\r\n1. Setup structlog\r\n2. Add the elasticapm.structlog_processor to the list of processors for structlog\r\n3. Send a logger (e.g. `logger.info(\"test\")`). \r\n\r\n**Environment (please complete the following information)**\r\n- OS: Debian (slim)\r\n- Python version: 3.9.7\r\n- Framework and version [e.g. Django 2.1]: 3.2.8\r\n- APM Server version: 7.15\r\n- Agent version: N/A\r\n\r\n\r\n**Additional context**\r\n\r\nWe use `python-logstash-async` for delivery of the logs to the logstash server. \r\n\r\nrequirements.txt (excerpts)\r\n```pip\r\nelastic-apm==6.5.0\r\nelasticsearch==7.15.0\r\ndjango-structlog==2.1.3\r\npython-logstash-async==2.3.0\r\nstructlog==21.1.0\r\nstructlog-sentry==1.4.0\r\n``` \n", "before_files": [{"content": "#  Copyright (c) 2019, Elasticsearch BV\n#  All rights reserved.\n#\n#  Redistribution and use in source and binary forms, with or without\n#  modification, are permitted provided that the following conditions are met:\n#\n#  * Redistributions of source code must retain the above copyright notice, this\n#    list of conditions and the following disclaimer.\n#\n#  * Redistributions in binary form must reproduce the above copyright notice,\n#    this list of conditions and the following disclaimer in the documentation\n#    and/or other materials provided with the distribution.\n#\n#  * Neither the name of the copyright holder nor the names of its\n#    contributors may be used to endorse or promote products derived from\n#    this software without specific prior written permission.\n#\n#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS \"AS IS\"\n#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE\n#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE\n#  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE\n#  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL\n#  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR\n#  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER\n#  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,\n#  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\n\n\nfrom __future__ import absolute_import\n\nfrom elasticapm import get_client\nfrom elasticapm.traces import execution_context\n\n\ndef structlog_processor(logger, method_name, event_dict):\n    \"\"\"\n    Add three new entries to the event_dict for any processed events:\n\n    * transaction.id\n    * trace.id\n    * span.id\n\n    Only adds non-None IDs.\n\n    :param logger:\n        Unused (logger instance in structlog)\n    :param method_name:\n        Unused (wrapped method_name)\n    :param event_dict:\n        Event dictionary for the event we're processing\n    :return:\n        `event_dict`, with three new entries.\n    \"\"\"\n    transaction = execution_context.get_transaction()\n    if transaction:\n        event_dict[\"transaction.id\"] = transaction.id\n    client = get_client()\n    if client:\n        event_dict[\"service.name\"] = client.config.service_name\n        event_dict[\"event.dataset\"] = f\"{client.config.service_name}\"\n    if transaction and transaction.trace_parent:\n        event_dict[\"trace.id\"] = transaction.trace_parent.trace_id\n    span = execution_context.get_span()\n    if span and span.id:\n        event_dict[\"span.id\"] = span.id\n    return event_dict\n", "path": "elasticapm/handlers/structlog.py"}], "after_files": [{"content": "#  Copyright (c) 2019, Elasticsearch BV\n#  All rights reserved.\n#\n#  Redistribution and use in source and binary forms, with or without\n#  modification, are permitted provided that the following conditions are met:\n#\n#  * Redistributions of source code must retain the above copyright notice, this\n#    list of conditions and the following disclaimer.\n#\n#  * Redistributions in binary form must reproduce the above copyright notice,\n#    this list of conditions and the following disclaimer in the documentation\n#    and/or other materials provided with the distribution.\n#\n#  * Neither the name of the copyright holder nor the names of its\n#    contributors may be used to endorse or promote products derived from\n#    this software without specific prior written permission.\n#\n#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS \"AS IS\"\n#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE\n#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE\n#  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE\n#  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL\n#  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR\n#  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER\n#  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,\n#  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\n\n\nfrom __future__ import absolute_import\n\nfrom elasticapm import get_client\nfrom elasticapm.traces import execution_context\n\n\ndef structlog_processor(logger, method_name, event_dict):\n    \"\"\"\n    Add three new entries to the event_dict for any processed events:\n\n    * transaction.id\n    * trace.id\n    * span.id\n\n    Only adds non-None IDs.\n\n    :param logger:\n        Unused (logger instance in structlog)\n    :param method_name:\n        Unused (wrapped method_name)\n    :param event_dict:\n        Event dictionary for the event we're processing\n    :return:\n        `event_dict`, with three new entries.\n    \"\"\"\n    transaction = execution_context.get_transaction()\n    if transaction:\n        event_dict[\"transaction.id\"] = transaction.id\n    client = get_client()\n    if client:\n        event_dict[\"service.name\"] = client.config.service_name\n        event_dict[\"event\"] = {\"dataset\": f\"{client.config.service_name}\"}\n    if transaction and transaction.trace_parent:\n        event_dict[\"trace.id\"] = transaction.trace_parent.trace_id\n    span = execution_context.get_span()\n    if span and span.id:\n        event_dict[\"span.id\"] = span.id\n    return event_dict\n", "path": "elasticapm/handlers/structlog.py"}]}