summaryrefslogtreecommitdiffstats
path: root/scripts/automation/trex_control_plane/client/trex_async_client.py
blob: 6793a4caab541f71bd5ffa948e9d71d824836ff5 (plain)
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()