Adds lag_in_actions to metrics

CHANGELOG.md

@@ -25,6 +25,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Added cluster_status to lavinmqctl [#787](https://github.com/cloudamqp/lavinmq/pull/787)
 - Added deliver_get to message_stats [#793](https://github.com/cloudamqp/lavinmq/pull/793)
+- Added lag_in_actions to metrics [#811](https://github.com/cloudamqp/lavinmq/pull/811)
 
 ## [2.0.0-rc.4] - 2024-08-21
 

src/lavinmq/clustering/follower.cr

@@ -177,6 +177,7 @@ module LavinMQ
           sent_bytes:         @sent_bytes,
           acked_bytes:        @acked_bytes,
           lag_in_bytes:       lag_in_bytes,
+          lag_in_actions:     lag_in_actions,
           compression_ratio:  @lz4.compression_ratio,
           uncompressed_bytes: @lz4.uncompressed_bytes,
           compressed_bytes:   @lz4.compressed_bytes,
@@ -187,6 +188,10 @@ module LavinMQ
       def lag_in_bytes : Int64
         @sent_bytes - @acked_bytes
       end
+
+      def lag_in_actions : Int32
+        @actions.size
+      end
     end
   end
 end

src/lavinmq/http/controller/prometheus.cr

@@ -270,6 +270,11 @@ module LavinMQ
                         value:  f.lag_in_bytes,
                         type:   "gauge",
                         help:   "Bytes that hasn't been synchronized with the follower yet"})
+          writer.write({name:   "follower_lag_in_actions",
+                        labels: {id: f.id.to_s(36)},
+                        value:  f.lag_in_actions,
+                        type:   "gauge",
+                        help:   "Actions that hasn't been synchronized with the follower yet"})
         end
       end
 

src/stdlib/channel.cr

@@ -47,4 +47,11 @@ class Channel(T)
   ensure
     @lock.unlock
   end
+
+  def size : Int32
+    if queue = @queue
+      return queue.size
+    end
+    0i32
+  end
 end