X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=adapter%2Fstatsclient%2Fstatsclient.go;h=a39cbd5595cdfdc4af76b562666dfe8c7e1da454;hb=e517439567ad843033257664fdfe90a9173b0aa6;hp=6381b9fd8a603a493bfe47a770482f5354ed8591;hpb=bde85d422c7949ec32fb067e9c36320ccc47fb9e;p=govpp.git

diff --git a/adapter/statsclient/statsclient.go b/adapter/statsclient/statsclient.go
index 6381b9f..a39cbd5 100644
--- a/adapter/statsclient/statsclient.go
+++ b/adapter/statsclient/statsclient.go
@@ -18,13 +18,15 @@ package statsclient
 import (
 	"bytes"
 	"fmt"
+	"net"
 	"os"
 	"regexp"
-	"unsafe"
-
-	logger "github.com/sirupsen/logrus"
+	"syscall"
+	"time"
 
 	"git.fd.io/govpp.git/adapter"
+	"github.com/ftrvxmtrx/fd"
+	logger "github.com/sirupsen/logrus"
 )
 
 const (
@@ -41,7 +43,7 @@ const socketMissing = `
 
  To enable it add following section to your VPP config:
    statseg {
-     default
+     socket-name /run/vpp/stats.sock
    }
 ------------------------------------------------------------
 `
@@ -63,11 +65,21 @@ func init() {
 	}
 }
 
+func debugf(f string, a ...interface{}) {
+	if Debug {
+		Log.Debugf(f, a...)
+	}
+}
+
+// implements StatsAPI
+var _ adapter.StatsAPI = (*StatsClient)(nil)
+
 // StatsClient is the pure Go implementation for VPP stats API.
 type StatsClient struct {
-	sockAddr string
+	sockAddr    string
+	headerData  []byte
+	isConnected bool
 
-	currentEpoch int64
 	statSegment
 }
 
@@ -80,144 +92,341 @@ func NewStatsClient(sockAddr string) *StatsClient {
 		sockAddr: sockAddr,
 	}
 }
