source: trunk/bin/sampler @ 8

Last change on this file since 8 was 5, checked in by tim, 9 years ago

.

  • Property svn:executable set to *
File size: 6.3 KB
Line 
1#!/usr/bin/env python3
2#-*- mode: Python;-*-
3
4import sys
5import os
6import time
7import random
8import tempfile
9import argparse
10import socket
11import json
12try:
13    import requests
14except:
15    sys.stderr.write('ERROR: Could not import requests module.  Ensure it is installed.\n')
16    sys.stderr.write('       Under Debian, the package name is "python3-requests"\n.')
17    sys.exit(1)
18
19VERSION = "{DEVELOPMENT}"
20if VERSION == "{DEVELOPMENT}":
21    script_dir = '.'
22    try:
23        script_dir = os.path.dirname(os.path.realpath(__file__))
24    except:
25        try:
26            script_dir = os.path.dirname(os.path.abspath(sys.argv[0]))
27        except:
28            pass
29    sys.path.append("%s/../lib" % script_dir)
30
31from nanownlib import *
32import nanownlib.storage
33
34parser = argparse.ArgumentParser(
35    description="")
36parser.add_argument('-c', dest='cases', type=str, default='{"short":10000,"long":13000}',
37                    help='JSON representation of echo timing cases.')
38parser.add_argument('--no-tcpts', action='store_true', help='Disbale TCP timestamp profiling')
39parser.add_argument('session_name', default=None,
40                    help='Name for the sampler session (used in output filenames)')
41parser.add_argument('sample_count', type=int, default=None,
42                    help='Number of samples to collect')
43parser.add_argument('host', default=None,
44                    help='IP address or host name of server')
45parser.add_argument('port', nargs='?', type=int, default=80,
46                    help='TCP port number of HTTP service (default: 80)')
47options = parser.parse_args()
48
49
50session_name = options.session_name
51num_samples=options.sample_count
52hostname = options.host
53port = options.port
54protocol = 'http'
55
56cases = json.loads(options.cases)
57
58def extractReportedRuntime(headers, body):
59    if body.startswith("waited: "):
60        return int(body.split("waited: ")[1], 10)
61
62
63def sendRequest(data=None):
64    method = 'GET'
65    path = '/data/clamav-audit-results.txt?t=' + data
66    url = "%s://%s:%d%s" % (protocol,hostname,port,path)
67    headers = {}
68    body = (b'')
69    req = requests.Request('GET', url).prepare()
70
71    retry = True
72    while retry:
73        try:
74            session = requests.Session()
75            response = session.send(req)
76            #print(repr(response.raw._original_response.local_address))
77            reported = extractReportedRuntime(response.headers, response.text)
78            retry = False
79        except Exception as e:
80            sys.stderr.write("ERROR: HTTP request problem: %s\n" % repr(e))
81            time.sleep(1.0)
82            sys.stderr.write("ERROR: retrying...\n")
83   
84    return {'userspace_rtt':response.elapsed.microseconds*1000,
85            'reported':reported,
86            'local_port':response.raw._original_response.local_address[1]}
87
88
89def fetch(probedata, data):
90    #   http://docs.python-requests.org/en/latest/api/#requests.Response
91    result = sendRequest("%d" % data)
92    result.update(probedata)
93   
94    return result
95
96
97setCPUAffinity()
98setTCPTimestamps()
99host_ip = socket.gethostbyname(hostname) #XXX: what about multiple A records?
100
101meta = {'tcpts_mean':None,'tcpts_stddev':None,'tcpts_slopes':None}
102if not options.no_tcpts:
103    sys.stderr.write("INFO: Probing target for TCP timestamp precision...\n")
104    sniffer_fp = tempfile.NamedTemporaryFile('w+t')
105    sniffer = startSniffer(host_ip, port, sniffer_fp.name)
106    time.sleep(1.0)
107    ports = runTimestampProbes(host_ip, port, hostname, 12)
108    time.sleep(1.0)
109    stopSniffer(sniffer)
110    sniffer_fp.seek(0)
111    mean,stddev,slopes = computeTimestampPrecision(sniffer_fp, ports)
112    meta = {'tcpts_mean':mean,'tcpts_stddev':stddev,'tcpts_slopes':json.dumps(slopes)}
113   
114if meta['tcpts_mean'] == None:
115    sys.stderr.write("INFO: TCP timestamps not supported.\n")
116elif meta['tcpts_stddev'] == None:
117    sys.stderr.write("INFO: Not enough TCP timestamp samples.\n")
118elif meta['tcpts_stddev']/meta['tcpts_mean'] > 0.05:
119    sys.stderr.write("INFO: TCP timestamp precision is inconsistent.\n")
120    sys.stderr.write("INFO:     mean=%(tcpts_mean)f,stddev=%(tcpts_stddev)f,samples=%(tcpts_slopes)s\n"
121                     % meta)
122else:
123    sys.stderr.write("INFO: Estimated TCP timestamp precision: %f (stddev: %f, %f%%)\n"
124                     % (meta['tcpts_mean'], meta['tcpts_stddev'], 100*meta['tcpts_stddev']/meta['tcpts_mean']))
125
126
127sniffer_fp = tempfile.NamedTemporaryFile('w+t')
128db_file = "%s.db" % session_name
129
130sniffer = startSniffer(host_ip, port, sniffer_fp.name)
131db = nanownlib.storage.db(db_file)
132db.addMeta(meta)
133time.sleep(0.5) # ensure sniffer is fully ready and our process is migrated
134
135sid = 0
136sample_types = [('train',int(num_samples*1/5)),
137                ('train_null',int(num_samples*2/5)),
138                ('test',int(num_samples*2/5)),
139                ('test_null',0)]
140report_interval = 20
141start = time.time()
142next_report = start+report_interval
143for st,count in sample_types:
144    if sniffer.poll() != None:
145        sys.stderr.write('ERROR: Sniffer quit, sender exiting...\n')
146        break
147       
148    for k in range(0,count):
149        sample_order = list(cases.items())
150        random.shuffle(sample_order)
151        if st.endswith('null'):
152            for i in range(1,len(sample_order)):
153                sample_order[i] = (sample_order[i][0],sample_order[0][1])
154            random.shuffle(sample_order)
155            #print('after', sample_order)
156           
157        results = []
158        now = int(time.time()*1000000000)
159        for i in range(len(sample_order)):
160            results.append(fetch({'sample':sid, 'test_case':sample_order[i][0],
161                                  'type':st, 'tc_order':i, 'time_of_day':now},
162                                 sample_order[i][1]))
163
164        #print(results)
165        db.addProbes(results)
166        db.conn.commit()
167        sid += 1
168
169        if (time.time() > next_report):
170            #s = time.time()
171            reportProgress(db, sample_types, start)
172            #print("reportProgress time:", time.time()-s)
173            next_report += report_interval
174
175print("probes complete in %f" % (time.time()-start))
176time.sleep(2.0) # Give sniffer a chance to collect remaining packets
177stopSniffer(sniffer)
178
179start = time.time()
180associatePackets(sniffer_fp, db)
181sniffer_fp.close()
182end = time.time()
183print("associate time:", end-start)
184
185start = time.time()
186num_probes = analyzeProbes(db)
187end = time.time()
188print("analyzed %d probes' packets in: %f" % (num_probes, end-start))
Note: See TracBrowser for help on using the repository browser.