Skip to content

Expose C_SEGMENT_SIZE_BYTES gauges #218

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

Open
lukas8219 wants to merge 14 commits into
base: main
Choose a base branch
from
Open

Expose C_SEGMENT_SIZE_BYTES gauges #218

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
faf46d1
Expose `SEGMENT_SIZE_BYTES` and `SEGMENT_SIZE_BYTES` gauges
lukas8219 May 19, 2026
7e01092
Adjusted `osiris_retention:schedule/3` and added simple test case for…
lukas8219 May 20, 2026
4530f93
Removed unnecessary commentaries
lukas8219 May 20, 2026
5a6721e
Adjusted counter after truncate call and tail-recursive optimizations
lukas8219 May 29, 2026
8a3c74f
Adjusted to Mike comments.
lukas8219 Jun 6, 2026
fa26dbe
Use precomputed IndexFiles instead of running `sorted_index_files` mu…
lukas8219 Jul 3, 2026
1345908
Adjusted unnecessary calls to `file_size_or_zero` during retention ch…
lukas8219 Jul 3, 2026
ddcd47d
Merge branch 'main' into osiris-segmenet-size-gauge
lukas8219 Jul 3, 2026
bdade52
Merge branch 'main' into osiris-segmenet-size-gauge
lukas8219 Jul 20, 2026
7f8fc12
Fix eval_max_bytes deleting the same segment repeatedly
lukas8219 Jul 27, 2026
c3f0367
Merge branch 'main' into osiris-segmenet-size-gauge
lukas8219 Aug 4, 2026
9a2daf9
Merge branch 'main' into osiris-segmenet-size-gauge
lukas8219 Aug 17, 2026
b6f0ea0
Merge branch 'main' into osiris-segmenet-size-gauge
lukas8219 Aug 18, 2026
796c0d4
Merge branch 'main' into osiris-segmenet-size-gauge
lukas8219 Sep 9, 2026
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
2 changes: 1 addition & 1 deletion src/osiris.hrl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@

-define(IS_STRING(S), is_list(S) orelse is_binary(S)).

-define(C_NUM_LOG_FIELDS, 5).
-define(C_NUM_LOG_FIELDS, 7).

-define(MAGIC, 5).
%% chunk format version
Expand Down
113 changes: 76 additions & 37 deletions src/osiris_log.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,14 +84,18 @@
-define(C_FIRST_TIMESTAMP, 3).
-define(C_CHUNKS, 4).
-define(C_SEGMENTS, 5).
-define(C_SEGMENT_SIZE_BYTES, 6).
-define(C_INDEX_SIZE_BYTES, 7).
-define(COUNTER_FIELDS,
[
{offset, ?C_OFFSET, counter,
"The last offset (not chunk id) in the log for writers. The last offset read for readers" },
{first_offset, ?C_FIRST_OFFSET, counter, "First offset, not updated for readers"},
{first_timestamp, ?C_FIRST_TIMESTAMP, counter, "First timestamp, not updated for readers"},
{chunks, ?C_CHUNKS, counter, "Number of chunks read or written, incremented even if a reader only reads the header"},
{segments, ?C_SEGMENTS, counter, "Number of segments"}
{segments, ?C_SEGMENTS, counter, "Number of segments"},
{segment_size_bytes, ?C_SEGMENT_SIZE_BYTES, counter, "Total size of all segment files in bytes"},
{index_size_bytes, ?C_INDEX_SIZE_BYTES, counter, "Total size of all index files in bytes"}

@the-mikedavis the-mikedavis May 20, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not sure it's worthwhile to track index data. It would mean an extra fstat for index files during recovery and retention to gather the size data. Since chunk headers are at minimum 48 bytes per chunk, the segment will always be larger (48 > 29 for each index record). If these metrics aim to answer a question of "which stream is taking up all of my data" I think segment size metrics are enough.

I think there's a separate question of "are my chunks so small they're hurting performance" and that could be answered by a bytes-per-chunk metric instead. Tracking index data here would indirectly hint towards that rather than answer it clearly.

@lukas8219 lukas8219 May 20, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am not strong on any claim here as I also believe that for big clusters, with loads of streams, that's not a cheap price to pay.

@lukas8219 lukas8219 Jun 2, 2026 •
edited
Loading

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@the-mikedavis
To track or not to track it: I thought maybe, if the overhead is not big, to keep going with both. Easier to catch bugs if users have this level of visibility.
Now that was myself assuming there's a chance the index file could have holes/get bloated somehow - which could have been naive as Streams are running for lots of years and there's not a single complaint about it.

I'll good with removing the index tracking from this PR

@the-mikedavis the-mikedavis May 28, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It looks like all of these are counter currently but I think most of them (maybe all of them?) should be gauge. Maybe it's outside the scope of this change though since the existing ones all use counter

@lukas8219 lukas8219 May 29, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agree for a future PR

]
).

