diff options
| -rw-r--r-- | adapter/statsclient/statsclient.go | 125 |
1 files changed, 100 insertions, 25 deletions
diff --git a/adapter/statsclient/statsclient.go b/adapter/statsclient/statsclient.go index a39cbd5..8693410 100644 --- a/adapter/statsclient/statsclient.go +++ b/adapter/statsclient/statsclient.go @@ -25,6 +25,7 @@ import ( "time" "git.fd.io/govpp.git/adapter" + "github.com/fsnotify/fsnotify" "github.com/ftrvxmtrx/fd" logger "github.com/sirupsen/logrus" ) @@ -80,6 +81,9 @@ type StatsClient struct { headerData []byte isConnected bool + // to quit socket monitor + done chan struct{} + statSegment } @@ -92,39 +96,29 @@ func NewStatsClient(sockAddr string) *StatsClient { sockAddr: sockAddr, } } -// 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.validate(); err != nil { + return err + } + sc.done = make(chan struct{}) + sc.monitorSocket() if sc.statSegment, err = sc.connect(); err != nil { return err } - sc.isConnected = true 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 - } - 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 + close(sc.done) + return sc.disconnect() } func (sc *StatsClient) ListStats(patterns ...string) ([]string, error) { @@ -293,7 +287,20 @@ func (sc *StatsClient) UpdateDir(dir *adapter.StatDir) (err error) { return nil } -func (sc *StatsClient) connect() (statSegment, error) { +// validate file presence by retrieving its file info +func (sc *StatsClient) validate() error { + 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) + } + 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, @@ -343,13 +350,81 @@ func (sc *StatsClient) connect() (statSegment, error) { 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) } + sc.isConnected = true + 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.validate(); err != nil { + return fmt.Errorf("error validating 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 { + 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 (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 { + if err := sc.reconnect(); err != nil { + Log.Errorf("error occurred during socket reconnect: %v", err) + } + // path must be re-added to the watcher + if err = watcher.Add(sc.sockAddr); err != nil { + Log.Errorf("failed to add socket address to the watcher: %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(sc.sockAddr); err != nil { + Log.Errorf("failed to add socket address to the watcher: %v", err) + } } // Starts monitoring 'inProgress' field. Returns stats segment |