raft info

rabbitmq · Jan 20, 2025 · 0548ffa · 0548ffa
1 parent a8c8cf2
commit 0548ffa
Show file tree

Hide file tree

Showing 4 changed files with 117 additions and 7 deletions.
diff --git a/deps/rabbit/src/rabbit_fifo.erl b/deps/rabbit/src/rabbit_fifo.erl
@@ -1029,9 +1029,10 @@ handle_aux(_RaftState, cast, {#return{msg_ids = MsgIds,
             %% for returns with a delivery limit set we can just return as before
             {no_reply, Aux0, RaAux0, [{append, Ret, {notify, Corr, Pid}}]}
     end;
-handle_aux(leader, _, {handle_tick, [QName, Overview0, Nodes]},
+handle_aux(leader, _, {handle_tick, [QName, MacOverview0, Nodes]},
            #?AUX{tick_pid = Pid} = Aux, RaAux) ->
-    Overview = Overview0#{members_info => ra_aux:members_info(RaAux)},
+    Overview = MacOverview0#{members_info => ra_aux:members_info(RaAux),
+                             ra_overview => ra_aux:overview(RaAux)},
     NewPid =
         case process_is_alive(Pid) of
             false ->

diff --git a/deps/rabbit/src/rabbit_quorum_queue.erl b/deps/rabbit/src/rabbit_quorum_queue.erl
@@ -560,12 +560,29 @@ handle_tick(QName,
               num_discarded := NumDiscarded,
               num_discard_checked_out  :=  NumDiscardedCheckedOut,
               discard_message_bytes := DiscardBytes,
-              discard_checkout_message_bytes := DiscardCheckoutBytes,
-              smallest_raft_index := _} = Overview,
-            Nodes) ->
+              discard_checkout_message_bytes := DiscardCheckoutBytes
+             } = Overview,
+            KnownNodes) ->
     %% this makes calls to remote processes so cannot be run inside the
     %% ra server
     Self = self(),
+    MembersInfo0 = maps:get(members_info, Overview, #{}),
+    RaOverview0 = maps:get(ra_overview, Overview, #{}),
+    RaOverview = maps:update_with(log,
+                                  fun (L) ->
+                                          maps:with([last_index,
+                                                     first_index,
+                                                     last_written_index,
+                                                     last_wal_write,
+                                                     lastest_checkpoint_index],
+                                                    case L of
+                                                        #{last_written_index_term := {I, _}} = LO ->
+                                                            LO#{last_written_index => I};
+                                                        LO ->
+                                                            LO
+                                                    end)
+                                  end, #{}, RaOverview0),
+
     spawn(
       fun() ->
               try
@@ -600,6 +617,17 @@ handle_tick(QName,
                              end, info(Q, Keys), Overview),
                   MsgBytesDiscarded = DiscardBytes + DiscardCheckoutBytes,
                   MsgBytes = EnqueueBytes + CheckoutBytes + MsgBytesDiscarded,
+                  MembersInfo = maps:fold(fun
+                                              ({_, N}, Val0, Acc) ->
+                                                  Val = maps:map(
+                                                          fun (status, {S, _}) ->
+                                                                  S;
+                                                              (_K, V) ->
+                                                                  V
+                                                          end, Val0),
+                                                  Acc#{N => Val}
+                                          end, #{}, MembersInfo0),
+
                   Infos = [{consumers, NumConsumers},
                            {publishers, NumEnqueuers},
                            {consumer_capacity, Util},
@@ -620,7 +648,12 @@ handle_tick(QName,
                                                     unlimited;
                                                 Limit ->
                                                     Limit
-                                            end}
+                                            end},
+                           {members_info, MembersInfo},
+                           {raft, maps:with([current_term,
+                                             commit_index,
+                                             last_applied,
+                                             log], RaOverview)}
                            | Infos0],
                   rabbit_core_metrics:queue_stats(QName, Infos),
                   ok = repair_leader_record(Q, Self),
@@ -631,7 +664,7 @@ handle_tick(QName,
                           rabbit_log:debug("Repaired quorum queue ~ts amqqueue record", [rabbit_misc:rs(QName)])
                   end,
                   ExpectedNodes = rabbit_nodes:list_members(),
-                  case Nodes -- ExpectedNodes of
+                  case KnownNodes -- ExpectedNodes of
                       [] ->
                           ok;
                       Stale when length(ExpectedNodes) > 0 ->

diff --git a/deps/rabbitmq_management/priv/www/js/tmpl/queue.ejs b/deps/rabbitmq_management/priv/www/js/tmpl/queue.ejs
@@ -421,6 +421,17 @@
 </div>
 <% } %>
 
+<% if(is_quorum(queue) && queue.hasOwnProperty('raft')) { %>
+<div class="section-hidden">
+  <h2 class="updatable">Raft Info (Advanced) </h2>
+  <div class="hider updatable">
+<%= format('quorum-members-info', {'mode': 'queue',
+                                   'queue': queue,
+                                   'members_info': queue.members_info}) %>
+  </div>
+</div>
+<% } %>
+
 <% if(queue.reductions || queue.garbage_collection) { %>
 <div class="section-hidden">
 <h2>Runtime Metrics (Advanced)</h2>

diff --git a/deps/rabbitmq_management/priv/www/js/tmpl/quorum-members-info.ejs b/deps/rabbitmq_management/priv/www/js/tmpl/quorum-members-info.ejs
@@ -0,0 +1,65 @@
+<% if (members_info) {
+var i = 0;
+var r = queue.raft;
+var l = r.log;
+delete members_info[queue.leader];
+%>
+<h3>Leader</h3>
+<table class="list">
+    <thead>
+        <tr>
+            <th>Node</th>
+            <th>Term</th>
+            <th>First Index</th>
+            <th>Last Index</th>
+            <th>Log Length</th>
+            <th>Last Applied</th>
+            <th>Commit Index</th>
+        </tr>
+    </thead>
+        <tr>
+       <td><%= fmt_string(queue.leader) %></td>
+       <td><%= fmt_string(r.current_term) %></td>
+       <td><%= fmt_string(l.first_index) %></td>
+       <td><%= fmt_string(l.last_index) %></td>
+       <td><%= fmt_string(l.last_index - l.first_index + 1) %></td>
+       <td><%= fmt_string(r.last_applied) %></td>
+       <td><%= fmt_string(r.commit_index) %></td>
+        </tr>
+
+</table>
+<h3>Followers </h3>
+<table class="list">
+    <thead>
+        <tr>
+            <th>Node</th>
+            <th>Status</th>
+            <th>Next Index</th>
+            <th>Match Index</th>
+            <th>Lag</th>
+            <th>Commit Index Sent</th>
+            <th>Voter Status</th>
+        </tr>
+    </thead>
+    <% for (var node in members_info) {
+       i++;
+       var m = members_info[node];
+       %>
+      <tr<%= alt_rows(i) %>>
+       <td><%= fmt_string(node) %></td>
+       <td><%= fmt_string(m.status) %></td>
+       <td><%= fmt_string(m.next_index) %></td>
+       <td><%= fmt_string(m.match_index) %></td>
+       <td><%= fmt_string(l.last_index - Math.max(l.first_index, m.match_index)) %></td>
+       <% if(m.hasOwnProperty('commit_index_sent')) { %>
+       <td><%= fmt_string(m.commit_index_sent) %></td>
+       <% } else { %>
+       <td><%= fmt_string("n/a") %></td>
+       <% } %>
+       <td><%= fmt_string(m.voter_status.membership) %></td>
+       </tr>
+       <% } %>
+</table>
+<% } else { %>
+<p>... no info ...</p>
+<% } %>