Skip to content

feat(spans): Evict spans during insert #92393

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jan-auer merged 3 commits into from
Jun 2, 2025
Merged

feat(spans): Evict spans during insert #92393

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f0fca4f
feat(spans): Evict spans during insert
jan-auer May 28, 2025
e80062c
fix: Typing
jan-auer May 28, 2025
eb9623b
Merge branch 'master' into ref/spans-evict-oldest
jan-auer Jun 2, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 15 additions & 5 deletions src/sentry/scripts/spans/add-buffer.lua
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ local main_redirect_key = string.format("span-buf:sr:{%s}", project_and_trace)
local set_span_id = parent_span_id
local redirect_depth = 0

for i = 0, 10000 do -- theoretically this limit means that segment trees of depth 10k may not be joined together correctly.
for i = 0, 1000 do
local new_set_span = redis.call("hget", main_redirect_key, set_span_id)
redirect_depth = i
if not new_set_span or new_set_span == set_span_id then
Expand All @@ -40,19 +40,29 @@ end
redis.call("hset", main_redirect_key, span_id, set_span_id)
redis.call("expire", main_redirect_key, set_timeout)

local span_count = 0

local set_key = string.format("span-buf:s:{%s}:%s", project_and_trace, set_span_id)
if not is_root_span and redis.call("scard", span_key) > 0 then
redis.call("sunionstore", set_key, set_key, span_key)
if not is_root_span and redis.call("zcard", span_key) > 0 then
span_count = redis.call("zunionstore", set_key, 2, set_key, span_key)
redis.call("unlink", span_key)
end

local parent_key = string.format("span-buf:s:{%s}:%s", project_and_trace, parent_span_id)
if set_span_id ~= parent_span_id and redis.call("scard", parent_key) > 0 then
redis.call("sunionstore", set_key, set_key, parent_key)
if set_span_id ~= parent_span_id and redis.call("zcard", parent_key) > 0 then
span_count = redis.call("zunionstore", set_key, 2, set_key, parent_key)
redis.call("unlink", parent_key)
end
redis.call("expire", set_key, set_timeout)

if span_count == 0 then
span_count = redis.call("zcard", set_key)
end

if span_count > 1000 then
redis.call("zpopmin", set_key, span_count - 1000)
end

local has_root_span_key = string.format("span-buf:hrs:%s", set_key)
local has_root_span = redis.call("get", has_root_span_key) == "1" or is_root_span
if has_root_span then
Expand Down
21 changes: 8 additions & 13 deletions src/sentry/spans/buffer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -116,6 +116,7 @@ class Span(NamedTuple):
parent_span_id: str | None
project_id: int
payload: bytes
end_timestamp_precise: float
is_segment_span: bool = False

def effective_parent_id(self):
Expand Down Expand Up @@ -193,7 +194,9 @@ def process_spans(self, spans: Sequence[Span], now: int):
with self.client.pipeline(transaction=False) as p:
for (project_and_trace, parent_span_id), subsegment in trees.items():
set_key = f"span-buf:s:{{{project_and_trace}}}:{parent_span_id}"
p.sadd(set_key, *[span.payload for span in subsegment])
p.zadd(
set_key, {span.payload: span.end_timestamp_precise for span in subsegment}
)

p.execute()

