Rename StreamInfo to StreamStats

acogoluegnes · mergify[bot] · commit d8ec9eface7e · 2022-08-09T08:41:11.000Z
Other changes: returns a map of int64, use the new osiris:get_stats/1 API. References #5412 (cherry picked from commit 93c33f2)
diff --git a/deps/rabbitmq_stream/docs/PROTOCOL.adoc b/deps/rabbitmq_stream/docs/PROTOCOL.adoc
@@ -226,7 +226,7 @@ used to make the difference between a request (0) and a response (1). Example fo
 |0x001b
 |Yes
 
-|<<streaminfo>>
+|<<streamstats>>
 |Client
 |0x001c
 |Yes
@@ -375,7 +375,7 @@ Deliver => Key Version SubscriptionId CommittedOffset OsirisChunk
   Key => uint16 // 0x0008
   Version => uint16
   SubscriptionId => uint8
-  CommittedOffset => uint64
+  CommittedChunkId => uint64
   OsirisChunk => MagicVersion NumEntries NumRecords Epoch ChunkFirstOffset ChunkCrc DataLength Messages
   MagicVersion => int8
   ChunkType => int8 // 0: user, 1: tracking delta, 2: tracking snapshot
@@ -707,24 +707,24 @@ CommandVersionsExchangeResponse => Key Version CorrelationId ResponseCode [Comma
   MaxVersion => uint16
 ```
 
-=== StreamInfo
+=== StreamStats
 
 ```
-StreamInfoRequest => Key Version CorrelationId Stream
+StreamStatsRequest => Key Version CorrelationId Stream
   Key => uint16 // 0x001c
   Version => uint16
   CorrelationId => uint32
   Stream => string
 
-StreamInfoResponse => Key Version CorrelationId ResponseCode Info
+StreamStatsResponse => Key Version CorrelationId ResponseCode Stats
   Key => uint16 // 0x801c
   Version => uint16
   CorrelationId => uint32
   ResponseCode => uint16
-  Info => [PieceOfInfo]
-  PieceOfInfo => Key Value
+  Stats => [Statistic]
+  Statistic => Key Value
   Key => string
-  Value => string
+  Value => int64
 ```
 
 
diff --git a/deps/rabbitmq_stream/src/rabbit_stream_reader.erl b/deps/rabbitmq_stream/src/rabbit_stream_reader.erl
@@ -2646,7 +2646,7 @@ handle_frame_post_auth(Transport,
                                           user = User} =
                            Connection,
                        State,
-                       {request, CorrelationId, {stream_info, Stream}}) ->
+                       {request, CorrelationId, {stream_stats, Stream}}) ->
     QueueResource =
         #resource{name = Stream,
                   kind = queue,
@@ -2661,38 +2661,28 @@ handle_frame_post_auth(Transport,
                         rabbit_global_counters:increase_protocol_counter(stream,
                                                                          ?STREAM_NOT_AVAILABLE,
                                                                          1),
-                        {stream_info, ?RESPONSE_CODE_STREAM_NOT_AVAILABLE, #{}};
+                        {stream_stats, ?RESPONSE_CODE_STREAM_NOT_AVAILABLE,
+                         #{}};
                     {error, not_found} ->
                         rabbit_global_counters:increase_protocol_counter(stream,
                                                                          ?STREAM_DOES_NOT_EXIST,
                                                                          1),
-                        {stream_info, ?RESPONSE_CODE_STREAM_DOES_NOT_EXIST,
+                        {stream_stats, ?RESPONSE_CODE_STREAM_DOES_NOT_EXIST,
                          #{}};
                     {ok, MemberPid} ->
-                        OffsetInfo =
-                            case gen:call(MemberPid, '$gen_call',
-                                          get_reader_context)
-                            of
-                                {ok, #{offset_ref := undefined}} ->
-                                    #{};
-                                {ok, #{offset_ref := OffsetRef}} ->
-                                    #{<<"first_offset">> =>
-                                          rabbit_data_coercion:to_binary(
-                                              atomics:get(OffsetRef, 2)),
-                                      <<"committed_offset">> =>
-                                          rabbit_data_coercion:to_binary(
-                                              atomics:get(OffsetRef, 1))};
-                                _ ->
-                                    #{}
-                            end,
-
-                        {stream_info, ?RESPONSE_CODE_OK, OffsetInfo}
+                        StreamStats =
+                            maps:fold(fun(K, V, Acc) ->
+                                         Acc#{rabbit_data_coercion:to_binary(K)
+                                                  => V}
+                                      end,
+                                      #{}, osiris:get_stats(MemberPid)),
+                        {stream_stats, ?RESPONSE_CODE_OK, StreamStats}
                 end;
             error ->
                 rabbit_global_counters:increase_protocol_counter(stream,
                                                                  ?ACCESS_REFUSED,
                                                                  1),
-                {stream_info, ?RESPONSE_CODE_ACCESS_REFUSED, #{}}
+                {stream_stats, ?RESPONSE_CODE_ACCESS_REFUSED, #{}}
         end,
     Frame = rabbit_stream_core:frame({response, CorrelationId, Response}),
     send(Transport, S, Frame),
@@ -3243,7 +3233,7 @@ send_file_callback(?VERSION_2,
     fun(#{chunk_id := FirstOffsetInChunk, num_entries := NumEntries},
         Size) ->
        FrameSize = 2 + 2 + 1 + 8 + Size,
-       CommittedOffset =
+       CommittedChunkId =
            case osiris_log:committed_offset(Log) of
                undefined -> 0;
                R -> R
@@ -3254,7 +3244,7 @@ send_file_callback(?VERSION_2,
              ?COMMAND_DELIVER:15,
              ?VERSION_2:16,
              SubscriptionId:8/unsigned,
-             CommittedOffset:64>>,
+             CommittedChunkId:64>>,
        Transport:send(S, FrameBeginning),
        atomics:add(Counter, 1, Size),
        increase_messages_consumed(Counters, NumEntries),
diff --git a/deps/rabbitmq_stream/src/rabbit_stream_utils.erl b/deps/rabbitmq_stream/src/rabbit_stream_utils.erl
@@ -276,4 +276,4 @@ command_versions() ->
      {heartbeat, ?VERSION_1, ?VERSION_1},
      {route, ?VERSION_1, ?VERSION_1},
      {partitions, ?VERSION_1, ?VERSION_1},
-     {stream_info, ?VERSION_1, ?VERSION_1}].
+     {stream_stats, ?VERSION_1, ?VERSION_1}].
diff --git a/deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl b/deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl
@@ -440,7 +440,7 @@ test_server(Transport, Config) ->
     C12 = test_deliver_v2(Transport, S, SubscriptionId2, 0, Body, C11),
     C13 = test_deliver_v2(Transport, S, SubscriptionId2, 1, Body, C12),
 
-    C14 = test_stream_info(Transport, S, Stream, C13),
+    C14 = test_stream_stats(Transport, S, Stream, C13),
 
     C15 = test_delete_stream(Transport, S, Stream, C14),
     _C16 = test_close(Transport, S, C15),
@@ -622,15 +622,15 @@ test_exchange_command_versions(Transport, S, C0) ->
                  Cmd),
     C.
 
-test_stream_info(Transport, S, Stream, C0) ->
-    SICmd = {request, 1, {stream_info, Stream}},
+test_stream_stats(Transport, S, Stream, C0) ->
+    SICmd = {request, 1, {stream_stats, Stream}},
     SIFrame = rabbit_stream_core:frame(SICmd),
     ok = Transport:send(S, SIFrame),
     {Cmd, C} = receive_commands(Transport, S, C0),
     ?assertEqual({response, 1,
-                  {stream_info, ?RESPONSE_CODE_OK,
-                   #{<<"committed_offset">> => <<"1">>,
-                     <<"first_offset">> => <<"0">>}}},
+                  {stream_stats, ?RESPONSE_CODE_OK,
+                   #{<<"first_chunk_id">> => 0,
+                     <<"committed_chunk_id">> => 1}}},
                  Cmd),
     C.
 
diff --git a/deps/rabbitmq_stream_common/include/rabbit_stream.hrl b/deps/rabbitmq_stream_common/include/rabbit_stream.hrl
@@ -25,7 +25,7 @@
 -define(COMMAND_PARTITIONS, 25).
 -define(COMMAND_CONSUMER_UPDATE, 26).
 -define(COMMAND_EXCHANGE_COMMAND_VERSIONS, 27).
--define(COMMAND_STREAM_INFO, 28).
+-define(COMMAND_STREAM_STATS, 28).
 
 -define(REQUEST, 0).
 -define(RESPONSE, 1).
diff --git a/deps/rabbitmq_stream_common/src/rabbit_stream_core.erl b/deps/rabbitmq_stream_common/src/rabbit_stream_core.erl
@@ -75,7 +75,7 @@
     {deliver, subscription_id(), Chunk :: binary()} |
     {deliver_v2,
      subscription_id(),
-     CommittedOffset :: osiris:offset(),
+     CommittedChunkId :: osiris:offset(),
      Chunk :: binary()} |
     {credit, subscription_id(), Credit :: non_neg_integer()} |
     {metadata_update, stream_name(), response_code()} |
@@ -109,7 +109,7 @@
      {exchange_command_versions,
       [{Command :: atom(), MinVersion :: command_version(),
         MaxVersion :: command_version()}]} |
-     {stream_info, Stream :: binary()}} |
+     {stream_stats, Stream :: binary()}} |
     {response, correlation_id(),
      {declare_publisher |
       delete_publisher |
@@ -140,7 +140,7 @@
      {exchange_command_versions, response_code(),
       [{Command :: atom(), MinVersion :: command_version(),
         MaxVersion :: command_version()}]} |
-     {stream_info, response_code(), Info :: #{binary() => binary()}}} |
+     {stream_stats, response_code(), Stats :: #{binary() => integer()}}} |
     {unknown, binary()}.
 
 -spec init(term()) -> state().
@@ -245,12 +245,12 @@ frame({deliver, SubscriptionId, Chunk}) ->
                      ?VERSION_1:16,
                      SubscriptionId:8>>,
                    Chunk]);
-frame({deliver_v2, SubscriptionId, CommittedOffset, Chunk}) ->
+frame({deliver_v2, SubscriptionId, CommittedChunkId, Chunk}) ->
     wrap_in_frame([<<?REQUEST:1,
                      ?COMMAND_DELIVER:15,
                      ?VERSION_2:16,
                      SubscriptionId:8,
-                     CommittedOffset:64>>,
+                     CommittedChunkId:64>>,
                    Chunk]);
 frame({metadata_update, Stream, ResponseCode}) ->
     StreamSize = byte_size(Stream),
@@ -474,19 +474,17 @@ response_body({exchange_command_versions = Tag, Code,
                     [], CommandVersions),
     {command_id(Tag),
      [<<Code:16, (length(CommandVersions)):32>>, CommandVersionsBin]};
-response_body({stream_info = Tag, Code, Info}) ->
-    Init = <<Code:16, (maps:size(Info)):32>>,
+response_body({stream_stats = Tag, Code, Stats}) ->
+    Init = <<Code:16, (maps:size(Stats)):32>>,
     {command_id(Tag),
      maps:fold(fun(Key, Value, Acc) ->
                   KeySize = byte_size(Key),
-                  ValueSize = byte_size(Value),
                   <<Acc/binary,
                     KeySize:16,
                     Key:KeySize/binary,
-                    ValueSize:16,
-                    Value:ValueSize/binary>>
+                    Value:64/signed>>
                end,
-               Init, Info)}.
+               Init, Stats)}.
 
 request_body({declare_publisher = Tag,
               PublisherId,
@@ -592,7 +590,7 @@ request_body({exchange_command_versions = Tag, CommandVersions}) ->
                     [], CommandVersions),
     CommandVersionsLength = length(CommandVersions),
     {Tag, [<<CommandVersionsLength:32>>, CommandVersionsBin]};
-request_body({stream_info = Tag, Stream}) ->
+request_body({stream_stats = Tag, Stream}) ->
     {Tag, <<?STRING(Stream)>>}.
 
 append_data(Prev, Data) when is_binary(Prev) ->
@@ -639,9 +637,9 @@ parse_request(<<?REQUEST:1,
                 ?COMMAND_DELIVER:15,
                 ?VERSION_2:16,
                 SubscriptionId:8,
-                CommittedOffset:64,
+                CommittedChunkId:64,
                 Chunk/binary>>) ->
-    {deliver_v2, SubscriptionId, CommittedOffset, Chunk};
+    {deliver_v2, SubscriptionId, CommittedChunkId, Chunk};
 parse_request(<<?REQUEST:1,
                 ?COMMAND_CREDIT:15,
                 ?VERSION_1:16,
@@ -861,11 +859,11 @@ parse_request(<<?REQUEST:1,
     CommandVersions = parse_command_versions(CommandVersionsBin),
     request(CorrelationId, {exchange_command_versions, CommandVersions});
 parse_request(<<?REQUEST:1,
-                ?COMMAND_STREAM_INFO:15,
+                ?COMMAND_STREAM_STATS:15,
                 ?VERSION_1:16,
                 CorrelationId:32,
                 ?STRING(StreamSize, Stream)>>) ->
-    request(CorrelationId, {stream_info, Stream});
+    request(CorrelationId, {stream_stats, Stream});
 parse_request(Bin) ->
     {unknown, Bin}.
 
@@ -952,10 +950,10 @@ parse_response_body(?COMMAND_EXCHANGE_COMMAND_VERSIONS,
                       CommandVersionsBin/binary>>) ->
     CommandVersions = parse_command_versions(CommandVersionsBin),
     {exchange_command_versions, ResponseCode, CommandVersions};
-parse_response_body(?COMMAND_STREAM_INFO,
-                    <<ResponseCode:16, _Count:32, InfoBin/binary>>) ->
-    Info = parse_map(InfoBin, #{}),
-    {stream_info, ResponseCode, Info}.
+parse_response_body(?COMMAND_STREAM_STATS,
+                    <<ResponseCode:16, _Count:32, StatsBin/binary>>) ->
+    Info = parse_int_map(StatsBin, #{}),
+    {stream_stats, ResponseCode, Info}.
 
 offset_spec(OffsetType, OffsetValueBin) ->
     case OffsetType of
@@ -1027,6 +1025,11 @@ parse_map(<<?STRING(KeySize, Key), ?STRING(ValSize, Value),
           Acc) ->
     parse_map(Rem, Acc#{Key => Value}).
 
+parse_int_map(<<>>, Acc) ->
+    Acc;
+parse_int_map(<<?STRING(KeySize, Key), Value:64, Rem/binary>>, Acc) ->
+    parse_int_map(Rem, Acc#{Key => Value}).
+
 generate_map(Map) ->
     maps:fold(fun(K, V, Acc) -> [<<?STRING(K), ?STRING(V)>> | Acc] end,
               [], Map).
@@ -1107,8 +1110,8 @@ command_id(consumer_update) ->
     ?COMMAND_CONSUMER_UPDATE;
 command_id(exchange_command_versions) ->
     ?COMMAND_EXCHANGE_COMMAND_VERSIONS;
-command_id(stream_info) ->
-    ?COMMAND_STREAM_INFO.
+command_id(stream_stats) ->
+    ?COMMAND_STREAM_STATS.
 
 parse_command_id(?COMMAND_DECLARE_PUBLISHER) ->
     declare_publisher;
@@ -1164,8 +1167,8 @@ parse_command_id(?COMMAND_CONSUMER_UPDATE) ->
     consumer_update;
 parse_command_id(?COMMAND_EXCHANGE_COMMAND_VERSIONS) ->
     exchange_command_versions;
-parse_command_id(?COMMAND_STREAM_INFO) ->
-    stream_info.
+parse_command_id(?COMMAND_STREAM_STATS) ->
+    stream_stats.
 
 element_index(Element, List) ->
     element_index(Element, List, 0).
diff --git a/deps/rabbitmq_stream_common/test/rabbit_stream_core_SUITE.erl b/deps/rabbitmq_stream_common/test/rabbit_stream_core_SUITE.erl
@@ -108,7 +108,7 @@ roundtrip(_Config) ->
     test_roundtrip({request, 99,
                     {exchange_command_versions,
                      [{deliver, ?VERSION_1, ?VERSION_1}]}}),
-    test_roundtrip({request, 99, {stream_info, <<"stream_name">>}}),
+    test_roundtrip({request, 99, {stream_stats, <<"stream_name">>}}),
 
     %% RESPONSES
     [test_roundtrip({response, 99, {Tag, 53}})
@@ -143,7 +143,7 @@ roundtrip(_Config) ->
                     {exchange_command_versions, 1,
                      [{publish, ?VERSION_1, ?VERSION_1}]}}),
     test_roundtrip({response, 99,
-                    {stream_info, 1, #{<<"committed_offset">> => <<"42">>}}}),
+                    {stream_stats, 1, #{<<"committed_offset">> => 42}}}),
     ok.
 
 roundtrip_metadata(_Config) ->
