Remove ra_metrics

Author: mkuratczyk

deps/rabbit/src/rabbit_observer_cli_quorum_queues.erl

@@ -123,11 +123,11 @@ sheet_header() ->
 sheet_body(PrevState) ->
     {_, RaStates} = rabbit_quorum_queue:all_replica_states(),
     Body = [begin
-                #resource{name = Name, virtual_host = Vhost} = R = amqqueue:get_name(Q),
+                #resource{name = Name, virtual_host = Vhost} = amqqueue:get_name(Q),
                 case rabbit_amqqueue:pid_of(Q) of
                     none ->
                         empty_row(Name);
-                    {QName, _QNode} = _QQ ->
+                    {QName, _QNode} = ServerId ->
                         case whereis(QName) of
                             undefined ->
                                 empty_row(Name);
@@ -139,7 +139,12 @@ sheet_body(PrevState) ->
                                     _ ->
                                         QQCounters = maps:get({QName, node()}, ra_counters:overview()),
                                         {ok, InternalName} = rabbit_queue_type_util:qname_to_internal_name(#resource{virtual_host = Vhost, name= Name}),
-                                        [{_, CT, SnapIdx, LA, CI, LW, CL}] = ets:lookup(ra_metrics, R),
+                                        #{snapshot_index := SnapIdx,
+                                            last_written_index := LW,
+                                            term := CT,
+                                            commit_latency := CL,
+                                            commit_index := CI,
+                                            last_applied := LA} = ra:key_metrics(ServerId),
                                         [
                                          Pid,
                                          QName,

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -1598,12 +1598,10 @@ transfer_leadership(Q, Destination) ->
     end.
 
 queue_length(Q) ->
-    Name = amqqueue:get_name(Q),
-    case ets:lookup(ra_metrics, Name) of
-        [] -> 0;
-        [{_, _, SnapIdx, _, _, LastIdx, _}] ->
-            LastIdx - SnapIdx
-    end.
+    ServerId = amqqueue:get_pid(Q),
+    #{snapshot_index := SnapIdx,
+      last_written_index := LastIdx} = key_metrics_rpc(ServerId),
+    LastIdx - SnapIdx.
 
 get_replicas(Q) ->
     get_nodes(Q).

deps/rabbitmq_management_agent/src/rabbit_mgmt_external_stats.erl

@@ -267,20 +267,8 @@ i(context_switches, State) ->
     {Sw, 0} = erlang:statistics(context_switches),
     {State, Sw};
 i(ra_open_file_metrics, State) ->
-    {State, [{ra_log_wal, ra_metrics(ra_log_wal)},
-             {ra_log_segment_writer, ra_metrics(ra_log_segment_writer)}]}.
-
-ra_metrics(K) ->
-    try
-        case ets:lookup(ra_open_file_metrics, whereis(K)) of
-            [] -> 0;
-            [{_, C}] -> C
-        end
-    catch
-        error:badarg ->
-            %% On startup the mgmt might start before ra does
-            0
-    end.
+    {State, [{ra_log_wal, 0},
+             {ra_log_segment_writer, 0}]}.
 
 resource_alarm_set(Source) ->
     lists:member({{resource_limit, Source, node()},[]},
@@ -421,7 +409,7 @@ update_state(State0) ->
 get_fhc_stats() ->
     dict:to_list(dict:merge(fun(_, V1, V2) -> V1 + V2 end,
                             dict:from_list(zero_fhc_stats()),
-                            dict:from_list(get_ra_io_metrics()))).
+                            dict:from_list(get_zero_ra_io_metrics()))).
 
 zero_fhc_stats() ->
     [{{Op, Counter}, 0} || Op <- [io_read, io_write],
@@ -435,5 +423,17 @@ zero_fhc_stats() ->
                                   queue_index_write, queue_index_read],
                            Counter <- [count]].
 
-get_ra_io_metrics() ->
-    lists:sort(ets:tab2list(ra_io_metrics)).
+get_zero_ra_io_metrics() ->
+    %% not tracked anymore
+    [{{io_file_handle_open_attempt,count},0},
+     {{io_file_handle_open_attempt,time},0},
+     {{io_read,bytes},0},
+     {{io_read,count},0},
+     {{io_read,time},0},
+     {{io_seek,count},0},
+     {{io_seek,time},0},
+     {{io_sync,count},0},
+     {{io_sync,time},0},
+     {{io_write,bytes},0},
+     {{io_write,count},0},
+     {{io_write,time},0}].

deps/rabbitmq_prometheus/docker/grafana/dashboards/RabbitMQ-Quorum-Queues-Raft.json

