read memory stats from per-heap couters
[govpp.git] / core / stats.go
1 package core
2
3 import (
4         "path"
5         "strings"
6         "time"
7
8         "git.fd.io/govpp.git/adapter"
9         "git.fd.io/govpp.git/api"
10 )
11
12 var (
13         RetryUpdateCount    = 10
14         RetryUpdateDelay    = time.Millisecond * 10
15         HealthCheckInterval = time.Second // default health check probe interval
16 )
17
18 const (
19         SystemStatsPrefix               = "/sys/"
20         SystemStats_VectorRate          = SystemStatsPrefix + "vector_rate"
21         SystemStats_NumWorkerThreads    = SystemStatsPrefix + "num_worker_threads"
22         SystemStats_VectorRatePerWorker = SystemStatsPrefix + "vector_rate_per_worker"
23         SystemStats_InputRate           = SystemStatsPrefix + "input_rate"
24         SystemStats_LastUpdate          = SystemStatsPrefix + "last_update"
25         SystemStats_LastStatsClear      = SystemStatsPrefix + "last_stats_clear"
26         SystemStats_Heartbeat           = SystemStatsPrefix + "heartbeat"
27
28         NodeStatsPrefix    = "/sys/node/"
29         NodeStats_Names    = NodeStatsPrefix + "names"
30         NodeStats_Clocks   = NodeStatsPrefix + "clocks"
31         NodeStats_Vectors  = NodeStatsPrefix + "vectors"
32         NodeStats_Calls    = NodeStatsPrefix + "calls"
33         NodeStats_Suspends = NodeStatsPrefix + "suspends"
34
35         BufferStatsPrefix     = "/buffer-pools/"
36         BufferStats_Cached    = "cached"
37         BufferStats_Used      = "used"
38         BufferStats_Available = "available"
39
40         CounterStatsPrefix = "/err/"
41
42         MemoryStatSegPrefix = "/mem/statseg"
43         MemoryStatPrefix    = "/mem/stat"
44         MemoryMainPrefix    = "/mem/main"
45         MemoryStats_Total   = "total"
46         MemoryStats_Used    = "used"
47
48         InterfaceStatsPrefix         = "/if/"
49         InterfaceStats_Names         = InterfaceStatsPrefix + "names"
50         InterfaceStats_Drops         = InterfaceStatsPrefix + "drops"
51         InterfaceStats_Punt          = InterfaceStatsPrefix + "punt"
52         InterfaceStats_IP4           = InterfaceStatsPrefix + "ip4"
53         InterfaceStats_IP6           = InterfaceStatsPrefix + "ip6"
54         InterfaceStats_RxNoBuf       = InterfaceStatsPrefix + "rx-no-buf"
55         InterfaceStats_RxMiss        = InterfaceStatsPrefix + "rx-miss"
56         InterfaceStats_RxError       = InterfaceStatsPrefix + "rx-error"
57         InterfaceStats_TxError       = InterfaceStatsPrefix + "tx-error"
58         InterfaceStats_Mpls          = InterfaceStatsPrefix + "mpls"
59         InterfaceStats_Rx            = InterfaceStatsPrefix + "rx"
60         InterfaceStats_RxUnicast     = InterfaceStatsPrefix + "rx-unicast"
61         InterfaceStats_RxMulticast   = InterfaceStatsPrefix + "rx-multicast"
62         InterfaceStats_RxBroadcast   = InterfaceStatsPrefix + "rx-broadcast"
63         InterfaceStats_Tx            = InterfaceStatsPrefix + "tx"
64         InterfaceStats_TxUnicast     = InterfaceStatsPrefix + "tx-unicast"
65         InterfaceStats_TxUnicastMiss = InterfaceStatsPrefix + "tx-unicast-miss"
66         InterfaceStats_TxMulticast   = InterfaceStatsPrefix + "tx-multicast"
67         InterfaceStats_TxBroadcast   = InterfaceStatsPrefix + "tx-broadcast"
68
69         // TODO: network stats
70         NetworkStatsPrefix     = "/net/"
71         NetworkStats_RouteTo   = NetworkStatsPrefix + "route/to"
72         NetworkStats_RouteVia  = NetworkStatsPrefix + "route/via"
73         NetworkStats_MRoute    = NetworkStatsPrefix + "mroute"
74         NetworkStats_Adjacency = NetworkStatsPrefix + "adjacency"
75         NetworkStats_Punt      = NetworkStatsPrefix + "punt"
76 )
77
78 type StatsConnection struct {
79         statsClient adapter.StatsAPI
80
81         maxAttempts int           // interval for reconnect attempts
82         recInterval time.Duration // maximum number of reconnect attempts
83
84         connChan chan ConnectionEvent // connection event channel
85         done     chan struct{}        // to terminate stats connection watcher
86
87         errorStatsData *adapter.StatDir
88         nodeStatsData  *adapter.StatDir
89         ifaceStatsData *adapter.StatDir
90         sysStatsData   *adapter.StatDir
91         bufStatsData   *adapter.StatDir
92         memStatsData   *adapter.StatDir
93 }
94
95 func newStatsConnection(stats adapter.StatsAPI, attempts int, interval time.Duration) *StatsConnection {
96         if attempts == 0 {
97                 attempts = DefaultMaxReconnectAttempts
98         }
99         if interval == 0 {
100                 interval = DefaultReconnectInterval
101         }
102
103         return &StatsConnection{
104                 statsClient: stats,
105                 maxAttempts: attempts,
106                 recInterval: interval,
107                 connChan:    make(chan ConnectionEvent, NotificationChanBufSize),
108                 done:        make(chan struct{}),
109         }
110 }
111
112 // ConnectStats connects to Stats API using specified adapter and returns a connection handle.
113 // This call blocks until it is either connected, or an error occurs.
114 // Only one connection attempt will be performed.
115 func ConnectStats(stats adapter.StatsAPI) (*StatsConnection, error) {
116         log.Debug("Connecting to stats..")
117         c := newStatsConnection(stats, DefaultMaxReconnectAttempts, DefaultReconnectInterval)
118
119         if err := c.statsClient.Connect(); err != nil {
120                 return nil, err
121         }
122         log.Debugf("Connected to stats.")
123
124         return c, nil
125 }
126
127 // AsyncConnectStats  connects to the VPP stats socket asynchronously and returns the connection
128 // handle with state channel. The call is non-blocking and the caller is expected to watch ConnectionEvent
129 // values from the channel and wait for connect/disconnect events. Connection loop tries to reconnect the
130 // socket in case the session was disconnected.
131 func AsyncConnectStats(stats adapter.StatsAPI, attempts int, interval time.Duration) (*StatsConnection, chan ConnectionEvent, error) {
132         log.Debug("Connecting to stats asynchronously..")
133         c := newStatsConnection(stats, attempts, interval)
134
135         go c.connectLoop()
136
137         return c, c.connChan, nil
138 }
139
140 func (c *StatsConnection) connectLoop() {
141         log.Debug("Asynchronously connecting to stats..")
142         var reconnectAttempts int
143
144         // loop until connected
145         for {
146                 if err := c.statsClient.Connect(); err == nil {
147                         c.sendStatsConnEvent(ConnectionEvent{Timestamp: time.Now(), State: Connected})
148                         break
149                 } else if reconnectAttempts < c.maxAttempts {
150                         reconnectAttempts++
151                         log.Warnf("connecting stats failed (attempt %d/%d): %v", reconnectAttempts, c.maxAttempts, err)
152                         time.Sleep(c.recInterval)
153                 } else {
154                         c.sendStatsConnEvent(ConnectionEvent{Timestamp: time.Now(), State: Failed, Error: err})
155                         return
156                 }
157         }
158         // start monitoring stats connection state
159         go c.monitorSocket()
160 }
161
162 // Disconnect disconnects from Stats API and releases all connection-related resources.
163 func (c *StatsConnection) Disconnect() {
164         if c == nil {
165                 return
166         }
167         if c.statsClient != nil {
168                 if err := c.statsClient.Disconnect(); err != nil {
169                         log.Debugf("disconnecting stats client failed: %v", err)
170                 }
171         }
172         close(c.connChan)
173         close(c.done)
174 }
175
176 func (c *StatsConnection) monitorSocket() {
177         var state, lastState ConnectionState
178         ticker := time.NewTicker(HealthCheckInterval)
179
180         for {
181                 select {
182                 case <-ticker.C:
183                         _, err := c.statsClient.ListStats(SystemStats_Heartbeat)
184                         state = Connected
185                         if err == adapter.ErrStatsDataBusy {
186                                 state = NotResponding
187                         }
188                         if err == adapter.ErrStatsDisconnected {
189                                 state = Disconnected
190                         }
191                         if err == adapter.ErrStatsAccessFailed {
192                                 state = Failed
193                         }
194                         if state == lastState {
195                                 continue
196                         }
197                         lastState = state
198                         c.sendStatsConnEvent(ConnectionEvent{Timestamp: time.Now(), State: state, Error: err})
199                 case <-c.done:
200                         log.Debugf("health check watcher closed")
201                         c.sendStatsConnEvent(ConnectionEvent{Timestamp: time.Now(), State: Disconnected, Error: nil})
202                         break
203                 }
204         }
205 }
206
207 func (c *StatsConnection) updateStats(statDir **adapter.StatDir, patterns ...string) error {
208         if statDir == nil {
209                 panic("statDir must not nil")
210         }
211         try := func() error {
212                 if (*statDir) == nil {
213                         dir, err := c.statsClient.PrepareDir(patterns...)
214                         if err != nil {
215                                 log.Debugln("preparing dir failed:", err)
216                                 return err
217                         }
218                         *statDir = dir
219                 } else {
220                         if err := c.statsClient.UpdateDir(*statDir); err != nil {
221                                 log.Debugln("updating dir failed:", err)
222                                 *statDir = nil
223                                 return err
224                         }
225                 }
226
227                 return nil
228         }
229         var err error
230         for r := 0; r < RetryUpdateCount; r++ {
231                 if err = try(); err == nil {
232                         if r > 0 {
233                                 log.Debugf("retry successfull (r=%d)", r)
234                         }
235                         return nil
236                 } else if err == adapter.ErrStatsDirStale || err == adapter.ErrStatsDataBusy {
237                         // retrying
238                         if r > 1 {
239                                 log.Debugf("sleeping for %v before next try", RetryUpdateDelay)
240                                 time.Sleep(RetryUpdateDelay)
241                         }
242                 } else {
243                         // error is not retryable
244                         break
245                 }
246         }
247         return err
248 }
249
250 // GetSystemStats retrieves VPP system stats.
251 func (c *StatsConnection) GetSystemStats(sysStats *api.SystemStats) (err error) {
252         if err := c.updateStats(&c.sysStatsData, SystemStatsPrefix); err != nil {
253                 return err
254         }
255
256         for _, stat := range c.sysStatsData.Entries {
257                 var val uint64
258                 if s, ok := stat.Data.(adapter.ScalarStat); ok {
259                         val = uint64(s)
260                 }
261                 switch string(stat.Name) {
262                 case SystemStats_VectorRate:
263                         sysStats.VectorRate = val
264                 case SystemStats_NumWorkerThreads:
265                         sysStats.NumWorkerThreads = val
266                 case SystemStats_VectorRatePerWorker:
267                         var vals []uint64
268                         if ss, ok := stat.Data.(adapter.SimpleCounterStat); ok {
269                                 vals = make([]uint64, len(ss))
270                                 for w := range ss {
271                                         if ss[w] == nil {
272                                                 continue
273                                         }
274                                         vals[w] = uint64(ss[w][0])
275                                 }
276                         }
277                         sysStats.VectorRatePerWorker = vals
278                 case SystemStats_InputRate:
279                         sysStats.InputRate = val
280                 case SystemStats_LastUpdate:
281                         sysStats.LastUpdate = val
282                 case SystemStats_LastStatsClear:
283                         sysStats.LastStatsClear = val
284                 case SystemStats_Heartbeat:
285                         sysStats.Heartbeat = val
286                 }
287         }
288
289         return nil
290 }
291
292 // GetErrorStats retrieves VPP error stats.
293 func (c *StatsConnection) GetErrorStats(errorStats *api.ErrorStats) (err error) {
294         if err := c.updateStats(&c.errorStatsData, CounterStatsPrefix); err != nil {
295                 return err
296         }
297
298         if errorStats.Errors == nil || len(errorStats.Errors) != len(c.errorStatsData.Entries) {
299                 errorStats.Errors = make([]api.ErrorCounter, len(c.errorStatsData.Entries))
300                 for i := 0; i < len(c.errorStatsData.Entries); i++ {
301                         errorStats.Errors[i].CounterName = string(c.errorStatsData.Entries[i].Name)
302                 }
303         }
304
305         for i, stat := range c.errorStatsData.Entries {
306                 if stat.Type != adapter.ErrorIndex {
307                         continue
308                 }
309                 if errStat, ok := stat.Data.(adapter.ErrorStat); ok {
310                         values := make([]uint64, len(errStat))
311                         for j, errStatW := range errStat {
312                                 values[j] = uint64(errStatW)
313                         }
314                         errorStats.Errors[i].Values = values
315                 }
316         }
317
318         return nil
319 }
320
321 func (c *StatsConnection) GetNodeStats(nodeStats *api.NodeStats) (err error) {
322         if err := c.updateStats(&c.nodeStatsData, NodeStatsPrefix); err != nil {
323                 return err
324         }
325
326         prepNodes := func(l int) {
327                 if nodeStats.Nodes == nil || len(nodeStats.Nodes) != l {
328                         nodeStats.Nodes = make([]api.NodeCounters, l)
329                         for i := 0; i < l; i++ {
330                                 nodeStats.Nodes[i].NodeIndex = uint32(i)
331                         }
332                 }
333         }
334         perNode := func(stat adapter.StatEntry, fn func(*api.NodeCounters, uint64)) {
335                 if s, ok := stat.Data.(adapter.SimpleCounterStat); ok {
336                         prepNodes(len(s[0]))
337                         for i := range nodeStats.Nodes {
338                                 val := adapter.ReduceSimpleCounterStatIndex(s, i)
339                                 fn(&nodeStats.Nodes[i], val)
340                         }
341                 }
342         }
343
344         for _, stat := range c.nodeStatsData.Entries {
345                 switch string(stat.Name) {
346                 case NodeStats_Names:
347                         if stat, ok := stat.Data.(adapter.NameStat); ok {
348                                 prepNodes(len(stat))
349                                 for i, nc := range nodeStats.Nodes {
350                                         if nc.NodeName != string(stat[i]) {
351                                                 nc.NodeName = string(stat[i])
352                                                 nodeStats.Nodes[i] = nc
353                                         }
354                                 }
355                         }
356                 case NodeStats_Clocks:
357                         perNode(stat, func(node *api.NodeCounters, val uint64) {
358                                 node.Clocks = val
359                         })
360                 case NodeStats_Vectors:
361                         perNode(stat, func(node *api.NodeCounters, val uint64) {
362                                 node.Vectors = val
363                         })
364                 case NodeStats_Calls:
365                         perNode(stat, func(node *api.NodeCounters, val uint64) {
366                                 node.Calls = val
367                         })
368                 case NodeStats_Suspends:
369                         perNode(stat, func(node *api.NodeCounters, val uint64) {
370                                 node.Suspends = val
371                         })
372                 }
373         }
374
375         return nil
376 }
377
378 // GetInterfaceStats retrieves VPP per interface stats.
379 func (c *StatsConnection) GetInterfaceStats(ifaceStats *api.InterfaceStats) (err error) {
380         if err := c.updateStats(&c.ifaceStatsData, InterfaceStatsPrefix); err != nil {
381                 return err
382         }
383
384         prep := func(l int) {
385                 if ifaceStats.Interfaces == nil || len(ifaceStats.Interfaces) != l {
386                         ifaceStats.Interfaces = make([]api.InterfaceCounters, l)
387                         for i := 0; i < l; i++ {
388                                 ifaceStats.Interfaces[i].InterfaceIndex = uint32(i)
389                         }
390                 }
391         }
392         perNode := func(stat adapter.StatEntry, fn func(*api.InterfaceCounters, uint64)) {
393                 if s, ok := stat.Data.(adapter.SimpleCounterStat); ok {
394                         prep(len(s[0]))
395                         for i := range ifaceStats.Interfaces {
396                                 val := adapter.ReduceSimpleCounterStatIndex(s, i)
397                                 fn(&ifaceStats.Interfaces[i], val)
398                         }
399                 }
400         }
401         perNodeComb := func(stat adapter.StatEntry, fn func(*api.InterfaceCounters, [2]uint64)) {
402                 if s, ok := stat.Data.(adapter.CombinedCounterStat); ok {
403                         prep(len(s[0]))
404                         for i := range ifaceStats.Interfaces {
405                                 val := adapter.ReduceCombinedCounterStatIndex(s, i)
406                                 fn(&ifaceStats.Interfaces[i], val)
407                         }
408                 }
409         }
410
411         for _, stat := range c.ifaceStatsData.Entries {
412                 switch string(stat.Name) {
413                 case InterfaceStats_Names:
414                         if stat, ok := stat.Data.(adapter.NameStat); ok {
415                                 prep(len(stat))
416                                 for i, nc := range ifaceStats.Interfaces {
417                                         if nc.InterfaceName != string(stat[i]) {
418                                                 nc.InterfaceName = string(stat[i])
419                                                 ifaceStats.Interfaces[i] = nc
420                                         }
421                                 }
422                         }
423                 case InterfaceStats_Drops:
424                         perNode(stat, func(iface *api.InterfaceCounters, val uint64) {
425                                 iface.Drops = val
426                         })
427                 case InterfaceStats_Punt:
428                         perNode(stat, func(iface *api.InterfaceCounters, val uint64) {
429                                 iface.Punts = val
430                         })
431                 case InterfaceStats_IP4:
432                         perNode(stat, func(iface *api.InterfaceCounters, val uint64) {
433                                 iface.IP4 = val
434                         })
435                 case InterfaceStats_IP6:
436                         perNode(stat, func(iface *api.InterfaceCounters, val uint64) {
437                                 iface.IP6 = val
438                         })
439                 case InterfaceStats_RxNoBuf:
440                         perNode(stat, func(iface *api.InterfaceCounters, val uint64) {
441                                 iface.RxNoBuf = val
442                         })
443                 case InterfaceStats_RxMiss:
444                         perNode(stat, func(iface *api.InterfaceCounters, val uint64) {
445                                 iface.RxMiss = val
446                         })
447                 case InterfaceStats_RxError:
448                         perNode(stat, func(iface *api.InterfaceCounters, val uint64) {
449                                 iface.RxErrors = val
450                         })
451                 case InterfaceStats_TxError:
452                         perNode(stat, func(iface *api.InterfaceCounters, val uint64) {
453                                 iface.TxErrors = val
454                         })
455                 case InterfaceStats_Mpls:
456                         perNode(stat, func(iface *api.InterfaceCounters, val uint64) {
457                                 iface.Mpls = val
458                         })
459                 case InterfaceStats_Rx:
460                         perNodeComb(stat, func(iface *api.InterfaceCounters, val [2]uint64) {
461                                 iface.Rx.Packets = val[0]
462                                 iface.Rx.Bytes = val[1]
463                         })
464                 case InterfaceStats_RxUnicast:
465                         perNodeComb(stat, func(iface *api.InterfaceCounters, val [2]uint64) {
466                                 iface.RxUnicast.Packets = val[0]
467                                 iface.RxUnicast.Bytes = val[1]
468                         })
469                 case InterfaceStats_RxMulticast:
470                         perNodeComb(stat, func(iface *api.InterfaceCounters, val [2]uint64) {
471                                 iface.RxMulticast.Packets = val[0]
472                                 iface.RxMulticast.Bytes = val[1]
473                         })
474                 case InterfaceStats_RxBroadcast:
475                         perNodeComb(stat, func(iface *api.InterfaceCounters, val [2]uint64) {
476                                 iface.RxBroadcast.Packets = val[0]
477                                 iface.RxBroadcast.Bytes = val[1]
478                         })
479                 case InterfaceStats_Tx:
480                         perNodeComb(stat, func(iface *api.InterfaceCounters, val [2]uint64) {
481                                 iface.Tx.Packets = val[0]
482                                 iface.Tx.Bytes = val[1]
483                         })
484                 case InterfaceStats_TxUnicastMiss:
485                         // tx-unicast-miss was a spelling mistake in older versions
486                         //
487                         fallthrough
488                 case InterfaceStats_TxUnicast:
489                         perNodeComb(stat, func(iface *api.InterfaceCounters, val [2]uint64) {
490                                 iface.TxUnicast.Packets = val[0]
491                                 iface.TxUnicast.Bytes = val[1]
492                         })
493                 case InterfaceStats_TxMulticast:
494                         perNodeComb(stat, func(iface *api.InterfaceCounters, val [2]uint64) {
495                                 iface.TxMulticast.Packets = val[0]
496                                 iface.TxMulticast.Bytes = val[1]
497                         })
498                 case InterfaceStats_TxBroadcast:
499                         perNodeComb(stat, func(iface *api.InterfaceCounters, val [2]uint64) {
500                                 iface.TxBroadcast.Packets = val[0]
501                                 iface.TxBroadcast.Bytes = val[1]
502                         })
503                 }
504         }
505
506         return nil
507 }
508
509 // GetBufferStats retrieves VPP buffer pools stats.
510 func (c *StatsConnection) GetBufferStats(bufStats *api.BufferStats) (err error) {
511         if err := c.updateStats(&c.bufStatsData, BufferStatsPrefix); err != nil {
512                 return err
513         }
514
515         if bufStats.Buffer == nil {
516                 bufStats.Buffer = make(map[string]api.BufferPool)
517         }
518
519         for _, stat := range c.bufStatsData.Entries {
520                 d, f := path.Split(string(stat.Name))
521                 d = strings.TrimSuffix(d, "/")
522
523                 name := strings.TrimPrefix(d, BufferStatsPrefix)
524                 b, ok := bufStats.Buffer[name]
525                 if !ok {
526                         b.PoolName = name
527                 }
528
529                 var val float64
530                 s, ok := stat.Data.(adapter.ScalarStat)
531                 if ok {
532                         val = float64(s)
533                 }
534                 switch f {
535                 case BufferStats_Cached:
536                         b.Cached = val
537                 case BufferStats_Used:
538                         b.Used = val
539                 case BufferStats_Available:
540                         b.Available = val
541                 }
542
543                 bufStats.Buffer[name] = b
544         }
545
546         return nil
547 }
548
549 func (c *StatsConnection) GetMemoryStats(memStats *api.MemoryStats) (err error) {
550         if err := c.updateStats(&c.memStatsData, MemoryStatSegPrefix, MemoryStatPrefix, MemoryMainPrefix); err != nil {
551                 return err
552         }
553         convertStats := func(stats []adapter.Counter) api.MemoryCounters {
554                 memUsg := make([]adapter.Counter, 7)
555                 copy(memUsg, stats)
556                 return api.MemoryCounters{
557                         Total: uint64(memUsg[0]), Used: uint64(memUsg[1]), Free: uint64(memUsg[2]), UsedMMap: uint64(memUsg[3]),
558                         TotalAlloc: uint64(memUsg[4]), FreeChunks: uint64(memUsg[5]), Releasable: uint64(memUsg[6]),
559                 }
560         }
561
562         for _, stat := range c.memStatsData.Entries {
563                 if strings.Contains(string(stat.Name), MemoryStatSegPrefix) {
564                         _, f := path.Split(string(stat.Name))
565                         var val float64
566                         m, ok := stat.Data.(adapter.ScalarStat)
567                         if ok {
568                                 val = float64(m)
569                         }
570                         switch f {
571                         case MemoryStats_Total:
572                                 memStats.Total = val
573                         case MemoryStats_Used:
574                                 memStats.Used = val
575                         }
576                 } else if strings.Contains(string(stat.Name), MemoryStatPrefix) {
577                         if perHeapStats, ok := stat.Data.(adapter.SimpleCounterStat); ok {
578                                 if memStats.Stat == nil {
579                                         memStats.Stat = make(map[int]api.MemoryCounters)
580                                 }
581                                 for heap, stats := range perHeapStats {
582                                         memStats.Stat[heap] = convertStats(stats)
583                                 }
584                         }
585                 } else if strings.Contains(string(stat.Name), MemoryMainPrefix) {
586                         if perHeapStats, ok := stat.Data.(adapter.SimpleCounterStat); ok {
587                                 if memStats.Main == nil {
588                                         memStats.Main = make(map[int]api.MemoryCounters)
589                                 }
590                                 for heap, stats := range perHeapStats {
591                                         memStats.Main[heap] = convertStats(stats)
592                                 }
593                         }
594                 }
595         }
596         return nil
597 }
598
599 func (c *StatsConnection) sendStatsConnEvent(event ConnectionEvent) {
600         select {
601         case c.connChan <- event:
602         default:
603                 log.Warn("Stats connection state channel is full, discarding value.")
604         }
605 }