adapter/statsclient/statsclient.go

   1 // Copyright (c) 2019 Cisco and/or its affiliates.
   2 //
   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:
   6 //
   7 //     http://www.apache.org/licenses/LICENSE-2.0
   8 //
   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.
  14
  15 // Package statsclient is pure Go implementation of VPP stats API client.
  16 package statsclient
  17
  18 import (
  19         "bytes"
  20         "fmt"
  21         "net"
  22         "os"
  23         "path/filepath"
  24         "regexp"
  25         "sync/atomic"
  26         "syscall"
  27         "time"
  28
  29         "git.fd.io/govpp.git/adapter"
  30         "github.com/fsnotify/fsnotify"
  31         "github.com/ftrvxmtrx/fd"
  32         logger "github.com/sirupsen/logrus"
  33 )
  34
  35 const (
  36         // DefaultSocketName is default VPP stats socket file path.
  37         DefaultSocketName = adapter.DefaultStatsSocket
  38
  39         // DefaultSocketRetryPeriod is the time period after the socket availability
  40         // will be re-checked
  41         DefaultSocketRetryPeriod = 50 * time.Millisecond
  42
  43         // DefaultSocketRetryTimeout is the maximum time for the stats socket
  44         DefaultSocketRetryTimeout = 3 * time.Second
  45 )
  46
  47 var (
  48         // Debug is global variable that determines debug mode
  49         Debug = os.Getenv("DEBUG_GOVPP_STATS") != ""
  50
  51         // Log is global logger
  52         Log = logger.New()
  53 )
  54
  55 // init initializes global logger, which logs debug level messages to stdout.
  56 func init() {
  57         Log.Out = os.Stdout
  58         if Debug {
  59                 Log.Level = logger.DebugLevel
  60                 Log.Debug("govpp/statsclient: enabled debug mode")
  61         }
  62 }
  63
  64 func debugf(f string, a ...interface{}) {
  65         if Debug {
  66                 Log.Debugf(f, a...)
  67         }
  68 }
  69
  70 // implements StatsAPI
  71 var _ adapter.StatsAPI = (*StatsClient)(nil)
  72
  73 // StatsClient is the pure Go implementation for VPP stats API.
  74 type StatsClient struct {
  75         socket       string
  76         retryPeriod  time.Duration
  77         retryTimeout time.Duration
  78
  79         headerData []byte
  80
  81         // defines the adapter connection state
  82         connected uint32
  83
  84         // to quit socket monitor
  85         done chan struct{}
  86
  87         statSegment
  88 }
  89
  90 // Option is a StatsClient option
  91 type Option func(*StatsClient)
  92
  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) {
  97                 c.retryPeriod = t
  98         }
  99 }
 100
 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) {
 105                 c.retryTimeout = t
 106         }
 107 }
 108
 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 {
 112         if socket == "" {
 113                 socket = DefaultSocketName
 114         }
 115         s := &StatsClient{
 116                 socket: socket,
 117         }
 118         for _, option := range options {
 119                 option(s)
 120         }
 121         if s.retryPeriod == 0 {
 122                 s.retryPeriod = DefaultSocketRetryPeriod
 123         }
 124         if s.retryTimeout == 0 {
 125                 s.retryTimeout = DefaultSocketRetryTimeout
 126         }
 127         return s
 128 }
 129
 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 {
 134                 return err
 135         }
 136         sc.done = make(chan struct{})
 137         if sc.statSegment, err = sc.connect(); err != nil {
 138                 return err
 139         }
 140         sc.monitorSocket()
 141         return nil
 142 }
 143
 144 // Disconnect from the socket, unmap shared memory and terminate
 145 // socket monitor
 146 func (sc *StatsClient) Disconnect() error {
 147         if sc.headerData == nil {
 148                 return nil
 149         }
 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)
 153         }
 154         sc.headerData = nil
 155         sc.done <- struct{}{}
 156
 157         Log.Debugf("successfully unmapped shared memory")
 158         return nil
 159 }
 160
 161 func (sc *StatsClient) ListStats(patterns ...string) (entries []adapter.StatIdentifier, err error) {
 162         if !sc.isConnected() {
 163                 return nil, adapter.ErrStatsDisconnected
 164         }
 165         accessEpoch := sc.accessStart()
 166         if accessEpoch == 0 {
 167                 return nil, adapter.ErrStatsAccessFailed
 168         }
 169
 170         entries, err = sc.getIdentifierEntries(patterns...)
 171         if err != nil {
 172                 return nil, err
 173         }
 174
 175         if !sc.accessEnd(accessEpoch) {
 176                 return nil, adapter.ErrStatsDataBusy
 177         }
 178         return entries, nil
 179 }
 180
 181 func (sc *StatsClient) DumpStats(patterns ...string) (entries []adapter.StatEntry, err error) {
 182         if !sc.isConnected() {
 183                 return nil, adapter.ErrStatsDisconnected
 184         }
 185
 186         accessEpoch := sc.accessStart()
 187         if accessEpoch == 0 {
 188                 return nil, adapter.ErrStatsAccessFailed
 189         }
 190
 191         entries, err = sc.getStatEntries(patterns...)
 192         if err != nil {
 193                 return nil, err
 194         }
 195
 196         if !sc.accessEnd(accessEpoch) {
 197                 return nil, adapter.ErrStatsDataBusy
 198         }
 199         return entries, nil
 200 }
 201
 202 func (sc *StatsClient) PrepareDir(patterns ...string) (*adapter.StatDir, error) {
 203         if !sc.isConnected() {
 204                 return nil, adapter.ErrStatsDisconnected
 205         }
 206
 207         accessEpoch := sc.accessStart()
 208         if accessEpoch == 0 {
 209                 return nil, adapter.ErrStatsAccessFailed
 210         }
 211
 212         entries, err := sc.getStatEntries(patterns...)
 213         if err != nil {
 214                 return nil, err
 215         }
 216
 217         if !sc.accessEnd(accessEpoch) {
 218                 return nil, adapter.ErrStatsDataBusy
 219         }
 220
 221         dir := &adapter.StatDir{
 222                 Epoch:   accessEpoch,
 223                 Entries: entries,
 224         }
 225
 226         return dir, nil
 227 }
 228
 229 func (sc *StatsClient) PrepareDirOnIndex(indexes ...uint32) (*adapter.StatDir, error) {
 230         if !sc.isConnected() {
 231                 return nil, adapter.ErrStatsDisconnected
 232         }
 233
 234         accessEpoch := sc.accessStart()
 235         if accessEpoch == 0 {
 236                 return nil, adapter.ErrStatsAccessFailed
 237         }
 238         vector := sc.GetDirectoryVector()
 239         if vector == nil {
 240                 return nil, fmt.Errorf("failed to prepare dir on index: directory vector is nil")
 241         }
 242         entries, err := sc.getStatEntriesOnIndex(vector, indexes...)
 243         if err != nil {
 244                 return nil, err
 245         }
 246
 247         if !sc.accessEnd(accessEpoch) {
 248                 return nil, adapter.ErrStatsDataBusy
 249         }
 250
 251         dir := &adapter.StatDir{
 252                 Epoch:   accessEpoch,
 253                 Entries: entries,
 254         }
 255
 256         return dir, nil
 257 }
 258
 259 // UpdateDir refreshes directory data for all counters
 260 func (sc *StatsClient) UpdateDir(dir *adapter.StatDir) (err error) {
 261         if !sc.isConnected() {
 262                 return adapter.ErrStatsDisconnected
 263         }
 264         epoch, _ := sc.GetEpoch()
 265         if dir.Epoch != epoch {
 266                 return adapter.ErrStatsDirStale
 267         }
 268
 269         accessEpoch := sc.accessStart()
 270         if accessEpoch == 0 {
 271                 return adapter.ErrStatsAccessFailed
 272         }
 273         dirVector := sc.GetDirectoryVector()
 274         if dirVector == nil {
 275                 return err
 276         }
 277         for i := 0; i < len(dir.Entries); i++ {
 278                 if err := sc.updateStatOnIndex(&dir.Entries[i], dirVector); err != nil {
 279                         return err
 280                 }
 281         }
 282         if !sc.accessEnd(accessEpoch) {
 283                 return adapter.ErrStatsDataBusy
 284         }
 285         return nil
 286 }
 287
 288 // checks the socket existence and waits for it for the designated
 289 // time if it is not available immediately
 290 func (sc *StatsClient) waitForSocket() error {
 291         if _, err := os.Stat(sc.socket); err != nil {
 292                 if os.IsNotExist(err) {
 293                         n := time.Now()
 294                         ticker := time.NewTicker(sc.retryPeriod)
 295                         timeout := time.After(sc.retryTimeout)
 296                         for {
 297                                 select {
 298                                 case <-ticker.C:
 299                                         if _, err := os.Stat(sc.socket); err == nil {
 300                                                 return nil
 301                                         }
 302                                 case <-timeout:
 303                                         return fmt.Errorf("stats socket file %s is not ready within timeout (after %.2f s) ",
 304                                                 sc.socket, time.Since(n).Seconds())
 305                                 }
 306                         }
 307                 } else {
 308                         return fmt.Errorf("stats socket error: %v", err)
 309                 }
 310         }
 311         return nil
 312 }
 313
 314 // connect to the socket and map it into the memory. According to the
 315 // header version info, an appropriate segment handler is returned
 316 func (sc *StatsClient) connect() (ss statSegment, err error) {
 317         addr := net.UnixAddr{
 318                 Net:  "unixpacket",
 319                 Name: sc.socket,
 320         }
 321         Log.Debugf("connecting to: %v", addr)
 322
 323         conn, err := net.DialUnix(addr.Net, nil, &addr)
 324         if err != nil {
 325                 Log.Warnf("connecting to socket %s failed: %s", addr, err)
 326                 return nil, err
 327         }
 328         defer func() {
 329                 if err := conn.Close(); err != nil {
 330                         Log.Warnf("closing socket failed: %v", err)
 331                 }
 332         }()
 333         Log.Debugf("connected to socket")
 334
 335         files, err := fd.Get(conn, 1, nil)
 336         if err != nil {
 337                 return nil, fmt.Errorf("getting file descriptor over socket failed: %v", err)
 338         }
 339         if len(files) == 0 {
 340                 return nil, fmt.Errorf("no files received over socket")
 341         }
 342
 343         file := files[0]
 344         defer func() {
 345                 if err := file.Close(); err != nil {
 346                         Log.Warnf("closing file failed: %v", err)
 347                 }
 348         }()
 349
 350         info, err := file.Stat()
 351         if err != nil {
 352                 return nil, err
 353         }
 354         size := info.Size()
 355
 356         sc.headerData, err = syscall.Mmap(int(file.Fd()), 0, int(size), syscall.PROT_READ, syscall.MAP_SHARED)
 357         if err != nil {
 358                 Log.Debugf("mapping shared memory failed: %v", err)
 359                 return nil, fmt.Errorf("mapping shared memory failed: %v", err)
 360         }
 361         Log.Debugf("successfully mmapped shared memory segment (size: %v) %v", size, len(sc.headerData))
 362
 363         version := getVersion(sc.headerData)
 364         switch version {
 365         case 1:
 366                 ss = newStatSegmentV1(sc.headerData, size)
 367         case 2:
 368                 ss = newStatSegmentV2(sc.headerData, size)
 369         default:
 370                 return nil, fmt.Errorf("stat segment version is not supported: %v (min: %v, max: %v)",
 371                         version, minVersion, maxVersion)
 372         }
 373
 374         // set connected
 375         atomic.CompareAndSwapUint32(&sc.connected, 0, 1)
 376
 377         return ss, nil
 378 }
 379
 380 // reconnect disconnects from the socket, re-validates it and
 381 // connects again
 382 func (sc *StatsClient) reconnect() (err error) {
 383         if err = sc.disconnect(); err != nil {
 384                 return fmt.Errorf("error disconnecting socket: %v", err)
 385         }
 386         if err = sc.waitForSocket(); err != nil {
 387                 return fmt.Errorf("error while waiting on socket: %v", err)
 388         }
 389         if sc.statSegment, err = sc.connect(); err != nil {
 390                 return fmt.Errorf("error connecting socket: %v", err)
 391         }
 392         return nil
 393 }
 394
 395 // disconnect unmaps socket data from the memory and resets the header
 396 func (sc *StatsClient) disconnect() error {
 397         if !atomic.CompareAndSwapUint32(&sc.connected, 1, 0) {
 398                 return fmt.Errorf("stats client is already disconnected")
 399         }
 400         if sc.headerData == nil {
 401                 return nil
 402         }
 403         if err := syscall.Munmap(sc.headerData); err != nil {
 404                 Log.Debugf("unmapping shared memory failed: %v", err)
 405                 return fmt.Errorf("unmapping shared memory failed: %v", err)
 406         }
 407         sc.headerData = nil
 408
 409         Log.Debugf("successfully unmapped shared memory")
 410         return nil
 411 }
 412
 413 func (sc *StatsClient) monitorSocket() {
 414         watcher, err := fsnotify.NewWatcher()
 415         if err != nil {
 416                 Log.Errorf("error starting socket monitor: %v", err)
 417                 return
 418         }
 419
 420         go func() {
 421                 for {
 422                         select {
 423                         case event := <-watcher.Events:
 424                                 if event.Op == fsnotify.Remove && event.Name == sc.socket {
 425                                         if err := sc.reconnect(); err != nil {
 426                                                 Log.Errorf("error occurred during socket reconnect: %v", err)
 427                                         }
 428                                 }
 429                         case err := <-watcher.Errors:
 430                                 Log.Errorf("socket monitor delivered error event: %v", err)
 431                         case <-sc.done:
 432                                 err := watcher.Close()
 433                                 Log.Debugf("socket monitor closed (error: %v)", err)
 434                                 return
 435                         }
 436                 }
 437         }()
 438
 439         if err := watcher.Add(filepath.Dir(sc.socket)); err != nil {
 440                 Log.Errorf("failed to add socket address to the watcher: %v", err)
 441         }
 442 }
 443
 444 // Starts monitoring 'inProgress' field. Returns stats segment
 445 // access epoch when completed, or zero value if not finished
 446 // within MaxWaitInProgress
 447 func (sc *StatsClient) accessStart() (epoch int64) {
 448         t := time.Now()
 449
 450         epoch, inProg := sc.GetEpoch()
 451         for inProg {
 452                 if time.Since(t) > MaxWaitInProgress {
 453                         return int64(0)
 454                 }
 455                 time.Sleep(CheckDelayInProgress)
 456                 epoch, inProg = sc.GetEpoch()
 457         }
 458         return epoch
 459 }
 460
 461 // AccessEnd returns true if stats data reading was finished, false
 462 // otherwise
 463 func (sc *StatsClient) accessEnd(accessEpoch int64) bool {
 464         epoch, inProgress := sc.GetEpoch()
 465         if accessEpoch != epoch || inProgress {
 466                 return false
 467         }
 468         return true
 469 }
 470
 471 // getStatEntries retrieves all stats matching desired patterns, or all stats if no pattern is provided.
 472 func (sc *StatsClient) getStatEntries(patterns ...string) (entries []adapter.StatEntry, err error) {
 473         vector := sc.GetDirectoryVector()
 474         if vector == nil {
 475                 return nil, fmt.Errorf("failed to get stat entries: directory vector is nil")
 476         }
 477         indexes, err := sc.listIndexes(vector, patterns...)
 478         if err != nil {
 479                 return nil, err
 480         }
 481         return sc.getStatEntriesOnIndex(vector, indexes...)
 482 }
 483
 484 // getIdentifierEntries retrieves all identifiers matching desired patterns, or all identifiers
 485 // if no pattern is provided.
 486 func (sc *StatsClient) getIdentifierEntries(patterns ...string) (identifiers []adapter.StatIdentifier, err error) {
 487         vector := sc.GetDirectoryVector()
 488         if vector == nil {
 489                 return nil, fmt.Errorf("failed to get identifier entries: directory vector is nil")
 490         }
 491         indexes, err := sc.listIndexes(vector, patterns...)
 492         if err != nil {
 493                 return nil, err
 494         }
 495         return sc.getIdentifierEntriesOnIndex(vector, indexes...)
 496 }
 497
 498 // getStatEntriesOnIndex retrieves stats on indexes, or all stats if indexes are not defined.
 499 func (sc *StatsClient) getStatEntriesOnIndex(vector dirVector, indexes ...uint32) (entries []adapter.StatEntry, err error) {
 500         dirLen := *(*uint32)(vectorLen(vector))
 501         for _, index := range indexes {
 502                 if index >= dirLen {
 503                         return nil, fmt.Errorf("stat entry index %d out of dir vector length (%d)", index, dirLen)
 504                 }
 505                 dirPtr, dirName, dirType := sc.GetStatDirOnIndex(vector, index)
 506                 if len(dirName) == 0 {
 507                         return
 508                 }
 509                 var t adapter.StatType
 510                 d := sc.CopyEntryData(dirPtr, ^uint32(0))
 511                 if d != nil {
 512                         t = d.Type()
 513                 }
 514                 entries = append(entries, adapter.StatEntry{
 515                         StatIdentifier: adapter.StatIdentifier{
 516                                 Index: index,
 517                                 Name:  dirName,
 518                         },
 519                         Type:    t,
 520                         Data:    d,
 521                         Symlink: adapter.StatType(dirType) == adapter.Symlink,
 522                 })
 523         }
 524         return entries, nil
 525 }
 526
 527 // getIdentifierEntriesOnIndex retrieves identifiers on indexes, or all identifiers if indexes are not defined.
 528 func (sc *StatsClient) getIdentifierEntriesOnIndex(vector dirVector, indexes ...uint32) (identifiers []adapter.StatIdentifier, err error) {
 529         dirLen := *(*uint32)(vectorLen(vector))
 530         for _, index := range indexes {
 531                 if index >= dirLen {
 532                         return nil, fmt.Errorf("stat entry index %d out of dir vector length (%d)", index, dirLen)
 533                 }
 534                 _, dirName, _ := sc.GetStatDirOnIndex(vector, index)
 535                 if len(dirName) == 0 {
 536                         return
 537                 }
 538                 identifiers = append(identifiers, adapter.StatIdentifier{
 539                         Index: index,
 540                         Name:  dirName,
 541                 })
 542         }
 543         return identifiers, nil
 544 }
 545
 546 // listIndexes lists indexes for all stat entries that match any of the regex patterns.
 547 func (sc *StatsClient) listIndexes(vector dirVector, patterns ...string) (indexes []uint32, err error) {
 548         if len(patterns) == 0 {
 549                 return sc.listIndexesFunc(vector, nil)
 550         }
 551         var regexes = make([]*regexp.Regexp, len(patterns))
 552         for i, pattern := range patterns {
 553                 r, err := regexp.Compile(pattern)
 554                 if err != nil {
 555                         return nil, fmt.Errorf("compiling regexp failed: %v", err)
 556                 }
 557                 regexes[i] = r
 558         }
 559         nameMatches := func(name dirName) bool {
 560                 for _, r := range regexes {
 561                         if r.Match(name) {
 562                                 return true
 563                         }
 564                 }
 565                 return false
 566         }
 567         return sc.listIndexesFunc(vector, nameMatches)
 568 }
 569
 570 // listIndexesFunc lists stats indexes. The optional function
 571 // argument filters returned values or returns all if empty
 572 func (sc *StatsClient) listIndexesFunc(vector dirVector, f func(name dirName) bool) (indexes []uint32, err error) {
 573         if f == nil {
 574                 // there is around ~3157 stats, so to avoid too many allocations
 575                 // we set capacity to 3200 when listing all stats
 576                 indexes = make([]uint32, 0, 3200)
 577         }
 578         vecLen := *(*uint32)(vectorLen(vector))
 579         for i := uint32(0); i < vecLen; i++ {
 580                 _, dirName, _ := sc.GetStatDirOnIndex(vector, i)
 581                 if f != nil {
 582                         if len(dirName) == 0 || !f(dirName) {
 583                                 continue
 584                         }
 585                 }
 586                 indexes = append(indexes, i)
 587         }
 588
 589         return indexes, nil
 590 }
 591
 592 func (sc *StatsClient) isConnected() bool {
 593         return atomic.LoadUint32(&sc.connected) == 1
 594 }
 595
 596 // updateStatOnIndex refreshes the entry data.
 597 func (sc *StatsClient) updateStatOnIndex(entry *adapter.StatEntry, vector dirVector) (err error) {
 598         dirLen := *(*uint32)(vectorLen(vector))
 599         if entry.Index >= dirLen {
 600                 return fmt.Errorf("stat entry index %d out of dir vector length (%d)", entry.Index, dirLen)
 601         }
 602         dirPtr, dirName, dirType := sc.GetStatDirOnIndex(vector, entry.Index)
 603         if len(dirName) == 0 ||
 604                 !bytes.Equal(dirName, entry.Name) ||
 605                 adapter.StatType(dirType) != entry.Type ||
 606                 entry.Data == nil {
 607                 return nil
 608         }
 609         if err := sc.UpdateEntryData(dirPtr, &entry.Data); err != nil {
 610                 err = fmt.Errorf("updating stat data for entry %s failed: %v", dirName, err)
 611         }
 612         return
 613 }