KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use vmstat method in autotest

Best Python code snippet using autotest_python

vmstat_collectd.py

Source:vmstat_collectd.py

1#!/usr/bin/env python2# coding=utf-83#4# vmstat-collectd5# ===============6# vmstat-collectd is a python based plugin for collectd that reports metrics7# collected from the unix command vmstat8# https://github.com/signalfx/vmstat-collectd9#10# This project is based on the SignalFx maintained fork of 11# collectd-iostat-python project.12# 13# https://github.com/signalfx/collectd-iostat-python14# 15# Information about the original collectd-iostat-python project: 16# -------------------------------------------------------------17# Collectd-iostat-python is an iostat plugin for collectd that allows you to18# graph Linux iostat metrics in Graphite or other output formats that are19# supported by collectd.20#21# https://github.com/powdahound/redis-collectd-plugin22#   - was used as template23# https://github.com/keirans/collectd-iostat/24#   - was used as inspiration and contains some code from25# https://bitbucket.org/jakamkon/python-iostat26#   - by Kuba KoÅczyk <jakamkon at users.sourceforge.net>27#28# This fork is maintained by SignalFx <support@signalfx.com>29import signal30import string31import subprocess32import sys33__version__ = '0.0.1'34__author__ = 'SignalFx'35# Fork maintained by SignalFx36__maintainer__ = 'SignalFx'37__email__ = 'support@signalfx.com'38class VMStatError(Exception):39    pass40class CmdError(VMStatError):41    pass42class ParseError(VMStatError):43    pass44class VMStat(object):45    def __init__(self, path='/usr/bin/vmstat', interval=2, count=2):46        self.path = path47        self.interval = interval48        self.count = count49    def parse_vmstats(self, input):50        """51        Parse vmstat.52        By default parse statistics for all avaliable devices.53        @type input: C{string}54        @param input: vmstat output55        @return: C{dictionary} contains per block device statistics.56        Statistics are in form of C{dictonary}.57        """58        vmstats = {}59        # Find the line with the column headers60        hdr_index = input.rfind('r  b')61        if hdr_index == -1:62            raise ParseError('Unknown input format: %r' % input)63        64        # data string of headers and values65        ds = input[hdr_index:].splitlines()66        # pop off the header string and store67        hdr = ds.pop(0).split()68        # For data point in data string69        for stats in ds:70            # If string of data points71            if stats:72                # split the data points73                stats = stats.split()74                vmstats = dict([(k, float(v)) for k, v in zip(hdr, stats)])75        return vmstats76    def _run(self, options=None):77        """78        Run vmstat command.79        """80        close_fds = 'posix' in sys.builtin_module_names81        args = '%s %s %s %s' % (82            self.path,83            ''.join(options),84            self.interval,85            self.count)86        return subprocess.Popen(87            args,88            bufsize=1,89            shell=True,90            stdout=subprocess.PIPE,91            close_fds=close_fds)92    @staticmethod93    def _get_childs_data(child):94        """95        Return child's data when avaliable.96        """97        (stdout, stderr) = child.communicate()98        ecode = child.poll()99        if ecode != 0:100            raise CmdError('Command %r returned %d' % (child.cmd, ecode))101        return stdout102    def get_vmstats(self):103        """104        Get all avaliable vmstats that we can get.105        """106        dstats = self._run(options=['-n -S K'])107        extdstats = self._run(options=['-an -S K'])108        dsd = self._get_childs_data(dstats)109        edd = self._get_childs_data(extdstats)110        ds = self.parse_vmstats(dsd)111        eds = self.parse_vmstats(edd)112        113        if len(eds) > 0:114            ds.update(eds)115        return ds116class VMMon(object):117    def __init__(self):118        self.plugin_name = 'vmstat'119        self.vmstat_path = '/usr/bin/vmstat'120        self.interval = 10.0121        self.vmstat_interval = 1122        self.vmstat_count = 2123        self.vmstat_nice_names = False124        self.verbose_logging = False125        self.include = set([])126        self.names = {127            'r': {'t': 'vmstat_process', 'ti': 'waiting'},128            'b': {'t': 'vmstat_process', 'ti': 'uninterruptible_sleep'},129            'swpd': {'t': 'vmstat_memory', 'ti': 'swap'},130            'free': {'t': 'vmstat_memory', 'ti': 'free'},131            'buff': {'t': 'vmstat_memory', 'ti': 'buffer'},132            'cache': {'t': 'vmstat_memory', 'ti': 'cache'},133            'inact': {'t': 'vmstat_memory', 'ti': 'inactive'},134            'active': {'t': 'vmstat_memory', 'ti': 'active'},135            'si': {'t': 'vmstat_swap', 'ti': 'in_per_second'},136            'so': {'t': 'vmstat_swap', 'ti': 'out_per_second'},137            'bi': {'t': 'vmstat_blocks', 'ti': 'received_per_second'},138            'bo': {'t': 'vmstat_blocks', 'ti': 'sent_per_second'},139            'in': {'t': 'vmstat_system', 'ti': 'interrupts_per_second'},140            'cs': {'t': 'vmstat_system', 'ti': 'context_switches_per_second'},141            'us': {'t': 'vmstat_cpu', 'ti': 'user'},142            'sy': {'t': 'vmstat_cpu', 'ti': 'system'},143            'id': {'t': 'vmstat_cpu', 'ti': 'idle'},144            'wa': {'t': 'vmstat_cpu', 'ti': 'wait'},145            'st': {'t': 'vmstat_cpu', 'ti': 'stolen'}146        }147    def log_verbose(self, msg):148        if not self.verbose_logging:149            return150        collectd.info('%s plugin [verbose]: %s' % (self.plugin_name, msg))151    def configure_callback(self, conf):152        """153        Receive configuration block154        """155        for node in conf.children:156            val = str(node.values[0])157            if node.key == 'Path':158                self.vmstat_path = val159            elif node.key == 'Interval':160                self.interval = float(val)161            elif node.key == 'VMStatInterval':162                self.vmstat_interval = int(float(val))163            elif node.key == 'Count':164                self.vmstat_count = int(float(val))165            elif node.key == 'NiceNames':166                self.vmstat_nice_names = val in ['True', 'true']167            elif node.key == 'PluginName':168                self.plugin_name = val169            elif node.key == 'Verbose':170                self.verbose_logging = val in ['True', 'true']171            elif node.key == 'Include':172                self.include.update(node.values)173            else:174                collectd.warning(175                    '%s plugin: Unknown config key: %s.' % (176                        self.plugin_name,177                        node.key))178        self.log_verbose(179            'Configured with vmstat=%s, interval=%s, count=%s, ' % (180                self.vmstat_path,181                self.vmstat_interval,182                self.vmstat_count))183        if len(self.include) != 0:184            self.log_verbose('Report only the following metrics: {0}'185                             .format(self.include))186        collectd.register_read(self.read_callback, self.interval)187    def dispatch_value(self, val_type, type_instance, value):188        """189        Dispatch a value to collectd190        """191        plugin_instance = "vmstat"192        self.log_verbose(193            'Sending value: %s-%s.%s=%s' % (194                self.plugin_name,195                plugin_instance,196                '-'.join([val_type, type_instance]),197                value))198        val = collectd.Values()199        val.plugin = self.plugin_name200        val.plugin_instance = plugin_instance201        val.type = val_type202        if len(type_instance):203            val.type_instance = type_instance204        val.values = [value, ]205        val.meta = {'0': True}206        val.dispatch()207    def read_callback(self):208        """209        Collectd read callback210        """211        self.log_verbose('Read callback called')212        vmstat = VMStat(213            path=self.vmstat_path,214            interval=self.vmstat_interval)215        ds = vmstat.get_vmstats()216        if not ds:217            self.log_verbose('%s plugin: No info received.' % self.plugin_name)218            return219        for name in ds:220            if self.vmstat_nice_names and name in self.names:221                val_type = self.names[name]['t']222                if 'ti' in self.names[name]:223                    type_instance = self.names[name]['ti']224                else:225                    type_instance = ''226                value = ds[name]227                if 'm' in self.names[name]:228                    value *= self.names[name]['m']229            else:230                val_type = 'gauge'231                type_instance = name232                value = ds[name]233            if len(self.include) == 0 \234                or (self.vmstat_nice_names is True235                    and ('.'.join([val_type, type_instance]) in self.include236                        or val_type in self.include)) \237                or (self.vmstat_nice_names is False and238                    type_instance in self.include):239                self.dispatch_value(val_type, type_instance, value)240def restore_sigchld():241    """242    Restore SIGCHLD handler for python <= v2.6243    It will BREAK exec plugin!!!244    See https://github.com/deniszh/collectd-iostat-python/issues/2 for details245    """246    if sys.version_info[0] == 2 and sys.version_info[1] <= 6:247        signal.signal(signal.SIGCHLD, signal.SIG_DFL)248if __name__ == '__main__':249    vmstat = VMStat()250    ds = vmstat.get_vmstats()251    for metric in ds:252        print("%s:%s" % (metric, ds[metric]))253    sys.exit(0)254else:255    import collectd256    vmmon = VMMon()257    # Register callbacks258    collectd.register_init(restore_sigchld)...

