#!/usr/bin/env python3 #-*- mode: Python;-*- import sys import os import time import random import tempfile import argparse import socket import json try: import requests except: sys.stderr.write('ERROR: Could not import requests module. Ensure it is installed.\n') sys.stderr.write(' Under Debian, the package name is "python3-requests"\n.') sys.exit(1) VERSION = "{DEVELOPMENT}" if VERSION == "{DEVELOPMENT}": script_dir = '.' try: script_dir = os.path.dirname(os.path.realpath(__file__)) except: try: script_dir = os.path.dirname(os.path.abspath(sys.argv[0])) except: pass sys.path.append("%s/../lib" % script_dir) from nanownlib import * from nanownlib.platform import * from nanownlib.tcpts import * import nanownlib.storage parser = argparse.ArgumentParser( description="") parser.add_argument('-c', dest='cases', type=str, default='{"short":10000,"long":13000}', help='JSON representation of echo timing cases.') parser.add_argument('--no-tcpts', action='store_true', help='Disable TCP timestamp profiling') parser.add_argument('--no-control', action='store_true', help='Do not collect separate control data. Instead, synthesize it from test and train data.') parser.add_argument('session_name', default=None, help='Name for the sampler session (used in output filenames)') parser.add_argument('sample_count', type=int, default=None, help='Number of samples to collect') parser.add_argument('host', default=None, help='IP address or host name of server') parser.add_argument('port', nargs='?', type=int, default=80, help='TCP port number of HTTP service (default: 80)') options = parser.parse_args() num_samples = options.sample_count hostname = options.host port = options.port protocol = 'http' cases = json.loads(options.cases) db_file = "%s.db" % options.session_name db = nanownlib.storage.db(db_file) def extractReportedRuntime(headers, body): if body.startswith("waited: "): return int(body.split("waited: ")[1], 10) def sendRequest(data=None): method = 'GET' path = '/?t=' + data url = "%s://%s:%d%s" % (protocol,hostname,port,path) headers = {} body = (b'') req = requests.Request('GET', url).prepare() retry = True while retry: try: session = requests.Session() response = session.send(req, allow_redirects=False) #print("sendRequest:", repr(response.raw._original_response.local_address)) reported = extractReportedRuntime(response.headers, response.text) retry = False except Exception as e: sys.stderr.write("ERROR: HTTP request problem: %s\n" % repr(e)) time.sleep(1.0) sys.stderr.write("ERROR: retrying...\n") return {'userspace_rtt':response.elapsed.microseconds*1000, 'reported':reported, 'local_port':response.raw._original_response.local_address[1]} def fetch(probedata, data): # http://docs.python-requests.org/en/latest/api/#requests.Response result = sendRequest("%d" % data) result.update(probedata) return result setCPUAffinity() time.sleep(0.25) # Give our process a chance to migrate to a different CPU if necessary setPowersave(False) # XXX: test this to see if it helps setLowLatency(True) # XXX: test this to see if it helps tcpts_previous = setTCPTimestamps(True) #XXX: what about multiple A records? # perform this during a script generation step, measuring lowest latency server # and hard-coding the IP and host name separately. However, including all # server IPs in comments host_ip = socket.gethostbyname(hostname) meta = {'tcpts_mean':None,'tcpts_stddev':None,'tcpts_slopes':None} if not options.no_tcpts: sys.stderr.write("INFO: Probing target for TCP timestamp precision...\n") sniffer_fp = tempfile.NamedTemporaryFile('w+t') sniffer = startSniffer(host_ip, port, sniffer_fp.name) time.sleep(1.0) ports = runTimestampProbes(host_ip, port, hostname, 12) time.sleep(1.0) stopSniffer(sniffer) sniffer_fp.seek(0) mean,stddev,slopes = computeTimestampPrecision(sniffer_fp, ports) meta = {'tcpts_mean':mean,'tcpts_stddev':stddev,'tcpts_slopes':json.dumps(slopes)} if meta['tcpts_mean'] == None: sys.stderr.write("INFO: TCP timestamps not supported.\n") elif meta['tcpts_stddev'] == None: sys.stderr.write("INFO: Not enough TCP timestamp samples.\n") elif meta['tcpts_stddev']/meta['tcpts_mean'] > 0.05: sys.stderr.write("INFO: TCP timestamp precision is inconsistent.\n") sys.stderr.write("INFO: mean=%(tcpts_mean)f,stddev=%(tcpts_stddev)f,samples=%(tcpts_slopes)s\n" % meta) else: sys.stderr.write("INFO: Estimated TCP timestamp precision: %f (stddev: %f, %f%%)\n" % (meta['tcpts_mean'], meta['tcpts_stddev'], 100*meta['tcpts_stddev']/meta['tcpts_mean'])) db.addMeta(meta) def findNextSampleID(db): cursor = db.conn.cursor() cursor.execute("SELECT max(sample) FROM probes") row = cursor.fetchone() if row != None and row[0] != None: return row[0]+1 return 0 def collectSamples(db, sample_type, count, sniffer): sniffer.start() if not sniffer.is_running(): sys.stderr.write('ERROR: Sniffer did not start...\n') return sid = findNextSampleID(db) for k in range(0,count): sample_order = list(cases.items()) random.shuffle(sample_order) if sample_type.endswith('null'): for i in range(1,len(sample_order)): sample_order[i] = (sample_order[i][0],sample_order[0][1]) random.shuffle(sample_order) results = [] now = int(time.time()*1000000000) for i in range(len(sample_order)): results.append(fetch({'sample':sid, 'test_case':sample_order[i][0], 'type':sample_type, 'tc_order':i, 'time_of_day':now}, sample_order[i][1])) print(results) db.addProbes(results) db.conn.commit() sid += 1 time.sleep(2.0) # Give sniffer a chance to collect remaining packets sniffer.stop() #print(sniffer.openPacketLog().read()) start = time.time() associatePackets(sniffer.openPacketLog(), db) end = time.time() print("associate time:", end-start) if options.no_control: num_control = 0 else: num_control = int(num_samples*2/5) num_train = int((num_samples-num_control)/3) num_test = num_samples-num_train-num_control sample_types = [('train',num_train), ('train_null',num_control), ('test',num_test)] sniffer = snifferProcess(host_ip, port) for st,count in sample_types: collectSamples(db, st,count,sniffer) #start = time.time() #report_interval = 20 #next_report = start+report_interval # if (time.time() > next_report): # reportProgress(db, sample_types, start) # next_report += report_interval if options.no_control: print("TODO: implement control synthesizing!") start = time.time() num_probes = analyzeProbes(db) end = time.time() print("analyzed %d probes' packets in: %f" % (num_probes, end-start)) setPowersave(True) # XXX: test this to see if it actually helps setLowLatency(False) # XXX: test this to see if it actually helps setTCPTimestamps(tcpts_previous)