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
223
224
225
226
227
|
#!/router/bin/python
try:
# support import for Python 2
import outer_packages
except ImportError:
# support import for Python 3
import client.outer_packages
from client_utils.jsonrpc_client import JsonRpcClient, BatchMessage
from common.text_opts import *
import json
import threading
import time
import datetime
import zmq
import re
from common.trex_stats import *
from common.trex_streams import *
# basic async stats class
class TrexAsyncStats(object):
def __init__ (self):
self.ref_point = None
self.current = {}
self.last_update_ts = datetime.datetime.now()
def __format_num (self, size, suffix = ""):
for unit in ['','K','M','G','T','P']:
if abs(size) < 1000.0:
return "%3.2f %s%s" % (size, unit, suffix)
size /= 1000.0
return "NaN"
def update (self, snapshot):
#update
self.last_update_ts = datetime.datetime.now()
self.current = snapshot
if self.ref_point == None:
self.ref_point = self.current
def get (self, field, format = False, suffix = ""):
if not field in self.current:
return "N/A"
if not format:
return self.current[field]
else:
return self.__format_num(self.current[field], suffix)
def get_rel (self, field, format = False, suffix = ""):
if not field in self.current:
return "N/A"
if not format:
return (self.current[field] - self.ref_point[field])
else:
return self.__format_num(self.current[field] - self.ref_point[field], suffix)
# return true if new data has arrived in the past 2 seconds
def is_online (self):
delta_ms = (datetime.datetime.now() - self.last_update_ts).total_seconds() * 1000
return (delta_ms < 2000)
# describes the general stats provided by TRex
class TrexAsyncStatsGeneral(TrexAsyncStats):
def __init__ (self):
super(TrexAsyncStatsGeneral, self).__init__()
# per port stats
class TrexAsyncStatsPort(TrexAsyncStats):
def __init__ (self):
super(TrexAsyncStatsPort, self).__init__()
def get_stream_stats (self, stream_id):
return None
# stats manager
class TrexAsyncStatsManager():
def __init__ (self):
self.general_stats = TrexAsyncStatsGeneral()
self.port_stats = {}
def get_general_stats (self):
return self.general_stats
def get_port_stats (self, port_id):
if not str(port_id) in self.port_stats:
return None
return self.port_stats[str(port_id)]
def update (self, data):
self.__handle_snapshot(data)
def __handle_snapshot (self, snapshot):
general_stats = {}
port_stats = {}
# filter the values per port and general
for key, value in snapshot.iteritems():
# match a pattern of ports
m = re.search('(.*)\-([0-8])', key)
if m:
port_id = m.group(2)
field_name = m.group(1)
if not port_id in port_stats:
port_stats[port_id] = {}
port_stats[port_id][field_name] = value
else:
# no port match - general stats
general_stats[key] = value
# update the general object with the snapshot
self.general_stats.update(general_stats)
# update all ports
for port_id, data in port_stats.iteritems():
if not port_id in self.port_stats:
self.port_stats[port_id] = TrexAsyncStatsPort()
self.port_stats[port_id].update(data)
class CTRexAsyncClient():
def __init__ (self, server, port, stateless_client):
self.port = port
self.server = server
self.stateless_client = stateless_client
self.raw_snapshot = {}
self.stats = TrexAsyncStatsManager()
self.tr = "tcp://{0}:{1}".format(self.server, self.port)
print "\nConnecting To ZMQ Publisher At {0}".format(self.tr)
self.active = True
self.t = threading.Thread(target = self.run)
# kill this thread on exit and don't add it to the join list
self.t.setDaemon(True)
self.t.start()
def run (self):
# Socket to talk to server
self.context = zmq.Context()
self.socket = self.context.socket(zmq.SUB)
self.socket.connect(self.tr)
self.socket.setsockopt(zmq.SUBSCRIBE, '')
self.socket.setsockopt(zmq.RCVTIMEO, 3000)
while self.active:
try:
line = self.socket.recv_string();
self.stateless_client.on_async_alive()
except zmq.Again:
self.stateless_client.on_async_dead()
continue
msg = json.loads(line)
name = msg['name']
data = msg['data']
type = msg['type']
self.raw_snapshot[name] = data
self.__dispatch(name, type, data)
def get_stats (self):
return self.stats
def get_raw_snapshot (self):
return self.raw_snapshot
# dispatch the message to the right place
def __dispatch (self, name, type, data):
# stats
if name == "trex-global":
self.stats.update(data)
# events
elif name == "trex-event":
self.stateless_client.handle_async_event(type, data)
else:
pass
def stop (self):
self.active = False
self.t.join()
|
