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
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
|
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# Copyright (c) 2017 Cisco and/or its affiliates.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import logging
import shlex
import socket
import subprocess
import threading
from netmodel.network.interface import Interface as BaseInterface
from netmodel.network.packet import Packet
from netmodel.network.prefix import Prefix
from netmodel.model.attribute import Attribute
from netmodel.model.filter import Filter
from netmodel.model.object import Object
from netmodel.model.query import Query, ACTION_UPDATE, ACTION2STR
from netmodel.model.query import ACTION_SUBSCRIBE, ACTION_UNSUBSCRIBE
from netmodel.model.query import FUNCTION_SUM
from netmodel.model.type import String, Integer, Double
log = logging.getLogger(__name__)
DEFAULT_INTERVAL = 1 # s
KEY_FIELD = 'device_name'
class Interface(Object):
__type__ = 'interface'
node = Attribute(String)
device_name = Attribute(String)
bw_upstream = Attribute(Double) # bytes
bw_downstream = Attribute(Double) # bytes
class Process(threading.Thread):
pass
class BWMThread(Process):
SEP=';'
CMD="stdbuf -oL bwm-ng -t 1000 -N -o csv -c 0 -C '%s'"
# Parsing information (from README, specs section)
# https://github.com/jgjl/bwm-ng/blob/master/README
#
# Type rate:
FIELDS_RATE = ['timestamp', 'iface_name', 'bytes_out_s', 'bytes_in_s',
'bytes_total_s', 'bytes_in', 'bytes_out', 'packets_out_s',
'packets_in_s', 'packets_total_s', 'packets_in', 'packets_out',
'errors_out_s', 'errors_in_s', 'errors_in', 'errors_out']
# Type svg, sum, max
FIELDS_SUM = ['timestamp', 'iface_name', 'bytes_out', 'bytes_in',
'bytes_total', 'packets_out', 'packets_in', 'packets_total',
'errors_out', 'errors_in']
def __init__(self, interfaces, callback):
threading.Thread.__init__(self)
# The list of interfaces is used for filtering
self.groups_of_interfaces = set(interfaces)
self._callback = callback
self._is_running = False
def run(self):
cmd = self.CMD % (self.SEP)
p = subprocess.Popen(shlex.split(cmd), stdout = subprocess.PIPE,
stderr = subprocess.STDOUT)
stdout = []
self._is_running = True
self.bwm_stats = dict()
while self._is_running:
line = p.stdout.readline().decode()
if line == '' and p.poll() is not None:
break
if line:
record = self._parse_line(line.strip())
# We use 'total' to push the statistics back to VICN
if record['iface_name'] == 'total':
for interfaces in self.groups_of_interfaces:
if not len(interfaces) > 1:
# If the tuple contains only one interface, grab
# the information from bwm_stats and sends it back
# to VICN
if interfaces[0] not in self.bwm_stats:
continue
interface = self.bwm_stats[interfaces[0]]
f_list = [[KEY_FIELD, '==', interface.device_name]]
query = Query(ACTION_UPDATE, Interface.__type__,
filter = Filter.from_list(f_list),
params = interface.get_attribute_dict())
self._callback(query)
else:
# Iterate over each tuple of interfaces to create
# the aggregated filter and paramters to send back
# Currently, we only support sum among the stats
# when VICN subscribes to a tuple of interfaces
aggregated_filters = list()
aggregated_interface = Interface(
node = socket.gethostname(),
device_name = 'sum',
bw_upstream = 0,
bw_downstream = 0)
predicate = list()
predicate.append(KEY_FIELD)
predicate.append('INCLUDED')
for interface in interfaces:
if interface not in self.bwm_stats:
continue
iface = self.bwm_stats[interface]
aggregated_filters.append(iface.device_name)
aggregated_interface.bw_upstream += \
iface.bw_upstream
aggregated_interface.bw_downstream += \
iface.bw_downstream
if not aggregated_filters:
continue
predicate.append(aggregated_filters)
# We support mulitple interfaces only if tied up
# with the SUM function. The update must have the
# sum function specified because it is used to
# match the subscribe query
attrs = aggregated_interface.get_attribute_dict()
query = Query(ACTION_UPDATE, Interface.__type__,
filter = Filter.from_list([predicate]),
params = attrs,
aggregate = FUNCTION_SUM)
self._callback(query)
else:
# Statistics from netmodel.network.interface will be stored
# in self.bwm_stats and used later to construct the update
# queries
interface = Interface(
node = socket.gethostname(),
device_name = record['iface_name'],
bw_upstream = float(record['bytes_out_s']),
bw_downstream = float(record['bytes_in_s']),
)
self.bwm_stats[record['iface_name']] = interface
rc = p.poll()
return rc
def stop(self):
self._is_running = False
def _parse_line(self, line):
return dict(zip(self.FIELDS_RATE, line.split(self.SEP)))
class BWMInterface(BaseInterface):
__interface__ = 'bwm'
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._thread = None
self.register_object(Interface)
def terminate(self):
self._thread.stop()
def _on_reply(self, reply):
packet = Packet.from_query(reply, reply = True)
self.receive(packet)
#--------------------------------------------------------------------------
# Router interface
#--------------------------------------------------------------------------
def send_impl(self, packet):
query = packet.to_query()
if query.action not in [ACTION_SUBSCRIBE]: # , ACTION_UNSUBSCRIBE]:
log.warning("Ignore unknown action {}".format(
ACTION2STR[query.action]))
return
# TODO: Add the sum operator. If sum the list of interfaces is
# added to the BWMThread as a tuple, otherwise every single
# interface will be added singularly
# We currently simply extract it from the filter
interfaces_list = [p.value for p in query.filter if p.key == KEY_FIELD]
# interfaces is a list of tuple. If someone sbscribe to mulitple
# interfaces interfaces will be a list of 1 tuple. The tuple will
# contain the set of interfaces
if len(interfaces_list) != 1:
log.warning("interfaces_list should have len = 1: {}".format(interfaces_list))
return
interfaces = interfaces_list[0] \
if isinstance(interfaces_list[0], tuple) \
else tuple([interfaces_list[0]])
# Check if interfaces is more than one. In this case, we only support
# The SUM function on the list of field.
if len(interfaces) > 1:
assert query.aggregate == FUNCTION_SUM
if self._thread is None:
self._thread = BWMThread(tuple([interfaces]), self._on_reply)
self._thread.start()
else:
self._thread.groups_of_interfaces.add(interfaces)
|