@@ -418,7 +418,7 @@
             "type": "prometheus",
             "uid": "${DS_PROMETHEUS}"
           },
-          "expr": "rabbitmq_raft_entry_commit_latency_seconds * on(instance, job) group_left(rabbitmq_cluster, rabbitmq_node) rabbitmq_identity_info{rabbitmq_cluster=\"$rabbitmq_cluster\", namespace=\"$namespace\",rabbitmq_endpoint=\"$endpoint\"}",
+          "expr": "rabbitmq_raft_commit_latency_seconds * on(instance, job) group_left(rabbitmq_cluster, rabbitmq_node) rabbitmq_identity_info{rabbitmq_cluster=\"$rabbitmq_cluster\", namespace=\"$namespace\",rabbitmq_endpoint=\"$endpoint\"}",
           "format": "time_series",
           "instant": false,
           "intervalFactor": 1,

deps/rabbitmq_prometheus/metrics.md

@@ -249,7 +249,7 @@ These metrics are specific to the stream protocol.
 
 | Metric                                     | Description                             |
 | ---                                        | ---                                     |
-| rabbitmq_raft_entry_commit_latency_seconds | Time taken for an entry to be committed |
+| rabbitmq_raft_commit_latency_seconds       | Time taken for an entry to be committed |
 | rabbitmq_raft_log_commit_index             | Raft log commit index                   |
 | rabbitmq_raft_log_last_applied_index       | Raft log last applied index             |
 | rabbitmq_raft_log_last_written_index       | Raft log last written index             |

deps/rabbitmq_prometheus/src/collectors/prometheus_rabbitmq_core_metrics_collector.erl

@@ -106,15 +106,6 @@
         {2, ?MICROSECOND, io_seek_time_seconds_total, counter, "Total I/O seek time", io_seek_time}
     ]},
 
