Fix merging dicts for duplicate dotted keys

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## Unreleased
+
+* Fixed an issue in `StructlogFormatter` caused by a conflict with `event`
+  (used for the log `message`) and `event.dataset` (a field provided by the
+  `elasticapm` integration) ([#46](https://github.com/elastic/ecs-logging-python/pull/46))
+
 ## 1.0.0 (2021-02-08)
 
 * Remove "beta" designation

ecs_logging/_structlog.py

@@ -29,14 +29,17 @@ class StructlogFormatter:
 
     def __call__(self, _, name, event_dict):
         # type: (Any, str, Dict[str, Any]) -> str
+
+        # Handle event -> message now so that stuff like `event.dataset` doesn't
+        # cause problems down the line
+        event_dict["message"] = str(event_dict.pop("event"))
         event_dict = normalize_dict(event_dict)
         event_dict.setdefault("log", {}).setdefault("level", name.lower())
         event_dict = self.format_to_ecs(event_dict)
         return json_dumps(event_dict)
 
     def format_to_ecs(self, event_dict):
         # type: (Dict[str, Any]) -> Dict[str, Any]
-        event_dict["message"] = event_dict.pop("event")
         if "@timestamp" not in event_dict:
             event_dict["@timestamp"] = (
                 datetime.datetime.utcfromtimestamp(time.time()).strftime(

ecs_logging/_utils.py

@@ -111,8 +111,16 @@ def merge_dicts(from_, into):
     When called has side-effects within 'destination'.
     """
     for key, value in from_.items():
-        if isinstance(value, dict):
-            merge_dicts(value, into.setdefault(key, {}))
+        into.setdefault(key, {})
+        if isinstance(value, dict) and isinstance(into[key], dict):
+            merge_dicts(value, into[key])
+        elif into[key] != {}:
+            raise TypeError(
+                "Type mismatch at key `{}`: merging dicts would replace value `{}` with `{}`. This is likely due to "
+                "dotted keys in the event dict being turned into nested dictionaries, causing a conflict.".format(
+                    key, into[key], value
+                )
+            )
         else:
             into[key] = value
     return into

tests/conftest.py

@@ -19,6 +19,9 @@
 import json
 import os
 import collections
+import sys
+import logging
+import elasticapm
 
 import pytest
 
@@ -27,6 +30,10 @@ class ValidationError(Exception):
     pass
 
 
+if sys.version_info[0] >= 3:
+    basestring = str
+
+
 @pytest.fixture
 def spec_validator():
     with open(
@@ -81,3 +88,16 @@ def validator(data_json):
         return data_json
 
     return validator
+
+
+@pytest.fixture
+def apm():
+    if sys.version_info < (3, 6):
+        pytest.skip("elasticapm only supports python 3.6+")
+    if sys.version_info[0] >= 3:
+        record_factory = logging.getLogRecordFactory()
+    apm = elasticapm.Client({"SERVICE_NAME": "apm-service", "DISABLE_SEND": True})
+    yield apm
+    apm.close()
+    if sys.version_info[0] >= 3:
+        logging.setLogRecordFactory(record_factory)

tests/test_apm.py

@@ -27,8 +27,7 @@
 from .compat import StringIO
 
 
-def test_elasticapm_structlog_log_correlation_ecs_fields(spec_validator):
-    apm = elasticapm.Client({"SERVICE_NAME": "apm-service", "DISABLE_SEND": True})
+def test_elasticapm_structlog_log_correlation_ecs_fields(spec_validator, apm):
     stream = StringIO()
     logger = structlog.PrintLogger(stream)
     logger = structlog.wrap_logger(
@@ -57,14 +56,11 @@ def test_elasticapm_structlog_log_correlation_ecs_fields(spec_validator):
         "trace": {"id": trace_id},
         "transaction": {"id": transaction_id},
         "service": {"name": "apm-service"},
+        "event": {"dataset": "apm-service.log"},
     }
 
 
-@pytest.mark.skipif(
-    sys.version_info < (3, 2), reason="elastic-apm uses logger factory in Python 3.2+"
-)
-def test_elastic_apm_stdlib_no_filter_log_correlation_ecs_fields():
-    apm = elasticapm.Client({"SERVICE_NAME": "apm-service", "DISABLE_SEND": True})
+def test_elastic_apm_stdlib_no_filter_log_correlation_ecs_fields(apm):
     stream = StringIO()
     logger = logging.getLogger("apm-logger")
     handler = logging.StreamHandler(stream)
@@ -104,11 +100,11 @@ def test_elastic_apm_stdlib_no_filter_log_correlation_ecs_fields():
         "trace": {"id": trace_id},
         "transaction": {"id": transaction_id},
         "service": {"name": "apm-service"},
+        "event": {"dataset": "apm-service.log"},
     }
 
 
-def test_elastic_apm_stdlib_with_filter_log_correlation_ecs_fields():
-    apm = elasticapm.Client({"SERVICE_NAME": "apm-service", "DISABLE_SEND": True})
+def test_elastic_apm_stdlib_with_filter_log_correlation_ecs_fields(apm):
     stream = StringIO()
     logger = logging.getLogger("apm-logger")
     handler = logging.StreamHandler(stream)
@@ -149,11 +145,11 @@ def test_elastic_apm_stdlib_with_filter_log_correlation_ecs_fields():
         "trace": {"id": trace_id},
         "transaction": {"id": transaction_id},
         "service": {"name": "apm-service"},
+        "event": {"dataset": "apm-service.log"},
     }
 
 
-def test_elastic_apm_stdlib_exclude_fields():
-    apm = elasticapm.Client({"SERVICE_NAME": "apm-service", "DISABLE_SEND": True})
+def test_elastic_apm_stdlib_exclude_fields(apm):
     stream = StringIO()
     logger = logging.getLogger("apm-logger")
     handler = logging.StreamHandler(stream)
@@ -195,4 +191,5 @@ def test_elastic_apm_stdlib_exclude_fields():
         "message": "test message",
         "trace": {"id": trace_id},
         "service": {"name": "apm-service"},
+        "event": {"dataset": "apm-service.log"},
     }

tests/test_structlog_formatter.py

@@ -20,9 +20,24 @@
 import mock
 from .compat import StringIO
 
+import pytest
+
 
 def make_event_dict():
-    return {"event": "test message", "log.logger": "logger-name"}
+    return {
+        "event": "test message",
+        "event.dataset": "agent.log",
+        "log.logger": "logger-name",
+        "foo": "bar",
+    }
+
+
+def test_conflicting_event_dict():
+    formatter = ecs_logging.StructlogFormatter()
+    event_dict = make_event_dict()
+    event_dict["foo.bar"] = "baz"
+    with pytest.raises(TypeError):
+        formatter(None, "debug", event_dict)
 
 
 @mock.patch("time.time")
@@ -33,6 +48,8 @@ def test_event_dict_formatted(time, spec_validator):
     assert spec_validator(formatter(None, "debug", make_event_dict())) == (
         '{"@timestamp":"2020-03-20T16:16:37.187Z","log.level":"debug",'
         '"message":"test message","ecs":{"version":"1.6.0"},'
+        '"event":{"dataset":"agent.log"},'
+        '"foo":"bar",'
         '"log":{"logger":"logger-name"}}'
     )