Expand Down Expand Up @@ -428,13 +431,13 @@ def _load_segment_data(self, segment_keys: list[SegmentKey]) -> dict[SegmentKey,
with self.client.pipeline(transaction=False) as p:
current_keys = []
for key, cursor in cursors.items():
p.sscan(key, cursor=cursor, count=self.segment_page_size)
p.zscan(key, cursor=cursor, count=self.segment_page_size)
current_keys.append(key)

results = p.execute()

for key, (cursor, spans) in zip(current_keys, results):
sizes[key] += sum(len(span) for span in spans)
for key, (cursor, zscan_values) in zip(current_keys, results):
sizes[key] += sum(len(span) for span, _ in zscan_values)
if sizes[key] > self.max_segment_bytes:
metrics.incr("spans.buffer.flush_segments.segment_size_exceeded")
logger.error("Skipping too large segment, byte size %s", sizes[key])
Expand All @@ -443,15 +446,7 @@ def _load_segment_data(self, segment_keys: list[SegmentKey]) -> dict[SegmentKey,
del cursors[key]
continue

payloads[key].extend(spans)
if len(payloads[key]) > self.max_segment_spans:
metrics.incr("spans.buffer.flush_segments.segment_span_count_exceeded")
logger.error("Skipping too large segment, span count %s", len(payloads[key]))

del payloads[key]
del cursors[key]
continue

payloads[key].extend(span for span, _ in zscan_values)
if cursor == 0:
del cursors[key]
else:
Expand Down
5 changes: 4 additions & 1 deletion src/sentry/spans/consumers/process/factory.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
import time
from collections.abc import Callable, Mapping
from functools import partial
from typing import cast

import rapidjson
from arroyo.backends.kafka.consumer import KafkaPayload
Expand All @@ -10,6 +11,7 @@
from arroyo.processing.strategies.commit import CommitOffsets
from arroyo.processing.strategies.run_task import RunTask
from arroyo.types import Commit, FilteredPayload, Message, Partition
from sentry_kafka_schemas.schema_types.ingest_spans_v1 import SpanEvent

from sentry.spans.buffer import Span, SpansBuffer
from sentry.spans.consumers.process.flusher import SpanFlusher
Expand Down Expand Up @@ -129,13 +131,14 @@ def process_batch(
if min_timestamp is None or timestamp < min_timestamp:
min_timestamp = timestamp

val = rapidjson.loads(payload.value)
val = cast(SpanEvent, rapidjson.loads(payload.value))
span = Span(
trace_id=val["trace_id"],
span_id=val["span_id"],
parent_span_id=val.get("parent_span_id"),
project_id=val["project_id"],
payload=payload.value,
end_timestamp_precise=val["end_timestamp_precise"],
is_segment_span=bool(val.get("parent_span_id") is None or val.get("is_remote")),
)
spans.append(span)
Expand Down
2 changes: 2 additions & 0 deletions tests/sentry/spans/consumers/process/test_consumer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ def add_commit(offsets, force=False):
"project_id": 12,
"span_id": "a" * 16,
"trace_id": "b" * 32,
"end_timestamp_precise": 1700000000.0,
}
).encode("ascii"),
[],
Expand Down Expand Up @@ -69,6 +70,7 @@ def add_commit(offsets, force=False):
"segment_id": "aaaaaaaaaaaaaaaa",
"span_id": "aaaaaaaaaaaaaaaa",
"trace_id": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
"end_timestamp_precise": 1700000000.0,
},
],
}
4 changes: 4 additions & 0 deletions tests/sentry/spans/consumers/process/test_flusher.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,20 +44,23 @@ def append(msg):
span_id="a" * 16,
parent_span_id="b" * 16,
project_id=1,
end_timestamp_precise=now,
),
Span(
payload=_payload(b"d" * 16),
trace_id=trace_id,
span_id="d" * 16,
parent_span_id="b" * 16,
project_id=1,
end_timestamp_precise=now,
),
Span(
payload=_payload(b"c" * 16),
trace_id=trace_id,
span_id="c" * 16,
parent_span_id="b" * 16,
project_id=1,
end_timestamp_precise=now,
),
Span(
payload=_payload(b"b" * 16),
Expand All @@ -66,6 +69,7 @@ def append(msg):
parent_span_id=None,
is_segment_span=True,
project_id=1,
end_timestamp_precise=now,
),
]

