Add meta snapshot metrics to jsz monitoring

Signed-off-by: Waldemar Quevedo <[email protected]>

Author: wallyqs

server/jetstream_cluster.go

@@ -70,6 +70,9 @@ type jetStreamCluster struct {
 	peerStreamCancelMove *subscription
 	// To pop out the monitorCluster before the raft layer.
 	qch chan struct{}
+	// Track last meta snapshot time and duration for monitoring (atomic)
+	lastMetaSnapTime     int64 // Unix nanoseconds, accessed atomically
+	lastMetaSnapDuration int64 // Duration in nanoseconds, accessed atomically
 }
 
 // Used to track inflight stream add requests to properly re-use same group and sync subject.
@@ -1405,11 +1408,20 @@ func (js *jetStream) monitorCluster() {
 		byNeither := !byEntries && !bySize
 		// For the meta layer we want to snapshot when over the above threshold (which could be 0 by default).
 		if ne, nsz := n.Size(); force || byNeither || (byEntries && ne > ethresh) || (bySize && nsz > szthresh) || n.NeedSnapshot() {
+			snapStart := time.Now()
 			snap, err := js.metaSnapshot()
 			if err != nil {
 				s.Warnf("Error generating JetStream cluster snapshot: %v", err)
 			} else if err = n.InstallSnapshot(snap); err == nil {
 				lastSnapTime = time.Now()
+				snapDuration := lastSnapTime.Sub(snapStart)
+				// Update cluster-level last snap time and duration for monitoring (atomic)
+				js.mu.RLock()
+				if js.cluster != nil {
+					atomic.StoreInt64(&js.cluster.lastMetaSnapTime, lastSnapTime.UnixNano())
+					atomic.StoreInt64(&js.cluster.lastMetaSnapDuration, int64(snapDuration))
+				}
+				js.mu.RUnlock()
 			} else if err != errNoSnapAvailable && err != errNodeClosed {
 				s.Warnf("Error snapshotting JetStream cluster state: %v", err)
 			}

server/monitor.go

@@ -2969,14 +2969,23 @@ type AccountDetail struct {
 	Streams []StreamDetail `json:"stream_detail,omitempty"`
 }
 
+// MetaSnapshotStats shows information about meta snapshots.
+type MetaSnapshotStats struct {
+	PendingEntries uint64        `json:"pending_entries"`         // PendingEntries is the count of pending entries in the meta layer
+	PendingSize    uint64        `json:"pending_size"`            // PendingSize is the size in bytes of pending entries in the meta layer
+	LastTime       time.Time     `json:"last_time,omitempty"`     // LastTime is when the last meta snapshot was taken
+	LastDuration   time.Duration `json:"last_duration,omitempty"` // LastDuration is how long the last meta snapshot took
+}
+
 // MetaClusterInfo shows information about the meta group.
 type MetaClusterInfo struct {
-	Name     string      `json:"name,omitempty"`     // Name is the name of the cluster
-	Leader   string      `json:"leader,omitempty"`   // Leader is the server name of the cluster leader
-	Peer     string      `json:"peer,omitempty"`     // Peer is unique ID of the leader
-	Replicas []*PeerInfo `json:"replicas,omitempty"` // Replicas is a list of known peers
-	Size     int         `json:"cluster_size"`       // Size is the known size of the cluster
-	Pending  int         `json:"pending"`            // Pending is how many RAFT messages are not yet processed
+	Name     string             `json:"name,omitempty"`     // Name is the name of the cluster
+	Leader   string             `json:"leader,omitempty"`   // Leader is the server name of the cluster leader
+	Peer     string             `json:"peer,omitempty"`     // Peer is unique ID of the leader
+	Replicas []*PeerInfo        `json:"replicas,omitempty"` // Replicas is a list of known peers
+	Size     int                `json:"cluster_size"`       // Size is the known size of the cluster
+	Pending  int                `json:"pending"`            // Pending is how many RAFT messages are not yet processed
+	Snapshot *MetaSnapshotStats `json:"snapshot"`           // Snapshot contains meta snapshot statistics
 }
 
 // JSInfo has detailed information on JetStream.
@@ -3181,13 +3190,33 @@ func (s *Server) Jsz(opts *JSzOptions) (*JSInfo, error) {
 
 	if mg := js.getMetaGroup(); mg != nil {
 		if ci := s.raftNodeToClusterInfo(mg); ci != nil {
+			entries, bytes := mg.Size()
 			jsi.Meta = &MetaClusterInfo{Name: ci.Name, Leader: ci.Leader, Peer: getHash(ci.Leader), Size: mg.ClusterSize()}
 			if isLeader {
 				jsi.Meta.Replicas = ci.Replicas
 			}
 			if ipq := s.jsAPIRoutedReqs; ipq != nil {
 				jsi.Meta.Pending = ipq.len()
 			}
+			// Add meta snapshot stats
+			jsi.Meta.Snapshot = &MetaSnapshotStats{
+				PendingEntries: entries,
+				PendingSize:    bytes,
+			}
+			// Read atomic snapshot timing (no lock needed for atomics)
+			js.mu.RLock()
+			cluster := js.cluster
+			js.mu.RUnlock()
+			if cluster != nil {
+				timeNanos := atomic.LoadInt64(&cluster.lastMetaSnapTime)
+				durationNanos := atomic.LoadInt64(&cluster.lastMetaSnapDuration)
+				if timeNanos > 0 {
+					jsi.Meta.Snapshot.LastTime = time.Unix(0, timeNanos).UTC()
+				}
+				if durationNanos > 0 {
+					jsi.Meta.Snapshot.LastDuration = time.Duration(durationNanos)
+				}
+			}
 		}
 	}
 

server/monitor_test.go

@@ -5306,6 +5306,25 @@ func TestMonitorJsz(t *testing.T) {
 			t.Fatalf("received cluster info from multiple nodes")
 		}
 	})
+	t.Run("meta-snapshot-stats", func(t *testing.T) {
+		for _, url := range []string{monUrl1, monUrl2} {
+			info := readJsInfo(url)
+			require_True(t, info.Meta != nil)
+			require_True(t, info.Meta.Snapshot != nil)
+
+			snapshot := info.Meta.Snapshot
+
+			require_True(t, snapshot.PendingEntries >= 0)
+			require_True(t, snapshot.PendingSize >= 0)
+
+			if !snapshot.LastTime.IsZero() {
+				require_Equal(t, snapshot.LastTime.Location(), time.UTC)
+				require_True(t, snapshot.LastDuration >= 0)
+				require_True(t, snapshot.LastDuration < 30*time.Second)
+				require_True(t, time.Since(snapshot.LastTime) < 5*time.Minute)
+			}
+		}
+	})
 	t.Run("account-non-existing", func(t *testing.T) {
 		for _, url := range []string{monUrl1, monUrl2} {
 			info := readJsInfo(url + "?acc=DOES_NOT_EXIST")