logger.py 2.6 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788
  1. # NOTE: this is the logger sentry exposes to users, not some generic logger.
  2. import functools
  3. import time
  4. from typing import Any, TYPE_CHECKING
  5. import sentry_sdk
  6. from sentry_sdk.utils import format_attribute, capture_internal_exceptions
  7. if TYPE_CHECKING:
  8. from sentry_sdk._types import Attributes
  9. OTEL_RANGES = [
  10. # ((severity level range), severity text)
  11. # https://opentelemetry.io/docs/specs/otel/logs/data-model
  12. ((1, 4), "trace"),
  13. ((5, 8), "debug"),
  14. ((9, 12), "info"),
  15. ((13, 16), "warn"),
  16. ((17, 20), "error"),
  17. ((21, 24), "fatal"),
  18. ]
  19. class _dict_default_key(dict): # type: ignore[type-arg]
  20. """dict that returns the key if missing."""
  21. def __missing__(self, key: str) -> str:
  22. return "{" + key + "}"
  23. def _capture_log(
  24. severity_text: str, severity_number: int, template: str, **kwargs: "Any"
  25. ) -> None:
  26. body = template
  27. attributes: "Attributes" = {}
  28. if "attributes" in kwargs:
  29. provided_attributes = kwargs.pop("attributes") or {}
  30. for attribute, value in provided_attributes.items():
  31. attributes[attribute] = format_attribute(value)
  32. for k, v in kwargs.items():
  33. attributes[f"sentry.message.parameter.{k}"] = format_attribute(v)
  34. if kwargs:
  35. # only attach template if there are parameters
  36. attributes["sentry.message.template"] = format_attribute(template)
  37. with capture_internal_exceptions():
  38. body = template.format_map(_dict_default_key(kwargs))
  39. sentry_sdk.get_current_scope()._capture_log(
  40. {
  41. "severity_text": severity_text,
  42. "severity_number": severity_number,
  43. "attributes": attributes,
  44. "body": body,
  45. "time_unix_nano": time.time_ns(),
  46. "trace_id": None,
  47. "span_id": None,
  48. }
  49. )
  50. trace = functools.partial(_capture_log, "trace", 1)
  51. debug = functools.partial(_capture_log, "debug", 5)
  52. info = functools.partial(_capture_log, "info", 9)
  53. warning = functools.partial(_capture_log, "warn", 13)
  54. error = functools.partial(_capture_log, "error", 17)
  55. fatal = functools.partial(_capture_log, "fatal", 21)
  56. def _otel_severity_text(otel_severity_number: int) -> str:
  57. for (lower, upper), severity in OTEL_RANGES:
  58. if lower <= otel_severity_number <= upper:
  59. return severity
  60. return "default"
  61. def _log_level_to_otel(level: int, mapping: "dict[Any, int]") -> "tuple[int, str]":
  62. for py_level, otel_severity_number in sorted(mapping.items(), reverse=True):
  63. if level >= py_level:
  64. return otel_severity_number, _otel_severity_text(otel_severity_number)
  65. return 0, "default"