𝗖𝗨𝗥𝗥𝗘𝗡𝗧 𝗙𝗜𝗟𝗘 : /opt/hc_python/lib/python3.12/site-packages/sentry_sdk//_log_batcher.py
from typing import TYPE_CHECKING
from sentry_sdk._batcher import Batcher
from sentry_sdk.envelope import Item, PayloadRef
from sentry_sdk.utils import serialize_attribute
if TYPE_CHECKING:
from typing import Any
from sentry_sdk._types import Log
class LogBatcher(Batcher["Log"]):
MAX_BEFORE_FLUSH = 100
MAX_BEFORE_DROP = 1_000
FLUSH_WAIT_TIME = 5.0
TYPE = "log"
CONTENT_TYPE = "application/vnd.sentry.items.log+json"
@staticmethod
def _to_transport_format(item: "Log") -> "Any":
if "sentry.severity_number" not in item["attributes"]:
item["attributes"]["sentry.severity_number"] = item["severity_number"]
if "sentry.severity_text" not in item["attributes"]:
item["attributes"]["sentry.severity_text"] = item["severity_text"]
res = {
"timestamp": int(item["time_unix_nano"]) / 1.0e9,
"trace_id": item.get("trace_id", "00000000-0000-0000-0000-000000000000"),
"span_id": item.get("span_id"),
"level": str(item["severity_text"]),
"body": str(item["body"]),
"attributes": {
k: serialize_attribute(v) for (k, v) in item["attributes"].items()
},
}
return res
def _record_lost(self, item: "Log") -> None:
# Construct log envelope item without sending it to report lost bytes
log_item = Item(
type=self.TYPE,
content_type=self.CONTENT_TYPE,
headers={
"item_count": 1,
},
payload=PayloadRef(
json={
"version": 2,
"items": [self._to_transport_format(item)],
}
),
)
self._record_lost_func(
reason="queue_overflow",
data_category="log_item",
item=log_item,
quantity=1,
)
