"fmt"
"net"
"os"
+ "path/filepath"
"regexp"
+ "sync/atomic"
"syscall"
"time"
"git.fd.io/govpp.git/adapter"
+ "github.com/fsnotify/fsnotify"
"github.com/ftrvxmtrx/fd"
logger "github.com/sirupsen/logrus"
)
const (
// DefaultSocketName is default VPP stats socket file path.
DefaultSocketName = adapter.DefaultStatsSocket
-)
-
-const socketMissing = `
-------------------------------------------------------------
- VPP stats socket file %s is missing!
- - is VPP running with stats segment enabled?
- - is the correct socket name configured?
+ // SocketRetryPeriod is the time period after the socket availability
+ // will be re-checked
+ SocketRetryPeriod = 50 * time.Millisecond
- To enable it add following section to your VPP config:
- statseg {
- socket-name /run/vpp/stats.sock
- }
-------------------------------------------------------------
-`
+ // SocketRetryTimeout is the maximum time for the stats socket
+ SocketRetryTimeout = 3 * time.Second
+)
var (
// Debug is global variable that determines debug mode
// StatsClient is the pure Go implementation for VPP stats API.
type StatsClient struct {
- sockAddr string
- headerData []byte
- isConnected bool
+ socket string
+
+ headerData []byte
+
+ // defines the adapter connection state
+ connected uint32
+
+ // to quit socket monitor
+ done chan struct{}
statSegment
}
-// NewStatsClient returns new VPP stats API client.
-func NewStatsClient(sockAddr string) *StatsClient {
- if sockAddr == "" {
- sockAddr = DefaultSocketName
+// NewStatsClient returns a new StatsClient using socket.
+// If socket is empty string DefaultSocketName is used.
+func NewStatsClient(socket string) *StatsClient {
+ if socket == "" {
+ socket = DefaultSocketName
}
return &StatsClient{
- sockAddr: sockAddr,
+ socket: socket,
}
}
-// Connect to the VPP stats socket
+
+// Connect to validated VPP stats socket and start monitoring
+// socket file changes
func (sc *StatsClient) Connect() (err error) {
- // check if socket exists
- 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 sc.isConnected {
- return fmt.Errorf("already connected")
+ if err := sc.waitForSocket(); err != nil {
+ return err
}
+ sc.done = make(chan struct{})
if sc.statSegment, err = sc.connect(); err != nil {
return err
}
- sc.isConnected = true
+ sc.monitorSocket()
return nil
}
-// Disconnect from the socket and unmap shared memory
+// Disconnect from the socket, unmap shared memory and terminate
+// socket monitor
func (sc *StatsClient) Disconnect() error {
- sc.isConnected = false
if sc.headerData == nil {
return nil
}
}
func (sc *StatsClient) ListStats(patterns ...string) ([]string, error) {
+ if !sc.isConnected() {
+ return nil, adapter.ErrStatsDisconnected
+ }
accessEpoch := sc.accessStart()
if accessEpoch == 0 {
return nil, adapter.ErrStatsAccessFailed
}
func (sc *StatsClient) DumpStats(patterns ...string) (entries []adapter.StatEntry, err error) {
+ if !sc.isConnected() {
+ return nil, adapter.ErrStatsDisconnected
+ }
accessEpoch := sc.accessStart()
if accessEpoch == 0 {
return nil, adapter.ErrStatsAccessFailed
}
func (sc *StatsClient) PrepareDir(patterns ...string) (*adapter.StatDir, error) {
+ if !sc.isConnected() {
+ return nil, adapter.ErrStatsDisconnected
+ }
dir := new(adapter.StatDir)
accessEpoch := sc.accessStart()
// UpdateDir refreshes directory data for all counters
func (sc *StatsClient) UpdateDir(dir *adapter.StatDir) (err error) {
+ if !sc.isConnected() {
+ return adapter.ErrStatsDisconnected
+ }
epoch, _ := sc.GetEpoch()
if dir.Epoch != epoch {
return adapter.ErrStatsDirStale
return nil
}
-func (sc *StatsClient) connect() (statSegment, error) {
+// checks the socket existence and waits for it for the designated
+// time if it is not available immediately
+func (sc *StatsClient) waitForSocket() error {
+ if _, err := os.Stat(sc.socket); err != nil {
+ if os.IsNotExist(err) {
+ ticker := time.NewTicker(SocketRetryPeriod)
+ for {
+ select {
+ case <-ticker.C:
+ if _, err := os.Stat(sc.socket); err == nil {
+ return nil
+ }
+ case <-time.After(SocketRetryTimeout):
+ return fmt.Errorf("stats socket file %s is not ready within timeout ", sc.socket)
+ }
+ }
+ } else {
+ return fmt.Errorf("stats socket error: %v", err)
+ }
+ }
+ return nil
+}
+
+// connect to the socket and map it into the memory. According to the
+// header version info, an appropriate segment handler is returned
+func (sc *StatsClient) connect() (ss statSegment, err error) {
addr := net.UnixAddr{
Net: "unixpacket",
- Name: sc.sockAddr,
+ Name: sc.socket,
}
Log.Debugf("connecting to: %v", addr)
version := getVersion(sc.headerData)
switch version {
case 1:
- return newStatSegmentV1(sc.headerData, size), nil
+ ss = newStatSegmentV1(sc.headerData, size)
case 2:
- return newStatSegmentV2(sc.headerData, size), nil
+ ss = newStatSegmentV2(sc.headerData, size)
default:
return nil, fmt.Errorf("stat segment version is not supported: %v (min: %v, max: %v)",
version, minVersion, maxVersion)
}
+
+ // set connected
+ atomic.CompareAndSwapUint32(&sc.connected, 0, 1)
+
+ return ss, nil
+}
+
+// reconnect disconnects from the socket, re-validates it and
+// connects again
+func (sc *StatsClient) reconnect() (err error) {
+ if err = sc.disconnect(); err != nil {
+ return fmt.Errorf("error disconnecting socket: %v", err)
+ }
+ if err = sc.waitForSocket(); err != nil {
+ return fmt.Errorf("error while waiting on socket: %v", err)
+ }
+ if sc.statSegment, err = sc.connect(); err != nil {
+ return fmt.Errorf("error connecting socket: %v", err)
+ }
+ return nil
+}
+
+// disconnect unmaps socket data from the memory and resets the header
+func (sc *StatsClient) disconnect() error {
+ if !atomic.CompareAndSwapUint32(&sc.connected, 1, 0) {
+ return fmt.Errorf("stats client is already disconnected")
+ }
+ 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 (sc *StatsClient) monitorSocket() {
+ watcher, err := fsnotify.NewWatcher()
+ if err != nil {
+ Log.Errorf("error starting socket monitor: %v", err)
+ return
+ }
+
+ go func() {
+ for {
+ select {
+ case event := <-watcher.Events:
+ if event.Op == fsnotify.Remove && event.Name == sc.socket {
+ if err := sc.reconnect(); err != nil {
+ Log.Errorf("error occurred during socket reconnect: %v", err)
+ }
+ }
+ case err := <-watcher.Errors:
+ Log.Errorf("socket monitor delivered error event: %v", err)
+ case <-sc.done:
+ err := watcher.Close()
+ Log.Debugf("socket monitor closed (error: %v)", err)
+ return
+ }
+ }
+ }()
+
+ if err := watcher.Add(filepath.Dir(sc.socket)); err != nil {
+ Log.Errorf("failed to add socket address to the watcher: %v", err)
+ }
}
// Starts monitoring 'inProgress' field. Returns stats segment
return indexes, nil
}
+
+func (sc *StatsClient) isConnected() bool {
+ return atomic.LoadUint32(&sc.connected) == 1
+}
Code Review / govpp.git / blobdiff