Expand Down
22 changes: 22 additions & 0 deletions tests/sentry/spans/test_buffer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -123,20 +123,23 @@ def process_spans(spans: Sequence[Span | _SplitBatch], buffer: SpansBuffer, now)
span_id="a" * 16,
parent_span_id="b" * 16,
project_id=1,
end_timestamp_precise=1700000000.0,
),
Span(
payload=_payload(b"d" * 16),
trace_id="a" * 32,
span_id="d" * 16,
parent_span_id="b" * 16,
project_id=1,
end_timestamp_precise=1700000000.0,
),
Span(
payload=_payload(b"c" * 16),
trace_id="a" * 32,
span_id="c" * 16,
parent_span_id="b" * 16,
project_id=1,
end_timestamp_precise=1700000000.0,
),
Span(
payload=_payload(b"b" * 16),
Expand All @@ -145,6 +148,7 @@ def process_spans(spans: Sequence[Span | _SplitBatch], buffer: SpansBuffer, now)
parent_span_id=None,
is_segment_span=True,
project_id=1,
end_timestamp_precise=1700000000.0,
),
]
)
Expand Down Expand Up @@ -188,6 +192,7 @@ def test_basic(buffer: SpansBuffer, spans):
span_id="d" * 16,
parent_span_id="b" * 16,
project_id=1,
end_timestamp_precise=1700000000.0,
),
_SplitBatch(),
Span(
Expand All @@ -196,6 +201,7 @@ def test_basic(buffer: SpansBuffer, spans):
span_id="b" * 16,
parent_span_id="a" * 16,
project_id=1,
end_timestamp_precise=1700000000.0,
),
Span(
payload=_payload(b"a" * 16),
Expand All @@ -204,13 +210,15 @@ def test_basic(buffer: SpansBuffer, spans):
parent_span_id=None,
is_segment_span=True,
project_id=1,
end_timestamp_precise=1700000000.0,
),
Span(
payload=_payload(b"c" * 16),
trace_id="a" * 32,
span_id="c" * 16,
parent_span_id="a" * 16,
project_id=1,
end_timestamp_precise=1700000000.0,
),
]
)
Expand Down Expand Up @@ -254,27 +262,31 @@ def test_deep(buffer: SpansBuffer, spans):
span_id="e" * 16,
parent_span_id="d" * 16,
project_id=1,
end_timestamp_precise=1700000000.0,
),
Span(
payload=_payload(b"d" * 16),
trace_id="a" * 32,
span_id="d" * 16,
parent_span_id="b" * 16,
project_id=1,
end_timestamp_precise=1700000000.0,
),
Span(
payload=_payload(b"b" * 16),
trace_id="a" * 32,
span_id="b" * 16,
parent_span_id="c" * 16,
project_id=1,
end_timestamp_precise=1700000000.0,
),
Span(
payload=_payload(b"c" * 16),
trace_id="a" * 32,
span_id="c" * 16,
parent_span_id="a" * 16,
project_id=1,
end_timestamp_precise=1700000000.0,
),
Span(
payload=_payload(b"a" * 16),
Expand All @@ -283,6 +295,7 @@ def test_deep(buffer: SpansBuffer, spans):
parent_span_id=None,
is_segment_span=True,
project_id=1,
end_timestamp_precise=1700000000.0,
),
]
)
Expand Down Expand Up @@ -327,20 +340,23 @@ def test_deep2(buffer: SpansBuffer, spans):
span_id="c" * 16,
parent_span_id="b" * 16,
project_id=1,
end_timestamp_precise=1700000000.0,
),
Span(
payload=_payload(b"d" * 16),
trace_id="a" * 32,
span_id="d" * 16,
parent_span_id="b" * 16,
project_id=1,
end_timestamp_precise=1700000000.0,
),
Span(
payload=_payload(b"e" * 16),
trace_id="a" * 32,
span_id="e" * 16,
parent_span_id="b" * 16,
project_id=1,
end_timestamp_precise=1700000000.0,
),
Span(
payload=_payload(b"b" * 16),
Expand All @@ -349,6 +365,7 @@ def test_deep2(buffer: SpansBuffer, spans):
parent_span_id=None,
is_segment_span=True,
project_id=2,
end_timestamp_precise=1700000000.0,
),
]
)
Expand Down Expand Up @@ -400,20 +417,23 @@ def test_parent_in_other_project(buffer: SpansBuffer, spans):
parent_span_id="d" * 16,
project_id=1,
is_segment_span=True,
end_timestamp_precise=1700000000.0,
),
Span(
payload=_payload(b"d" * 16),
trace_id="a" * 32,
span_id="d" * 16,
parent_span_id="b" * 16,
project_id=1,
end_timestamp_precise=1700000000.0,
),
Span(
payload=_payload(b"e" * 16),
trace_id="a" * 32,
span_id="e" * 16,
parent_span_id="b" * 16,
project_id=1,
end_timestamp_precise=1700000000.0,
),
Span(
payload=_payload(b"b" * 16),
Expand All @@ -422,6 +442,7 @@ def test_parent_in_other_project(buffer: SpansBuffer, spans):
parent_span_id=None,
is_segment_span=True,
project_id=2,
end_timestamp_precise=1700000000.0,
),
]
),
Expand Down Expand Up @@ -476,6 +497,7 @@ def test_flush_rebalance(buffer: SpansBuffer):
parent_span_id=None,
project_id=1,
is_segment_span=True,
end_timestamp_precise=1700000000.0,
)
]

Expand Down