1 // Copyright (c) 2019 Cisco and/or its affiliates.
3 // Licensed under the Apache License, Version 2.0 (the "License");
4 // you may not use this file except in compliance with the License.
5 // You may obtain a copy of the License at:
7 // http://www.apache.org/licenses/LICENSE-2.0
9 // Unless required by applicable law or agreed to in writing, software
10 // distributed under the License is distributed on an "AS IS" BASIS,
11 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 // See the License for the specific language governing permissions and
13 // limitations under the License.
15 // Package statsclient is pure Go implementation of VPP stats API client.
29 "git.fd.io/govpp.git/adapter"
30 "github.com/fsnotify/fsnotify"
31 "github.com/ftrvxmtrx/fd"
32 logger "github.com/sirupsen/logrus"
36 // DefaultSocketName is default VPP stats socket file path.
37 DefaultSocketName = adapter.DefaultStatsSocket
39 // DefaultSocketRetryPeriod is the time period after the socket availability
41 DefaultSocketRetryPeriod = 50 * time.Millisecond
43 // DefaultSocketRetryTimeout is the maximum time for the stats socket
44 DefaultSocketRetryTimeout = 3 * time.Second
48 // Debug is global variable that determines debug mode
49 Debug = os.Getenv("DEBUG_GOVPP_STATS") != ""
51 // Log is global logger
55 // init initializes global logger, which logs debug level messages to stdout.
59 Log.Level = logger.DebugLevel
60 Log.Debug("govpp/statsclient: enabled debug mode")
64 func debugf(f string, a ...interface{}) {
70 // implements StatsAPI
71 var _ adapter.StatsAPI = (*StatsClient)(nil)
73 // StatsClient is the pure Go implementation for VPP stats API.
74 type StatsClient struct {
76 retryPeriod time.Duration
77 retryTimeout time.Duration
81 // defines the adapter connection state
84 // to quit socket monitor
90 // Option is a StatsClient option
91 type Option func(*StatsClient)
93 // SetSocketRetryPeriod is and optional parameter to define a custom
94 // retry period while waiting for the VPP socket
95 func SetSocketRetryPeriod(t time.Duration) Option {
96 return func(c *StatsClient) {
101 // SetSocketRetryTimeout is and optional parameter to define a custom
102 // timeout while waiting for the VPP socket
103 func SetSocketRetryTimeout(t time.Duration) Option {
104 return func(c *StatsClient) {
109 // NewStatsClient returns a new StatsClient using socket.
110 // If socket is empty string DefaultSocketName is used.
111 func NewStatsClient(socket string, options ...Option) *StatsClient {
113 socket = DefaultSocketName
118 for _, option := range options {
121 if s.retryPeriod == 0 {
122 s.retryPeriod = DefaultSocketRetryPeriod
124 if s.retryTimeout == 0 {
125 s.retryTimeout = DefaultSocketRetryTimeout
130 // Connect to validated VPP stats socket and start monitoring
131 // socket file changes
132 func (sc *StatsClient) Connect() (err error) {
133 if err := sc.waitForSocket(); err != nil {
136 sc.done = make(chan struct{})
137 if sc.statSegment, err = sc.connect(); err != nil {
144 // Disconnect from the socket, unmap shared memory and terminate
146 func (sc *StatsClient) Disconnect() error {
147 if sc.headerData == nil {
150 if err := syscall.Munmap(sc.headerData); err != nil {
151 Log.Debugf("unmapping shared memory failed: %v", err)
152 return fmt.Errorf("unmapping shared memory failed: %v", err)
156 Log.Debugf("successfully unmapped shared memory")
160 func (sc *StatsClient) ListStats(patterns ...string) (entries []adapter.StatIdentifier, err error) {
161 if !sc.isConnected() {
162 return nil, adapter.ErrStatsDisconnected
164 accessEpoch := sc.accessStart()
165 if accessEpoch == 0 {
166 return nil, adapter.ErrStatsAccessFailed
169 entries, err = sc.getIdentifierEntries(patterns...)
174 if !sc.accessEnd(accessEpoch) {
175 return nil, adapter.ErrStatsDataBusy
180 func (sc *StatsClient) DumpStats(patterns ...string) (entries []adapter.StatEntry, err error) {
181 if !sc.isConnected() {
182 return nil, adapter.ErrStatsDisconnected
185 accessEpoch := sc.accessStart()
186 if accessEpoch == 0 {
187 return nil, adapter.ErrStatsAccessFailed
190 entries, err = sc.getStatEntries(patterns...)
195 if !sc.accessEnd(accessEpoch) {
196 return nil, adapter.ErrStatsDataBusy
201 func (sc *StatsClient) PrepareDir(patterns ...string) (*adapter.StatDir, error) {
202 if !sc.isConnected() {
203 return nil, adapter.ErrStatsDisconnected
206 accessEpoch := sc.accessStart()
207 if accessEpoch == 0 {
208 return nil, adapter.ErrStatsAccessFailed
211 entries, err := sc.getStatEntries(patterns...)
216 if !sc.accessEnd(accessEpoch) {
217 return nil, adapter.ErrStatsDataBusy
220 dir := &adapter.StatDir{
228 func (sc *StatsClient) PrepareDirOnIndex(indexes ...uint32) (*adapter.StatDir, error) {
229 if !sc.isConnected() {
230 return nil, adapter.ErrStatsDisconnected
233 accessEpoch := sc.accessStart()
234 if accessEpoch == 0 {
235 return nil, adapter.ErrStatsAccessFailed
237 vector := sc.GetDirectoryVector()
239 return nil, fmt.Errorf("failed to prepare dir on index: directory vector is nil")
241 entries, err := sc.getStatEntriesOnIndex(vector, indexes...)
246 if !sc.accessEnd(accessEpoch) {
247 return nil, adapter.ErrStatsDataBusy
250 dir := &adapter.StatDir{
258 // UpdateDir refreshes directory data for all counters
259 func (sc *StatsClient) UpdateDir(dir *adapter.StatDir) (err error) {
260 if !sc.isConnected() {
261 return adapter.ErrStatsDisconnected
263 epoch, _ := sc.GetEpoch()
264 if dir.Epoch != epoch {
265 return adapter.ErrStatsDirStale
268 accessEpoch := sc.accessStart()
269 if accessEpoch == 0 {
270 return adapter.ErrStatsAccessFailed
272 dirVector := sc.GetDirectoryVector()
273 if dirVector == nil {
276 for i := 0; i < len(dir.Entries); i++ {
277 if err := sc.updateStatOnIndex(&dir.Entries[i], dirVector); err != nil {
281 if !sc.accessEnd(accessEpoch) {
282 return adapter.ErrStatsDataBusy
287 // checks the socket existence and waits for it for the designated
288 // time if it is not available immediately
289 func (sc *StatsClient) waitForSocket() error {
290 if _, err := os.Stat(sc.socket); err != nil {
291 if os.IsNotExist(err) {
293 ticker := time.NewTicker(sc.retryPeriod)
294 timeout := time.After(sc.retryTimeout)
298 if _, err := os.Stat(sc.socket); err == nil {
302 return fmt.Errorf("stats socket file %s is not ready within timeout (after %.2f s) ",
303 sc.socket, time.Since(n).Seconds())
307 return fmt.Errorf("stats socket error: %v", err)
313 // connect to the socket and map it into the memory. According to the
314 // header version info, an appropriate segment handler is returned
315 func (sc *StatsClient) connect() (ss statSegment, err error) {
316 addr := net.UnixAddr{
320 Log.Debugf("connecting to: %v", addr)
322 conn, err := net.DialUnix(addr.Net, nil, &addr)
324 Log.Warnf("connecting to socket %s failed: %s", addr, err)
328 if err := conn.Close(); err != nil {
329 Log.Warnf("closing socket failed: %v", err)
332 Log.Debugf("connected to socket")
334 files, err := fd.Get(conn, 1, nil)
336 return nil, fmt.Errorf("getting file descriptor over socket failed: %v", err)
339 return nil, fmt.Errorf("no files received over socket")
344 if err := file.Close(); err != nil {
345 Log.Warnf("closing file failed: %v", err)
349 info, err := file.Stat()
355 sc.headerData, err = syscall.Mmap(int(file.Fd()), 0, int(size), syscall.PROT_READ, syscall.MAP_SHARED)
357 Log.Debugf("mapping shared memory failed: %v", err)
358 return nil, fmt.Errorf("mapping shared memory failed: %v", err)
360 Log.Debugf("successfully mmapped shared memory segment (size: %v) %v", size, len(sc.headerData))
362 version := getVersion(sc.headerData)
365 ss = newStatSegmentV1(sc.headerData, size)
367 ss = newStatSegmentV2(sc.headerData, size)
369 return nil, fmt.Errorf("stat segment version is not supported: %v (min: %v, max: %v)",
370 version, minVersion, maxVersion)
374 atomic.CompareAndSwapUint32(&sc.connected, 0, 1)
379 // reconnect disconnects from the socket, re-validates it and
381 func (sc *StatsClient) reconnect() (err error) {
382 if err = sc.disconnect(); err != nil {
383 return fmt.Errorf("error disconnecting socket: %v", err)
385 if err = sc.waitForSocket(); err != nil {
386 return fmt.Errorf("error while waiting on socket: %v", err)
388 if sc.statSegment, err = sc.connect(); err != nil {
389 return fmt.Errorf("error connecting socket: %v", err)
394 // disconnect unmaps socket data from the memory and resets the header
395 func (sc *StatsClient) disconnect() error {
396 if !atomic.CompareAndSwapUint32(&sc.connected, 1, 0) {
397 return fmt.Errorf("stats client is already disconnected")
399 if sc.headerData == nil {
402 if err := syscall.Munmap(sc.headerData); err != nil {
403 Log.Debugf("unmapping shared memory failed: %v", err)
404 return fmt.Errorf("unmapping shared memory failed: %v", err)
408 Log.Debugf("successfully unmapped shared memory")
412 func (sc *StatsClient) monitorSocket() {
413 watcher, err := fsnotify.NewWatcher()
415 Log.Errorf("error starting socket monitor: %v", err)
422 case event := <-watcher.Events:
423 if event.Op == fsnotify.Remove && event.Name == sc.socket {
424 if err := sc.reconnect(); err != nil {
425 Log.Errorf("error occurred during socket reconnect: %v", err)
428 case err := <-watcher.Errors:
429 Log.Errorf("socket monitor delivered error event: %v", err)
431 err := watcher.Close()
432 Log.Debugf("socket monitor closed (error: %v)", err)
438 if err := watcher.Add(filepath.Dir(sc.socket)); err != nil {
439 Log.Errorf("failed to add socket address to the watcher: %v", err)
443 // Starts monitoring 'inProgress' field. Returns stats segment
444 // access epoch when completed, or zero value if not finished
445 // within MaxWaitInProgress
446 func (sc *StatsClient) accessStart() (epoch int64) {
449 epoch, inProg := sc.GetEpoch()
451 if time.Since(t) > MaxWaitInProgress {
454 time.Sleep(CheckDelayInProgress)
455 epoch, inProg = sc.GetEpoch()
460 // AccessEnd returns true if stats data reading was finished, false
462 func (sc *StatsClient) accessEnd(accessEpoch int64) bool {
463 epoch, inProgress := sc.GetEpoch()
464 if accessEpoch != epoch || inProgress {
470 // getStatEntries retrieves all stats matching desired patterns, or all stats if no pattern is provided.
471 func (sc *StatsClient) getStatEntries(patterns ...string) (entries []adapter.StatEntry, err error) {
472 vector := sc.GetDirectoryVector()
474 return nil, fmt.Errorf("failed to get stat entries: directory vector is nil")
476 indexes, err := sc.listIndexes(vector, patterns...)
480 return sc.getStatEntriesOnIndex(vector, indexes...)
483 // getIdentifierEntries retrieves all identifiers matching desired patterns, or all identifiers
484 // if no pattern is provided.
485 func (sc *StatsClient) getIdentifierEntries(patterns ...string) (identifiers []adapter.StatIdentifier, err error) {
486 vector := sc.GetDirectoryVector()
488 return nil, fmt.Errorf("failed to get identifier entries: directory vector is nil")
490 indexes, err := sc.listIndexes(vector, patterns...)
494 return sc.getIdentifierEntriesOnIndex(vector, indexes...)
497 // getStatEntriesOnIndex retrieves stats on indexes, or all stats if indexes are not defined.
498 func (sc *StatsClient) getStatEntriesOnIndex(vector dirVector, indexes ...uint32) (entries []adapter.StatEntry, err error) {
499 dirLen := *(*uint32)(vectorLen(vector))
500 for _, index := range indexes {
502 return nil, fmt.Errorf("stat entry index %d out of dir vector length (%d)", index, dirLen)
504 dirPtr, dirName, dirType := sc.GetStatDirOnIndex(vector, index)
505 if len(dirName) == 0 {
508 var t adapter.StatType
509 d := sc.CopyEntryData(dirPtr, ^uint32(0))
513 entries = append(entries, adapter.StatEntry{
514 StatIdentifier: adapter.StatIdentifier{
520 Symlink: adapter.StatType(dirType) == adapter.Symlink,
526 // getIdentifierEntriesOnIndex retrieves identifiers on indexes, or all identifiers if indexes are not defined.
527 func (sc *StatsClient) getIdentifierEntriesOnIndex(vector dirVector, indexes ...uint32) (identifiers []adapter.StatIdentifier, err error) {
528 dirLen := *(*uint32)(vectorLen(vector))
529 for _, index := range indexes {
531 return nil, fmt.Errorf("stat entry index %d out of dir vector length (%d)", index, dirLen)
533 _, dirName, _ := sc.GetStatDirOnIndex(vector, index)
534 if len(dirName) == 0 {
537 identifiers = append(identifiers, adapter.StatIdentifier{
542 return identifiers, nil
545 // listIndexes lists indexes for all stat entries that match any of the regex patterns.
546 func (sc *StatsClient) listIndexes(vector dirVector, patterns ...string) (indexes []uint32, err error) {
547 if len(patterns) == 0 {
548 return sc.listIndexesFunc(vector, nil)
550 var regexes = make([]*regexp.Regexp, len(patterns))
551 for i, pattern := range patterns {
552 r, err := regexp.Compile(pattern)
554 return nil, fmt.Errorf("compiling regexp failed: %v", err)
558 nameMatches := func(name dirName) bool {
559 for _, r := range regexes {
566 return sc.listIndexesFunc(vector, nameMatches)
569 // listIndexesFunc lists stats indexes. The optional function
570 // argument filters returned values or returns all if empty
571 func (sc *StatsClient) listIndexesFunc(vector dirVector, f func(name dirName) bool) (indexes []uint32, err error) {
573 // there is around ~3157 stats, so to avoid too many allocations
574 // we set capacity to 3200 when listing all stats
575 indexes = make([]uint32, 0, 3200)
577 vecLen := *(*uint32)(vectorLen(vector))
578 for i := uint32(0); i < vecLen; i++ {
579 _, dirName, _ := sc.GetStatDirOnIndex(vector, i)
581 if len(dirName) == 0 || !f(dirName) {
585 indexes = append(indexes, i)
591 func (sc *StatsClient) isConnected() bool {
592 return atomic.LoadUint32(&sc.connected) == 1
595 // updateStatOnIndex refreshes the entry data.
596 func (sc *StatsClient) updateStatOnIndex(entry *adapter.StatEntry, vector dirVector) (err error) {
597 dirLen := *(*uint32)(vectorLen(vector))
598 if entry.Index >= dirLen {
599 return fmt.Errorf("stat entry index %d out of dir vector length (%d)", entry.Index, dirLen)
601 dirPtr, dirName, dirType := sc.GetStatDirOnIndex(vector, entry.Index)
602 if len(dirName) == 0 ||
603 !bytes.Equal(dirName, entry.Name) ||
604 adapter.StatType(dirType) != entry.Type ||
608 if err := sc.UpdateEntryData(dirPtr, &entry.Data); err != nil {
609 err = fmt.Errorf("updating stat data for entry %s failed: %v", dirName, err)