-    {ra_metrics, [
-        {2, undefined, raft_term_total, counter, "Current Raft term number"},
-        {3, undefined, raft_log_snapshot_index, gauge, "Raft log snapshot index"},
-        {4, undefined, raft_log_last_applied_index, gauge, "Raft log last applied index"},
-        {5, undefined, raft_log_commit_index, gauge, "Raft log commit index"},
-        {6, undefined, raft_log_last_written_index, gauge, "Raft log last written index"},
-        {7, ?MILLISECOND, raft_entry_commit_latency_seconds, gauge, "Time taken for a log entry to be committed"}
-    ]},
-
     {auth_attempt_metrics, [
         {2, undefined, auth_attempts_total, counter, "Total number of authentication attempts"},
         {3, undefined, auth_attempts_succeeded_total, counter, "Total number of successful authentication attempts"},
@@ -701,7 +692,6 @@ get_data(Table, false, VHostsFilter) when Table == channel_exchange_metrics;
                            Table == exchange_metrics;
                            Table == queue_exchange_metrics;
                            Table == channel_queue_exchange_metrics;
-                           Table == ra_metrics;
                            Table == channel_process_metrics ->
     Result = ets:foldl(fun
                   %% For queue_coarse_metrics
@@ -723,33 +713,10 @@ get_data(Table, false, VHostsFilter) when Table == channel_exchange_metrics;
                        {T, V1 + A1, V2 + A2, V3 + A3, V4 + A4};
                   ({_, V1, V2, V3, V4}, {T, A1, A2, A3, A4}) ->
                        {T, V1 + A1, V2 + A2, V3 + A3, V4 + A4};
-                  ({_, V1, V2, V3, V4, V5, V6}, {T, A1, A2, A3, A4, A5, A6}) ->
-                       %% ra_metrics: raft_entry_commit_latency_seconds needs to be an average
-                       {T, V1 + A1, V2 + A2, V3 + A3, V4 + A4, V5 + A5, accumulate_count_and_sum(V6, A6)};
                   ({_, V1, V2, V3, V4, V5, V6, V7, _}, {T, A1, A2, A3, A4, A5, A6, A7}) ->
                        {T, V1 + A1, V2 + A2, V3 + A3, V4 + A4, V5 + A5, V6 + A6, V7 + A7}
                end, empty(Table), Table),
-    case Table of
-        %% raft_entry_commit_latency_seconds needs to be an average
-        ra_metrics ->
-            {Count, Sum} = element(7, Result),
-            [setelement(7, Result, division(Sum, Count))];
-        _ ->
-            [Result]
-    end;
-get_data(ra_metrics = Table, true, _) ->
-    ets:foldl(
-      fun ({#resource{kind = queue}, _, _, _, _, _, _} = Row, Acc) ->
-              %% Metrics for QQ records use the queue resource as the table
-              %% key. The queue name and vhost will be rendered as tags.
-              [Row | Acc];
-          ({ClusterName, _, _, _, _, _, _} = Row, Acc) when is_atom(ClusterName) ->
-              %% Other Ra clusters like Khepri and the stream coordinator use
-              %% the cluster name as the metrics key. Transform this into a
-              %% value that can be rendered as a "raft_cluster" tag.
-              Row1 = setelement(1, Row, #{<<"raft_cluster">> => atom_to_binary(ClusterName, utf8)}),
-              [Row1 | Acc]
-      end, [], Table);
+    [Result];
 get_data(exchange_metrics = Table, true, VHostsFilter) when is_map(VHostsFilter)->
     ets:foldl(fun
         ({#resource{kind = exchange, virtual_host = VHost}, _, _, _, _, _} = Row, Acc) when
@@ -912,22 +879,12 @@ sum_queue_metrics(Props, {T, A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11,
      sum(proplists:get_value(segments, Props), A17)
     }.
 
-division(0, 0) ->
-    0;
-division(A, B) ->
-    A / B.
-
-accumulate_count_and_sum(Value, {Count, Sum}) ->
-    {Count + 1, Sum + Value}.
-
 empty(T) when T == channel_queue_exchange_metrics; T == queue_exchange_metrics; T == channel_process_metrics; T == queue_consumer_count ->
     {T, 0};
 empty(T) when T == connection_coarse_metrics; T == auth_attempt_metrics; T == auth_attempt_detailed_metrics ->
     {T, 0, 0, 0};
 empty(T) when T == channel_exchange_metrics; T == exchange_metrics; T == queue_coarse_metrics; T == connection_metrics ->
     {T, 0, 0, 0, 0};
-empty(T) when T == ra_metrics ->
-    {T, 0, 0, 0, 0, 0, {0, 0}};
 empty(T) when T == channel_queue_metrics; T == queue_delivery_metrics; T == channel_metrics ->
     {T, 0, 0, 0, 0, 0, 0, 0};
 empty(queue_metrics = T) ->

deps/rabbitmq_prometheus/test/rabbit_prometheus_http_SUITE.erl

@@ -392,7 +392,6 @@ aggregated_metrics_test(Config) ->
     ?assertEqual(match, re:run(Body, "^rabbitmq_process_open_fds ", [{capture, none}, multiline])),
     ?assertEqual(match, re:run(Body, "^rabbitmq_process_max_fds ", [{capture, none}, multiline])),
     ?assertEqual(match, re:run(Body, "^rabbitmq_io_read_ops_total ", [{capture, none}, multiline])),
-    ?assertEqual(match, re:run(Body, "^rabbitmq_raft_term ", [{capture, none}, multiline])),
     ?assertEqual(match, re:run(Body, "^rabbitmq_queue_messages_ready ", [{capture, none}, multiline])),
     ?assertEqual(match, re:run(Body, "^rabbitmq_queue_consumers ", [{capture, none}, multiline])),
     ?assertEqual(match, re:run(Body, "TYPE rabbitmq_auth_attempts_total", [{capture, none}, multiline])),
@@ -403,7 +402,7 @@ aggregated_metrics_test(Config) ->
     %% Check the first TOTALS metric value
     ?assertEqual(match, re:run(Body, "^rabbitmq_connections ", [{capture, none}, multiline])),
     %% Check raft_entry_commit_latency_seconds because we are aggregating it
-    ?assertEqual(match, re:run(Body, "^rabbitmq_raft_entry_commit_latency_seconds ", [{capture, none}, multiline])).
+    ?assertEqual(match, re:run(Body, "^rabbitmq_raft_commit_latency_seconds ", [{capture, none}, multiline])).
 
 endpoint_per_object_metrics(Config) ->
     per_object_metrics_test(Config, "/metrics/per-object").
@@ -439,7 +438,7 @@ per_object_metrics_test(Config, Path) ->
     %% Check the first metric value in each ETS table that requires converting
     ?assertEqual(match, re:run(Body, "^rabbitmq_erlang_uptime_seconds ", [{capture, none}, multiline])),
     ?assertEqual(match, re:run(Body, "^rabbitmq_io_read_time_seconds_total ", [{capture, none}, multiline])),
-    ?assertEqual(match, re:run(Body, "^rabbitmq_raft_entry_commit_latency_seconds{", [{capture, none}, multiline])),
+    ?assertEqual(match, re:run(Body, "^rabbitmq_raft_commit_latency_seconds{", [{capture, none}, multiline])),
     %% Check the first TOTALS metric value
     ?assertEqual(match, re:run(Body, "^rabbitmq_connections ", [{capture, none}, multiline])).