diff options
Diffstat (limited to 'src/vpp-api/python/vpp_papi/vpp_stats.py')
| -rwxr-xr-x | src/vpp-api/python/vpp_papi/vpp_stats.py | 380 |
1 files changed, 189 insertions, 191 deletions
diff --git a/src/vpp-api/python/vpp_papi/vpp_stats.py b/src/vpp-api/python/vpp_papi/vpp_stats.py index 3831a82f428..aa9ff85b3c7 100755 --- a/src/vpp-api/python/vpp_papi/vpp_stats.py +++ b/src/vpp-api/python/vpp_papi/vpp_stats.py @@ -14,7 +14,7 @@ # limitations under the License. # -''' +""" This module implement Python access to the VPP statistics segment. It accesses the data structures directly in shared memory. VPP uses optimistic locking, so data structures may change underneath @@ -39,7 +39,7 @@ stat['/if/rx'][:, 1].sum_packets() - returns the sum of packet counters for interface 1 on all threads stat['/if/rx-miss'][:, 1].sum() - returns the sum of packet counters for interface 1 on all threads for simple counters -''' +""" import os import socket @@ -50,31 +50,36 @@ import time import unittest import re + def recv_fd(sock): - '''Get file descriptor for memory map''' - fds = array.array("i") # Array of ints - _, ancdata, _, _ = sock.recvmsg(0, socket.CMSG_LEN(4)) + """Get file descriptor for memory map""" + fds = array.array("i") # Array of ints + _, ancdata, _, _ = sock.recvmsg(0, socket.CMSG_SPACE(4)) for cmsg_level, cmsg_type, cmsg_data in ancdata: if cmsg_level == socket.SOL_SOCKET and cmsg_type == socket.SCM_RIGHTS: - fds.frombytes(cmsg_data[:len(cmsg_data) - (len(cmsg_data) % fds.itemsize)]) + fds.frombytes(cmsg_data[: len(cmsg_data) - (len(cmsg_data) % fds.itemsize)]) return list(fds)[0] -VEC_LEN_FMT = Struct('I') + +VEC_LEN_FMT = Struct("I") + + def get_vec_len(stats, vector_offset): - '''Equivalent to VPP vec_len()''' + """Equivalent to VPP vec_len()""" return VEC_LEN_FMT.unpack_from(stats.statseg, vector_offset - 8)[0] + def get_string(stats, ptr): - '''Get a string from a VPP vector''' + """Get a string from a VPP vector""" namevector = ptr - stats.base namevectorlen = get_vec_len(stats, namevector) if namevector + namevectorlen >= stats.size: - raise IOError('String overruns stats segment') - return stats.statseg[namevector:namevector+namevectorlen-1].decode('ascii') + raise IOError("String overruns stats segment") + return stats.statseg[namevector : namevector + namevectorlen - 1].decode("ascii") class StatsVector: - '''A class representing a VPP vector''' + """A class representing a VPP vector""" def __init__(self, stats, ptr, fmt): self.vec_start = ptr - stats.base @@ -86,28 +91,35 @@ class StatsVector: self.stats = stats if self.vec_start + self.vec_len * self.elementsize >= stats.size: - raise IOError('Vector overruns stats segment') + raise IOError("Vector overruns stats segment") def __iter__(self): with self.stats.lock: - return self.struct.iter_unpack(self.statseg[self.vec_start:self.vec_start + - self.elementsize*self.vec_len]) + return self.struct.iter_unpack( + self.statseg[ + self.vec_start : self.vec_start + self.elementsize * self.vec_len + ] + ) def __getitem__(self, index): if index > self.vec_len: - raise IOError('Index beyond end of vector') + raise IOError("Index beyond end of vector") with self.stats.lock: if self.fmtlen == 1: - return self.struct.unpack_from(self.statseg, self.vec_start + - (index * self.elementsize))[0] - return self.struct.unpack_from(self.statseg, self.vec_start + - (index * self.elementsize)) + return self.struct.unpack_from( + self.statseg, self.vec_start + (index * self.elementsize) + )[0] + return self.struct.unpack_from( + self.statseg, self.vec_start + (index * self.elementsize) + ) + + +class VPPStats: + """Main class implementing Python access to the VPP statistics segment""" -class VPPStats(): - '''Main class implementing Python access to the VPP statistics segment''' # pylint: disable=too-many-instance-attributes - shared_headerfmt = Struct('QPQQPP') - default_socketname = '/run/vpp/stats.sock' + shared_headerfmt = Struct("QPQQPP") + default_socketname = "/run/vpp/stats.sock" def __init__(self, socketname=default_socketname, timeout=10): self.socketname = socketname @@ -117,89 +129,87 @@ class VPPStats(): self.connected = False self.size = 0 self.last_epoch = 0 - self.error_vectors = 0 self.statseg = 0 def connect(self): - '''Connect to stats segment''' + """Connect to stats segment""" if self.connected: return sock = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) + + # Our connect races the corresponding recv_fds call in VPP, if we beat + # VPP then we will try (unsuccessfully) to receive file descriptors and + # will have gone away before VPP can respond to our connect. A short + # timeout here stops this error occurring. + sock.settimeout(1) sock.connect(self.socketname) mfd = recv_fd(sock) sock.close() stat_result = os.fstat(mfd) - self.statseg = mmap.mmap(mfd, stat_result.st_size, mmap.PROT_READ, mmap.MAP_SHARED) + self.statseg = mmap.mmap( + mfd, stat_result.st_size, mmap.PROT_READ, mmap.MAP_SHARED + ) os.close(mfd) self.size = stat_result.st_size if self.version != 2: - raise Exception('Incompatbile stat segment version {}' - .format(self.version)) + raise Exception("Incompatbile stat segment version {}".format(self.version)) self.refresh() self.connected = True def disconnect(self): - '''Disconnect from stats segment''' + """Disconnect from stats segment""" if self.connected: self.statseg.close() self.connected = False @property def version(self): - '''Get version of stats segment''' + """Get version of stats segment""" return self.shared_headerfmt.unpack_from(self.statseg)[0] @property def base(self): - '''Get base pointer of stats segment''' + """Get base pointer of stats segment""" return self.shared_headerfmt.unpack_from(self.statseg)[1] @property def epoch(self): - '''Get current epoch value from stats segment''' + """Get current epoch value from stats segment""" return self.shared_headerfmt.unpack_from(self.statseg)[2] @property def in_progress(self): - '''Get value of in_progress from stats segment''' + """Get value of in_progress from stats segment""" return self.shared_headerfmt.unpack_from(self.statseg)[3] @property def directory_vector(self): - '''Get pointer of directory vector''' + """Get pointer of directory vector""" return self.shared_headerfmt.unpack_from(self.statseg)[4] - @property - def error_vector(self): - '''Get pointer of error vector''' - return self.shared_headerfmt.unpack_from(self.statseg)[5] - - elementfmt = 'IQ128s' + elementfmt = "IQ128s" def refresh(self, blocking=True): - '''Refresh directory vector cache (epoch changed)''' + """Refresh directory vector cache (epoch changed)""" directory = {} directory_by_idx = {} while True: try: with self.lock: self.last_epoch = self.epoch - for i, direntry in enumerate(StatsVector(self, self.directory_vector, self.elementfmt)): - path_raw = direntry[2].find(b'\x00') - path = direntry[2][:path_raw].decode('ascii') + for i, direntry in enumerate( + StatsVector(self, self.directory_vector, self.elementfmt) + ): + path_raw = direntry[2].find(b"\x00") + path = direntry[2][:path_raw].decode("ascii") directory[path] = StatsEntry(direntry[0], direntry[1]) directory_by_idx[i] = path self.directory = directory self.directory_by_idx = directory_by_idx - - # Cache the error index vectors - self.error_vectors = [] - for threads in StatsVector(self, self.error_vector, 'P'): - self.error_vectors.append(StatsVector(self, threads[0], 'Q')) return except IOError: if not blocking: @@ -222,58 +232,39 @@ class VPPStats(): return iter(self.directory.items()) def set_errors(self, blocking=True): - '''Return dictionary of error counters > 0''' + """Return dictionary of error counters > 0""" if not self.connected: self.connect() - errors = {k:v for k, v in self.directory.items() if k.startswith("/err/")} + errors = {k: v for k, v in self.directory.items() if k.startswith("/err/")} result = {} - while True: + for k in errors: try: - if self.last_epoch != self.epoch: - self.refresh(blocking) - with self.lock: - for k, entry in errors.items(): - total = 0 - i = entry.value - for per_thread in self.error_vectors: - total += per_thread[i] - if total: - result[k] = total - return result - except IOError: - if not blocking: - raise + total = self[k].sum() + if total: + result[k] = total + except KeyError: + pass + return result def set_errors_str(self, blocking=True): - '''Return all errors counters > 0 pretty printed''' - error_string = ['ERRORS:'] + """Return all errors counters > 0 pretty printed""" + error_string = ["ERRORS:"] error_counters = self.set_errors(blocking) for k in sorted(error_counters): - error_string.append('{:<60}{:>10}'.format(k, error_counters[k])) - return '%s\n' % '\n'.join(error_string) + error_string.append("{:<60}{:>10}".format(k, error_counters[k])) + return "%s\n" % "\n".join(error_string) def get_counter(self, name, blocking=True): - '''Alternative call to __getitem__''' + """Alternative call to __getitem__""" return self.__getitem__(name, blocking) def get_err_counter(self, name, blocking=True): - '''Return a single value (sum of all threads)''' - if not self.connected: - self.connect() - if name.startswith("/err/"): - while True: - try: - if self.last_epoch != self.epoch: - self.refresh(blocking) - with self.lock: - return sum(self.directory[name].get_counter(self)) - except IOError: - if not blocking: - raise + """Alternative call to __getitem__""" + return self.__getitem__(name, blocking).sum() def ls(self, patterns): - '''Returns list of counters matching pattern''' + """Returns list of counters matching pattern""" # pylint: disable=invalid-name if not self.connected: self.connect() @@ -283,20 +274,24 @@ class VPPStats(): if self.last_epoch != self.epoch: self.refresh() - return [k for k, v in self.directory.items() - if any(re.match(pattern, k) for pattern in regex)] + return [ + k + for k, v in self.directory.items() + if any(re.match(pattern, k) for pattern in regex) + ] def dump(self, counters, blocking=True): - '''Given a list of counters return a dictionary of results''' + """Given a list of counters return a dictionary of results""" if not self.connected: self.connect() result = {} for cnt in counters: - result[cnt] = self.__getitem__(cnt,blocking) + result[cnt] = self.__getitem__(cnt, blocking) return result -class StatsLock(): - '''Stat segment optimistic locking''' + +class StatsLock: + """Stat segment optimistic locking""" def __init__(self, stats): self.stats = stats @@ -311,7 +306,7 @@ class StatsLock(): self.release() def acquire(self, blocking=True, timeout=-1): - '''Acquire the lock. Await in progress to go false. Record epoch.''' + """Acquire the lock. Await in progress to go false. Record epoch.""" self.epoch = self.stats.epoch if timeout > 0: start = time.monotonic() @@ -324,46 +319,49 @@ class StatsLock(): return True def release(self): - '''Check if data read while locked is valid''' + """Check if data read while locked is valid""" if self.stats.in_progress or self.stats.epoch != self.epoch: - raise IOError('Optimistic lock failed, retry') + raise IOError("Optimistic lock failed, retry") def locked(self): - '''Not used''' + """Not used""" class StatsCombinedList(list): - '''Column slicing for Combined counters list''' + """Column slicing for Combined counters list""" def __getitem__(self, item): - '''Supports partial numpy style 2d support. Slice by column [:,1]''' + """Supports partial numpy style 2d support. Slice by column [:,1]""" if isinstance(item, int): return list.__getitem__(self, item) return CombinedList([row[item[1]] for row in self]) + class CombinedList(list): - '''Combined Counters 2-dimensional by thread by index of packets/octets''' + """Combined Counters 2-dimensional by thread by index of packets/octets""" def packets(self): - '''Return column (2nd dimension). Packets for all threads''' + """Return column (2nd dimension). Packets for all threads""" return [pair[0] for pair in self] def octets(self): - '''Return column (2nd dimension). Octets for all threads''' + """Return column (2nd dimension). Octets for all threads""" return [pair[1] for pair in self] def sum_packets(self): - '''Return column (2nd dimension). Sum of all packets for all threads''' + """Return column (2nd dimension). Sum of all packets for all threads""" return sum(self.packets()) def sum_octets(self): - '''Return column (2nd dimension). Sum of all octets for all threads''' + """Return column (2nd dimension). Sum of all octets for all threads""" return sum(self.octets()) + class StatsTuple(tuple): - '''A Combined vector tuple (packets, octets)''' + """A Combined vector tuple (packets, octets)""" + def __init__(self, data): - self.dictionary = {'packets': data[0], 'bytes': data[1]} + self.dictionary = {"packets": data[0], "bytes": data[1]} super().__init__() def __repr__(self): @@ -372,28 +370,32 @@ class StatsTuple(tuple): def __getitem__(self, item): if isinstance(item, int): return tuple.__getitem__(self, item) - if item == 'packets': + if item == "packets": return tuple.__getitem__(self, 0) return tuple.__getitem__(self, 1) + class StatsSimpleList(list): - '''Simple Counters 2-dimensional by thread by index of packets''' + """Simple Counters 2-dimensional by thread by index of packets""" def __getitem__(self, item): - '''Supports partial numpy style 2d support. Slice by column [:,1]''' + """Supports partial numpy style 2d support. Slice by column [:,1]""" if isinstance(item, int): return list.__getitem__(self, item) return SimpleList([row[item[1]] for row in self]) + class SimpleList(list): - '''Simple counter''' + """Simple counter""" def sum(self): - '''Sum the vector''' + """Sum the vector""" return sum(self) -class StatsEntry(): - '''An individual stats entry''' + +class StatsEntry: + """An individual stats entry""" + # pylint: disable=unused-argument,no-self-use def __init__(self, stattype, statvalue): @@ -407,140 +409,135 @@ class StatsEntry(): elif stattype == 3: self.function = self.combined elif stattype == 4: - self.function = self.error - elif stattype == 5: self.function = self.name - elif stattype == 7: + elif stattype == 6: self.function = self.symlink else: self.function = self.illegal def illegal(self, stats): - '''Invalid or unknown counter type''' + """Invalid or unknown counter type""" return None def scalar(self, stats): - '''Scalar counter''' + """Scalar counter""" return self.value def simple(self, stats): - '''Simple counter''' + """Simple counter""" counter = StatsSimpleList() - for threads in StatsVector(stats, self.value, 'P'): - clist = [v[0] for v in StatsVector(stats, threads[0], 'Q')] + for threads in StatsVector(stats, self.value, "P"): + clist = [v[0] for v in StatsVector(stats, threads[0], "Q")] counter.append(clist) return counter def combined(self, stats): - '''Combined counter''' + """Combined counter""" counter = StatsCombinedList() - for threads in StatsVector(stats, self.value, 'P'): - clist = [StatsTuple(cnt) for cnt in StatsVector(stats, threads[0], 'QQ')] + for threads in StatsVector(stats, self.value, "P"): + clist = [StatsTuple(cnt) for cnt in StatsVector(stats, threads[0], "QQ")] counter.append(clist) return counter - def error(self, stats): - '''Error counter''' - counter = SimpleList() - for clist in stats.error_vectors: - counter.append(clist[self.value]) - return counter - def name(self, stats): - '''Name counter''' + """Name counter""" counter = [] - for name in StatsVector(stats, self.value, 'P'): + for name in StatsVector(stats, self.value, "P"): if name[0]: counter.append(get_string(stats, name[0])) return counter - SYMLINK_FMT1 = Struct('II') - SYMLINK_FMT2 = Struct('Q') + SYMLINK_FMT1 = Struct("II") + SYMLINK_FMT2 = Struct("Q") + def symlink(self, stats): - '''Symlink counter''' + """Symlink counter""" b = self.SYMLINK_FMT2.pack(self.value) index1, index2 = self.SYMLINK_FMT1.unpack(b) name = stats.directory_by_idx[index1] - return stats[name][:,index2] + return stats[name][:, index2] def get_counter(self, stats): - '''Return a list of counters''' + """Return a list of counters""" if stats: return self.function(stats) + class TestStats(unittest.TestCase): - '''Basic statseg tests''' + """Basic statseg tests""" def setUp(self): - '''Connect to statseg''' + """Connect to statseg""" self.stat = VPPStats() self.stat.connect() self.profile = cProfile.Profile() self.profile.enable() def tearDown(self): - '''Disconnect from statseg''' + """Disconnect from statseg""" self.stat.disconnect() profile = Stats(self.profile) profile.strip_dirs() - profile.sort_stats('cumtime') + profile.sort_stats("cumtime") profile.print_stats() print("\n--->>>") def test_counters(self): - '''Test access to statseg''' - - print('/err/abf-input-ip4/missed', self.stat['/err/abf-input-ip4/missed']) - print('/sys/heartbeat', self.stat['/sys/heartbeat']) - print('/if/names', self.stat['/if/names']) - print('/if/rx-miss', self.stat['/if/rx-miss']) - print('/if/rx-miss', self.stat['/if/rx-miss'][1]) - print('/nat44-ed/out2in/slowpath/drops', self.stat['/nat44-ed/out2in/slowpath/drops']) - print('Set Errors', self.stat.set_errors()) + """Test access to statseg""" + + print("/err/abf-input-ip4/missed", self.stat["/err/abf-input-ip4/missed"]) + print("/sys/heartbeat", self.stat["/sys/heartbeat"]) + print("/if/names", self.stat["/if/names"]) + print("/if/rx-miss", self.stat["/if/rx-miss"]) + print("/if/rx-miss", self.stat["/if/rx-miss"][1]) + print( + "/nat44-ed/out2in/slowpath/drops", + self.stat["/nat44-ed/out2in/slowpath/drops"], + ) with self.assertRaises(KeyError): - print('NO SUCH COUNTER', self.stat['foobar']) - print('/if/rx', self.stat.get_counter('/if/rx')) - print('/err/ethernet-input/no error', - self.stat.get_err_counter('/err/ethernet-input/no error')) + print("NO SUCH COUNTER", self.stat["foobar"]) + print("/if/rx", self.stat.get_counter("/if/rx")) + print( + "/err/ethernet-input/no_error", + self.stat.get_counter("/err/ethernet-input/no_error"), + ) def test_column(self): - '''Test column slicing''' - - print('/if/rx-miss', self.stat['/if/rx-miss']) - print('/if/rx', self.stat['/if/rx']) # All interfaces for thread #1 - print('/if/rx thread #1', self.stat['/if/rx'][0]) # All interfaces for thread #1 - print('/if/rx thread #1, interface #1', - self.stat['/if/rx'][0][1]) # All interfaces for thread #1 - print('/if/rx if_index #1', self.stat['/if/rx'][:, 1]) - print('/if/rx if_index #1 packets', self.stat['/if/rx'][:, 1].packets()) - print('/if/rx if_index #1 packets', self.stat['/if/rx'][:, 1].sum_packets()) - print('/if/rx if_index #1 packets', self.stat['/if/rx'][:, 1].octets()) - print('/if/rx-miss', self.stat['/if/rx-miss']) - print('/if/rx-miss if_index #1 packets', self.stat['/if/rx-miss'][:, 1].sum()) - print('/if/rx if_index #1 packets', self.stat['/if/rx'][0][1]['packets']) - - def test_error(self): - '''Test the error vector''' - - print('/err/ethernet-input', self.stat['/err/ethernet-input/no error']) - print('/err/nat44-ei-ha/pkts-processed', self.stat['/err/nat44-ei-ha/pkts-processed']) - print('/err/ethernet-input', self.stat.get_err_counter('/err/ethernet-input/no error')) - print('/err/ethernet-input', self.stat['/err/ethernet-input/no error'].sum()) + """Test column slicing""" + + print("/if/rx-miss", self.stat["/if/rx-miss"]) + print("/if/rx", self.stat["/if/rx"]) # All interfaces for thread #1 + print( + "/if/rx thread #1", self.stat["/if/rx"][0] + ) # All interfaces for thread #1 + print( + "/if/rx thread #1, interface #1", self.stat["/if/rx"][0][1] + ) # All interfaces for thread #1 + print("/if/rx if_index #1", self.stat["/if/rx"][:, 1]) + print("/if/rx if_index #1 packets", self.stat["/if/rx"][:, 1].packets()) + print("/if/rx if_index #1 packets", self.stat["/if/rx"][:, 1].sum_packets()) + print("/if/rx if_index #1 packets", self.stat["/if/rx"][:, 1].octets()) + print("/if/rx-miss", self.stat["/if/rx-miss"]) + print("/if/rx-miss if_index #1 packets", self.stat["/if/rx-miss"][:, 1].sum()) + print("/if/rx if_index #1 packets", self.stat["/if/rx"][0][1]["packets"]) def test_nat44(self): - '''Test the nat counters''' + """Test the nat counters""" - print('/nat44-ei/ha/del-event-recv', self.stat['/nat44-ei/ha/del-event-recv']) - print('/err/nat44-ei-ha/pkts-processed', self.stat['/err/nat44-ei-ha/pkts-processed'].sum()) + print("/nat44-ei/ha/del-event-recv", self.stat["/nat44-ei/ha/del-event-recv"]) + print( + "/err/nat44-ei-ha/pkts-processed", + self.stat["/err/nat44-ei-ha/pkts-processed"].sum(), + ) def test_legacy(self): - '''Legacy interface''' + """Legacy interface""" directory = self.stat.ls(["^/if", "/err/ip4-input", "/sys/node/ip4-input"]) data = self.stat.dump(directory) print(data) - print('Looking up sys node') + print("Looking up sys node") directory = self.stat.ls(["^/sys/node"]) - print('Dumping sys node') + print("Dumping sys node") data = self.stat.dump(directory) print(data) directory = self.stat.ls(["^/foobar"]) @@ -548,18 +545,19 @@ class TestStats(unittest.TestCase): print(data) def test_sys_nodes(self): - '''Test /sys/nodes''' - counters = self.stat.ls('^/sys/node') - print('COUNTERS:', counters) - print('/sys/node', self.stat.dump(counters)) - print('/net/route/to', self.stat['/net/route/to']) + """Test /sys/nodes""" + counters = self.stat.ls("^/sys/node") + print("COUNTERS:", counters) + print("/sys/node", self.stat.dump(counters)) + print("/net/route/to", self.stat["/net/route/to"]) def test_symlink(self): - '''Symbolic links''' - print('/interface/local0/rx', self.stat['/interfaces/local0/rx']) - print('/sys/nodes/unix-epoll-input', self.stat['/nodes/unix-epoll-input/calls']) + """Symbolic links""" + print("/interface/local0/rx", self.stat["/interfaces/local0/rx"]) + print("/sys/nodes/unix-epoll-input", self.stat["/nodes/unix-epoll-input/calls"]) + -if __name__ == '__main__': +if __name__ == "__main__": import cProfile from pstats import Stats |