in ecs_logging/_stdlib.py [0:0]
def format_to_ecs(self, record: logging.LogRecord) -> Dict[str, Any]:
"""Function that can be overridden to add additional fields to
(or remove fields from) the JSON before being dumped into a string.
.. code-block: python
class MyFormatter(StdlibFormatter):
def format_to_ecs(self, record):
result = super().format_to_ecs(record)
del result["log"]["original"] # remove unwanted field(s)
result["my_field"] = "my_value" # add custom field
return result
"""
extractors: Dict[str, Callable[[logging.LogRecord], Any]] = {
"@timestamp": self._record_timestamp,
"ecs.version": lambda _: ECS_VERSION,
"log.level": lambda r: (r.levelname.lower() if r.levelname else None),
"log.origin.function": self._record_attribute("funcName"),
"log.origin.file.line": self._record_attribute("lineno"),
"log.origin.file.name": self._record_attribute("filename"),
"log.original": lambda r: r.getMessage(),
"log.logger": self._record_attribute("name"),
"process.pid": self._record_attribute("process"),
"process.name": self._record_attribute("processName"),
"process.thread.id": self._record_attribute("thread"),
"process.thread.name": self._record_attribute("threadName"),
"error.type": self._record_error_type,
"error.message": self._record_error_message,
"error.stack_trace": self._record_error_stack_trace,
}
result: Dict[str, Any] = {}
for field in set(extractors.keys()).difference(self._exclude_fields):
if self._is_field_excluded(field):
continue
value = extractors[field](record)
if value is not None:
# special case ecs.version that should not be de-dotted
if field == "ecs.version":
field_dict = {field: value}
else:
field_dict = de_dot(field, value)
merge_dicts(field_dict, result)
available = record.__dict__
# This is cleverness because 'message' is NOT a member
# key of ``record.__dict__`` the ``getMessage()`` method
# is effectively ``msg % args`` (actual keys) By manually
# adding 'message' to ``available``, it simplifies the code
available["message"] = record.getMessage()
# Pull all extras and flatten them to be sent into '_is_field_excluded'
# since they can be defined as 'extras={"http": {"method": "GET"}}'
extra_keys = set(available).difference(self._LOGRECORD_DICT)
extras = flatten_dict({key: available[key] for key in extra_keys})
# Merge in any global extra's
if self._extra is not None:
for field, value in self._extra.items():
merge_dicts(de_dot(field, value), extras)
# Pop all Elastic APM extras and add them
# to standard tracing ECS fields.
extras.setdefault("span.id", extras.pop("elasticapm_span_id", None))
extras.setdefault(
"transaction.id", extras.pop("elasticapm_transaction_id", None)
)
extras.setdefault("trace.id", extras.pop("elasticapm_trace_id", None))
extras.setdefault("service.name", extras.pop("elasticapm_service_name", None))
extras.setdefault(
"service.environment", extras.pop("elasticapm_service_environment", None)
)
# Merge in any keys that were set within 'extra={...}'
for field, value in extras.items():
if field.startswith("elasticapm_labels."):
continue # Unconditionally remove, we don't need this info.
if value is None or self._is_field_excluded(field):
continue
merge_dicts(de_dot(field, value), result)
# The following is mostly for the ecs format. You can't have 2x
# 'message' keys in _WANTED_ATTRS, so we set the value to
# 'log.original' in ecs, and this code block guarantees it
# still appears as 'message' too.
if not self._is_field_excluded("message"):
result.setdefault("message", available["message"])
return result