src/vpp-api/python/vpp_papi/vpp_stats.py

   1 #!/usr/bin/env python3
   2 #
   3 # Copyright (c) 2021 Cisco and/or its affiliates.
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at:
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15 #
  16
  17 """
  18 This module implement Python access to the VPP statistics segment. It
  19 accesses the data structures directly in shared memory.
  20 VPP uses optimistic locking, so data structures may change underneath
  21 us while we are reading. Data is copied out and it's important to
  22 spend as little time as possible "holding the lock".
  23
  24 Counters are stored in VPP as a two dimensional array.
  25 Index by thread and index (typically sw_if_index).
  26 Simple counters count only packets, Combined counters count packets
  27 and octets.
  28
  29 Counters can be accessed in either dimension.
  30 stat['/if/rx'] - returns 2D lists
  31 stat['/if/rx'][0] - returns counters for all interfaces for thread 0
  32 stat['/if/rx'][0][1] - returns counter for interface 1 on thread 0
  33 stat['/if/rx'][0][1]['packets'] - returns the packet counter
  34                                   for interface 1 on thread 0
  35 stat['/if/rx'][:, 1] - returns the counters for interface 1 on all threads
  36 stat['/if/rx'][:, 1].packets() - returns the packet counters for
  37                                  interface 1 on all threads
  38 stat['/if/rx'][:, 1].sum_packets() - returns the sum of packet counters for
  39                                      interface 1 on all threads
  40 stat['/if/rx-miss'][:, 1].sum() - returns the sum of packet counters for
  41                                   interface 1 on all threads for simple counters
  42 """
  43
  44 import os
  45 import socket
  46 import array
  47 import mmap
  48 from struct import Struct
  49 import time
  50 import unittest
  51 import re
  52
  53
  54 def recv_fd(sock):
  55     """Get file descriptor for memory map"""
  56     fds = array.array("i")  # Array of ints
  57     _, ancdata, _, _ = sock.recvmsg(0, socket.CMSG_SPACE(4))
  58     for cmsg_level, cmsg_type, cmsg_data in ancdata:
  59         if cmsg_level == socket.SOL_SOCKET and cmsg_type == socket.SCM_RIGHTS:
  60             fds.frombytes(cmsg_data[: len(cmsg_data) - (len(cmsg_data) % fds.itemsize)])
  61     return list(fds)[0]
  62
  63
  64 VEC_LEN_FMT = Struct("I")
  65
  66
  67 def get_vec_len(stats, vector_offset):
  68     """Equivalent to VPP vec_len()"""
  69     return VEC_LEN_FMT.unpack_from(stats.statseg, vector_offset - 8)[0]
  70
  71
  72 def get_string(stats, ptr):
  73     """Get a string from a VPP vector"""
  74     namevector = ptr - stats.base
  75     namevectorlen = get_vec_len(stats, namevector)
  76     if namevector + namevectorlen >= stats.size:
  77         raise IOError("String overruns stats segment")
  78     return stats.statseg[namevector : namevector + namevectorlen - 1].decode("ascii")
  79
  80
  81 class StatsVector:
  82     """A class representing a VPP vector"""
  83
  84     def __init__(self, stats, ptr, fmt):
  85         self.vec_start = ptr - stats.base
  86         self.vec_len = get_vec_len(stats, ptr - stats.base)
  87         self.struct = Struct(fmt)
  88         self.fmtlen = len(fmt)
  89         self.elementsize = self.struct.size
  90         self.statseg = stats.statseg
  91         self.stats = stats
  92
  93         if self.vec_start + self.vec_len * self.elementsize >= stats.size:
  94             raise IOError("Vector overruns stats segment")
  95
  96     def __iter__(self):
  97         with self.stats.lock:
  98             return self.struct.iter_unpack(
  99                 self.statseg[
 100                     self.vec_start : self.vec_start + self.elementsize * self.vec_len
 101                 ]
 102             )
 103
 104     def __getitem__(self, index):
 105         if index > self.vec_len:
 106             raise IOError("Index beyond end of vector")
 107         with self.stats.lock:
 108             if self.fmtlen == 1:
 109                 return self.struct.unpack_from(
 110                     self.statseg, self.vec_start + (index * self.elementsize)
 111                 )[0]
 112             return self.struct.unpack_from(
 113                 self.statseg, self.vec_start + (index * self.elementsize)
 114             )
 115
 116
 117 class VPPStats:
 118     """Main class implementing Python access to the VPP statistics segment"""
 119
 120     # pylint: disable=too-many-instance-attributes
 121     shared_headerfmt = Struct("QPQQPP")
 122     default_socketname = "/run/vpp/stats.sock"
 123
 124     def __init__(self, socketname=default_socketname, timeout=10):
 125         self.socketname = socketname
 126         self.timeout = timeout
 127         self.directory = {}
 128         self.lock = StatsLock(self)
 129         self.connected = False
 130         self.size = 0
 131         self.last_epoch = 0
 132         self.statseg = 0
 133
 134     def connect(self):
 135         """Connect to stats segment"""
 136         if self.connected:
 137             return
 138         sock = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
 139         sock.connect(self.socketname)
 140
 141         mfd = recv_fd(sock)
 142         sock.close()
 143
 144         stat_result = os.fstat(mfd)
 145         self.statseg = mmap.mmap(
 146             mfd, stat_result.st_size, mmap.PROT_READ, mmap.MAP_SHARED
 147         )
 148         os.close(mfd)
 149
 150         self.size = stat_result.st_size
 151         if self.version != 2:
 152             raise Exception("Incompatbile stat segment version {}".format(self.version))
 153
 154         self.refresh()
 155         self.connected = True
 156
 157     def disconnect(self):
 158         """Disconnect from stats segment"""
 159         if self.connected:
 160             self.statseg.close()
 161             self.connected = False
 162
 163     @property
 164     def version(self):
 165         """Get version of stats segment"""
 166         return self.shared_headerfmt.unpack_from(self.statseg)[0]
 167
 168     @property
 169     def base(self):
 170         """Get base pointer of stats segment"""
 171         return self.shared_headerfmt.unpack_from(self.statseg)[1]
 172
 173     @property
 174     def epoch(self):
 175         """Get current epoch value from stats segment"""
 176         return self.shared_headerfmt.unpack_from(self.statseg)[2]
 177
 178     @property
 179     def in_progress(self):
 180         """Get value of in_progress from stats segment"""
 181         return self.shared_headerfmt.unpack_from(self.statseg)[3]
 182
 183     @property
 184     def directory_vector(self):
 185         """Get pointer of directory vector"""
 186         return self.shared_headerfmt.unpack_from(self.statseg)[4]
 187
 188     elementfmt = "IQ128s"
 189
 190     def refresh(self, blocking=True):
 191         """Refresh directory vector cache (epoch changed)"""
 192         directory = {}
 193         directory_by_idx = {}
 194         while True:
 195             try:
 196                 with self.lock:
 197                     self.last_epoch = self.epoch
 198                     for i, direntry in enumerate(
 199                         StatsVector(self, self.directory_vector, self.elementfmt)
 200                     ):
 201                         path_raw = direntry[2].find(b"\x00")
 202                         path = direntry[2][:path_raw].decode("ascii")
 203                         directory[path] = StatsEntry(direntry[0], direntry[1])
 204                         directory_by_idx[i] = path
 205                     self.directory = directory
 206                     self.directory_by_idx = directory_by_idx
 207                     return
 208             except IOError:
 209                 if not blocking:
 210                     raise
 211
 212     def __getitem__(self, item, blocking=True):
 213         if not self.connected:
 214             self.connect()
 215         while True:
 216             try:
 217                 if self.last_epoch != self.epoch:
 218                     self.refresh(blocking)
 219                 with self.lock:
 220                     return self.directory[item].get_counter(self)
 221             except IOError:
 222                 if not blocking:
 223                     raise
 224
 225     def __iter__(self):
 226         return iter(self.directory.items())
 227
 228     def set_errors(self, blocking=True):
 229         """Return dictionary of error counters > 0"""
 230         if not self.connected:
 231             self.connect()
 232
 233         errors = {k: v for k, v in self.directory.items() if k.startswith("/err/")}
 234         result = {}
 235         for k in errors:
 236             try:
 237                 total = self[k].sum()
 238                 if total:
 239                     result[k] = total
 240             except KeyError:
 241                 pass
 242         return result
 243
 244     def set_errors_str(self, blocking=True):
 245         """Return all errors counters > 0 pretty printed"""
 246         error_string = ["ERRORS:"]
 247         error_counters = self.set_errors(blocking)
 248         for k in sorted(error_counters):
 249             error_string.append("{:<60}{:>10}".format(k, error_counters[k]))
 250         return "%s\n" % "\n".join(error_string)
 251
 252     def get_counter(self, name, blocking=True):
 253         """Alternative call to __getitem__"""
 254         return self.__getitem__(name, blocking)
 255
 256     def get_err_counter(self, name, blocking=True):
 257         """Alternative call to __getitem__"""
 258         return self.__getitem__(name, blocking).sum()
 259
 260     def ls(self, patterns):
 261         """Returns list of counters matching pattern"""
 262         # pylint: disable=invalid-name
 263         if not self.connected:
 264             self.connect()
 265         if not isinstance(patterns, list):
 266             patterns = [patterns]
 267         regex = [re.compile(i) for i in patterns]
 268         if self.last_epoch != self.epoch:
 269             self.refresh()
 270
 271         return [
 272             k
 273             for k, v in self.directory.items()
 274             if any(re.match(pattern, k) for pattern in regex)
 275         ]
 276
 277     def dump(self, counters, blocking=True):
 278         """Given a list of counters return a dictionary of results"""
 279         if not self.connected:
 280             self.connect()
 281         result = {}
 282         for cnt in counters:
 283             result[cnt] = self.__getitem__(cnt, blocking)
 284         return result
 285
 286
 287 class StatsLock:
 288     """Stat segment optimistic locking"""
 289
 290     def __init__(self, stats):
 291         self.stats = stats
 292         self.epoch = 0
 293
 294     def __enter__(self):
 295         acquired = self.acquire(blocking=True)
 296         assert acquired, "Lock wasn't acquired, but blocking=True"
 297         return self
 298
 299     def __exit__(self, exc_type=None, exc_value=None, traceback=None):
 300         self.release()
 301
 302     def acquire(self, blocking=True, timeout=-1):
 303         """Acquire the lock. Await in progress to go false. Record epoch."""
 304         self.epoch = self.stats.epoch
 305         if timeout > 0:
 306             start = time.monotonic()
 307         while self.stats.in_progress:
 308             if not blocking:
 309                 time.sleep(0.01)
 310                 if timeout > 0:
 311                     if start + time.monotonic() > timeout:
 312                         return False
 313         return True
 314
 315     def release(self):
 316         """Check if data read while locked is valid"""
 317         if self.stats.in_progress or self.stats.epoch != self.epoch:
 318             raise IOError("Optimistic lock failed, retry")
 319
 320     def locked(self):
 321         """Not used"""
 322
 323
 324 class StatsCombinedList(list):
 325     """Column slicing for Combined counters list"""
 326
 327     def __getitem__(self, item):
 328         """Supports partial numpy style 2d support. Slice by column [:,1]"""
 329         if isinstance(item, int):
 330             return list.__getitem__(self, item)
 331         return CombinedList([row[item[1]] for row in self])
 332
 333
 334 class CombinedList(list):
 335     """Combined Counters 2-dimensional by thread by index of packets/octets"""
 336
 337     def packets(self):
 338         """Return column (2nd dimension). Packets for all threads"""
 339         return [pair[0] for pair in self]
 340
 341     def octets(self):
 342         """Return column (2nd dimension). Octets for all threads"""
 343         return [pair[1] for pair in self]
 344
 345     def sum_packets(self):
 346         """Return column (2nd dimension). Sum of all packets for all threads"""
 347         return sum(self.packets())
 348
 349     def sum_octets(self):
 350         """Return column (2nd dimension). Sum of all octets for all threads"""
 351         return sum(self.octets())
 352
 353
 354 class StatsTuple(tuple):
 355     """A Combined vector tuple (packets, octets)"""
 356
 357     def __init__(self, data):
 358         self.dictionary = {"packets": data[0], "bytes": data[1]}
 359         super().__init__()
 360
 361     def __repr__(self):
 362         return dict.__repr__(self.dictionary)
 363
 364     def __getitem__(self, item):
 365         if isinstance(item, int):
 366             return tuple.__getitem__(self, item)
 367         if item == "packets":
 368             return tuple.__getitem__(self, 0)
 369         return tuple.__getitem__(self, 1)
 370
 371
 372 class StatsSimpleList(list):
 373     """Simple Counters 2-dimensional by thread by index of packets"""
 374
 375     def __getitem__(self, item):
 376         """Supports partial numpy style 2d support. Slice by column [:,1]"""
 377         if isinstance(item, int):
 378             return list.__getitem__(self, item)
 379         return SimpleList([row[item[1]] for row in self])
 380
 381
 382 class SimpleList(list):
 383     """Simple counter"""
 384
 385     def sum(self):
 386         """Sum the vector"""
 387         return sum(self)
 388
 389
 390 class StatsEntry:
 391     """An individual stats entry"""
 392
 393     # pylint: disable=unused-argument,no-self-use
 394
 395     def __init__(self, stattype, statvalue):
 396         self.type = stattype
 397         self.value = statvalue
 398
 399         if stattype == 1:
 400             self.function = self.scalar
 401         elif stattype == 2:
 402             self.function = self.simple
 403         elif stattype == 3:
 404             self.function = self.combined
 405         elif stattype == 4:
 406             self.function = self.name
 407         elif stattype == 6:
 408             self.function = self.symlink
 409         else:
 410             self.function = self.illegal
 411
 412     def illegal(self, stats):
 413         """Invalid or unknown counter type"""
 414         return None
 415
 416     def scalar(self, stats):
 417         """Scalar counter"""
 418         return self.value
 419
 420     def simple(self, stats):
 421         """Simple counter"""
 422         counter = StatsSimpleList()
 423         for threads in StatsVector(stats, self.value, "P"):
 424             clist = [v[0] for v in StatsVector(stats, threads[0], "Q")]
 425             counter.append(clist)
 426         return counter
 427
 428     def combined(self, stats):
 429         """Combined counter"""
 430         counter = StatsCombinedList()
 431         for threads in StatsVector(stats, self.value, "P"):
 432             clist = [StatsTuple(cnt) for cnt in StatsVector(stats, threads[0], "QQ")]
 433             counter.append(clist)
 434         return counter
 435
 436     def name(self, stats):
 437         """Name counter"""
 438         counter = []
 439         for name in StatsVector(stats, self.value, "P"):
 440             if name[0]:
 441                 counter.append(get_string(stats, name[0]))
 442         return counter
 443
 444     SYMLINK_FMT1 = Struct("II")
 445     SYMLINK_FMT2 = Struct("Q")
 446
 447     def symlink(self, stats):
 448         """Symlink counter"""
 449         b = self.SYMLINK_FMT2.pack(self.value)
 450         index1, index2 = self.SYMLINK_FMT1.unpack(b)
 451         name = stats.directory_by_idx[index1]
 452         return stats[name][:, index2]
 453
 454     def get_counter(self, stats):
 455         """Return a list of counters"""
 456         if stats:
 457             return self.function(stats)
 458
 459
 460 class TestStats(unittest.TestCase):
 461     """Basic statseg tests"""
 462
 463     def setUp(self):
 464         """Connect to statseg"""
 465         self.stat = VPPStats()
 466         self.stat.connect()
 467         self.profile = cProfile.Profile()
 468         self.profile.enable()
 469
 470     def tearDown(self):
 471         """Disconnect from statseg"""
 472         self.stat.disconnect()
 473         profile = Stats(self.profile)
 474         profile.strip_dirs()
 475         profile.sort_stats("cumtime")
 476         profile.print_stats()
 477         print("\n--->>>")
 478
 479     def test_counters(self):
 480         """Test access to statseg"""
 481
 482         print("/err/abf-input-ip4/missed", self.stat["/err/abf-input-ip4/missed"])
 483         print("/sys/heartbeat", self.stat["/sys/heartbeat"])
 484         print("/if/names", self.stat["/if/names"])
 485         print("/if/rx-miss", self.stat["/if/rx-miss"])
 486         print("/if/rx-miss", self.stat["/if/rx-miss"][1])
 487         print(
 488             "/nat44-ed/out2in/slowpath/drops",
 489             self.stat["/nat44-ed/out2in/slowpath/drops"],
 490         )
 491         with self.assertRaises(KeyError):
 492             print("NO SUCH COUNTER", self.stat["foobar"])
 493         print("/if/rx", self.stat.get_counter("/if/rx"))
 494         print(
 495             "/err/ethernet-input/no_error",
 496             self.stat.get_counter("/err/ethernet-input/no_error"),
 497         )
 498
 499     def test_column(self):
 500         """Test column slicing"""
 501
 502         print("/if/rx-miss", self.stat["/if/rx-miss"])
 503         print("/if/rx", self.stat["/if/rx"])  # All interfaces for thread #1
 504         print(
 505             "/if/rx thread #1", self.stat["/if/rx"][0]
 506         )  # All interfaces for thread #1
 507         print(
 508             "/if/rx thread #1, interface #1", self.stat["/if/rx"][0][1]
 509         )  # All interfaces for thread #1
 510         print("/if/rx if_index #1", self.stat["/if/rx"][:, 1])
 511         print("/if/rx if_index #1 packets", self.stat["/if/rx"][:, 1].packets())
 512         print("/if/rx if_index #1 packets", self.stat["/if/rx"][:, 1].sum_packets())
 513         print("/if/rx if_index #1 packets", self.stat["/if/rx"][:, 1].octets())
 514         print("/if/rx-miss", self.stat["/if/rx-miss"])
 515         print("/if/rx-miss if_index #1 packets", self.stat["/if/rx-miss"][:, 1].sum())
 516         print("/if/rx if_index #1 packets", self.stat["/if/rx"][0][1]["packets"])
 517
 518     def test_nat44(self):
 519         """Test the nat counters"""
 520
 521         print("/nat44-ei/ha/del-event-recv", self.stat["/nat44-ei/ha/del-event-recv"])
 522         print(
 523             "/err/nat44-ei-ha/pkts-processed",
 524             self.stat["/err/nat44-ei-ha/pkts-processed"].sum(),
 525         )
 526
 527     def test_legacy(self):
 528         """Legacy interface"""
 529         directory = self.stat.ls(["^/if", "/err/ip4-input", "/sys/node/ip4-input"])
 530         data = self.stat.dump(directory)
 531         print(data)
 532         print("Looking up sys node")
 533         directory = self.stat.ls(["^/sys/node"])
 534         print("Dumping sys node")
 535         data = self.stat.dump(directory)
 536         print(data)
 537         directory = self.stat.ls(["^/foobar"])
 538         data = self.stat.dump(directory)
 539         print(data)
 540
 541     def test_sys_nodes(self):
 542         """Test /sys/nodes"""
 543         counters = self.stat.ls("^/sys/node")
 544         print("COUNTERS:", counters)
 545         print("/sys/node", self.stat.dump(counters))
 546         print("/net/route/to", self.stat["/net/route/to"])
 547
 548     def test_symlink(self):
 549         """Symbolic links"""
 550         print("/interface/local0/rx", self.stat["/interfaces/local0/rx"])
 551         print("/sys/nodes/unix-epoll-input", self.stat["/nodes/unix-epoll-input/calls"])
 552
 553
 554 if __name__ == "__main__":
 555     import cProfile
 556     from pstats import Stats
 557
 558     unittest.main()