-
-func (c *StatsClient) Connect() error {
+// Connect to the VPP stats socket
+func (sc *StatsClient) Connect() (err error) {
 	// check if socket exists
-	if _, err := os.Stat(c.sockAddr); os.IsNotExist(err) {
-		fmt.Fprintf(os.Stderr, socketMissing, c.sockAddr)
-		return fmt.Errorf("stats socket file %s does not exist", c.sockAddr)
+	if _, err := os.Stat(sc.sockAddr); os.IsNotExist(err) {
+		fmt.Fprintf(os.Stderr, socketMissing, sc.sockAddr)
+		return fmt.Errorf("stats socket file %s does not exist", sc.sockAddr)
 	} else if err != nil {
 		return fmt.Errorf("stats socket error: %v", err)
 	}
-
-	if err := c.statSegment.connect(c.sockAddr); err != nil {
+	if sc.isConnected {
+		return fmt.Errorf("already connected")
+	}
+	if sc.statSegment, err = sc.connect(); err != nil {
 		return err
 	}
+	sc.isConnected = true
+	return nil
+}
 
-	ver := c.readVersion()
-	Log.Debugf("stat segment version: %v", ver)
-
-	if err := checkVersion(ver); err != nil {
-		return err
+// Disconnect from the socket and unmap shared memory
+func (sc *StatsClient) Disconnect() error {
+	sc.isConnected = false
+	if sc.headerData == nil {
+		return nil
+	}
+	if err := syscall.Munmap(sc.headerData); err != nil {
+		Log.Debugf("unmapping shared memory failed: %v", err)
+		return fmt.Errorf("unmapping shared memory failed: %v", err)
 	}
+	sc.headerData = nil
 
+	Log.Debugf("successfully unmapped shared memory")
 	return nil
 }
 
-func (c *StatsClient) Disconnect() error {
-	if err := c.statSegment.disconnect(); err != nil {
-		return err
+func (sc *StatsClient) ListStats(patterns ...string) ([]string, error) {
+	accessEpoch := sc.accessStart()
+	if accessEpoch == 0 {
+		return nil, adapter.ErrStatsAccessFailed
 	}
 
-	return nil
+	indexes, err := sc.listIndexes(patterns...)
+	if err != nil {
+		return nil, err
+	}
+
+	dirVector := sc.GetDirectoryVector()
+	if dirVector == nil {
+		return nil, fmt.Errorf("failed to list stats: %v", err)
+	}
+	vecLen := *(*uint32)(vectorLen(dirVector))
+
+	var names []string
+	for _, index := range indexes {
+		if index >= vecLen {
+			return nil, fmt.Errorf("stat entry index %d out of dir vector len (%d)", index, vecLen)
+		}
+		_, dirName, _ := sc.GetStatDirOnIndex(dirVector, index)
+		names = append(names, string(dirName))
+	}
+
+	if !sc.accessEnd(accessEpoch) {
+		return nil, adapter.ErrStatsDataBusy
+	}
+
+	return names, nil
 }
 
-func (c *StatsClient) ListStats(patterns ...string) (statNames []string, err error) {
-	sa := c.accessStart()
-	if sa == nil {
-		return nil, fmt.Errorf("access failed")
+func (sc *StatsClient) DumpStats(patterns ...string) (entries []adapter.StatEntry, err error) {
+	accessEpoch := sc.accessStart()
+	if accessEpoch == 0 {
+		return nil, adapter.ErrStatsAccessFailed
 	}
 
-	dirOffset, _, _ := c.readOffsets()
-	Log.Debugf("dirOffset: %v", dirOffset)
+	indexes, err := sc.listIndexes(patterns...)
+	if err != nil {
+		return nil, err
+	}
 
-	vecLen := vectorLen(unsafe.Pointer(&c.sharedHeader[dirOffset]))
-	Log.Debugf("vecLen: %v", vecLen)
-	Log.Debugf("unsafe.Sizeof(statSegDirectoryEntry{}): %v", unsafe.Sizeof(statSegDirectoryEntry{}))
+	dirVector := sc.GetDirectoryVector()
+	if dirVector == nil {
+		return nil, err
+	}
+	dirLen := *(*uint32)(vectorLen(dirVector))
 
-	for i := uint64(0); i < vecLen; i++ {
-		offset := uintptr(i) * unsafe.Sizeof(statSegDirectoryEntry{})
-		dirEntry := (*statSegDirectoryEntry)(add(unsafe.Pointer(&c.sharedHeader[dirOffset]), offset))
+	debugf("dumping entries for %d indexes", len(indexes))
 
-		nul := bytes.IndexByte(dirEntry.name[:], '\x00')
-		if nul < 0 {
-			Log.Debugf("no zero byte found for: %q", dirEntry.name[:])
-			continue
+	entries = make([]adapter.StatEntry, 0, len(indexes))
+	for _, index := range indexes {
+		if index >= dirLen {
+			return nil, fmt.Errorf("stat entry index %d out of dir vector length (%d)", index, dirLen)
 		}
-		name := string(dirEntry.name[:nul])
-		if name == "" {
-			Log.Debugf("entry with empty name found (%d)", i)
+		dirPtr, dirName, dirType := sc.GetStatDirOnIndex(dirVector, index)
+		if len(dirName) == 0 {
 			continue
 		}
+		entry := adapter.StatEntry{
+			Name: append([]byte(nil), dirName...),
+			Type: adapter.StatType(dirType),
+			Data: sc.CopyEntryData(dirPtr),
+		}
+		entries = append(entries, entry)
+	}
 
-		Log.Debugf(" %80q (type: %v, data: %d, offset: %d) ", name, dirEntry.directoryType, dirEntry.unionData, dirEntry.offsetVector)
+	if !sc.accessEnd(accessEpoch) {
+		return nil, adapter.ErrStatsDataBusy
+	}
 
-		if nameMatches(name, patterns) {
-			statNames = append(statNames, name)
-		}
+	return entries, nil
+}
+
+func (sc *StatsClient) PrepareDir(patterns ...string) (*adapter.StatDir, error) {
+	dir := new(adapter.StatDir)
 
-		// TODO: copy the listed entries elsewhere
+	accessEpoch := sc.accessStart()
+	if accessEpoch == 0 {
+		return nil, adapter.ErrStatsAccessFailed
 	}
 
-	if !c.accessEnd(sa) {
-		return nil, adapter.ErrStatDirBusy
+	indexes, err := sc.listIndexes(patterns...)
+	if err != nil {
+		return nil, err
 	}
+	dir.Indexes = indexes
 
-	c.currentEpoch = sa.epoch
+	dirVector := sc.GetDirectoryVector()
+	if dirVector == nil {
+		return nil, err
+	}
+	dirLen := *(*uint32)(vectorLen(dirVector))
 
-	return statNames, nil
-}
+	debugf("dumping entries for %d indexes", len(indexes))
 
-func (c *StatsClient) DumpStats(patterns ...string) (entries []*adapter.StatEntry, err error) {
-	epoch, _ := c.readEpoch()
-	if c.currentEpoch > 0 && c.currentEpoch != epoch { // TODO: do list stats before dump
-		return nil, fmt.Errorf("old data")
+	entries := make([]adapter.StatEntry, 0, len(indexes))
+	for _, index := range indexes {
+		if index >= dirLen {
+			return nil, fmt.Errorf("stat entry index %d out of dir vector length (%d)", index, dirLen)
+		}
+		dirPtr, dirName, dirType := sc.GetStatDirOnIndex(dirVector, index)
+		if len(dirName) == 0 {
+			continue
+		}
+		entry := adapter.StatEntry{
+			Name: append([]byte(nil), dirName...),
+			Type: adapter.StatType(dirType),
+			Data: sc.CopyEntryData(dirPtr),
+		}
+		entries = append(entries, entry)
 	}
+	dir.Entries = entries
 
-	sa := c.accessStart()
-	if sa == nil {
-		return nil, fmt.Errorf("access failed")
+	if !sc.accessEnd(accessEpoch) {
+		return nil, adapter.ErrStatsDataBusy
 	}
+	dir.Epoch = accessEpoch
+
+	return dir, nil
+}
 
-	dirOffset, _, _ := c.readOffsets()
-	vecLen := vectorLen(unsafe.Pointer(&c.sharedHeader[dirOffset]))
+// UpdateDir refreshes directory data for all counters
+func (sc *StatsClient) UpdateDir(dir *adapter.StatDir) (err error) {
+	epoch, _ := sc.GetEpoch()
+	if dir.Epoch != epoch {
+		return adapter.ErrStatsDirStale
+	}
 
-	for i := uint64(0); i < vecLen; i++ {
-		offset := uintptr(i) * unsafe.Sizeof(statSegDirectoryEntry{})
-		dirEntry := (*statSegDirectoryEntry)(add(unsafe.Pointer(&c.sharedHeader[dirOffset]), offset))
+	accessEpoch := sc.accessStart()
+	if accessEpoch == 0 {
+		return adapter.ErrStatsAccessFailed
+	}
 
-		nul := bytes.IndexByte(dirEntry.name[:], '\x00')
-		if nul < 0 {
-			Log.Debugf("no zero byte found for: %q", dirEntry.name[:])
+	dirVector := sc.GetDirectoryVector()
+	if dirVector == nil {
+		return err
+	}
+	for i, index := range dir.Indexes {
+		statSegDir, dirName, dirType := sc.GetStatDirOnIndex(dirVector, index)
+		if len(dirName) == 0 {
+			continue
+		}
+		entry := &dir.Entries[i]
+		if !bytes.Equal(dirName, entry.Name) {
+			continue
+		}
+		if adapter.StatType(dirType) != entry.Type {
 			continue
 		}
-		name := string(dirEntry.name[:nul])
-		if name == "" {
-			Log.Debugf("entry with empty name found (%d)", i)
+		if entry.Data == nil {
 			continue
 		}
+		if err := sc.UpdateEntryData(statSegDir, &entry.Data); err != nil {
+			return fmt.Errorf("updating stat data for entry %s failed: %v", dirName, err)
+		}
+	}
+	if !sc.accessEnd(accessEpoch) {
+		return adapter.ErrStatsDataBusy
+	}
 
-		Log.Debugf(" - %s (type: %v, data: %v, offset: %v) ", name, dirEntry.directoryType, dirEntry.unionData, dirEntry.offsetVector)
+	return nil
+}
 
-		entry := adapter.StatEntry{
-			Name: name,
-			Type: adapter.StatType(dirEntry.directoryType),
-			Data: c.copyData(dirEntry),
+func (sc *StatsClient) connect() (statSegment, error) {
+	addr := net.UnixAddr{
+		Net:  "unixpacket",
+		Name: sc.sockAddr,
+	}
+	Log.Debugf("connecting to: %v", addr)
+
+	conn, err := net.DialUnix(addr.Net, nil, &addr)
+	if err != nil {
+		Log.Warnf("connecting to socket %s failed: %s", addr, err)
+		return nil, err
+	}
+	defer func() {
+		if err := conn.Close(); err != nil {
+			Log.Warnf("closing socket failed: %v", err)
 		}
+	}()
+	Log.Debugf("connected to socket")
 
-		Log.Debugf("\tentry data: %+v %#v (%T)", entry.Data, entry.Data, entry.Data)
+	files, err := fd.Get(conn, 1, nil)
+	if err != nil {
+		return nil, fmt.Errorf("getting file descriptor over socket failed: %v", err)
+	}
+	if len(files) == 0 {
+		return nil, fmt.Errorf("no files received over socket")
+	}
 
-		if nameMatches(entry.Name, patterns) {
-			entries = append(entries, &entry)
+	file := files[0]
+	defer func() {
+		if err := file.Close(); err != nil {
+			Log.Warnf("closing file failed: %v", err)
 		}
+	}()
+
+	info, err := file.Stat()
+	if err != nil {
+		return nil, err
 	}
+	size := info.Size()
 
-	if !c.accessEnd(sa) {
-		return nil, adapter.ErrStatDumpBusy
+	sc.headerData, err = syscall.Mmap(int(file.Fd()), 0, int(size), syscall.PROT_READ, syscall.MAP_SHARED)
+	if err != nil {
+		Log.Debugf("mapping shared memory failed: %v", err)
+		return nil, fmt.Errorf("mapping shared memory failed: %v", err)
+	}
+	Log.Debugf("successfully mmapped shared memory segment (size: %v) %v", size, len(sc.headerData))
+
+	version := getVersion(sc.headerData)
+	switch version {
+	case 1:
+		return newStatSegmentV1(sc.headerData, size), nil
+	case 2:
+		return newStatSegmentV2(sc.headerData, size), nil
+	default:
+		return nil, fmt.Errorf("stat segment version is not supported: %v (min: %v, max: %v)",
+			version, minVersion, maxVersion)
 	}
+}
 
-	return entries, nil
+// Starts monitoring 'inProgress' field. Returns stats segment
+// access epoch when completed, or zero value if not finished
+// within MaxWaitInProgress
+func (sc *StatsClient) accessStart() (epoch int64) {
+	t := time.Now()
+
+	epoch, inProg := sc.GetEpoch()
+	for inProg {
+		if time.Since(t) > MaxWaitInProgress {
+			return int64(0)
+		}
+		time.Sleep(CheckDelayInProgress)
+		epoch, inProg = sc.GetEpoch()
+	}
+	return epoch
+}
+
+// AccessEnd returns true if stats data reading was finished, false
+// otherwise
+func (sc *StatsClient) accessEnd(accessEpoch int64) bool {
+	epoch, inProgress := sc.GetEpoch()
+	if accessEpoch != epoch || inProgress {
+		return false
+	}
+	return true
 }
 
-func nameMatches(name string, patterns []string) bool {
+// listIndexes lists indexes for all stat entries that match any of the regex patterns.
+func (sc *StatsClient) listIndexes(patterns ...string) (indexes []uint32, err error) {
 	if len(patterns) == 0 {
-		return true
+		return sc.listIndexesFunc(nil)
 	}
-	for _, pattern := range patterns {
-		matched, err := regexp.MatchString(pattern, name)
-		if err == nil && matched {
-			return true
+	var regexes = make([]*regexp.Regexp, len(patterns))
+	for i, pattern := range patterns {
+		r, err := regexp.Compile(pattern)
+		if err != nil {
+			return nil, fmt.Errorf("compiling regexp failed: %v", err)
 		}
+		regexes[i] = r
 	}
-	return false
+	nameMatches := func(name []byte) bool {
+		for _, r := range regexes {
+			if r.Match(name) {
+				return true
+			}
+		}
+		return false
+	}
+	return sc.listIndexesFunc(nameMatches)
+}
+
+// listIndexesFunc lists stats indexes. The optional function
+// argument filters returned values or returns all if empty
+func (sc *StatsClient) listIndexesFunc(f func(name []byte) bool) (indexes []uint32, err error) {
+	if f == nil {
+		// there is around ~3157 stats, so to avoid too many allocations
+		// we set capacity to 3200 when listing all stats
+		indexes = make([]uint32, 0, 3200)
+	}
+
+	dirVector := sc.GetDirectoryVector()
+	if dirVector == nil {
+		return nil, err
+	}
+	vecLen := *(*uint32)(vectorLen(dirVector))
+
+	for i := uint32(0); i < vecLen; i++ {
+		_, dirName, _ := sc.GetStatDirOnIndex(dirVector, i)
+		if f != nil {
+			if len(dirName) == 0 || !f(dirName) {
+				continue
+			}
+		}
+		indexes = append(indexes, i)
+	}
+
+	return indexes, nil
 }