Expose C_SEGMENT_SIZE_BYTES gauges

src/osiris.hrl

@@ -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, 6).
 
 -define(MAGIC, 5).
 %% chunk format version

src/osiris_log.erl

@@ -84,14 +84,16 @@
 -define(C_FIRST_TIMESTAMP, 3).
 -define(C_CHUNKS, 4).
 -define(C_SEGMENTS, 5).
+-define(C_SEGMENT_SIZE_BYTES, 6).
 -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"}
         ]
        ).
 
@@ -612,6 +614,8 @@ init(#{dir := Dir,
             %% at a valid chunk we can now truncate the segment to size in
             %% case there is trailing data
             ok = file:truncate(SegFd),
+            InitSegBytes = sum_log_sizes(Config),
+            counters:put(Cnt, ?C_SEGMENT_SIZE_BYTES, InitSegBytes),
             {ok, IdxFd} = open(IdxFilename, ?FILE_OPTS_WRITE),
             {ok, IdxEof} = file:position(IdxFd, eof),
             NumChunks = (IdxEof - ?IDX_HEADER_SIZE) div ?INDEX_RECORD_SIZE_B,
@@ -632,6 +636,7 @@ 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),
             %% the segment could potentially have trailing data here so we'll
             %% do a truncate just in case. The index would have been truncated
             %% earlier
@@ -924,16 +929,17 @@ 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),
     ok = prim_file:delete(Index),
     ok = prim_file:delete(File),
-    ok.
+    SegSize.
 
 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
@@ -958,8 +964,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],
                             [];
                         false ->
                             %% there is overlap
@@ -1001,7 +1006,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
@@ -2252,38 +2257,49 @@ update_retention(Retention,
     State = State0#?MODULE{cfg = Cfg#cfg{retention = Retention}},
     trigger_retention_eval(State).
 
+-type retention_result() ::
+    #{range := range(),
+      first_timestamp := osiris:timestamp(),
+      num_remaining_segments := non_neg_integer(),
+      deleted_segment_bytes := non_neg_integer()}.
 -spec evaluate_retention(file:filename_all(), [retention_spec()]) ->
-    {range(), FirstTimestamp :: osiris:timestamp(),
-     NumRemainingFiles :: non_neg_integer()}.
+    retention_result().
 evaluate_retention(Dir, Specs) when is_list(Dir) ->
     % convert to binary for faster operations later
     % mostly in segment_from_index_file/1
     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),
-                               OffsetRange = offset_range_from_idx_files(IdxFiles),
-                               FirstTs = first_timestamp_from_index_files(IdxFiles),
-                               {OffsetRange, FirstTs, length(IdxFiles)}
+                               {IdxFiles, DelSeg} =
+                                   evaluate_retention0(IdxFiles0, Specs),
+                               #{range => offset_range_from_idx_files(IdxFiles),
+                                 first_timestamp => first_timestamp_from_index_files(IdxFiles),
+                                 num_remaining_segments => length(IdxFiles),
+                                 deleted_segment_bytes => DelSeg}
                        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).
-
-eval_age([_] = IdxFiles, _Age) ->
-    IdxFiles;
-eval_age([IdxFile | IdxFiles] = AllIdxFiles, Age) ->
+evaluate_retention0(IdxFiles, Specs) ->
+    evaluate_retention0(IdxFiles, Specs, 0).
+
+evaluate_retention0(IdxFiles, [], DeletedBytes) ->
+    {IdxFiles, DeletedBytes};
+evaluate_retention0(IdxFiles, [{max_bytes, MaxSize} | Specs], Acc) ->
+    {RemIdxFiles, DelSeg} = eval_max_bytes(IdxFiles, MaxSize),
+    evaluate_retention0(RemIdxFiles, Specs, Acc + DelSeg);
+evaluate_retention0(IdxFiles, [{max_age, Age} | Specs], Acc) ->
+    {RemIdxFiles, DelSeg} = eval_age(IdxFiles, Age),
+    evaluate_retention0(RemIdxFiles, Specs, Acc + DelSeg).
+
+eval_age(IdxFiles, Age) ->
+    eval_age(IdxFiles, Age, 0).
+
+eval_age([_] = IdxFiles, _Age, Acc) ->
+    {IdxFiles, Acc};
+eval_age([IdxFile | IdxFiles] = AllIdxFiles, Age, Acc) ->
     case last_timestamp_in_index_file(IdxFile) of
         {ok, Ts} ->
             Now = erlang:system_time(millisecond),
