@@ -30,6 +30,8 @@ type Producer struct {
3030 TCPPort int `json:"tcp_port"`
3131 HTTPPort int `json:"http_port"`
3232 Version string `json:"version"`
33+ TopologyZone string `json:"topology_zone,omitempty"`
34+ TopologyRegion string `json:"topology_region,omitempty"`
3335 VersionObj semver.Version `json:"-"`
3436 Topics ProducerTopics `json:"topics"`
3537 OutOfDate bool `json:"out_of_date"`
@@ -46,6 +48,8 @@ func (p *Producer) UnmarshalJSON(b []byte) error {
4648 Version string `json:"version"`
4749 Topics []string `json:"topics"`
4850 Tombstoned []bool `json:"tombstones"`
51+ TopologyZone string `json:"topology_zone,omitempty"`
52+ TopologyRegion string `json:"topology_region,omitempty"`
4953 }
5054 if err := json .Unmarshal (b , & r ); err != nil {
5155 return err
@@ -57,6 +61,8 @@ func (p *Producer) UnmarshalJSON(b []byte) error {
5761 TCPPort : r .TCPPort ,
5862 HTTPPort : r .HTTPPort ,
5963 Version : r .Version ,
64+ TopologyZone : r .TopologyZone ,
65+ TopologyRegion : r .TopologyRegion ,
6066 }
6167 for i , t := range r .Topics {
6268 p .Topics = append (p .Topics , ProducerTopic {Topic : t , Tombstoned : r .Tombstoned [i ]})
@@ -92,16 +98,20 @@ func (p *Producer) IsInconsistent(numLookupd int) bool {
9298}
9399
94100type TopicStats struct {
95- Node string `json:"node"`
96- Hostname string `json:"hostname"`
97- TopicName string `json:"topic_name"`
98- Depth int64 `json:"depth"`
99- MemoryDepth int64 `json:"memory_depth"`
100- BackendDepth int64 `json:"backend_depth"`
101- MessageCount int64 `json:"message_count"`
102- NodeStats []* TopicStats `json:"nodes"`
103- Channels []* ChannelStats `json:"channels"`
104- Paused bool `json:"paused"`
101+ Node string `json:"node"`
102+ Hostname string `json:"hostname"`
103+ TopicName string `json:"topic_name"`
104+ Depth int64 `json:"depth"`
105+ MemoryDepth int64 `json:"memory_depth"`
106+ BackendDepth int64 `json:"backend_depth"`
107+ MessageCount int64 `json:"message_count"`
108+ DeliveryMsgCount int64 `json:"delivery_msg_count"`
109+ ZoneLocalMsgCount int64 `json:"zone_local_msg_count,omitempty"`
110+ RegionLocalMsgCount int64 `json:"region_local_msg_count,omitempty"`
111+ GlobalMsgCount int64 `json:"global_msg_count,omitempty"`
112+ NodeStats []* TopicStats `json:"nodes"`
113+ Channels []* ChannelStats `json:"channels"`
114+ Paused bool `json:"paused"`
105115
106116 E2eProcessingLatency * quantile.E2eProcessingLatencyAggregate `json:"e2e_processing_latency"`
107117}
@@ -112,6 +122,10 @@ func (t *TopicStats) Add(a *TopicStats) {
112122 t .MemoryDepth += a .MemoryDepth
113123 t .BackendDepth += a .BackendDepth
114124 t .MessageCount += a .MessageCount
125+ t .DeliveryMsgCount += a .DeliveryMsgCount
126+ t .ZoneLocalMsgCount += a .ZoneLocalMsgCount
127+ t .RegionLocalMsgCount += a .RegionLocalMsgCount
128+ t .GlobalMsgCount += a .GlobalMsgCount
115129 if a .Paused {
116130 t .Paused = a .Paused
117131 }
@@ -139,23 +153,27 @@ func (t *TopicStats) Add(a *TopicStats) {
139153}
140154
141155type ChannelStats struct {
142- Node string `json:"node"`
143- Hostname string `json:"hostname"`
144- TopicName string `json:"topic_name"`
145- ChannelName string `json:"channel_name"`
146- Depth int64 `json:"depth"`
147- MemoryDepth int64 `json:"memory_depth"`
148- BackendDepth int64 `json:"backend_depth"`
149- InFlightCount int64 `json:"in_flight_count"`
150- DeferredCount int64 `json:"deferred_count"`
151- RequeueCount int64 `json:"requeue_count"`
152- TimeoutCount int64 `json:"timeout_count"`
153- MessageCount int64 `json:"message_count"`
154- ClientCount int `json:"client_count"`
155- Selected bool `json:"-"`
156- NodeStats []* ChannelStats `json:"nodes"`
157- Clients []* ClientStats `json:"clients"`
158- Paused bool `json:"paused"`
156+ Node string `json:"node"`
157+ Hostname string `json:"hostname"`
158+ TopicName string `json:"topic_name"`
159+ ChannelName string `json:"channel_name"`
160+ Depth int64 `json:"depth"`
161+ MemoryDepth int64 `json:"memory_depth"`
162+ BackendDepth int64 `json:"backend_depth"`
163+ InFlightCount int64 `json:"in_flight_count"`
164+ DeferredCount int64 `json:"deferred_count"`
165+ RequeueCount int64 `json:"requeue_count"`
166+ TimeoutCount int64 `json:"timeout_count"`
167+ MessageCount int64 `json:"message_count"`
168+ DeliveryMsgCount int64 `json:"delivery_msg_count,omitempty"`
169+ ZoneLocalMsgCount int64 `json:"zone_local_msg_count,omitempty"`
170+ RegionLocalMsgCount int64 `json:"region_local_msg_count,omitempty"`
171+ GlobalMsgCount int64 `json:"global_msg_count,omitempty"`
172+ ClientCount int `json:"client_count"`
173+ Selected bool `json:"-"`
174+ NodeStats []* ChannelStats `json:"nodes"`
175+ Clients []* ClientStats `json:"clients"`
176+ Paused bool `json:"paused"`
159177
160178 E2eProcessingLatency * quantile.E2eProcessingLatencyAggregate `json:"e2e_processing_latency"`
161179}
@@ -170,6 +188,10 @@ func (c *ChannelStats) Add(a *ChannelStats) {
170188 c .RequeueCount += a .RequeueCount
171189 c .TimeoutCount += a .TimeoutCount
172190 c .MessageCount += a .MessageCount
191+ c .DeliveryMsgCount += a .DeliveryMsgCount
192+ c .ZoneLocalMsgCount += a .ZoneLocalMsgCount
193+ c .RegionLocalMsgCount += a .RegionLocalMsgCount
194+ c .GlobalMsgCount += a .GlobalMsgCount
173195 c .ClientCount += a .ClientCount
174196 if a .Paused {
175197 c .Paused = a .Paused
@@ -189,25 +211,29 @@ func (c *ChannelStats) Add(a *ChannelStats) {
189211}
190212
191213type ClientStats struct {
192- Node string `json:"node"`
193- RemoteAddress string `json:"remote_address"`
194- Version string `json:"version"`
195- ClientID string `json:"client_id"`
196- Hostname string `json:"hostname"`
197- UserAgent string `json:"user_agent"`
198- ConnectTs int64 `json:"connect_ts"`
199- ConnectedDuration time.Duration `json:"connected"`
200- InFlightCount int `json:"in_flight_count"`
201- ReadyCount int `json:"ready_count"`
202- FinishCount int64 `json:"finish_count"`
203- RequeueCount int64 `json:"requeue_count"`
204- MessageCount int64 `json:"message_count"`
205- SampleRate int32 `json:"sample_rate"`
206- Deflate bool `json:"deflate"`
207- Snappy bool `json:"snappy"`
208- Authed bool `json:"authed"`
209- AuthIdentity string `json:"auth_identity"`
210- AuthIdentityURL string `json:"auth_identity_url"`
214+ Node string `json:"node"`
215+ RemoteAddress string `json:"remote_address"`
216+ Version string `json:"version"`
217+ ClientID string `json:"client_id"`
218+ Hostname string `json:"hostname"`
219+ UserAgent string `json:"user_agent"`
220+ ConnectTs int64 `json:"connect_ts"`
221+ ConnectedDuration time.Duration `json:"connected"`
222+ InFlightCount int `json:"in_flight_count"`
223+ ReadyCount int `json:"ready_count"`
224+ FinishCount int64 `json:"finish_count"`
225+ RequeueCount int64 `json:"requeue_count"`
226+ MessageCount int64 `json:"message_count"`
227+ SampleRate int32 `json:"sample_rate"`
228+ Deflate bool `json:"deflate"`
229+ Snappy bool `json:"snappy"`
230+ Authed bool `json:"authed"`
231+ AuthIdentity string `json:"auth_identity"`
232+ AuthIdentityURL string `json:"auth_identity_url"`
233+ NodeTopologyRegion string `json:"node_topology_region,omitempty"`
234+ NodeTopologyZone string `json:"node_topology_zone,omitempty"`
235+ TopologyRegion string `json:"topology_region,omitempty"`
236+ TopologyZone string `json:"topology_zone,omitempty"`
211237
212238 TLS bool `json:"tls"`
213239 CipherSuite string `json:"tls_cipher_suite"`
@@ -262,6 +288,35 @@ func (c ClientsByHost) Less(i, j int) bool {
262288 return c .ClientStatsList [i ].Hostname < c .ClientStatsList [j ].Hostname
263289}
264290
291+ type ClientStatsByNodeTopology struct {
292+ ClientStatsList
293+ }
294+
295+ func (c ClientStatsByNodeTopology ) Less (i , j int ) bool {
296+ // if its the same node, sort by topology
297+ if c .ClientStatsList [i ].Node == c .ClientStatsList [j ].Node {
298+ region := c .ClientStatsList [i ].NodeTopologyRegion
299+ zone := c .ClientStatsList [i ].NodeTopologyZone
300+
301+ switch {
302+ case c .ClientStatsList [i ].TopologyRegion == region && c .ClientStatsList [i ].TopologyZone == zone :
303+ return true
304+ case c .ClientStatsList [j ].TopologyRegion == region && c .ClientStatsList [j ].TopologyZone == zone :
305+ return false
306+ case c .ClientStatsList [i ].TopologyRegion == region :
307+ return true
308+ case c .ClientStatsList [j ].TopologyRegion == region :
309+ return false
310+ default :
311+ if c .ClientStatsList [i ].TopologyRegion == c .ClientStatsList [j ].TopologyRegion {
312+ return c .ClientStatsList [i ].TopologyZone < c .ClientStatsList [j ].TopologyZone
313+ }
314+ return c .ClientStatsList [i ].TopologyRegion < c .ClientStatsList [j ].TopologyRegion
315+ }
316+ }
317+ return c .ClientStatsList [i ].Node < c .ClientStatsList [j ].Node
318+ }
319+
265320type TopicStatsList []* TopicStats
266321
267322func (t TopicStatsList ) Len () int { return len (t ) }
0 commit comments