operational.py
View Options

	# Copyright 2019 VyOS maintainers and contributors <maintainers@vyos.io>
	#
	# This library is free software; you can redistribute it and/or
	# modify it under the terms of the GNU Lesser General Public
	# License as published by the Free Software Foundation; either
	# version 2.1 of the License, or (at your option) any later version.
	#
	# This library is distributed in the hope that it will be useful,
	# but WITHOUT ANY WARRANTY; without even the implied warranty of
	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
	# Lesser General Public License for more details.
	#
	# You should have received a copy of the GNU Lesser General Public
	# License along with this library. If not, see <http://www.gnu.org/licenses/>.

	import os

	from time import time
	from datetime import datetime
	from functools import reduce
	from tabulate import tabulate

	from vyos.ifconfig import Control

	class Operational(Control):
	"""
	A class able to load Interface statistics
	"""

	cache_magic = 'XYZZYX'

	_stat_names = {
	'rx': ['bytes', 'packets', 'errors', 'dropped', 'overrun', 'mcast'],
	'tx': ['bytes', 'packets', 'errors', 'dropped', 'carrier', 'collisions'],
	}

	_stats_dir = {
	'rx': ['rx_bytes', 'rx_packets', 'rx_errors', 'rx_dropped', 'rx_over_errors', 'multicast'],
	'tx': ['tx_bytes', 'tx_packets', 'tx_errors', 'tx_dropped', 'tx_carrier_errors', 'collisions'],
	}

	# a list made of the content of _stats_dir['rx'] + _stats_dir['tx']
	_stats_all = reduce(lambda x, y: x+y, _stats_dir.values())

	# this is not an interface but will be able to be controlled like one
	_sysfs_get = {
	'oper_state':{
	'location': '/sys/class/net/{ifname}/operstate',
	},
	}


	@classmethod
	def cachefile (cls, ifname):
	# the file where we are saving the counters
	return f'/var/run/vyatta/{ifname}.stats'


	def __init__(self, ifname):
	"""
	Operational provide access to the counters of an interface
	It behave like an interface when it comes to access sysfs

	interface is an instance of the interface for which we want
	to look at (a subclass of Interface, such as EthernetIf)
	"""

	# add a self.config to minic Interface behaviour and make
	# coding similar. Perhaps part of class Interface could be
	# moved into a shared base class.
	self.config = {
	'ifname': ifname,
	'create': False,
	'debug': False,
	}
	super().__init__(**self.config)
	self.ifname = ifname

	# adds all the counters of an interface
	for stat in self._stats_all:
	self._sysfs_get[stat] = {
	'location': '/sys/class/net/{ifname}/statistics/'+stat,
	}

	def get_state(self):
	"""
	Get interface operational state

	Example:
	>>> from vyos.ifconfig import Interface
	>>> Interface('eth0').operational.get_sate()
	'up'
	"""
	# https://www.kernel.org/doc/Documentation/ABI/testing/sysfs-class-net
	# "unknown", "notpresent", "down", "lowerlayerdown", "testing", "dormant", "up"
	return self.get_interface('oper_state')

	@classmethod
	def strtime (cls, epoc):
	"""
	represent an epoc/unix date in the format used by operation commands
	"""
	return datetime.fromtimestamp(epoc).strftime("%a %b %d %R:%S %Z %Y")

	def save_counters(self, stats):
	"""
	record the provided stats to a file keeping vyatta compatibility
	"""

	with open(self.cachefile(self.ifname), 'w') as f:
	f.write(self.cache_magic)
	f.write('\n')
	f.write(str(int(time())))
	f.write('\n')
	for k,v in stats.items():
	if v:
	f.write(f'{k},{v}\n')

	def load_counters(self):
	"""
	load the stats from a file keeping vyatta compatibility
	return a dict() with the value for each interface counter for the cache
	"""
	ifname = self.config['ifname']

	stats = {}
	no_stats = {}
	for name in self._stats_all:
	stats[name] = 0
	no_stats[name] = 0

	try:
	with open(self.cachefile(self.ifname),'r') as f:
	magic = f.readline().strip()
	if magic != self.cache_magic:
	print(f'bad magic {ifname}')
	return no_stats
	stats['timestamp'] = f.readline().strip()
	for line in f:
	k, v = line.split(',')
	stats[k] = int(v)
	return stats
	except IOError:
	return no_stats

	def clear_counters(self, counters=None):
	clear = self._stats_all if counters is None else []
	stats = self.load_counters()
	for counter, value in stats.items():
	stats[counter] = 0 if counter in clear else value
	self.save_counters(stats)

	def reset_counters(self):
	os.remove(self.cachefile(self.ifname))

	def get_stats(self):
	""" return a dict() with the value for each interface counter """
	stats = {}
	for counter in self._stats_all:
	stats[counter] = int(self.get_interface(counter))
	return stats

	def formated_stats(self, indent=4):
	tabs = []
	stats = self.get_stats()
	for rtx in self._stats_dir:
	tabs.append([f'{rtx.upper()}:', ] + [_ for _ in self._stat_names[rtx]])
	tabs.append(['', ] + [stats[_] for _ in self._stats_dir[rtx]])

	s = tabulate(
	tabs,
	stralign="right",
	numalign="right",
	tablefmt="plain"
	)

	p = ' '*indent
	return f'{p}' + s.replace('\n', f'\n{p}')

File Metadata

Mime Type: text/x-script.python
Expires: Sat, Sep 26, 9:06 AM (1 d, 7 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 4278269
Default Alt Text: operational.py (5 KB)

operational.py
No OneTemporary
Actions

operational.py
View Options

File Metadata

Event Timeline

operational.pyNo OneTemporaryActions

operational.pyView Options

File Metadata

Event Timeline

operational.py
No OneTemporary
Actions

operational.py
View Options