def check(request):
return {
'hostname': socket.gethostname(),
'ips': ips,
'cpus': psutil.cpu_count(),
'uptime': timesince(datetime.fromtimestamp(psutil.boot_time())),
'memory': {
'total': filesizeformat(psutil.virtual_memory().total),
'available': filesizeformat(psutil.virtual_memory().available),
'used': filesizeformat(psutil.virtual_memory().used),
'free': filesizeformat(psutil.virtual_memory().free),
'percent': psutil.virtual_memory().percent
},
'swap': {
'total': filesizeformat(psutil.swap_memory().total),
'used': filesizeformat(psutil.swap_memory().used),
'free': filesizeformat(psutil.swap_memory().free),
'percent': psutil.swap_memory().percent
}
}
python类swap_memory()的实例源码
def test_serialization(self):
def check(ret):
if json is not None:
json.loads(json.dumps(ret))
a = pickle.dumps(ret)
b = pickle.loads(a)
self.assertEqual(ret, b)
check(psutil.Process().as_dict())
check(psutil.virtual_memory())
check(psutil.swap_memory())
check(psutil.cpu_times())
check(psutil.cpu_times_percent(interval=0))
check(psutil.net_io_counters())
if LINUX and not os.path.exists('/proc/diskstats'):
pass
else:
if not APPVEYOR:
check(psutil.disk_io_counters())
check(psutil.disk_partitions())
check(psutil.disk_usage(os.getcwd()))
check(psutil.users())
def test_swap_memory(self):
out = sh('env PATH=/usr/sbin:/sbin:%s swap -l' % os.environ['PATH'])
lines = out.strip().split('\n')[1:]
if not lines:
raise ValueError('no swap device(s) configured')
total = free = 0
for line in lines:
line = line.split()
t, f = line[-2:]
total += int(int(t) * 512)
free += int(int(f) * 512)
used = total - free
psutil_swap = psutil.swap_memory()
self.assertEqual(psutil_swap.total, total)
self.assertEqual(psutil_swap.used, used)
self.assertEqual(psutil_swap.free, free)
def get_memory():
memory = dict()
virtual = psutil.virtual_memory()
swap = psutil.swap_memory()
memory['virtual_total'] = '%.2fMB' % (virtual.total/1024/1024)
memory['virtual_available'] = '%.2fMB' % (virtual.available/1024/1024)
memory['virtual_percent'] = str(virtual.percent)+'%'
memory['virtual_used'] = '%.2fMB' % (virtual.used/1024/1024)
memory['virtual_free'] = '%.2fMB' % (virtual.free/1024/1024)
memory['virtual_active'] = '%.2fMB' % (virtual.active/1024/1024)
memory['virtual_inactive'] = '%.2fMB' % (virtual.inactive/1024/1024)
memory['virtual_buffers'] = '%.2fMB' % (virtual.buffers/1024/1024)
memory['virtual_cached'] = '%.2fMB' % (virtual.cached/1024/1024)
memory['virtual_shared'] = '%.2fMB' % (virtual.shared/1024/1024)
memory['swap_total'] = '%.2fMB' % (swap.total/1024/1024)
memory['swap_used'] = '%.2fMB' % (swap.used/1024/1024)
memory['swap_free'] = '%.2fMB' % (swap.free/1024/1024)
memory['swap_percent'] = str(swap.percent)+'%'
memory['swap_sin'] = '%.2fMB' % (swap.sin/1024/1024)
memory['swap_sout'] = '%.2fMB' % (swap.sout/1024/1024)
return memory
def _get_memory_stats(self):
mem_info = {'virtual': {}, 'swap': {}}
v_mem = ps.virtual_memory()
s_mem = ps.swap_memory()
_c = mem_info['virtual']
_c['total'] = v_mem.total
_c['used'] = v_mem.used
_c['free'] = v_mem.free
_c['used_percent'] = v_mem.percent
_c = mem_info['swap']
_c['total'] = s_mem.total
_c['used'] = s_mem.used
_c['free'] = s_mem.free
_c['used_percent'] = s_mem.percent
return mem_info
def memory():
c = statsd.StatsClient(STATSD_HOST, 8125, prefix=PREFIX + 'system.memory')
while True:
swap = psutil.swap_memory()
c.gauge('swap.total', swap.total)
c.gauge('swap.used', swap.used)
c.gauge('swap.free', swap.free)
c.gauge('swap.percent', swap.percent)
virtual = psutil.virtual_memory()
c.gauge('virtual.total', virtual.total)
c.gauge('virtual.available', virtual.available)
c.gauge('virtual.used', virtual.used)
c.gauge('virtual.free', virtual.free)
c.gauge('virtual.percent', virtual.percent)
c.gauge('virtual.active', virtual.active)
c.gauge('virtual.inactive', virtual.inactive)
c.gauge('virtual.buffers', virtual.buffers)
c.gauge('virtual.cached', virtual.cached)
time.sleep(GRANULARITY)
def test_swapmem_sout(self):
vmstat_val = vm_stat("Pageout")
psutil_val = psutil.swap_memory().sout
self.assertEqual(psutil_val, vmstat_val)
# Not very reliable.
# def test_swapmem_total(self):
# out = sh('sysctl vm.swapusage')
# out = out.replace('vm.swapusage: ', '')
# total, used, free = re.findall('\d+.\d+\w', out)
# psutil_smem = psutil.swap_memory()
# self.assertEqual(psutil_smem.total, human2bytes(total))
# self.assertEqual(psutil_smem.used, human2bytes(used))
# self.assertEqual(psutil_smem.free, human2bytes(free))
# --- network
def test_serialization(self):
def check(ret):
if json is not None:
json.loads(json.dumps(ret))
a = pickle.dumps(ret)
b = pickle.loads(a)
self.assertEqual(ret, b)
check(psutil.Process().as_dict())
check(psutil.virtual_memory())
check(psutil.swap_memory())
check(psutil.cpu_times())
check(psutil.cpu_times_percent(interval=0))
check(psutil.net_io_counters())
if LINUX and not os.path.exists('/proc/diskstats'):
pass
else:
if not APPVEYOR:
check(psutil.disk_io_counters())
check(psutil.disk_partitions())
check(psutil.disk_usage(os.getcwd()))
check(psutil.users())
def test_swap_memory(self):
out = sh('env PATH=/usr/sbin:/sbin:%s swap -l' % os.environ['PATH'])
lines = out.strip().split('\n')[1:]
if not lines:
raise ValueError('no swap device(s) configured')
total = free = 0
for line in lines:
line = line.split()
t, f = line[-2:]
total += int(int(t) * 512)
free += int(int(f) * 512)
used = total - free
psutil_swap = psutil.swap_memory()
self.assertEqual(psutil_swap.total, total)
self.assertEqual(psutil_swap.used, used)
self.assertEqual(psutil_swap.free, free)
def test_swapmem_sout(self):
vmstat_val = vm_stat("Pageout")
psutil_val = psutil.swap_memory().sout
self.assertEqual(psutil_val, vmstat_val)
# Not very reliable.
# def test_swapmem_total(self):
# out = sh('sysctl vm.swapusage')
# out = out.replace('vm.swapusage: ', '')
# total, used, free = re.findall('\d+.\d+\w', out)
# psutil_smem = psutil.swap_memory()
# self.assertEqual(psutil_smem.total, human2bytes(total))
# self.assertEqual(psutil_smem.used, human2bytes(used))
# self.assertEqual(psutil_smem.free, human2bytes(free))
# --- network
def test_serialization(self):
def check(ret):
if json is not None:
json.loads(json.dumps(ret))
a = pickle.dumps(ret)
b = pickle.loads(a)
self.assertEqual(ret, b)
check(psutil.Process().as_dict())
check(psutil.virtual_memory())
check(psutil.swap_memory())
check(psutil.cpu_times())
check(psutil.cpu_times_percent(interval=0))
check(psutil.net_io_counters())
if LINUX and not os.path.exists('/proc/diskstats'):
pass
else:
if not APPVEYOR:
check(psutil.disk_io_counters())
check(psutil.disk_partitions())
check(psutil.disk_usage(os.getcwd()))
check(psutil.users())
def test_swap_memory(self):
out = sh('env PATH=/usr/sbin:/sbin:%s swap -l' % os.environ['PATH'])
lines = out.strip().split('\n')[1:]
if not lines:
raise ValueError('no swap device(s) configured')
total = free = 0
for line in lines:
line = line.split()
t, f = line[-2:]
total += int(int(t) * 512)
free += int(int(f) * 512)
used = total - free
psutil_swap = psutil.swap_memory()
self.assertEqual(psutil_swap.total, total)
self.assertEqual(psutil_swap.used, used)
self.assertEqual(psutil_swap.free, free)
def main():
virt = psutil.virtual_memory()
swap = psutil.swap_memory()
templ = "%-7s %10s %10s %10s %10s %10s %10s"
print(templ % ('', 'total', 'used', 'free', 'shared', 'buffers', 'cache'))
print(templ % (
'Mem:',
int(virt.total / 1024),
int(virt.used / 1024),
int(virt.free / 1024),
int(getattr(virt, 'shared', 0) / 1024),
int(getattr(virt, 'buffers', 0) / 1024),
int(getattr(virt, 'cached', 0) / 1024)))
print(templ % (
'Swap:', int(swap.total / 1024),
int(swap.used / 1024),
int(swap.free / 1024),
'',
'',
''))
def get_memo_usage_available():
psy_mem = psutil.virtual_memory()
vrt_mem = psutil.swap_memory()
return psy_mem.available, vrt_mem.free + psy_mem.available
def get_init_resource():
cpu = cpu_count() * 100
psy_mem = psutil.virtual_memory()
vrt_mem = psutil.swap_memory()
disk = list(psutil.disk_usage(get_config("env", "workspace")))[0]
return cpu, psy_mem.total, disk, vrt_mem.total + psy_mem.total
def get_total_swap(self):
return psutil.swap_memory()[0]
def swapinfo(self):
mem = psutil.swap_memory()
values = {'total': mem.total/1024/1024,
'used': mem.used/1024/1024,
'free': mem.free/1024/1024,
'sin': mem.sin/1024/1024,
'sout': mem.sout/1024/1024,
}
return values
def memory(self):
"""
System virtual memory and swap usage
Converting byte to mb and rounded result
Return dict
"""
mem = psutil.virtual_memory()
mem_total = self.hr(mem.total)
mem_used = self.hr(mem.used)
mem_free = self.hr(mem.free)
mem_percent = mem.percent
swap_mem = psutil.swap_memory()
swap_total = self.hr(swap_mem.total)
swap_used = self.hr(swap_mem.used)
swap_free = self.hr(swap_mem.free)
swap_percent = swap_mem.percent
mem = {
'virtual': {
'total': mem_total,
'used': mem_used,
'free': mem_free,
'percent': mem_percent
},
'swap': {
'total': swap_total,
'used': swap_used,
'free': swap_free,
'percent': swap_percent
}
}
return mem
def test_swap_memory(self):
self.execute(psutil.swap_memory)
def test_total(self):
free_value = free_swap().total
psutil_value = psutil.swap_memory().total
return self.assertAlmostEqual(
free_value, psutil_value, delta=MEMORY_TOLERANCE)
def test_used(self):
free_value = free_swap().used
psutil_value = psutil.swap_memory().used
return self.assertAlmostEqual(
free_value, psutil_value, delta=MEMORY_TOLERANCE)
def test_free(self):
free_value = free_swap().free
psutil_value = psutil.swap_memory().free
return self.assertAlmostEqual(
free_value, psutil_value, delta=MEMORY_TOLERANCE)
def test_warnings_mocked(self):
with mock.patch('psutil._pslinux.open', create=True) as m:
with warnings.catch_warnings(record=True) as ws:
warnings.simplefilter("always")
ret = psutil.swap_memory()
assert m.called
self.assertEqual(len(ws), 1)
w = ws[0]
self.assertTrue(w.filename.endswith('psutil/_pslinux.py'))
self.assertIn(
"'sin' and 'sout' swap memory stats couldn't "
"be determined", str(w.message))
self.assertEqual(ret.sin, 0)
self.assertEqual(ret.sout, 0)
def test_swapmem_sin(self):
vmstat_val = vm_stat("Pageins")
psutil_val = psutil.swap_memory().sin
self.assertEqual(psutil_val, vmstat_val)
def test_swapmem_sout(self):
vmstat_val = vm_stat("Pageout")
psutil_val = psutil.swap_memory().sout
self.assertEqual(psutil_val, vmstat_val)
# Not very reliable.
# def test_swapmem_total(self):
# out = sh('sysctl vm.swapusage')
# out = out.replace('vm.swapusage: ', '')
# total, used, free = re.findall('\d+.\d+\w', out)
# psutil_smem = psutil.swap_memory()
# self.assertEqual(psutil_smem.total, human2bytes(total))
# self.assertEqual(psutil_smem.used, human2bytes(used))
# self.assertEqual(psutil_smem.free, human2bytes(free))
def test_swap_memory(self):
mem = psutil.swap_memory()
assert mem.total >= 0, mem
assert mem.used >= 0, mem
if mem.total > 0:
# likely a system with no swap partition
assert mem.free > 0, mem
else:
assert mem.free == 0, mem
assert 0 <= mem.percent <= 100, mem
assert mem.sin >= 0, mem
assert mem.sout >= 0, mem
def test_swapmem_free(self):
self.assertAlmostEqual(
psutil.swap_memory().free, self.parse_meminfo("SwapFree:"),
delta=MEMORY_TOLERANCE)
def test_swapmem_used(self):
smem = psutil.swap_memory()
self.assertEqual(smem.used, smem.total - smem.free)
def getMemoryInfo(self):
"""Get a dict containing the memory info
Returned dict example::
{
'used': 377036800,
'total': 903979008,
'buffers': 129654784,
'cached': 135168000,
'processes': 112214016,
'free': 526942208,
'swap': {
'used': 0,
'free': 104853504,
'total': 104853504
}
}
"""
memory = {}
try:
vmem = psutil.virtual_memory()
memory['total'] = vmem.total
memory['free'] = vmem.available
memory['used'] = memory['total'] - memory['free']
memory['buffers'] = vmem.buffers
memory['cached'] = vmem.cached
memory['processes'] = memory['used']
swap = psutil.swap_memory()
memory['swap'] = {}
memory['swap']['total'] = swap.total
memory['swap']['free'] = swap.free
memory['swap']['used'] = swap.used
except Exception as e:
exception('Error getting memory info')
return memory
def run(self, *unused):
swap = {}
mem = psutil.swap_memory()
for name in mem._fields:
swap[name] = getattr(mem, name)
return swap
