How to use vmstat method in autotest

Best Python code snippet using autotest_python

vmstat_collectd.py

Source:vmstat_collectd.py Github

copy

Full Screen

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)...

Full Screen

Full Screen

hosts.py

Source:hosts.py Github

copy

Full Screen

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,...

Full Screen

Full Screen

vmstat_tab.py

Source:vmstat_tab.py Github

copy

Full Screen

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")...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run autotest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful