Add out-of-sync replica (OSR) tracking to clustering

src/lavinmq/clustering/follower.cr

@@ -224,6 +224,7 @@ module LavinMQ
           uncompressed_bytes: @lz4.uncompressed_bytes,
           compressed_bytes:   @lz4.compressed_bytes,
           id:                 @id.to_s(36),
+          state:              @state.to_s,
         }.to_json(json)
       end
 

src/lavinmq/clustering/server.cr

@@ -32,6 +32,7 @@ module LavinMQ
       @password : String
       @files = Hash(String, MFile?).new
       @dirty_isr = true
+      @dirty_osr = true
       @id : Int32
       @config : Config
 
@@ -180,19 +181,22 @@ module LavinMQ
             stale_follower.close
           end
           @followers << follower # Starts in Syncing state
+          update_osr
         end
         follower.full_sync # sync the bulk
         @lock.synchronize do
           follower.full_sync    # sync the last
           follower.mark_synced! # Change state to Synced
           update_isr
+          update_osr
         end
         begin
           follower.action_loop
         ensure
           @lock.synchronize do
             @followers.delete(follower)
             @dirty_isr = true
+            @dirty_osr = true
           end
         end
       rescue ex : AuthenticationError
@@ -222,6 +226,21 @@ module LavinMQ
         @dirty_isr = false
       end
 
+      private def update_osr
+        osr_key = "#{@config.clustering_etcd_prefix}/osr"
+        ids = String.build do |str|
+          @followers.each do |f|
+            next unless f.syncing?
+            f.id.to_s(str, 36)
+            str << ","
+          end
+        end
+        ids = ids.rstrip(',')
+        Log.info { "Out-of-sync replicas: #{ids}" }
+        @etcd.put(osr_key, ids)
+        @dirty_osr = false
+      end
+
       def close
         @listeners.each &.close
         @lock.synchronize do
@@ -235,6 +254,7 @@ module LavinMQ
       private def each_follower(& : Follower -> Nil) : Nil
         @lock.synchronize do
           update_isr if @dirty_isr
+          update_osr if @dirty_osr
           @followers.each do |f|
             next if f.syncing? # Performing a full sync
             yield f

src/lavinmq/http/controller/nodes.cr

@@ -79,7 +79,7 @@ module LavinMQ
           proc_used:          Fiber.count,
           run_queue:          0,
           sockets_used:       @amqp_server.vhosts.sum { |_, v| v.connections.size },
-          followers:          @amqp_server.followers,
+          followers:          @amqp_server.all_followers,
         }
       end
 

static/js/nodes.js

@@ -182,10 +182,19 @@ Table.renderTable('followers', followersTableOpts, (tr, item, firstRender) => {
   if (firstRender) {
     Table.renderCell(tr, 0, item.id)
   }
-  Table.renderCell(tr, 1, item.remote_address)
-  Table.renderCell(tr, 2, humanizeBytes(item.sent_bytes), 'right')
-  Table.renderCell(tr, 3, humanizeBytes(item.acked_bytes), 'right')
-  Table.renderCell(tr, 4, humanizeBytes(item.sent_bytes - item.acked_bytes), 'right')
+  // Display state from follower object
+  const state = item.state || 'Unknown'
+  const statusCell = Table.renderCell(tr, 1, state === 'Syncing' ? 'Out-of-sync' : 'In-sync')
+  if (state === 'Syncing') {
+    statusCell.style.color = '#ff6b6b'
+    statusCell.style.fontWeight = 'bold'
+  } else if (state === 'Synced') {
+    statusCell.style.color = '#51cf66'
+  }
+  Table.renderCell(tr, 2, item.remote_address)
+  Table.renderCell(tr, 3, humanizeBytes(item.sent_bytes), 'right')
+  Table.renderCell(tr, 4, humanizeBytes(item.acked_bytes), 'right')
+  Table.renderCell(tr, 5, humanizeBytes(item.sent_bytes - item.acked_bytes), 'right')
 })
 
 function updateCharts (response) {
@@ -243,6 +252,7 @@ function updateCharts (response) {
     }
     Chart.update(queueChurnChart, queueChurnStats)
   }
+
   followersDataSource.update(response[0].followers)
 }
 

views/nodes.ecr

@@ -88,6 +88,7 @@
             <thead>
               <tr>
                 <th class="left">ID</th>
+                <th class="left">Status</th>
                 <th class="left">Remote address</th>
                 <th class="right">Sent bytes</th>
                 <th class="right">Acked bytes</th>