Expand Down Expand Up @@ -596,10 +600,13 @@ init(#{dir := Dir,
TailInfo = {LastChId + LastNum,
{LastEpoch, LastChId, LastTs}},

{InitSegBytes, InitIdxBytes} = sum_log_sizes(Config),
counters:put(Cnt, ?C_FIRST_OFFSET, FstChId),
counters:put(Cnt, ?C_FIRST_TIMESTAMP, FstTs),
counters:put(Cnt, ?C_OFFSET, LastChId + LastNum - 1),
counters:put(Cnt, ?C_SEGMENTS, NumSegments),
Comment thread
lukas8219 marked this conversation as resolved.
Outdated
Comment thread
lukas8219 marked this conversation as resolved.
Outdated
counters:put(Cnt, ?C_SEGMENT_SIZE_BYTES, InitSegBytes),
counters:put(Cnt, ?C_INDEX_SIZE_BYTES, InitIdxBytes),
Comment thread
the-mikedavis marked this conversation as resolved.
Outdated
osiris_log_shared:set_first_chunk_id(Shared, FstChId),
osiris_log_shared:set_last_chunk_id(Shared, LastChId),
?DEBUG_(Name, " next offset ~b first offset ~b",
Expand Down Expand Up @@ -632,6 +639,8 @@ init(#{dir := Dir,
{ok, IdxFd} = open(IdxFilename, ?FILE_OPTS_WRITE),
{ok, _} = file:position(SegFd, ?LOG_HEADER_SIZE),
counters:put(Cnt, ?C_SEGMENTS, 1),
counters:put(Cnt, ?C_SEGMENT_SIZE_BYTES, ?LOG_HEADER_SIZE),
counters:put(Cnt, ?C_INDEX_SIZE_BYTES, ?IDX_HEADER_SIZE),
%% the segment could potentially have trailing data here so we'll
%% do a truncate just in case. The index would have been truncated
%% earlier
Expand Down Expand Up @@ -924,16 +933,18 @@ chunk_id_index_scan0(Fd, ChunkId) ->
delete_segment_from_index(Index) ->
File = segment_from_index_file(Index),
?DEBUG("osiris_log: deleting segment ~ts", [File]),
SegSize = file_size_or_zero(File),

@kjnilsson kjnilsson Jul 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this incurs an additional syscall for all calls of this function, even if they don't make use of it. Perhaps we need two different functions or the file size if got independently where needed.

@lukas8219 lukas8219 Jul 4, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I've used the Size from previous read. What I am not completely sure is if, there's any chance the segment size would change between the first stat and the delete

IdxSize = file_size_or_zero(Index),
ok = prim_file:delete(Index),
ok = prim_file:delete(File),
ok.
{SegSize, IdxSize}.

truncate_to(_Name, _Range, _EpochOffsets, []) ->
%% the target log is empty
[];
truncate_to(_Name, _Range, [], IdxFiles) ->
%% ????? this means the entire log is out
[begin ok = delete_segment_from_index(I) end || I <- IdxFiles],
[delete_segment_from_index(I) || I <- IdxFiles],
[];
truncate_to(Name, RemoteRange, [{E, ChId} | NextEOs], IdxFiles) ->
case find_segment_for_offset(ChId, IdxFiles) of
Expand All @@ -958,8 +969,7 @@ truncate_to(Name, RemoteRange, [{E, ChId} | NextEOs], IdxFiles) ->
true ->
%% there is no overlap, need to delete all
%% local segments
[begin ok = delete_segment_from_index(I) end
|| I <- IdxFiles],
[delete_segment_from_index(I) || I <- IdxFiles],

@lukas8219 lukas8219 May 19, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure we pattern match from above is really necessary?

@the-mikedavis the-mikedavis May 20, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah looks like it isn't since delete_segment_from_index/1 already hard matches on ok

[];
false ->
%% there is overlap
Expand Down Expand Up @@ -1001,7 +1011,7 @@ truncate_to(Name, RemoteRange, [{E, ChId} | NextEOs], IdxFiles) ->
fun (I) ->
case index_file_first_offset(I) > ChId of
true ->
ok = delete_segment_from_index(I),
_ = delete_segment_from_index(I),
false;
false ->
true
Expand Down Expand Up @@ -2254,35 +2264,34 @@ update_retention(Retention,

-spec evaluate_retention(file:filename_all(), [retention_spec()]) ->
{range(), FirstTimestamp :: osiris:timestamp(),
NumRemainingFiles :: non_neg_integer()}.
NumRemainingFiles :: non_neg_integer(),
{DeletedSegBytes :: non_neg_integer(), DeletedIndexBytes :: non_neg_integer()}}.
Comment thread
lukas8219 marked this conversation as resolved.
Outdated
Comment thread
the-mikedavis marked this conversation as resolved.
Outdated
evaluate_retention(Dir, Specs) when is_list(Dir) ->
% convert to binary for faster operations later
% mostly in segment_from_index_file/1
Comment thread
the-mikedavis marked this conversation as resolved.
evaluate_retention(unicode:characters_to_binary(Dir), Specs);
evaluate_retention(Dir, Specs) when is_binary(Dir) ->

{Time, Result} = timer:tc(
fun() ->
IdxFiles0 = sorted_index_files(Dir),
IdxFiles = evaluate_retention0(IdxFiles0, Specs),
{IdxFiles, DeletedBytes} =
evaluate_retention0(IdxFiles0, Specs, {0, 0}),
OffsetRange = offset_range_from_idx_files(IdxFiles),
FirstTs = first_timestamp_from_index_files(IdxFiles),
{OffsetRange, FirstTs, length(IdxFiles)}
{OffsetRange, FirstTs, length(IdxFiles), DeletedBytes}
end),
?DEBUG_(<<>>," (~w) completed in ~fms", [Specs, Time/1_000]),
Result.

evaluate_retention0(IdxFiles, []) ->
IdxFiles;
evaluate_retention0(IdxFiles, [{max_bytes, MaxSize} | Specs]) ->
RemIdxFiles = eval_max_bytes(IdxFiles, MaxSize),
evaluate_retention0(RemIdxFiles, Specs);
evaluate_retention0(IdxFiles, [{max_age, Age} | Specs]) ->
RemIdxFiles = eval_age(IdxFiles, Age),
evaluate_retention0(RemIdxFiles, Specs).
evaluate_retention0(IdxFiles, [], DeletedBytes) ->
{IdxFiles, DeletedBytes};
evaluate_retention0(IdxFiles, [{max_bytes, MaxSize} | Specs], {AccSeg, AccIdx}) ->
{RemIdxFiles, {DelSeg, DelIdx}} = eval_max_bytes(IdxFiles, MaxSize),
evaluate_retention0(RemIdxFiles, Specs, {AccSeg + DelSeg, AccIdx + DelIdx});
evaluate_retention0(IdxFiles, [{max_age, Age} | Specs], {AccSeg, AccIdx}) ->
{RemIdxFiles, {DelSeg, DelIdx}} = eval_age(IdxFiles, Age),
evaluate_retention0(RemIdxFiles, Specs, {AccSeg + DelSeg, AccIdx + DelIdx}).
Comment thread
lukas8219 marked this conversation as resolved.
Outdated

eval_age([_] = IdxFiles, _Age) ->
IdxFiles;
{IdxFiles, {0, 0}};
eval_age([IdxFile | IdxFiles] = AllIdxFiles, Age) ->
case last_timestamp_in_index_file(IdxFile) of
{ok, Ts} ->
Comment thread
lukas8219 marked this conversation as resolved.
Outdated
Comment thread
lukas8219 marked this conversation as resolved.
Outdated
Expand All @@ -2292,42 +2301,66 @@ eval_age([IdxFile | IdxFiles] = AllIdxFiles, Age) ->
%% the oldest timestamp is older than retention
%% and there are other segments available
%% we can delete
ok = delete_segment_from_index(IdxFile),
eval_age(IdxFiles, Age);
{SegSize, IdxSize} = delete_segment_from_index(IdxFile),
{RemIdxFiles, {AccSeg, AccIdx}} = eval_age(IdxFiles, Age),
{RemIdxFiles, {AccSeg + SegSize, AccIdx + IdxSize}};
false ->
AllIdxFiles
{AllIdxFiles, {0, 0}}
end;
_Err ->
AllIdxFiles
{AllIdxFiles, {0, 0}}
end;
eval_age([], _Age) ->
%% this could happen if retention is evaluated whilst
%% a stream is being deleted
[].
{[], {0, 0}}.

eval_max_bytes([], _) -> [];
eval_max_bytes([], _) -> {[], {0, 0}};
Comment thread
lukas8219 marked this conversation as resolved.
Outdated
eval_max_bytes(IdxFiles, MaxSize) ->
[Latest|Older] = lists:reverse(IdxFiles),
[Latest | Older] = lists:reverse(IdxFiles),
eval_max_bytes(Older,
%% for retention eval it is ok to use a file size function
%% that implicitly return 0 when file is not found
MaxSize - file_size_or_zero(
segment_from_index_file(Latest)),
[Latest]).
[Latest],
{0, 0}).

eval_max_bytes([], _, Acc) ->
Acc;
eval_max_bytes([IdxFile | Rest], Limit, Acc) ->
eval_max_bytes([], _, Acc, DeletedBytes) ->
{Acc, DeletedBytes};
eval_max_bytes([IdxFile | Rest], Limit, Acc, {AccSeg, AccIdx}) ->
SegFile = segment_from_index_file(IdxFile),
Size = file_size(SegFile),
case Size =< Limit of
true ->
eval_max_bytes(Rest, Limit - Size, [IdxFile | Acc]);
eval_max_bytes(Rest, Limit - Size, [IdxFile | Acc], {AccSeg, AccIdx});
false ->
[ok = delete_segment_from_index(Seg) || Seg <- [IdxFile | Rest]],
Acc
{TotalSeg, TotalIdx} =
lists:foldl(fun(Seg, {S, I}) ->
{SegSize, IdxSize} = delete_segment_from_index(Seg),
{S + SegSize, I + IdxSize}
end,
{AccSeg, AccIdx},
[IdxFile | Rest]),
{Acc, {TotalSeg, TotalIdx}}
end.

%% Sums the on-disk sizes of all segment and index files in a log directory.
%% Used once at writer/replica init to initialise the size counters.
sum_log_sizes(Config) ->
IdxFiles = case Config of
#{index_files := F} -> F;
#{dir := Dir} -> sorted_index_files(Dir)
end,
lists:foldl(
fun(IdxFile, {SegAcc, IdxAcc}) ->
SegFile = segment_from_index_file(IdxFile),
{SegAcc + file_size_or_zero(SegFile),
IdxAcc + file_size_or_zero(IdxFile)}
end,
{0, 0},
IdxFiles).

file_size(Path) ->
case prim_file:read_file_info(Path) of
{ok, #file_info{size = Size}} ->
Expand Down Expand Up @@ -2571,6 +2604,8 @@ write_chunk(Chunk,
%% update counters
counters:put(CntRef, ?C_OFFSET, NextOffset - 1),
counters:add(CntRef, ?C_CHUNKS, 1),
counters:add(CntRef, ?C_SEGMENT_SIZE_BYTES, Size),
counters:add(CntRef, ?C_INDEX_SIZE_BYTES, ?INDEX_RECORD_SIZE_B),
maybe_set_first_offset(Next, Cfg),
State#?MODULE{mode =
Write#write{tail_info = {NextOffset,
Expand Down Expand Up @@ -2806,6 +2841,8 @@ open_new_segment(#?MODULE{cfg = #cfg{name = Name,
{ok, _} = file:position(Fd, eof),
{ok, _} = file:position(IdxFd, eof),
counters:add(Cnt, ?C_SEGMENTS, 1),
counters:add(Cnt, ?C_SEGMENT_SIZE_BYTES, ?LOG_HEADER_SIZE),
counters:add(Cnt, ?C_INDEX_SIZE_BYTES, ?IDX_HEADER_SIZE),

State0#?MODULE{current_file = Filename,
fd = Fd,
Expand Down Expand Up @@ -3170,13 +3207,15 @@ trigger_retention_eval(#?MODULE{cfg =

%% updates first offset and first timestamp
%% after retention has been evaluated
EvalFun = fun ({{FstOff, _}, FstTs, NumSegLeft})
EvalFun = fun ({{FstOff, _}, FstTs, NumSegLeft, {DelSegBytes, DelIdxBytes}})
when is_integer(FstOff),
is_integer(FstTs) ->
osiris_log_shared:set_first_chunk_id(Shared, FstOff),
counters:put(Cnt, ?C_FIRST_OFFSET, FstOff),
counters:put(Cnt, ?C_FIRST_TIMESTAMP, FstTs),
counters:put(Cnt, ?C_SEGMENTS, NumSegLeft);
counters:put(Cnt, ?C_SEGMENTS, NumSegLeft),
counters:sub(Cnt, ?C_SEGMENT_SIZE_BYTES, DelSegBytes),
counters:sub(Cnt, ?C_INDEX_SIZE_BYTES, DelIdxBytes);
(_) ->
ok
end,
Expand Down
Loading
Loading