@@ -2292,41 +2308,59 @@ 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 = delete_segment_from_index(IdxFile),
+                    eval_age(IdxFiles, Age, Acc + SegSize);
                 false ->
-                    AllIdxFiles
+                    {AllIdxFiles, Acc}
             end;
         _Err ->
-            AllIdxFiles
+            {AllIdxFiles, Acc}
     end;
-eval_age([], _Age) ->
+eval_age([], _Age, Acc) ->
     %% this could happen if retention is evaluated whilst
     %% a stream is being deleted
-    [].
+    {[], Acc}.
 
-eval_max_bytes([], _) -> [];
+eval_max_bytes([], _) -> {[], 0};
 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).
 
-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) ->
     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);
         false ->
-            [ok = delete_segment_from_index(Seg) || Seg <- [IdxFile | Rest]],
-            Acc
-    end.
+            Total = lists:foldl(fun(Seg, S) ->
+                                        S + delete_segment_from_index(Seg)
+                                end,
+                                AccSeg,
+                                [IdxFile | Rest]),
+            {Acc, Total}
+    end.
+
+sum_log_sizes(Config) ->
+    IdxFiles = case Config of
+                   #{index_files := F} -> F;
+                   #{dir := Dir} -> sorted_index_files(Dir)
+               end,
+    lists:foldl(
+      fun(IdxFile, SegAcc) ->
+              SegFile = segment_from_index_file(IdxFile),
+              SegAcc + file_size_or_zero(SegFile)
+      end,
+      0,
+      IdxFiles).
 
 file_size(Path) ->
     case prim_file:read_file_info(Path) of
@@ -2571,6 +2605,7 @@ 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),
             maybe_set_first_offset(Next, Cfg),
             State#?MODULE{mode =
                           Write#write{tail_info = {NextOffset,
@@ -2806,6 +2841,7 @@ 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),
 
     State0#?MODULE{current_file = Filename,
                    fd = Fd,
@@ -3170,13 +3206,17 @@ trigger_retention_eval(#?MODULE{cfg =
 
     %% updates first offset and first timestamp
     %% after retention has been evaluated
-    EvalFun = fun ({{FstOff, _}, FstTs, NumSegLeft})
+    EvalFun = fun (#{range := {FstOff, _},
+                     first_timestamp := FstTs,
+                     num_remaining_segments := NumSegLeft,
+                     deleted_segment_bytes := DelSegBytes})
                     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);
                   (_) ->
                       ok
               end,

src/osiris_retention.erl

@@ -97,7 +97,7 @@ evaluate_retention({eval, Pid, Name, Dir, Specs, Fun} = Eval, State) ->
     end.
 
 schedule({eval, _Pid, Name, _Dir, Specs, _Fun} = Eval,
-         {_, _, NumSegmentRemaining},
+         #{num_remaining_segments := NumSegmentRemaining},
          #state{scheduled = Scheduled0} = State) ->
     %% we need to check the scheduled map even if the current specs do not
     %% include max_age as the retention config could have changed

test/osiris_SUITE.erl

@@ -59,6 +59,7 @@ all_tests() ->
      replica_unknown_command,
      diverged_replica,
      retention,
+     retention_size_counters,
      retention_max_age_eventually,
      retention_max_age_update_retention,
      retention_max_age_noproc,
@@ -1221,6 +1222,31 @@ retention(Config) ->
     osiris:stop_cluster(Conf1),
     ok.
 
+retention_size_counters(Config) ->
+    Name = ?config(cluster_name, Config),
+    SegSize = 100 * 1000,
+    Conf0 =
+        #{name => Name,
+          epoch => 1,
+          leader_node => node(),
+          retention => [{max_bytes, SegSize * 100}],
+          max_segment_size_bytes => SegSize,
+          replica_nodes => []},
+    {ok, #{leader_pid := Leader} = Conf1} = osiris:start_cluster(Conf0),
+    write_n(Leader, 1000, 0, 500 * 8, #{}),
+    Key = {osiris_writer, Name},
+    #{segment_size_bytes := SegBefore} =
+        osiris_counters:counters(Key, [segment_size_bytes]),
+    ok = osiris:update_retention(Leader, [{max_bytes, SegSize}]),
+    await_condition(
+        fun() ->
+            #{segment_size_bytes := S} =
+                osiris_counters:counters(Key, [segment_size_bytes]),
+            S < SegBefore
+        end, 200, 50),
+    osiris:stop_cluster(Conf1),
+    ok.
+
 retention_max_age_eventually(Config) ->
     DataDir = ?config(data_dir, Config),
     Num = 150000,