X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=core%2Fstats.go;h=3218f1e760be10ed8fb28d77e7589de9ef2252cd;hb=4e16c7100cc7f8dddca051ff393460d7a1a77c98;hp=4cbd9f25a0ba986298bd533bb990c74069b7538f;hpb=df1b888a2bfadefadc7dbfce59d34f811ff002ec;p=govpp.git

diff --git a/core/stats.go b/core/stats.go
index 4cbd9f2..3218f1e 100644
--- a/core/stats.go
+++ b/core/stats.go
@@ -1,23 +1,29 @@
 package core
 
 import (
-	"fmt"
+	"path"
 	"strings"
-	"sync/atomic"
+	"time"
 
 	"git.fd.io/govpp.git/adapter"
 	"git.fd.io/govpp.git/api"
 )
 
-const (
-	CounterStatsPrefix = "/err/"
+var (
+	RetryUpdateCount    = 10
+	RetryUpdateDelay    = time.Millisecond * 10
+	HealthCheckInterval = time.Second // default health check probe interval
+)
 
-	SystemStatsPrefix          = "/sys/"
-	SystemStats_VectorRate     = SystemStatsPrefix + "vector_rate"
-	SystemStats_InputRate      = SystemStatsPrefix + "input_rate"
-	SystemStats_LastUpdate     = SystemStatsPrefix + "last_update"
-	SystemStats_LastStatsClear = SystemStatsPrefix + "last_stats_clear"
-	SystemStats_Heartbeat      = SystemStatsPrefix + "heartbeat"
+const (
+	SystemStatsPrefix               = "/sys/"
+	SystemStats_VectorRate          = SystemStatsPrefix + "vector_rate"
+	SystemStats_NumWorkerThreads    = SystemStatsPrefix + "num_worker_threads"
+	SystemStats_VectorRatePerWorker = SystemStatsPrefix + "vector_rate_per_worker"
+	SystemStats_InputRate           = SystemStatsPrefix + "input_rate"
+	SystemStats_LastUpdate          = SystemStatsPrefix + "last_update"
+	SystemStats_LastStatsClear      = SystemStatsPrefix + "last_stats_clear"
+	SystemStats_Heartbeat           = SystemStatsPrefix + "heartbeat"
 
 	NodeStatsPrefix    = "/sys/node/"
 	NodeStats_Names    = NodeStatsPrefix + "names"
@@ -26,6 +32,17 @@ const (
 	NodeStats_Calls    = NodeStatsPrefix + "calls"
 	NodeStats_Suspends = NodeStatsPrefix + "suspends"
 
+	BufferStatsPrefix     = "/buffer-pools/"
+	BufferStats_Cached    = "cached"
+	BufferStats_Used      = "used"
+	BufferStats_Available = "available"
+
+	CounterStatsPrefix = "/err/"
+
+	MemoryStatPrefix  = "/mem/statseg"
+	MemoryStats_Total = "total"
+	MemoryStats_Used  = "used"
+
 	InterfaceStatsPrefix         = "/if/"
 	InterfaceStats_Names         = InterfaceStatsPrefix + "names"
 	InterfaceStats_Drops         = InterfaceStatsPrefix + "drops"
@@ -36,61 +53,108 @@ const (
 	InterfaceStats_RxMiss        = InterfaceStatsPrefix + "rx-miss"
 	InterfaceStats_RxError       = InterfaceStatsPrefix + "rx-error"
 	InterfaceStats_TxError       = InterfaceStatsPrefix + "tx-error"
+	InterfaceStats_Mpls          = InterfaceStatsPrefix + "mpls"
 	InterfaceStats_Rx            = InterfaceStatsPrefix + "rx"
 	InterfaceStats_RxUnicast     = InterfaceStatsPrefix + "rx-unicast"
 	InterfaceStats_RxMulticast   = InterfaceStatsPrefix + "rx-multicast"
 	InterfaceStats_RxBroadcast   = InterfaceStatsPrefix + "rx-broadcast"
 	InterfaceStats_Tx            = InterfaceStatsPrefix + "tx"
+	InterfaceStats_TxUnicast     = InterfaceStatsPrefix + "tx-unicast"
 	InterfaceStats_TxUnicastMiss = InterfaceStatsPrefix + "tx-unicast-miss"
 	InterfaceStats_TxMulticast   = InterfaceStatsPrefix + "tx-multicast"
 	InterfaceStats_TxBroadcast   = InterfaceStatsPrefix + "tx-broadcast"
 
-	NetworkStatsPrefix = "/net/"
 	// TODO: network stats
+	NetworkStatsPrefix     = "/net/"
 	NetworkStats_RouteTo   = NetworkStatsPrefix + "route/to"
 	NetworkStats_RouteVia  = NetworkStatsPrefix + "route/via"
 	NetworkStats_MRoute    = NetworkStatsPrefix + "mroute"
 	NetworkStats_Adjacency = NetworkStatsPrefix + "adjacency"
+	NetworkStats_Punt      = NetworkStatsPrefix + "punt"
 )
 
 type StatsConnection struct {
 	statsClient adapter.StatsAPI
 
-	connected uint32 // non-zero if the adapter is connected to VPP
+	maxAttempts int           // interval for reconnect attempts
+	recInterval time.Duration // maximum number of reconnect attempts
+
+	connChan chan ConnectionEvent // connection event channel
+	done     chan struct{}        // to terminate stats connection watcher
+
+	errorStatsData *adapter.StatDir
+	nodeStatsData  *adapter.StatDir
+	ifaceStatsData *adapter.StatDir
+	sysStatsData   *adapter.StatDir
+	bufStatsData   *adapter.StatDir
+	memStatsData   *adapter.StatDir
 }
 
-func newStatsConnection(stats adapter.StatsAPI) *StatsConnection {
+func newStatsConnection(stats adapter.StatsAPI, attempts int, interval time.Duration) *StatsConnection {
+	if attempts == 0 {
+		attempts = DefaultMaxReconnectAttempts
+	}
+	if interval == 0 {
+		interval = DefaultReconnectInterval
+	}
+
 	return &StatsConnection{
 		statsClient: stats,
+		maxAttempts: attempts,
+		recInterval: interval,
+		connChan:    make(chan ConnectionEvent, NotificationChanBufSize),
+		done:        make(chan struct{}),
 	}
 }
 
-// Connect connects to Stats API using specified adapter and returns a connection handle.
+// ConnectStats connects to Stats API using specified adapter and returns a connection handle.
 // This call blocks until it is either connected, or an error occurs.
 // Only one connection attempt will be performed.
 func ConnectStats(stats adapter.StatsAPI) (*StatsConnection, error) {
-	c := newStatsConnection(stats)
+	log.Debug("Connecting to stats..")
+	c := newStatsConnection(stats, DefaultMaxReconnectAttempts, DefaultReconnectInterval)
 
-	if err := c.connectClient(); err != nil {
+	if err := c.statsClient.Connect(); err != nil {
 		return nil, err
 	}
+	log.Debugf("Connected to stats.")
 
 	return c, nil
 }
 
-func (c *StatsConnection) connectClient() error {
-	log.Debug("Connecting to stats..")
+// AsyncConnectStats  connects to the VPP stats socket asynchronously and returns the connection
+// handle with state channel. The call is non-blocking and the caller is expected to watch ConnectionEvent
+// values from the channel and wait for connect/disconnect events. Connection loop tries to reconnect the
+// socket in case the session was disconnected.
+func AsyncConnectStats(stats adapter.StatsAPI, attempts int, interval time.Duration) (*StatsConnection, chan ConnectionEvent, error) {
+	log.Debug("Connecting to stats asynchronously..")
+	c := newStatsConnection(stats, attempts, interval)
 
-	if err := c.statsClient.Connect(); err != nil {
-		return err
-	}
+	go c.connectLoop()
 
-	log.Debugf("Connected to stats.")
-
-	// store connected state
-	atomic.StoreUint32(&c.connected, 1)
+	return c, c.connChan, nil
+}
 
-	return nil
+func (c *StatsConnection) connectLoop() {
+	log.Debug("Asynchronously connecting to stats..")
+	var reconnectAttempts int
+
+	// loop until connected
+	for {
+		if err := c.statsClient.Connect(); err == nil {
+			c.sendStatsConnEvent(ConnectionEvent{Timestamp: time.Now(), State: Connected})
+			break
+		} else if reconnectAttempts < c.maxAttempts {
+			reconnectAttempts++
+			log.Warnf("connecting stats failed (attempt %d/%d): %v", reconnectAttempts, c.maxAttempts, err)
+			time.Sleep(c.recInterval)
+		} else {
+			c.sendStatsConnEvent(ConnectionEvent{Timestamp: time.Now(), State: Failed, Error: err})
+			return
+		}
+	}
+	// start monitoring stats connection state
+	go c.monitorSocket()
 }
 
 // Disconnect disconnects from Stats API and releases all connection-related resources.
@@ -98,333 +162,414 @@ func (c *StatsConnection) Disconnect() {
 	if c == nil {
 		return
 	}
-
 	if c.statsClient != nil {
-		c.disconnectClient()
+		if err := c.statsClient.Disconnect(); err != nil {
+			log.Debugf("disconnecting stats client failed: %v", err)
+		}
 	}
+	close(c.connChan)
+	close(c.done)
 }
 
-func (c *StatsConnection) disconnectClient() {
-	if atomic.CompareAndSwapUint32(&c.connected, 1, 0) {
-		c.statsClient.Disconnect()
+func (c *StatsConnection) monitorSocket() {
+	var state, lastState ConnectionState
+	ticker := time.NewTicker(HealthCheckInterval)
+
+	for {
+		select {
+		case <-ticker.C:
+			_, err := c.statsClient.ListStats(SystemStats_Heartbeat)
+			state = Connected
+			if err == adapter.ErrStatsDataBusy {
+				state = NotResponding
+			}
+			if err == adapter.ErrStatsDisconnected {
+				state = Disconnected
+			}
+			if err == adapter.ErrStatsAccessFailed {
+				state = Failed
+			}
+			if state == lastState {
+				continue
+			}
+			lastState = state
+			c.sendStatsConnEvent(ConnectionEvent{Timestamp: time.Now(), State: state, Error: err})
+		case <-c.done:
+			log.Debugf("health check watcher closed")
+			c.sendStatsConnEvent(ConnectionEvent{Timestamp: time.Now(), State: Disconnected, Error: nil})
+			break
+		}
 	}
 }
 
-// GetSystemStats retrieves VPP system stats.
-func (c *StatsConnection) GetSystemStats() (*api.SystemStats, error) {
-	stats, err := c.statsClient.DumpStats(SystemStatsPrefix)
-	if err != nil {
-		return nil, err
+func (c *StatsConnection) updateStats(statDir **adapter.StatDir, patterns ...string) error {
+	if statDir == nil {
+		panic("statDir must not nil")
+	}
+	try := func() error {
+		if (*statDir) == nil {
+			dir, err := c.statsClient.PrepareDir(patterns...)
+			if err != nil {
+				log.Debugln("preparing dir failed:", err)
+				return err
+			}
+			*statDir = dir
+		} else {
+			if err := c.statsClient.UpdateDir(*statDir); err != nil {
+				log.Debugln("updating dir failed:", err)
+				*statDir = nil
+				return err
+			}
+		}
+
+		return nil
 	}
+	var err error
+	for r := 0; r < RetryUpdateCount; r++ {
+		if err = try(); err == nil {
+			if r > 0 {
+				log.Debugf("retry successfull (r=%d)", r)
+			}
+			return nil
+		} else if err == adapter.ErrStatsDirStale || err == adapter.ErrStatsDataBusy {
+			// retrying
+			if r > 1 {
+				log.Debugf("sleeping for %v before next try", RetryUpdateDelay)
+				time.Sleep(RetryUpdateDelay)
+			}
+		} else {
+			// error is not retryable
+			break
+		}
+	}
+	return err
+}
 
-	sysStats := &api.SystemStats{}
+// GetSystemStats retrieves VPP system stats.
+func (c *StatsConnection) GetSystemStats(sysStats *api.SystemStats) (err error) {
+	if err := c.updateStats(&c.sysStatsData, SystemStatsPrefix); err != nil {
+		return err
+	}
 
-	for _, stat := range stats {
-		switch stat.Name {
+	for _, stat := range c.sysStatsData.Entries {
+		var val uint64
+		if s, ok := stat.Data.(adapter.ScalarStat); ok {
+			val = uint64(s)
+		}
+		switch string(stat.Name) {
 		case SystemStats_VectorRate:
-			sysStats.VectorRate = scalarStatToFloat64(stat.Data)
+			sysStats.VectorRate = val
+		case SystemStats_NumWorkerThreads:
+			sysStats.NumWorkerThreads = val
+		case SystemStats_VectorRatePerWorker:
+			var vals []uint64
+			if ss, ok := stat.Data.(adapter.SimpleCounterStat); ok {
+				vals = make([]uint64, len(ss))
+				for w := range ss {
+					if ss[w] == nil {
+						continue
+					}
+					vals[w] = uint64(ss[w][0])
+				}
+			}
+			sysStats.VectorRatePerWorker = vals
 		case SystemStats_InputRate:
-			sysStats.InputRate = scalarStatToFloat64(stat.Data)
+			sysStats.InputRate = val
 		case SystemStats_LastUpdate:
-			sysStats.LastUpdate = scalarStatToFloat64(stat.Data)
+			sysStats.LastUpdate = val
 		case SystemStats_LastStatsClear:
-			sysStats.LastStatsClear = scalarStatToFloat64(stat.Data)
+			sysStats.LastStatsClear = val
 		case SystemStats_Heartbeat:
-			sysStats.Heartbeat = scalarStatToFloat64(stat.Data)
+			sysStats.Heartbeat = val
 		}
 	}
 
-	return sysStats, nil
+	return nil
 }
 
 // GetErrorStats retrieves VPP error stats.
-func (c *StatsConnection) GetErrorStats(names ...string) (*api.ErrorStats, error) {
-	var patterns []string
-	if len(names) > 0 {
-		patterns = make([]string, len(names))
-		for i, name := range names {
-			patterns[i] = CounterStatsPrefix + name
-		}
-	} else {
-		// retrieve all error counters by default
-		patterns = []string{CounterStatsPrefix}
+func (c *StatsConnection) GetErrorStats(errorStats *api.ErrorStats) (err error) {
+	if err := c.updateStats(&c.errorStatsData, CounterStatsPrefix); err != nil {
+		return err
 	}
-	stats, err := c.statsClient.DumpStats(patterns...)
-	if err != nil {
-		return nil, err
+
+	if errorStats.Errors == nil || len(errorStats.Errors) != len(c.errorStatsData.Entries) {
+		errorStats.Errors = make([]api.ErrorCounter, len(c.errorStatsData.Entries))
+		for i := 0; i < len(c.errorStatsData.Entries); i++ {
+			errorStats.Errors[i].CounterName = string(c.errorStatsData.Entries[i].Name)
+		}
 	}
 
-	var errorStats = &api.ErrorStats{}
-
-	for _, stat := range stats {
-		statName := strings.TrimPrefix(stat.Name, CounterStatsPrefix)
-
-		/* TODO: deal with stats that contain '/' in node/counter name
-		parts := strings.Split(statName, "/")
-		var nodeName, counterName string
-		switch len(parts) {
-		case 2:
-			nodeName = parts[0]
-			counterName = parts[1]
-		case 3:
-			nodeName = parts[0] + parts[1]
-			counterName = parts[2]
-		}*/
-
-		errorStats.Errors = append(errorStats.Errors, api.ErrorCounter{
-			CounterName: statName,
-			Value:       errorStatToUint64(stat.Data),
-		})
+	for i, stat := range c.errorStatsData.Entries {
+		if stat.Type != adapter.ErrorIndex {
+			continue
+		}
+		if errStat, ok := stat.Data.(adapter.ErrorStat); ok {
+			values := make([]uint64, len(errStat))
+			for j, errStatW := range errStat {
+				values[j] = uint64(errStatW)
+			}
+			errorStats.Errors[i].Values = values
+		}
 	}
 
-	return errorStats, nil
+	return nil
 }
 
-// GetNodeStats retrieves VPP per node stats.
-func (c *StatsConnection) GetNodeStats() (*api.NodeStats, error) {
-	stats, err := c.statsClient.DumpStats(NodeStatsPrefix)
-	if err != nil {
-		return nil, err
+func (c *StatsConnection) GetNodeStats(nodeStats *api.NodeStats) (err error) {
+	if err := c.updateStats(&c.nodeStatsData, NodeStatsPrefix); err != nil {
+		return err
 	}
 
-	nodeStats := &api.NodeStats{}
-	var setPerNode = func(perNode []uint64, fn func(c *api.NodeCounters, v uint64)) {
-		if nodeStats.Nodes == nil {
-			nodeStats.Nodes = make([]api.NodeCounters, len(perNode))
-			for i := range perNode {
+	prepNodes := func(l int) {
+		if nodeStats.Nodes == nil || len(nodeStats.Nodes) != l {
+			nodeStats.Nodes = make([]api.NodeCounters, l)
+			for i := 0; i < l; i++ {
 				nodeStats.Nodes[i].NodeIndex = uint32(i)
 			}
 		}
-		for i, v := range perNode {
-			nodeCounters := nodeStats.Nodes[i]
-			fn(&nodeCounters, v)
-			nodeStats.Nodes[i] = nodeCounters
+	}
+	perNode := func(stat adapter.StatEntry, fn func(*api.NodeCounters, uint64)) {
+		if s, ok := stat.Data.(adapter.SimpleCounterStat); ok {
+			prepNodes(len(s[0]))
+			for i := range nodeStats.Nodes {
+				val := adapter.ReduceSimpleCounterStatIndex(s, i)
+				fn(&nodeStats.Nodes[i], val)
+			}
 		}
 	}
 
-	for _, stat := range stats {
-		switch stat.Name {
+	for _, stat := range c.nodeStatsData.Entries {
+		switch string(stat.Name) {
 		case NodeStats_Names:
-			if names, ok := stat.Data.(adapter.NameStat); !ok {
-				return nil, fmt.Errorf("invalid stat type for %s", stat.Name)
-			} else {
-				if nodeStats.Nodes == nil {
-					nodeStats.Nodes = make([]api.NodeCounters, len(names))
-					for i := range names {
-						nodeStats.Nodes[i].NodeIndex = uint32(i)
+			if stat, ok := stat.Data.(adapter.NameStat); ok {
+				prepNodes(len(stat))
+				for i, nc := range nodeStats.Nodes {
+					if nc.NodeName != string(stat[i]) {
+						nc.NodeName = string(stat[i])
+						nodeStats.Nodes[i] = nc
 					}
 				}
-				for i, name := range names {
-					nodeStats.Nodes[i].NodeName = string(name)
-				}
 			}
 		case NodeStats_Clocks:
-			setPerNode(reduceSimpleCounterStat(stat.Data), func(c *api.NodeCounters, v uint64) {
-				c.Clocks = v
+			perNode(stat, func(node *api.NodeCounters, val uint64) {
+				node.Clocks = val
 			})
 		case NodeStats_Vectors:
-			setPerNode(reduceSimpleCounterStat(stat.Data), func(c *api.NodeCounters, v uint64) {
-				c.Vectors = v
+			perNode(stat, func(node *api.NodeCounters, val uint64) {
+				node.Vectors = val
 			})
 		case NodeStats_Calls:
-			setPerNode(reduceSimpleCounterStat(stat.Data), func(c *api.NodeCounters, v uint64) {
-				c.Calls = v
+			perNode(stat, func(node *api.NodeCounters, val uint64) {
+				node.Calls = val
 			})
 		case NodeStats_Suspends:
-			setPerNode(reduceSimpleCounterStat(stat.Data), func(c *api.NodeCounters, v uint64) {
-				c.Suspends = v
+			perNode(stat, func(node *api.NodeCounters, val uint64) {
+				node.Suspends = val
 			})
 		}
 	}
 
-	return nodeStats, nil
+	return nil
 }
 
 // GetInterfaceStats retrieves VPP per interface stats.
-func (c *StatsConnection) GetInterfaceStats() (*api.InterfaceStats, error) {
-	stats, err := c.statsClient.DumpStats(InterfaceStatsPrefix)
-	if err != nil {
-		return nil, err
+func (c *StatsConnection) GetInterfaceStats(ifaceStats *api.InterfaceStats) (err error) {
+	if err := c.updateStats(&c.ifaceStatsData, InterfaceStatsPrefix); err != nil {
+		return err
 	}
 
-	ifStats := &api.InterfaceStats{}
-
-	var setPerIf = func(perIf []uint64, fn func(c *api.InterfaceCounters, v uint64)) {
-		if ifStats.Interfaces == nil {
-			ifStats.Interfaces = make([]api.InterfaceCounters, len(perIf))
-			for i := range perIf {
-				ifStats.Interfaces[i].InterfaceIndex = uint32(i)
+	prep := func(l int) {
+		if ifaceStats.Interfaces == nil || len(ifaceStats.Interfaces) != l {
+			ifaceStats.Interfaces = make([]api.InterfaceCounters, l)
+			for i := 0; i < l; i++ {
+				ifaceStats.Interfaces[i].InterfaceIndex = uint32(i)
 			}
 		}
-		for i, v := range perIf {
-			ifCounters := ifStats.Interfaces[i]
-			fn(&ifCounters, v)
-			ifStats.Interfaces[i] = ifCounters
+	}
+	perNode := func(stat adapter.StatEntry, fn func(*api.InterfaceCounters, uint64)) {
+		if s, ok := stat.Data.(adapter.SimpleCounterStat); ok {
+			prep(len(s[0]))
+			for i := range ifaceStats.Interfaces {
+				val := adapter.ReduceSimpleCounterStatIndex(s, i)
+				fn(&ifaceStats.Interfaces[i], val)
+			}
+		}
+	}
+	perNodeComb := func(stat adapter.StatEntry, fn func(*api.InterfaceCounters, [2]uint64)) {
+		if s, ok := stat.Data.(adapter.CombinedCounterStat); ok {
+			prep(len(s[0]))
+			for i := range ifaceStats.Interfaces {
+				val := adapter.ReduceCombinedCounterStatIndex(s, i)
+				fn(&ifaceStats.Interfaces[i], val)
+			}
 		}
 	}
 
-	for _, stat := range stats {
-		switch stat.Name {
+	for _, stat := range c.ifaceStatsData.Entries {
+		switch string(stat.Name) {
 		case InterfaceStats_Names:
-			if names, ok := stat.Data.(adapter.NameStat); !ok {
-				return nil, fmt.Errorf("invalid stat type for %s", stat.Name)
-			} else {
-				if ifStats.Interfaces == nil {
-					ifStats.Interfaces = make([]api.InterfaceCounters, len(names))
-					for i := range names {
-						ifStats.Interfaces[i].InterfaceIndex = uint32(i)
+			if stat, ok := stat.Data.(adapter.NameStat); ok {
+				prep(len(stat))
+				for i, nc := range ifaceStats.Interfaces {
+					if nc.InterfaceName != string(stat[i]) {
+						nc.InterfaceName = string(stat[i])
+						ifaceStats.Interfaces[i] = nc
 					}
 				}
-				for i, name := range names {
-					ifStats.Interfaces[i].InterfaceName = string(name)
-				}
 			}
 		case InterfaceStats_Drops:
-			setPerIf(reduceSimpleCounterStat(stat.Data), func(c *api.InterfaceCounters, v uint64) {
-				c.Drops = v
+			perNode(stat, func(iface *api.InterfaceCounters, val uint64) {
+				iface.Drops = val
 			})
 		case InterfaceStats_Punt:
-			setPerIf(reduceSimpleCounterStat(stat.Data), func(c *api.InterfaceCounters, v uint64) {
-				c.Punts = v
+			perNode(stat, func(iface *api.InterfaceCounters, val uint64) {
+				iface.Punts = val
 			})
 		case InterfaceStats_IP4:
-			setPerIf(reduceSimpleCounterStat(stat.Data), func(c *api.InterfaceCounters, v uint64) {
-				c.IP4 = v
+			perNode(stat, func(iface *api.InterfaceCounters, val uint64) {
+				iface.IP4 = val
 			})
 		case InterfaceStats_IP6:
-			setPerIf(reduceSimpleCounterStat(stat.Data), func(c *api.InterfaceCounters, v uint64) {
-				c.IP6 = v
+			perNode(stat, func(iface *api.InterfaceCounters, val uint64) {
+				iface.IP6 = val
 			})
 		case InterfaceStats_RxNoBuf:
-			setPerIf(reduceSimpleCounterStat(stat.Data), func(c *api.InterfaceCounters, v uint64) {
-				c.RxNoBuf = v
+			perNode(stat, func(iface *api.InterfaceCounters, val uint64) {
+				iface.RxNoBuf = val
 			})
 		case InterfaceStats_RxMiss:
-			setPerIf(reduceSimpleCounterStat(stat.Data), func(c *api.InterfaceCounters, v uint64) {
-				c.RxMiss = v
+			perNode(stat, func(iface *api.InterfaceCounters, val uint64) {
+				iface.RxMiss = val
 			})
 		case InterfaceStats_RxError:
-			setPerIf(reduceSimpleCounterStat(stat.Data), func(c *api.InterfaceCounters, v uint64) {
-				c.RxErrors = v
+			perNode(stat, func(iface *api.InterfaceCounters, val uint64) {
+				iface.RxErrors = val
 			})
 		case InterfaceStats_TxError:
-			setPerIf(reduceSimpleCounterStat(stat.Data), func(c *api.InterfaceCounters, v uint64) {
-				c.TxErrors = v
+			perNode(stat, func(iface *api.InterfaceCounters, val uint64) {
+				iface.TxErrors = val
 			})
-		case InterfaceStats_Rx:
-			per := reduceCombinedCounterStat(stat.Data)
-			setPerIf(per[0], func(c *api.InterfaceCounters, v uint64) {
-				c.RxPackets = v
+		case InterfaceStats_Mpls:
+			perNode(stat, func(iface *api.InterfaceCounters, val uint64) {
+				iface.Mpls = val
 			})
-			setPerIf(per[1], func(c *api.InterfaceCounters, v uint64) {
-				c.RxBytes = v
+		case InterfaceStats_Rx:
+			perNodeComb(stat, func(iface *api.InterfaceCounters, val [2]uint64) {
+				iface.Rx.Packets = val[0]
+				iface.Rx.Bytes = val[1]
 			})
 		case InterfaceStats_RxUnicast:
-			per := reduceCombinedCounterStat(stat.Data)
-			setPerIf(per[0], func(c *api.InterfaceCounters, v uint64) {
-				c.RxUnicast[0] = v
-			})
-			setPerIf(per[1], func(c *api.InterfaceCounters, v uint64) {
-				c.RxUnicast[1] = v
+			perNodeComb(stat, func(iface *api.InterfaceCounters, val [2]uint64) {
+				iface.RxUnicast.Packets = val[0]
+				iface.RxUnicast.Bytes = val[1]
 			})
 		case InterfaceStats_RxMulticast:
-			per := reduceCombinedCounterStat(stat.Data)
-			setPerIf(per[0], func(c *api.InterfaceCounters, v uint64) {
-				c.RxMulticast[0] = v
-			})
-			setPerIf(per[1], func(c *api.InterfaceCounters, v uint64) {
-				c.RxMulticast[1] = v
+			perNodeComb(stat, func(iface *api.InterfaceCounters, val [2]uint64) {
+				iface.RxMulticast.Packets = val[0]
+				iface.RxMulticast.Bytes = val[1]
 			})
 		case InterfaceStats_RxBroadcast:
-			per := reduceCombinedCounterStat(stat.Data)
-			setPerIf(per[0], func(c *api.InterfaceCounters, v uint64) {
-				c.RxBroadcast[0] = v
-			})
-			setPerIf(per[1], func(c *api.InterfaceCounters, v uint64) {
-				c.RxBroadcast[1] = v
+			perNodeComb(stat, func(iface *api.InterfaceCounters, val [2]uint64) {
+				iface.RxBroadcast.Packets = val[0]
+				iface.RxBroadcast.Bytes = val[1]
 			})
 		case InterfaceStats_Tx:
-			per := reduceCombinedCounterStat(stat.Data)
-			setPerIf(per[0], func(c *api.InterfaceCounters, v uint64) {
-				c.TxPackets = v
-			})
-			setPerIf(per[1], func(c *api.InterfaceCounters, v uint64) {
-				c.TxBytes = v
+			perNodeComb(stat, func(iface *api.InterfaceCounters, val [2]uint64) {
+				iface.Tx.Packets = val[0]
+				iface.Tx.Bytes = val[1]
 			})
 		case InterfaceStats_TxUnicastMiss:
-			per := reduceCombinedCounterStat(stat.Data)
-			setPerIf(per[0], func(c *api.InterfaceCounters, v uint64) {
-				c.TxUnicastMiss[0] = v
-			})
-			setPerIf(per[1], func(c *api.InterfaceCounters, v uint64) {
-				c.TxUnicastMiss[1] = v
+			// tx-unicast-miss was a spelling mistake in older versions
+			//
+			fallthrough
+		case InterfaceStats_TxUnicast:
+			perNodeComb(stat, func(iface *api.InterfaceCounters, val [2]uint64) {
+				iface.TxUnicast.Packets = val[0]
+				iface.TxUnicast.Bytes = val[1]
 			})
 		case InterfaceStats_TxMulticast:
-			per := reduceCombinedCounterStat(stat.Data)
-			setPerIf(per[0], func(c *api.InterfaceCounters, v uint64) {
-				c.TxMulticast[0] = v
-			})
-			setPerIf(per[1], func(c *api.InterfaceCounters, v uint64) {
-				c.TxMulticast[1] = v
+			perNodeComb(stat, func(iface *api.InterfaceCounters, val [2]uint64) {
+				iface.TxMulticast.Packets = val[0]
+				iface.TxMulticast.Bytes = val[1]
 			})
 		case InterfaceStats_TxBroadcast:
-			per := reduceCombinedCounterStat(stat.Data)
-			setPerIf(per[0], func(c *api.InterfaceCounters, v uint64) {
-				c.TxBroadcast[0] = v
-			})
-			setPerIf(per[1], func(c *api.InterfaceCounters, v uint64) {
-				c.TxBroadcast[1] = v
+			perNodeComb(stat, func(iface *api.InterfaceCounters, val [2]uint64) {
+				iface.TxBroadcast.Packets = val[0]
+				iface.TxBroadcast.Bytes = val[1]
 			})
 		}
 	}
 
-	return ifStats, nil
+	return nil
 }
 
-func scalarStatToFloat64(stat adapter.Stat) float64 {
-	if s, ok := stat.(adapter.ScalarStat); ok {
-		return float64(s)
+// GetBufferStats retrieves VPP buffer pools stats.
+func (c *StatsConnection) GetBufferStats(bufStats *api.BufferStats) (err error) {
+	if err := c.updateStats(&c.bufStatsData, BufferStatsPrefix); err != nil {
+		return err
 	}
-	return 0
-}
 
-func errorStatToUint64(stat adapter.Stat) uint64 {
-	if s, ok := stat.(adapter.ErrorStat); ok {
-		return uint64(s)
+	if bufStats.Buffer == nil {
+		bufStats.Buffer = make(map[string]api.BufferPool)
 	}
-	return 0
-}
 
-func reduceSimpleCounterStat(stat adapter.Stat) []uint64 {
-	if s, ok := stat.(adapter.SimpleCounterStat); ok {
-		if len(s) == 0 {
-			return []uint64{}
+	for _, stat := range c.bufStatsData.Entries {
+		d, f := path.Split(string(stat.Name))
+		d = strings.TrimSuffix(d, "/")
+
+		name := strings.TrimPrefix(d, BufferStatsPrefix)
+		b, ok := bufStats.Buffer[name]
+		if !ok {
+			b.PoolName = name
 		}
-		var per = make([]uint64, len(s[0]))
-		for _, w := range s {
-			for i, n := range w {
-				per[i] += uint64(n)
-			}
+
+		var val float64
+		s, ok := stat.Data.(adapter.ScalarStat)
+		if ok {
+			val = float64(s)
 		}
-		return per
+		switch f {
+		case BufferStats_Cached:
+			b.Cached = val
+		case BufferStats_Used:
+			b.Used = val
+		case BufferStats_Available:
+			b.Available = val
+		}
+
+		bufStats.Buffer[name] = b
 	}
+
 	return nil
 }
 
-func reduceCombinedCounterStat(stat adapter.Stat) [2][]uint64 {
-	if s, ok := stat.(adapter.CombinedCounterStat); ok {
-		if len(s) == 0 {
-			return [2][]uint64{{}, {}}
+func (c *StatsConnection) GetMemoryStats(memStats *api.MemoryStats) (err error) {
+	if err := c.updateStats(&c.memStatsData, MemoryStatPrefix); err != nil {
+		return err
+	}
+
+	for _, stat := range c.memStatsData.Entries {
+		_, f := path.Split(string(stat.Name))
+		var val float64
+		m, ok := stat.Data.(adapter.ScalarStat)
+		if ok {
+			val = float64(m)
 		}
-		var perPackets = make([]uint64, len(s[0]))
-		var perBytes = make([]uint64, len(s[0]))
-		for _, w := range s {
-			for i, n := range w {
-				perPackets[i] += uint64(n.Packets)
-				perBytes[i] += uint64(n.Bytes)
-			}
+		switch f {
+		case MemoryStats_Total:
+			memStats.Total = val
+		case MemoryStats_Used:
+			memStats.Used = val
 		}
-		return [2][]uint64{perPackets, perBytes}
 	}
-	return [2][]uint64{}
+	return nil
+}
+
+func (c *StatsConnection) sendStatsConnEvent(event ConnectionEvent) {
+	select {
+	case c.connChan <- event:
+	default:
+		log.Warn("Stats connection state channel is full, discarding value.")
+	}
 }