1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
|
# 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}')
|
