Skip to content

Instantly share code, notes, and snippets.

@Freaky
Last active Mar 30, 2022
Embed
What would you like to do?
numastat - Python program to monitor FreeBSD NUMA domain memory use
#!/usr/bin/env python3
# BSD 2-Clause License
#
# Copyright (c) 2022, Openoid LLC, on behalf of the r/zfs community
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice, this
# list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
import signal
import time
# We have little interesting to do with these signals - restore default handlers
# so our WIFSIGNALED() state propagates properly to the shell
signal.signal(signal.SIGINT, signal.SIG_DFL)
# Docs warn against this, but we're not writing to anything but stdout/err.
# If we ever need to restore standard Python behaviour, catch BrokenPipeError in
# our print() calls, restore the handler and then re-raise the signal.
signal.signal(signal.SIGPIPE, signal.SIG_DFL)
################################################################################
# Taken from ioztat
class Column:
"""
Formats (translates raw value to string) and optionally justifies (adjusts
the string to fit within a given width) a single column.
"""
def __init__(self, key, heading, width=0, format=str, just=None):
self.key = key
self.heading = heading
self.width = width
self.format = format
self.just = just
def render(self, values):
return self.justify(str(self.format(values[self.key])))
def render_heading(self):
return self.justify(self.heading)
def justify(self, string):
return self.just(string, self.width) if self.just else string
class ColumnGroup:
"""
A simplified column that justifies a string to span the set of Column
objects it is meant to cover. Does not support formatting.
Width here is the size of separators not covered by the raw width of the
spanned columns themselves.
"""
def __init__(self, heading, columns, width=0, just=str.center):
self.heading = heading
self.columns = columns
self.width = width
self.just = just
def group_width(self):
return self.width + sum((c.width for c in self.columns))
def render_heading(self):
return self.justify(self.heading)
def justify(self, string):
return self.just(string, self.group_width()) if self.just else string
class ColumnFormatter:
"""
A buffered formatter for columnar data with optional group headings.
"""
def __init__(self, column_separator=' ', row_separator='-', buffer=None):
self.columns = []
self.column_index = {}
self.groups = None
self.column_separator = column_separator
self.row_separator = row_separator
self.buffer = buffer if buffer is not None else []
def add_column(self, name, cls=Column, **args):
self.column_index[name] = len(self.columns)
self.columns.append(cls(name, **args))
def add_group(self, name='', colspan=1, just=str.center):
"""Add a group heading that spans the previous colspan columns"""
if self.groups is None:
self.groups = ColumnFormatter(column_separator=self.column_separator,
buffer=self.buffer)
self.groups.add_column(name, cls=ColumnGroup, columns=self.columns[-colspan:],
width=len(self.column_separator) * (colspan - 1), just=just)
def set_column_width(self, key, width):
"""Set the width of the column"""
self.columns[self.column_index[key]].width = width
def get_printed_width(self, exclude=()):
"""
Return the printed width of columns and their separators, excluding the
keys in exclude.
"""
return sum((c.width + len(self.column_separator)
for c in self.columns
if c.key not in exclude))
def print_row(self, **data):
"""Add a row of data to be formatted into the internal buffer"""
formatted = (column.render(data) for column in self.columns)
self.print(self.column_separator.join(formatted))
def print_header(self):
"""Add headings and optionally group headings to the internal buffer"""
if self.groups:
self.groups.print_header()
formatted = (column.render_heading() for column in self.columns)
self.print(self.column_separator.join(formatted))
def print_divider(self):
"""
Add a line of column_separator in the space given to each column to the
internal buffer.
"""
self.print(
self.column_separator.join(
[''.ljust(c.width, self.row_separator) for c in self.columns]))
def print(self, string, end="\n"):
"""
Adds the specified string to the internal buffer.
Must eventually be followed by a call to flush()
"""
self.buffer.append(str(string) + end)
def flush(self, lines=None):
"""Print the internal buffer, up to the optional lines limit."""
buffer = ''.join(self.buffer)
if lines:
buffer = "\n".join(buffer.split("\n", lines)[:lines])
print(buffer, end='')
self.buffer.clear()
class NumberToHuman:
"""Number formatting functions"""
SIZE_PREFIX = ('', 'K', 'M', 'G', 'T', 'P', 'E')
FORMATS = ('{:.2f}{}', '{:.1f}{}', '{:.0f}{}')
@classmethod
def formatter(cls, length, decimal=False):
"""
Return a function that will format a number to a shortened SI decimal or
binary form that fits within a given length.
"""
divisor = 1000 if decimal else 1024
formats = cls.FORMATS
size_prefix = cls.SIZE_PREFIX
size_prefix_max = len(size_prefix) - 1
powers = [pow(divisor, i) for i in range(len(size_prefix))]
def fmt(num):
n = num
index = 0
while n >= divisor and index < size_prefix_max:
n /= divisor
index += 1
u = size_prefix[index]
if index == 0 or num % powers[index] == 0:
return str(int(n)) + u
for fmt in formats:
ret = fmt.format(n, u)
if len(ret) <= length:
return ret
return ret
return fmt
################################################################################
try:
import sysctl
def fetch_sysctl(oids):
return (value for oid in oids for value in sysctl.filter(oid))
except ImportError:
import subprocess
from collections import namedtuple
SysctlValue = namedtuple('SysctlValue', ['name', 'value'])
def fetch_sysctl(oids):
r = subprocess.run(['/sbin/sysctl', '-e', '--', *oids],
capture_output=True, check=False, encoding='latin-1')
stats = (line.split("=", 2) for line in r.stdout.split("\n"))
return (SysctlValue(*nv) for nv in stats if len(nv) == 2)
from collections import defaultdict
def remove_prefix(text, prefix):
return text[len(prefix):] if text.startswith(prefix) else text
def fetch_domains():
domains = defaultdict(dict)
for ctl in fetch_sysctl(['vm.domain.']):
name = remove_prefix(ctl.name, 'vm.domain.').split('.')
if len(name) == 3:
domain, kind, stat = name
domains[int(domain)]['.'.join((kind, stat))] = ctl.value
return domains
def create_column_formatter():
field_width = 8
bytes_format = NumberToHuman.formatter(length=field_width, decimal=False)
pages_format = lambda x: bytes_format(4096 * int(x))
page = {'width': field_width, 'just': str.rjust, 'format': pages_format}
formatter= ColumnFormatter()
formatter.add_column('domain', heading='DOMAIN', width=6, just=str.rjust)
formatter.add_column('stats.active', heading='ACTIVE', **page)
formatter.add_column('stats.inactive', heading='INACTIVE', **page)
formatter.add_column('stats.laundry', heading='LAUNDRY', **page)
formatter.add_column('stats.free_count', heading='FREE', **page)
return formatter
def interval(seconds):
deadline = time.monotonic() + seconds
while True:
yield
sleep = deadline - time.monotonic()
if sleep > 0:
time.sleep(sleep)
deadline += seconds
else:
deadline = time.monotonic() + seconds
def main():
formatter = create_column_formatter()
formatter.print_header()
for domains in (fetch_domains() for _ in interval(1.0)):
formatter.print_divider()
for domain, stats in sorted(domains.items()):
formatter.print_row(domain=domain, **stats)
formatter.flush()
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment