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         "regexp"
  24         "syscall"
  25         "time"
  26
  27         "git.fd.io/govpp.git/adapter"
  28         "github.com/ftrvxmtrx/fd"
  29         logger "github.com/sirupsen/logrus"
  30 )
  31
  32 const (
  33         // DefaultSocketName is default VPP stats socket file path.
  34         DefaultSocketName = adapter.DefaultStatsSocket
  35 )
  36
  37 const socketMissing = `
  38 ------------------------------------------------------------
  39  VPP stats socket file %s is missing!
  40
  41   - is VPP running with stats segment enabled?
  42   - is the correct socket name configured?
  43
  44  To enable it add following section to your VPP config:
  45    statseg {
  46      socket-name /run/vpp/stats.sock
  47    }
  48 ------------------------------------------------------------
  49 `
  50
  51 var (
  52         // Debug is global variable that determines debug mode
  53         Debug = os.Getenv("DEBUG_GOVPP_STATS") != ""
  54
  55         // Log is global logger
  56         Log = logger.New()
  57 )
  58
  59 // init initializes global logger, which logs debug level messages to stdout.
  60 func init() {
  61         Log.Out = os.Stdout
  62         if Debug {
  63                 Log.Level = logger.DebugLevel
  64                 Log.Debug("govpp/statsclient: enabled debug mode")
  65         }
  66 }
  67
  68 func debugf(f string, a ...interface{}) {
  69         if Debug {
  70                 Log.Debugf(f, a...)
  71         }
  72 }
  73
  74 // implements StatsAPI
  75 var _ adapter.StatsAPI = (*StatsClient)(nil)
  76
  77 // StatsClient is the pure Go implementation for VPP stats API.
  78 type StatsClient struct {
  79         sockAddr    string
  80         headerData  []byte
  81         isConnected bool
  82
  83         statSegment
  84 }
  85
  86 // NewStatsClient returns new VPP stats API client.
  87 func NewStatsClient(sockAddr string) *StatsClient {
  88         if sockAddr == "" {
  89                 sockAddr = DefaultSocketName
  90         }
  91         return &StatsClient{
  92                 sockAddr: sockAddr,
  93         }
  94 }
  95 // Connect to the VPP stats socket
  96 func (sc *StatsClient) Connect() (err error) {
  97         // check if socket exists
  98         if _, err := os.Stat(sc.sockAddr); os.IsNotExist(err) {
  99                 fmt.Fprintf(os.Stderr, socketMissing, sc.sockAddr)
 100                 return fmt.Errorf("stats socket file %s does not exist", sc.sockAddr)
 101         } else if err != nil {
 102                 return fmt.Errorf("stats socket error: %v", err)
 103         }
 104         if sc.isConnected {
 105                 return fmt.Errorf("already connected")
 106         }
 107         if sc.statSegment, err = sc.connect(); err != nil {
 108                 return err
 109         }
 110         sc.isConnected = true
 111         return nil
 112 }
 113
 114 // Disconnect from the socket and unmap shared memory
 115 func (sc *StatsClient) Disconnect() error {
 116         sc.isConnected = false
 117         if sc.headerData == nil {
 118                 return nil
 119         }
 120         if err := syscall.Munmap(sc.headerData); err != nil {
 121                 Log.Debugf("unmapping shared memory failed: %v", err)
 122                 return fmt.Errorf("unmapping shared memory failed: %v", err)
 123         }
 124         sc.headerData = nil
 125
 126         Log.Debugf("successfully unmapped shared memory")
 127         return nil
 128 }
 129
 130 func (sc *StatsClient) ListStats(patterns ...string) ([]string, error) {
 131         accessEpoch := sc.accessStart()
 132         if accessEpoch == 0 {
 133                 return nil, adapter.ErrStatsAccessFailed
 134         }
 135
 136         indexes, err := sc.listIndexes(patterns...)
 137         if err != nil {
 138                 return nil, err
 139         }
 140
 141         dirVector := sc.GetDirectoryVector()
 142         if dirVector == nil {
 143                 return nil, fmt.Errorf("failed to list stats: %v", err)
 144         }
 145         vecLen := *(*uint32)(vectorLen(dirVector))
 146
 147         var names []string
 148         for _, index := range indexes {
 149                 if index >= vecLen {
 150                         return nil, fmt.Errorf("stat entry index %d out of dir vector len (%d)", index, vecLen)
 151                 }
 152                 _, dirName, _ := sc.GetStatDirOnIndex(dirVector, index)
 153                 names = append(names, string(dirName))
 154         }
 155
 156         if !sc.accessEnd(accessEpoch) {
 157                 return nil, adapter.ErrStatsDataBusy
 158         }
 159
 160         return names, nil
 161 }
 162
 163 func (sc *StatsClient) DumpStats(patterns ...string) (entries []adapter.StatEntry, err error) {
 164         accessEpoch := sc.accessStart()
 165         if accessEpoch == 0 {
 166                 return nil, adapter.ErrStatsAccessFailed
 167         }
 168
 169         indexes, err := sc.listIndexes(patterns...)
 170         if err != nil {
 171                 return nil, err
 172         }
 173
 174         dirVector := sc.GetDirectoryVector()
 175         if dirVector == nil {
 176                 return nil, err
 177         }
 178         dirLen := *(*uint32)(vectorLen(dirVector))
 179
 180         debugf("dumping entries for %d indexes", len(indexes))
 181
 182         entries = make([]adapter.StatEntry, 0, len(indexes))
 183         for _, index := range indexes {
 184                 if index >= dirLen {
 185                         return nil, fmt.Errorf("stat entry index %d out of dir vector length (%d)", index, dirLen)
 186                 }
 187                 dirPtr, dirName, dirType := sc.GetStatDirOnIndex(dirVector, index)
 188                 if len(dirName) == 0 {
 189                         continue
 190                 }
 191                 entry := adapter.StatEntry{
 192                         Name: append([]byte(nil), dirName...),
 193                         Type: adapter.StatType(dirType),
 194                         Data: sc.CopyEntryData(dirPtr),
 195                 }
 196                 entries = append(entries, entry)
 197         }
 198
 199         if !sc.accessEnd(accessEpoch) {
 200                 return nil, adapter.ErrStatsDataBusy
 201         }
 202
 203         return entries, nil
 204 }
 205
 206 func (sc *StatsClient) PrepareDir(patterns ...string) (*adapter.StatDir, error) {
 207         dir := new(adapter.StatDir)
 208
 209         accessEpoch := sc.accessStart()
 210         if accessEpoch == 0 {
 211                 return nil, adapter.ErrStatsAccessFailed
 212         }
 213
 214         indexes, err := sc.listIndexes(patterns...)
 215         if err != nil {
 216                 return nil, err
 217         }
 218         dir.Indexes = indexes
 219
 220         dirVector := sc.GetDirectoryVector()
 221         if dirVector == nil {
 222                 return nil, err
 223         }
 224         dirLen := *(*uint32)(vectorLen(dirVector))
 225
 226         debugf("dumping entries for %d indexes", len(indexes))
 227
 228         entries := make([]adapter.StatEntry, 0, len(indexes))
 229         for _, index := range indexes {
 230                 if index >= dirLen {
 231                         return nil, fmt.Errorf("stat entry index %d out of dir vector length (%d)", index, dirLen)
 232                 }
 233                 dirPtr, dirName, dirType := sc.GetStatDirOnIndex(dirVector, index)
 234                 if len(dirName) == 0 {
 235                         continue
 236                 }
 237                 entry := adapter.StatEntry{
 238                         Name: append([]byte(nil), dirName...),
 239                         Type: adapter.StatType(dirType),
 240                         Data: sc.CopyEntryData(dirPtr),
 241                 }
 242                 entries = append(entries, entry)
 243         }
 244         dir.Entries = entries
 245
 246         if !sc.accessEnd(accessEpoch) {
 247                 return nil, adapter.ErrStatsDataBusy
 248         }
 249         dir.Epoch = accessEpoch
 250
 251         return dir, nil
 252 }
 253
 254 // UpdateDir refreshes directory data for all counters
 255 func (sc *StatsClient) UpdateDir(dir *adapter.StatDir) (err error) {
 256         epoch, _ := sc.GetEpoch()
 257         if dir.Epoch != epoch {
 258                 return adapter.ErrStatsDirStale
 259         }
 260
 261         accessEpoch := sc.accessStart()
 262         if accessEpoch == 0 {
 263                 return adapter.ErrStatsAccessFailed
 264         }
 265
 266         dirVector := sc.GetDirectoryVector()
 267         if dirVector == nil {
 268                 return err
 269         }
 270         for i, index := range dir.Indexes {
 271                 statSegDir, dirName, dirType := sc.GetStatDirOnIndex(dirVector, index)
 272                 if len(dirName) == 0 {
 273                         continue
 274                 }
 275                 entry := &dir.Entries[i]
 276                 if !bytes.Equal(dirName, entry.Name) {
 277                         continue
 278                 }
 279                 if adapter.StatType(dirType) != entry.Type {
 280                         continue
 281                 }
 282                 if entry.Data == nil {
 283                         continue
 284                 }
 285                 if err := sc.UpdateEntryData(statSegDir, &entry.Data); err != nil {
 286                         return fmt.Errorf("updating stat data for entry %s failed: %v", dirName, err)
 287                 }
 288         }
 289         if !sc.accessEnd(accessEpoch) {
 290                 return adapter.ErrStatsDataBusy
 291         }
 292
 293         return nil
 294 }
 295
 296 func (sc *StatsClient) connect() (statSegment, error) {
 297         addr := net.UnixAddr{
 298                 Net:  "unixpacket",
 299                 Name: sc.sockAddr,
 300         }
 301         Log.Debugf("connecting to: %v", addr)
 302
 303         conn, err := net.DialUnix(addr.Net, nil, &addr)
 304         if err != nil {
 305                 Log.Warnf("connecting to socket %s failed: %s", addr, err)
 306                 return nil, err
 307         }
 308         defer func() {
 309                 if err := conn.Close(); err != nil {
 310                         Log.Warnf("closing socket failed: %v", err)
 311                 }
 312         }()
 313         Log.Debugf("connected to socket")
 314
 315         files, err := fd.Get(conn, 1, nil)
 316         if err != nil {
 317                 return nil, fmt.Errorf("getting file descriptor over socket failed: %v", err)
 318         }
 319         if len(files) == 0 {
 320                 return nil, fmt.Errorf("no files received over socket")
 321         }
 322
 323         file := files[0]
 324         defer func() {
 325                 if err := file.Close(); err != nil {
 326                         Log.Warnf("closing file failed: %v", err)
 327                 }
 328         }()
 329
 330         info, err := file.Stat()
 331         if err != nil {
 332                 return nil, err
 333         }
 334         size := info.Size()
 335
 336         sc.headerData, err = syscall.Mmap(int(file.Fd()), 0, int(size), syscall.PROT_READ, syscall.MAP_SHARED)
 337         if err != nil {
 338                 Log.Debugf("mapping shared memory failed: %v", err)
 339                 return nil, fmt.Errorf("mapping shared memory failed: %v", err)
 340         }
 341         Log.Debugf("successfully mmapped shared memory segment (size: %v) %v", size, len(sc.headerData))
 342
 343         version := getVersion(sc.headerData)
 344         switch version {
 345         case 1:
 346                 return newStatSegmentV1(sc.headerData, size), nil
 347         case 2:
 348                 return newStatSegmentV2(sc.headerData, size), nil
 349         default:
 350                 return nil, fmt.Errorf("stat segment version is not supported: %v (min: %v, max: %v)",
 351                         version, minVersion, maxVersion)
 352         }
 353 }
 354
 355 // Starts monitoring 'inProgress' field. Returns stats segment
 356 // access epoch when completed, or zero value if not finished
 357 // within MaxWaitInProgress
 358 func (sc *StatsClient) accessStart() (epoch int64) {
 359         t := time.Now()
 360
 361         epoch, inProg := sc.GetEpoch()
 362         for inProg {
 363                 if time.Since(t) > MaxWaitInProgress {
 364                         return int64(0)
 365                 }
 366                 time.Sleep(CheckDelayInProgress)
 367                 epoch, inProg = sc.GetEpoch()
 368         }
 369         return epoch
 370 }
 371
 372 // AccessEnd returns true if stats data reading was finished, false
 373 // otherwise
 374 func (sc *StatsClient) accessEnd(accessEpoch int64) bool {
 375         epoch, inProgress := sc.GetEpoch()
 376         if accessEpoch != epoch || inProgress {
 377                 return false
 378         }
 379         return true
 380 }
 381
 382 // listIndexes lists indexes for all stat entries that match any of the regex patterns.
 383 func (sc *StatsClient) listIndexes(patterns ...string) (indexes []uint32, err error) {
 384         if len(patterns) == 0 {
 385                 return sc.listIndexesFunc(nil)
 386         }
 387         var regexes = make([]*regexp.Regexp, len(patterns))
 388         for i, pattern := range patterns {
 389                 r, err := regexp.Compile(pattern)
 390                 if err != nil {
 391                         return nil, fmt.Errorf("compiling regexp failed: %v", err)
 392                 }
 393                 regexes[i] = r
 394         }
 395         nameMatches := func(name []byte) bool {
 396                 for _, r := range regexes {
 397                         if r.Match(name) {
 398                                 return true
 399                         }
 400                 }
 401                 return false
 402         }
 403         return sc.listIndexesFunc(nameMatches)
 404 }
 405
 406 // listIndexesFunc lists stats indexes. The optional function
 407 // argument filters returned values or returns all if empty
 408 func (sc *StatsClient) listIndexesFunc(f func(name []byte) bool) (indexes []uint32, err error) {
 409         if f == nil {
 410                 // there is around ~3157 stats, so to avoid too many allocations
 411                 // we set capacity to 3200 when listing all stats
 412                 indexes = make([]uint32, 0, 3200)
 413         }
 414
 415         dirVector := sc.GetDirectoryVector()
 416         if dirVector == nil {
 417                 return nil, err
 418         }
 419         vecLen := *(*uint32)(vectorLen(dirVector))
 420
 421         for i := uint32(0); i < vecLen; i++ {
 422                 _, dirName, _ := sc.GetStatDirOnIndex(dirVector, i)
 423                 if f != nil {
 424                         if len(dirName) == 0 || !f(dirName) {
 425                                 continue
 426                         }
 427                 }
 428                 indexes = append(indexes, i)
 429         }
 430
 431         return indexes, nil
 432 }