hosts.py

Source:hosts.py

1import dg.conn2import dg.xtable3def xt_vmstat(conn, nsec=5):4    if nsec < 2 or nsec > 60:5        raise ValueError("A reasonable nsec values should be in [2, 60]")6    7    sql = """8select 9--10-- vmstat output columns, whatever does that mean11-- r b swpd free buff cache si so bi bo in cs us sy id wa st12-- 13dg_utils.transducer_column_int4(1) as vmstat_segid,14dg_utils.transducer_column_int4(2) as vmstat_sec, 15dg_utils.transducer_column_int8(3) as vmstat_r, 16dg_utils.transducer_column_int8(4) as vmstat_b, 17dg_utils.transducer_column_int8(5) as vmstat_swpd, 18dg_utils.transducer_column_int8(6) as vmstat_free, 19dg_utils.transducer_column_int8(7) as vmstat_buff, 20dg_utils.transducer_column_int8(8) as vmstat_cache, 21dg_utils.transducer_column_int8(9) as vmstat_si, 22dg_utils.transducer_column_int8(10) as vmstat_so, 23dg_utils.transducer_column_int8(11) as vmstat_bi, 24dg_utils.transducer_column_int8(12) as vmstat_bo, 25dg_utils.transducer_column_int8(13) as vmstat_in, 26dg_utils.transducer_column_int8(14) as vmstat_cs, 27dg_utils.transducer_column_int8(15) as vmstat_us, 28dg_utils.transducer_column_int8(16) as vmstat_sy, 29dg_utils.transducer_column_int8(17) as vmstat_id, 30dg_utils.transducer_column_int8(18) as vmstat_wa, 31dg_utils.transducer_column_int8(19) as vmstat_st, 32--33-- transducer function call, as UDF.34--35dg_utils.transducer($PHI$PhiExec python236## A valid python program below37import vitessedata.phi38import subprocess39## Python declare input/output types.40vitessedata.phi.DeclareTypes('''41//42// BEGIN INPUT TYPES43// segid int3244// nsec int3245// END INPUT TYPES46//47// BEGIN OUTPUT TYPES48// vmstat_segid int3249// vmstat_sec int3250// vmstat_r int6451// vmstat_b int6452// vmstat_swpd int6453// vmstat_free int6454// vmstat_buff int6455// vmstat_cache int6456// vmstat_si int6457// vmstat_so int6458// vmstat_bi int6459// vmstat_bo int6460// vmstat_in int6461// vmstat_cs int6462// vmstat_us int6463// vmstat_sy int6464// vmstat_id int6465// vmstat_wa int6466// vmstat_st int6467// END OUTPUT TYPES68//69''')70def do_x():71    segid = 072    nsec = 073    cnt = 074    while True: 75        rec = vitessedata.phi.NextInput()76        if not rec:77            break78        segid = rec[0]79        nsec = rec[1]80    if nsec <= 0:81        raise ValueError("Must supply a good nsec value")82    output = subprocess.Popen(["/usr/bin/vmstat", "1", str(nsec)], 83        stdout=subprocess.PIPE).communicate()[0]84    lines = output.split("\\n")85    for line in lines[2:]:86        if line.strip() == '':87            continue88        fields = line.split()89        outrec = [segid, cnt]90        cnt += 191        for field in fields:92            outrec.append(int(field))93        ## Output94        vitessedata.phi.WriteOutput(outrec)95    ## Write None to signal end of output96    vitessedata.phi.WriteOutput(None)97if __name__ == '__main__':98    do_x()99$PHI$), 100-- 101-- Input to Transducer102-- 103t.*104--105-- From, any SQL table/view subquery106-- 107from (108    select gp_segment_id::int4, {0}::int4 from dg_utils.eachseg109) t 110""".format(nsec)111    return dg.xtable.fromQuery(conn, sql)112def xt_segs(conn):113    return dg.xtable.fromQuery(conn, "select * from gp_segment_configuration")114    115def xt_vmstat_hosts(conn, nsec=5):116    vt = xt_vmstat(conn, nsec)117    segs = xt_segs(conn)118    vh = dg.xtable.fromQuery(conn, """select #1.address# as addr, #0.vmstat_sec# as n, 119        avg(#0.vmstat_r#) as vmstat_r,120        avg(#0.vmstat_b#) as vmstat_b,121        avg(#0.vmstat_swpd#) as vmstat_swpd,122        avg(#0.vmstat_free#) as vmstat_free,123        avg(#0.vmstat_buff#) as vmstat_buff,124        avg(#0.vmstat_cache#) as vmstat_cache,125        avg(#0.vmstat_si#) as vmstat_si,126        avg(#0.vmstat_so#) as vmstat_so,127        avg(#0.vmstat_bi#) as vmstat_bi,128        avg(#0.vmstat_bo#) as vmstat_bo,129        avg(#0.vmstat_in#) as vmstat_in,130        avg(#0.vmstat_cs#) as vmstat_cs,...

vmstat_tab.py

Source:vmstat_tab.py

1# pandas and numpy for data manipulation2import pandas as pd3import numpy as np4import sqlite35from bokeh.plotting import Figure6from bokeh.models import (7    CategoricalColorMapper,8    HoverTool,9    ColumnDataSource,10    Panel,11    FuncTickFormatter,12    SingleIntervalTicker,13    LinearAxis,14    Legend,15)16from bokeh.models.widgets import (17    CheckboxGroup,18    Slider,19    RangeSlider,20    Tabs,21    CheckboxButtonGroup,22    TableColumn,23    DataTable,24    Select,25)26from bokeh.layouts import column, row, WidgetBox27import matplotlib28matplotlib.use("Agg")29import matplotlib.pyplot as plt30import matplotlib.colors as colors31def vmstat_tab(db):32    def make_dataset(vmstat_list):33        newdf = vmstat[vmstat_list]34        # Convert dataframe to column data source35        return ColumnDataSource(newdf)36    def make_plot(src):37        # Blank plot with correct labels38        p = Figure(39            plot_width=1024,40            plot_height=768,41            x_axis_type="datetime",42            title="vmstat",43            output_backend="webgl",44        )45        cm = plt.get_cmap("gist_rainbow")46        numlines = len(vmstat.columns)47        mypal = [cm(1.0 * i / numlines) for i in range(numlines)]48        mypal = list(map(lambda x: colors.rgb2hex(x), mypal))49        col = 050        legenditems = []51        for key in src.data.keys():52            if key == "datetime":53                continue54            l = key + " "55            col = col + 156            cline = p.line(57                vmstat.index.values,58                vmstat[key],59                line_width=1,60                alpha=0.8,61                color=mypal[col],62            )63            legenditems += [(key, [cline])]64        p.legend.click_policy = "hide"65        legend = Legend(items=legenditems, location=(0, -30))66        p.add_layout(legend, "right")67        return p68    def update(attr, old, new):69        vmstats_to_plot = [vmstat_selection.labels[i] for i in vmstat_selection.active]70        new_src = make_dataset(vmstats_to_plot)71        plot = make_plot(new_src)72        layout.children[1] = plot73    # get data from DB, setup index74    cur = db.cursor()75    cur.execute(76        "SELECT name FROM sqlite_master WHERE type='table' AND name=?", ["vmstat"]77    )78    if len(cur.fetchall()) == 0:79        return None80    vmstat = pd.read_sql_query("select * from vmstat", db)81    vmstat.index = pd.to_datetime(vmstat["datetime"])82    vmstat = vmstat.drop(["datetime"], axis=1)83    vmstat.index.name = "datetime"84    vmstat_selection = CheckboxGroup(labels=list(vmstat.columns), active=[0, 5])85    vmstat_list = [vmstat_selection.labels[i] for i in vmstat_selection.active]86    src = make_dataset(vmstat_list)87    vmstat_selection.on_change("active", update)88    plot = make_plot(src)89    controls = WidgetBox(vmstat_selection)90    layout = row(controls, plot)91    tab = Panel(child=layout, title="vmstat")...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.