You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@trafficserver.apache.org by pa...@apache.org on 2017/04/21 19:20:43 UTC
[trafficserver] branch master updated: Traffic-replay client
This is an automated email from the ASF dual-hosted git repository.
paziz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/trafficserver.git
The following commit(s) were added to refs/heads/master by this push:
new 619039c Traffic-replay client
619039c is described below
commit 619039c5b1412103cc7e0cf33f6abf71b4f29f95
Author: Persia Aziz <pe...@yahoo-inc.com>
AuthorDate: Tue Apr 11 10:23:12 2017 -0500
Traffic-replay client
---
tests/tools/README.md | 42 +++
tests/tools/lib/result.py | 91 ++++++
tests/tools/sessionvalidation/session.py | 3 +
.../tools/{readme.txt => traffic-replay/Config.py} | 17 +-
tests/tools/traffic-replay/RandomReplay.py | 167 +++++++++++
tests/tools/traffic-replay/SSLReplay.py | 213 ++++++++++++++
tests/tools/traffic-replay/Scheduler.py | 62 ++++
tests/tools/traffic-replay/WorkerTask.py | 45 +++
.../{readme.txt => traffic-replay/__main__.py} | 21 +-
tests/tools/traffic-replay/extractHeader.py | 64 ++++
tests/tools/traffic-replay/h2Replay.py | 327 +++++++++++++++++++++
tests/tools/traffic-replay/mainProcess.py | 63 ++++
12 files changed, 1109 insertions(+), 6 deletions(-)
diff --git a/tests/tools/README.md b/tests/tools/README.md
new file mode 100644
index 0000000..d6d17c0
--- /dev/null
+++ b/tests/tools/README.md
@@ -0,0 +1,42 @@
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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.
+
+These tools are meant to become it own repository in the future. They are here at the moment to help accelerate progress at getting everything working.
+
+Note these Tools require python 3.4 or better.
+
+# Traffic-Replay
+
+Replay client to replay session logs.
+
+Usage:
+python3.5 trafficreplay_v2/ -type <ssl|h2|random> -log_dir /path/to/log -v
+
+Session Log format (in JSON):
+
+ {"version": "0.1",
+ "txns": [
+ {"request": {"headers": "POST \u2026\u2026\r\n\r\n", "timestamp": "..", "body": ".."},
+ "response": {"headers": "HTTP/1.1..\r\n\r\n", "timestamp": "..", "body": ".."},
+ "uuid": "1"},
+ {"request": {"headers": "POST ..\u2026.\r\n\r\n", "timestamp": "..", "body": ".."},
+ "response": {"headers": "HTTP/1.1..\r\nr\n", "timestamp": "..", "body": ".."},
+ "uuid": "2"}
+ ],
+ "timestamp": "....",
+ "encoding": "...."}
+ Configuration: The configuration required to run traffic-replay can be specified in traffic-replay/Config.py
diff --git a/tests/tools/lib/result.py b/tests/tools/lib/result.py
new file mode 100644
index 0000000..992879b
--- /dev/null
+++ b/tests/tools/lib/result.py
@@ -0,0 +1,91 @@
+#!/bin/env python3
+'''
+'''
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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 sys
+class TermColors:
+ ''' Collection of colors for printing out to terminal '''
+ HEADER = '\033[95m'
+ OKBLUE = '\033[94m'
+ OKGREEN = '\033[92m'
+ WARNING = '\033[93m'
+ FAIL = '\033[91m'
+ BOLD = '\033[1m'
+ UNDERLINE = '\033[4m'
+ ENDC = '\033[0m'
+
+ignoredFields = {'age', 'set-cookie', 'server', 'date', 'last-modified', 'via', 'expires', 'cahe-control'} #all lower case
+class Result(object):
+ ''' Result encapsulates the result of a single session replay '''
+
+ def __init__(self, test_name, expected_response, received_response, recv_resp_body=None):
+ ''' expected_response and received_response can be any datatype the caller wants as long as they are the same datatype '''
+ self._test_name = test_name
+ self._expected_response = expected_response
+ self._received_response = received_response
+ self._received_response_body = recv_resp_body
+
+ def getTestName(self):
+ return self._test_name
+
+ def getResultBool(self):
+ return self._expected_response == self._received_response
+
+ def getRespBody(self):
+ if self._received_response_body:
+ return self._received_response_body
+ else:
+ return ""
+ def Compare(self, received_dict, expected_dict):
+ global ignoredFields
+ try:
+ for key in received_dict:
+ #print(key)
+ if key.lower() in expected_dict and key.lower() not in ignoredFields:
+ #print("{0} ==? {1}".format(expected_dict[key.lower()],received_dict[key]))
+ if received_dict[key]!=expected_dict[key.lower()]:
+ print("{0}Difference in the field \"{1}\": \n received:\n{2}\n expected:\n{3}{4}".format(TermColors.FAIL,key,received_dict[key],expected_dict[key],TermColors.ENDC))
+ return False
+
+ except:
+ e=sys.exc_info()
+ print("Error in comparing key ",e,key,expected_dict[key.lower()],received_dict[key])
+ return True
+
+ def getResultString(self, received_dict, expected_dict, colorize=False ):
+ global ignoredFields
+ ''' Return a nicely formatted result string with color if requested '''
+ if self.getResultBool() and self.Compare(received_dict,expected_dict):
+ if colorize:
+ outstr = "{0}PASS{1}".format(
+ TermColors.OKGREEN, TermColors.ENDC)
+
+ else:
+ outstr = "PASS"
+
+ else:
+ if colorize:
+ outstr = "{0}FAIL{1}: expected {2}, received {3}, session file: {4}".format(
+ TermColors.FAIL, TermColors.ENDC, self._expected_response, self._received_response, self._test_name)
+
+ else:
+ outstr = "FAIL: expected {0}, received {1}".format(
+ self._expected_response, self._received_response)
+ sys.exit(0)
+
+ return outstr
diff --git a/tests/tools/sessionvalidation/session.py b/tests/tools/sessionvalidation/session.py
index 4bcb2c6..a36ae26 100644
--- a/tests/tools/sessionvalidation/session.py
+++ b/tests/tools/sessionvalidation/session.py
@@ -28,6 +28,9 @@ class Session(object):
''' Returns an iterator of transaction objects '''
return iter(self._transaction_list)
+ def returnFirstTransaction(self):
+ return self._transaction_list[0]
+
def __repr__(self):
return "<Session {{'filename': {0}, 'version': {1}, 'timestamp: {2}, 'encoding': {3}, 'transaction_list': {4}}}>".format(
self._filename, self._version, self._timestamp, self._encoding, repr(self._transaction_list)
diff --git a/tests/tools/readme.txt b/tests/tools/traffic-replay/Config.py
similarity index 77%
copy from tests/tools/readme.txt
copy to tests/tools/traffic-replay/Config.py
index 9bf71d6..b4e2759 100644
--- a/tests/tools/readme.txt
+++ b/tests/tools/traffic-replay/Config.py
@@ -1,4 +1,6 @@
-
+#!/bin/env python3
+'''
+'''
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
@@ -15,6 +17,15 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-These tools are meant to become it own repository in the furture. They are here at the moment to help accelerate progress at getting everything working.
+#SSL config
+ca_certs = "/path/to/certfile"
+keyfile = "/path/to/keyfile"
+
+#Proxy config
+proxy_host = "127.0.0.1"
+proxy_ssl_port = 443
+proxy_nonssl_port = 8080
-Note these Tools require python 3.4 or better.
+#process and thread config
+nProcess = 4
+nThread = 4
diff --git a/tests/tools/traffic-replay/RandomReplay.py b/tests/tools/traffic-replay/RandomReplay.py
new file mode 100644
index 0000000..8c3de6f
--- /dev/null
+++ b/tests/tools/traffic-replay/RandomReplay.py
@@ -0,0 +1,167 @@
+#!/bin/env python3
+'''
+'''
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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 gevent
+import socket
+import requests
+import os
+from threading import Thread
+import sys
+from multiprocessing import current_process
+import sessionvalidation.sessionvalidation as sv
+import lib.result as result
+import extractHeader
+import mainProcess
+import json
+import gzip
+bSTOP = False
+def createDummyBodywithLength(numberOfbytes):
+ if numberOfbytes<=0:
+ return None
+ body= 'a'
+ while numberOfbytes!=1:
+ body += 'b'
+ numberOfbytes -= 1
+ return body
+
+def handleResponse(response,*args, **kwargs):
+ print(response.status_code)
+ #resp=args[0]
+ #expected_output_split = resp.getHeaders().split('\r\n')[ 0].split(' ', 2)
+ #expected_output = (int(expected_output_split[1]), str( expected_output_split[2]))
+ #r = result.Result(session_filename, expected_output[0], response.status_code)
+ #print(r.getResultString(colorize=True))
+# make sure len of the message body is greater than length
+def gen():
+ yield 'pforpersia,champaignurbana'.encode('utf-8')
+ yield 'there'.encode('utf-8')
+
+def txn_replay(session_filename, txn, proxy, result_queue, request_session):
+ """ Replays a single transaction
+ :param request_session: has to be a valid requests session"""
+ req = txn.getRequest()
+ resp = txn.getResponse()
+
+ # Construct HTTP request & fire it off
+ txn_req_headers = req.getHeaders()
+ txn_req_headers_dict = extractHeader.header_to_dict(txn_req_headers)
+ txn_req_headers_dict['Content-MD5'] = txn._uuid # used as unique identifier
+ if 'body' in txn_req_headers_dict:
+ del txn_req_headers_dict['body']
+
+ #print("Replaying session")
+ try:
+ #response = request_session.request(extractHeader.extract_txn_req_method(txn_req_headers),
+ # 'http://' + extractHeader.extract_host(txn_req_headers) + extractHeader.extract_GET_path(txn_req_headers),
+ # headers=txn_req_headers_dict,stream=False) # making stream=False raises contentdecoding exception? kill me
+ method = extractHeader.extract_txn_req_method(txn_req_headers)
+ response = None
+ body=None
+ content=None
+ if 'Transfer-Encoding' in txn_req_headers_dict:
+ # deleting the host key, since the STUPID post/get functions are going to add host field anyway, so there will be multiple host fields in the header
+ # This confuses the ATS and it returns 400 "Invalid HTTP request". I don't believe this
+ # BUT, this is not a problem if the data is not chunked encoded.. Strange, huh?
+ del txn_req_headers_dict['Host']
+ if 'Content-Length' in txn_req_headers_dict:
+ #print("ewww !")
+ del txn_req_headers_dict['Content-Length']
+ body = gen()
+ if 'Content-Length' in txn_req_headers_dict:
+ nBytes=int(txn_req_headers_dict['Content-Length'])
+ body = createDummyBodywithLength(nBytes)
+ #print("request session is",id(request_session))
+ if method == 'GET':
+ response = request_session.get('http://' + extractHeader.extract_host(txn_req_headers) + extractHeader.extract_GET_path(txn_req_headers),
+ headers=txn_req_headers_dict, stream=False, allow_redirects=False,data=body)
+ if 'Content-Length' in response.headers:
+ content = response.raw
+ #print("len: {0} received {1}".format(response.headers['Content-Length'],content))
+
+ elif method == 'POST':
+ response = request_session.post('http://' + extractHeader.extract_host(txn_req_headers) + extractHeader.extract_GET_path(txn_req_headers),
+ headers=txn_req_headers_dict, stream=False, data=body, allow_redirects=False)
+
+ if 'Content-Length' in response.headers:
+ content = response.raw
+ #print("reading==========>>>>>>>>>>>>>.")
+ #print(content.data)
+ #print("len: {0} received {1}".format(response.headers['Content-Length'],content))
+ elif method == 'HEAD':
+ response = request_session.head('http://' + extractHeader.extract_host(txn_req_headers) + extractHeader.extract_GET_path(txn_req_headers),
+ headers=txn_req_headers_dict, stream=False)
+
+ #gzip_file = gzip.GzipFile(fileobj=content)
+ #shutil.copyfileobj(gzip_file, f)
+ expected=extractHeader.responseHeader_to_dict(resp.getHeaders())
+ #print(expected)
+ if mainProcess.verbose:
+ expected_output_split = resp.getHeaders().split('\r\n')[ 0].split(' ', 2)
+ expected_output = (int(expected_output_split[1]), str( expected_output_split[2]))
+ r = result.Result(session_filename, expected_output[0], response.status_code)
+ print(r.getResultString(response.headers,expected,colorize=True))
+ r.Compare(response.headers,expected)
+ #result_queue.put(r)
+ except UnicodeEncodeError as e:
+ # these unicode errors are due to the interaction between Requests and our wiretrace data.
+ # TODO fix
+ print("UnicodeEncodeError exception")
+
+ except requests.exceptions.ContentDecodingError as e:
+ print("ContentDecodingError",e)
+ except:
+ e=sys.exc_info()
+ print("ERROR in requests: ",e,response, session_filename)
+
+def session_replay(input, proxy, result_queue):
+ global bSTOP
+ ''' Replay all transactions in session
+
+ This entire session will be replayed in one requests.Session (so one socket / TCP connection)'''
+ #if timing_control:
+ # time.sleep(float(session._timestamp)) # allow other threads to run
+ while bSTOP == False:
+ for session in iter(input.get, 'STOP'):
+ #print(bSTOP)
+ if session == 'STOP':
+ print("Queue is empty")
+ bSTOP = True
+ break
+ with requests.Session() as request_session:
+ request_session.proxies = proxy
+ for txn in session.getTransactionIter():
+ try:
+ txn_replay(session._filename, txn, proxy, result_queue, request_session)
+ except:
+ e=sys.exc_info()
+ print("ERROR in replaying: ",e,txn.getRequest().getHeaders())
+ bSTOP = True
+ print("Queue is empty")
+ input.put('STOP')
+ break
+
+def client_replay(input, proxy, result_queue, nThread):
+ Threads = []
+ for i in range(nThread):
+ t = Thread(target=session_replay, args=[input, proxy, result_queue])
+ t.start()
+ Threads.append(t)
+
+ for t1 in Threads:
+ t1.join()
diff --git a/tests/tools/traffic-replay/SSLReplay.py b/tests/tools/traffic-replay/SSLReplay.py
new file mode 100644
index 0000000..dfc7d5c
--- /dev/null
+++ b/tests/tools/traffic-replay/SSLReplay.py
@@ -0,0 +1,213 @@
+#!/bin/env python3
+'''
+'''
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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 http.client
+import socket, ssl, pprint
+import gevent
+import requests
+import os
+#import threading
+import sys
+from multiprocessing import current_process
+import sessionvalidation.sessionvalidation as sv
+import lib.result as result
+import extractHeader
+from gevent import monkey, sleep
+from threading import Thread
+import mainProcess
+import json
+import extractHeader
+import time
+import Config
+bSTOP = False
+
+class ProxyHTTPSConnection(http.client.HTTPSConnection):
+ "This class allows communication via SSL."
+
+ default_port = http.client.HTTPS_PORT
+
+ # XXX Should key_file and cert_file be deprecated in favour of context?
+
+ def __init__(self, host, port=None, key_file=None, cert_file=None,
+ timeout=socket._GLOBAL_DEFAULT_TIMEOUT,
+ source_address=None, *, context=None,
+ check_hostname=None,server_name = None):
+ #http.client.HTTPSConnection.__init__(self)
+ super().__init__(host, port, key_file,cert_file, timeout, source_address,context=context,check_hostname=check_hostname)
+ '''
+ self.key_file = key_file
+ self.cert_file = cert_file
+ if context is None:
+ context = ssl._create_default_https_context()
+ will_verify = context.verify_mode != ssl.CERT_NONE
+ if check_hostname is None:
+ check_hostname = context.check_hostname
+ if check_hostname and not will_verify:
+ raise ValueError("check_hostname needs a SSL context with "
+ "either CERT_OPTIONAL or CERT_REQUIRED")
+ if key_file or cert_file:
+ context.load_cert_chain(cert_file, key_file)
+ self._context = context
+ self._check_hostname = check_hostname
+ '''
+ self.server_name = server_name
+
+ def connect(self):
+ "Connect to a host on a given (SSL) port."
+ http.client.HTTPConnection.connect(self)
+
+ if self._tunnel_host:
+ server_hostname = self._tunnel_host
+ else:
+ server_hostname = self.server_name
+ self.sock = self._context.wrap_socket(self.sock,
+ do_handshake_on_connect=True,
+ server_side=False,
+ server_hostname=server_hostname)
+ if not self._context.check_hostname and self._check_hostname:
+ try:
+ ssl.match_hostname(self.sock.getpeercert(), server_hostname)
+ except Exception:
+ self.sock.shutdown(socket.SHUT_RDWR)
+ self.sock.close()
+ raise
+
+
+def txn_replay(session_filename, txn, proxy, result_queue, request_session):
+ """ Replays a single transaction
+ :param request_session: has to be a valid requests session"""
+ req = txn.getRequest()
+ resp = txn.getResponse()
+ responseDict = {}
+ # Construct HTTP request & fire it off
+ txn_req_headers = req.getHeaders()
+ txn_req_headers_dict = extractHeader.header_to_dict(txn_req_headers)
+ txn_req_headers_dict['Content-MD5'] = txn._uuid # used as unique identifier
+ if 'body' in txn_req_headers_dict:
+ del txn_req_headers_dict['body']
+
+ #print("Replaying session")
+ try:
+ #response = request_session.request(extractHeader.extract_txn_req_method(txn_req_headers),
+ # 'http://' + extractHeader.extract_host(txn_req_headers) + extractHeader.extract_GET_path(txn_req_headers),
+ # headers=txn_req_headers_dict,stream=False) # making stream=False raises contentdecoding exception? kill me
+ method = extractHeader.extract_txn_req_method(txn_req_headers)
+ response = None
+ body=None
+ content=None
+ if 'Transfer-Encoding' in txn_req_headers_dict:
+ # deleting the host key, since the STUPID post/get functions are going to add host field anyway, so there will be multiple host fields in the header
+ # This confuses the ATS and it returns 400 "Invalid HTTP request". I don't believe this
+ # BUT, this is not a problem if the data is not chunked encoded.. Strange, huh?
+ del txn_req_headers_dict['Host']
+ if 'Content-Length' in txn_req_headers_dict:
+ #print("ewww !")
+ del txn_req_headers_dict['Content-Length']
+ body = gen()
+ if 'Content-Length' in txn_req_headers_dict:
+ nBytes=int(txn_req_headers_dict['Content-Length'])
+ body = createDummyBodywithLength(nBytes)
+ #print("request session is",id(request_session))
+ if method == 'GET':
+ request_session.request('GET','https://' + extractHeader.extract_host(txn_req_headers) + extractHeader.extract_GET_path(txn_req_headers),
+ headers=txn_req_headers_dict,body=body)
+ r1 = request_session.getresponse()
+ responseHeaders = r1.getheaders()
+ responseContent = r1.read()
+
+ elif method == 'POST':
+ request_session.request('POST','https://' + extractHeader.extract_host(txn_req_headers) + extractHeader.extract_GET_path(txn_req_headers),
+ headers=txn_req_headers_dict,body=body)
+ r1 = request_session.getresponse()
+ responseHeaders = r1.getheaders()
+ responseContent = r1.read()
+
+ elif method == 'HEAD':
+ request_session.request('HEAD','https://' + extractHeader.extract_host(txn_req_headers) + extractHeader.extract_GET_path(txn_req_headers),
+ headers=txn_req_headers_dict,body=body)
+ r1 = request_session.getresponse()
+ responseHeaders = r1.getheaders()
+ responseContent = r1.read()
+ for key,value in responseHeaders:
+ responseDict[key.lower()] = value
+ expected=extractHeader.responseHeader_to_dict(resp.getHeaders())
+ #print(responseDict)
+ if mainProcess.verbose:
+ expected_output_split = resp.getHeaders().split('\r\n')[ 0].split(' ', 2)
+ expected_output = (int(expected_output_split[1]), str( expected_output_split[2]))
+ r = result.Result(session_filename, expected_output[0],r1.status)
+ print(r.getResultString(responseDict,expected,colorize=True))
+ r.Compare(responseDict,expected)
+ #result_queue.put(r)
+ except UnicodeEncodeError as e:
+ # these unicode errors are due to the interaction between Requests and our wiretrace data.
+ # TODO fix
+ print("UnicodeEncodeError exception")
+
+ except requests.exceptions.ContentDecodingError as e:
+ print("ContentDecodingError",e)
+ except:
+ e=sys.exc_info()
+ print("ERROR in requests: ",e,response, session_filename)
+def client_replay(input, proxy, result_queue, nThread):
+ Threads = []
+ for i in range(nThread):
+ t = Thread(target=session_replay, args=[input, proxy, result_queue])
+ t.start()
+ Threads.append(t)
+
+ for t1 in Threads:
+ t1.join()
+
+def session_replay(input, proxy, result_queue):
+
+ ''' Replay all transactions in session
+
+ This entire session will be replayed in one requests.Session (so one socket / TCP connection)'''
+ #if timing_control:
+ # time.sleep(float(session._timestamp)) # allow other threads to run
+ global bSTOP
+ sslSocks = []
+ while bSTOP == False:
+ for session in iter(input.get, 'STOP'):
+ txn = session.returnFirstTransaction()
+ req = txn.getRequest()
+ # Construct HTTP request & fire it off
+ txn_req_headers = req.getHeaders()
+ txn_req_headers_dict = extractHeader.header_to_dict(txn_req_headers)
+ sc = ssl.SSLContext(protocol=ssl.PROTOCOL_SSLv23)
+ sc.load_cert_chain(Config.ca_certs,keyfile=Config.keyfile)
+ conn = ProxyHTTPSConnection(Config.proxy_host,Config.proxy_ssl_port,cert_file=Config.ca_certs,key_file=Config.keyfile,context=sc,server_name=txn_req_headers_dict['Host'])
+ for txn in session.getTransactionIter():
+ try:
+ #print(txn._uuid)
+ txn_replay(session._filename, txn, proxy, result_queue, conn)
+ except:
+ e=sys.exc_info()
+ print("ERROR in replaying: ",e,txn.getRequest().getHeaders())
+ #sslSocket.bStop = False
+
+ bSTOP = True
+ print("stopping now")
+ input.put('STOP')
+ break
+
+ #time.sleep(0.5)
+ for sslSock in sslSocks:
+ sslSock.ssl_sock.close()
diff --git a/tests/tools/traffic-replay/Scheduler.py b/tests/tools/traffic-replay/Scheduler.py
new file mode 100644
index 0000000..8d2d6d0
--- /dev/null
+++ b/tests/tools/traffic-replay/Scheduler.py
@@ -0,0 +1,62 @@
+#!/bin/env python3
+'''
+'''
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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 time
+import random
+import json
+from multiprocessing import Process, Queue, current_process
+from progress.bar import Bar
+import sessionvalidation.sessionvalidation as sv
+import WorkerTask
+import time
+
+
+def LaunchWorkers(path,nProcess,proxy,replay_type, nThread):
+ ms1=time.time()
+ s = sv.SessionValidator(path)
+ sessions = s.getSessionList()
+ sessions.sort(key=lambda session: session._timestamp)
+ Processes=[]
+ Qsize = 25000 #int (1.5 * len(sessions)/(nProcess))
+ QList=[Queue(Qsize) for i in range(nProcess)]
+ print("Dropped {0} sessions for being malformed. Number of correct sessions {1}".format(len(s.getBadSessionList()),len(sessions)))
+ print(range(nProcess))
+ OutputQ=Queue();
+ #======================================== Pre-load queues
+ for session in sessions:
+ #if nProcess == 1:
+ # QList[0].put(session)
+ #else:
+ QList[random.randint(0,nProcess-1)].put(session)
+ #if QList[0].qsize() > 10 :
+ # break
+ #=============================================== Launch Processes
+ print("size",QList[0].qsize())
+ for i in range(nProcess):
+ QList[i].put('STOP')
+ for i in range(nProcess):
+ p=Process(target=WorkerTask.worker, args=[QList[i],OutputQ,proxy,replay_type, nThread])
+ p.daemon=False
+ Processes.append(p);
+ p.start()
+
+ for p in Processes:
+ p.join()
+ ms2=time.time()
+ print("OK enough, it is time to exit, running time in seconds", (ms2-ms1))
diff --git a/tests/tools/traffic-replay/WorkerTask.py b/tests/tools/traffic-replay/WorkerTask.py
new file mode 100644
index 0000000..e5ba97d
--- /dev/null
+++ b/tests/tools/traffic-replay/WorkerTask.py
@@ -0,0 +1,45 @@
+#!/bin/env python3
+'''
+'''
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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 socket
+import requests
+import os
+#import threading
+import sys
+from multiprocessing import current_process
+import sessionvalidation.sessionvalidation as sv
+import lib.result as result
+from progress.bar import Bar
+import extractHeader
+import RandomReplay
+import SSLReplay
+import h2Replay
+def worker(input,output,proxy,replay_type,nThread):
+ #progress_bar = Bar(" Replaying sessions {0}".format(current_process().name), max=input.qsize())
+ #print("playing {0}=>{1}:{2}".format(current_process().name,session._timestamp,proxy))
+ if replay_type == 'random':
+ RandomReplay.client_replay(input, proxy, output, nThread)
+ elif replay_type == 'ssl':
+ SSLReplay.client_replay(input, proxy, output,nThread)
+ elif replay_type == 'h2':
+ h2Replay.client_replay(input, proxy, output,nThread)
+ #progress_bar.next()
+ #progress_bar.finish()
+ print("process{0} has exited".format(current_process().name))
+
diff --git a/tests/tools/readme.txt b/tests/tools/traffic-replay/__main__.py
similarity index 53%
rename from tests/tools/readme.txt
rename to tests/tools/traffic-replay/__main__.py
index 9bf71d6..7f2fcc5 100644
--- a/tests/tools/readme.txt
+++ b/tests/tools/traffic-replay/__main__.py
@@ -1,4 +1,6 @@
-
+#!/bin/env python3
+'''
+'''
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
@@ -15,6 +17,19 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-These tools are meant to become it own repository in the furture. They are here at the moment to help accelerate progress at getting everything working.
+from __future__ import absolute_import, division, print_function
+import mainProcess
+import argparse
+
+if __name__ == '__main__':
+
+ parser = argparse.ArgumentParser()
+ parser.add_argument("-type",action='store', dest='replay_type', help="Replay type: ssl/random/h2")
+ parser.add_argument("-log_dir",type=str, help="directory of JSON replay files")
+ parser.add_argument("-v", dest="verbose", help="verify response status code", action="store_true")
+
+ args = parser.parse_args()
-Note these Tools require python 3.4 or better.
+ # Let 'er loose
+ #main(args.log_dir, args.hostname, int(args.port), args.threads, args.timing, args.verbose)
+ mainProcess.main(args.log_dir, args.replay_type,args.verbose)
diff --git a/tests/tools/traffic-replay/extractHeader.py b/tests/tools/traffic-replay/extractHeader.py
new file mode 100644
index 0000000..5a52f05
--- /dev/null
+++ b/tests/tools/traffic-replay/extractHeader.py
@@ -0,0 +1,64 @@
+#!/bin/env python3
+'''
+'''
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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 sessionvalidation
+def extract_txn_req_method(headers):
+ ''' Extracts the HTTP request method from the header in a string format '''
+ line = (headers.split('\r\n'))[0]
+ return (line.split(' '))[0]
+
+def extract_host(headers):
+ ''' Returns the host header from the given headers '''
+ lines = headers.split('\r\n')
+ for line in lines:
+ if 'Host:' in line:
+ return line.split(' ')[1]
+ return "notfound"
+
+def responseHeader_to_dict(header):
+ headerFields = header.split('\r\n',1)[1]
+ fields =headerFields.split('\r\n')
+ header = [x for x in fields if (x != u'')]
+ headers = {}
+ for line in header:
+ split_here = line.find(":")
+ headers[line[:split_here].lower()] = line[(split_here + 1):].strip()
+
+ return headers
+
+def header_to_dict(header):
+ ''' Convert a HTTP header in string format to a python dictionary
+ Returns a dictionary of header values
+ '''
+ header = header.split('\r\n')
+ header = [x for x in header if (x != u'')]
+ headers = {}
+ for line in header:
+ if 'GET' in line or 'POST' in line or 'HEAD' in line: # ignore initial request line
+ continue
+
+ split_here = line.find(":")
+ headers[line[:split_here]] = line[(split_here + 1):].strip()
+
+ return headers
+
+def extract_GET_path(headers):
+ ''' Extracts the HTTP request URL from the header in a string format '''
+ line = (headers.split('\r\n'))[0]
+ return (line.split(' '))[1]
diff --git a/tests/tools/traffic-replay/h2Replay.py b/tests/tools/traffic-replay/h2Replay.py
new file mode 100644
index 0000000..31224bc
--- /dev/null
+++ b/tests/tools/traffic-replay/h2Replay.py
@@ -0,0 +1,327 @@
+#!/bin/env python3
+'''
+'''
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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 gevent
+import os
+from threading import Thread
+import sys
+from multiprocessing import current_process
+import sessionvalidation.sessionvalidation as sv
+import lib.result as result
+import extractHeader
+import mainProcess
+import json
+from hyper import HTTP20Connection
+from hyper.tls import wrap_socket, H2_NPN_PROTOCOLS, H2C_PROTOCOL
+from hyper.common.bufsocket import BufferedSocket
+import hyper
+import socket
+import logging
+import h2
+from h2.connection import H2Configuration
+import threading
+import Config
+
+log = logging.getLogger(__name__)
+bSTOP = False
+hyper.tls._context = hyper.tls.init_context()
+hyper.tls._context.check_hostname = False
+hyper.tls._context.verify_mode = hyper.compat.ssl.CERT_NONE
+class _LockedObject(object):
+ """
+ A wrapper class that hides a specific object behind a lock.
+
+ The goal here is to provide a simple way to protect access to an object
+ that cannot safely be simultaneously accessed from multiple threads. The
+ intended use of this class is simple: take hold of it with a context
+ manager, which returns the protected object.
+ """
+ def __init__(self, obj):
+ self.lock = threading.RLock()
+ self._obj = obj
+
+ def __enter__(self):
+ self.lock.acquire()
+ return self._obj
+
+ def __exit__(self, _exc_type, _exc_val, _exc_tb):
+ self.lock.release()
+
+
+class h2ATS(HTTP20Connection):
+
+
+ def __init_state(self):
+ """
+ Initializes the 'mutable state' portions of the HTTP/2 connection
+ object.
+
+ This method exists to enable HTTP20Connection objects to be reused if
+ they're closed, by resetting the connection object to its basic state
+ whenever it ends up closed. Any situation that needs to recreate the
+ connection can call this method and it will be done.
+
+ This is one of the only methods in hyper that is truly private, as
+ users should be strongly discouraged from messing about with connection
+ objects themselves.
+ """
+
+ config1 = H2Configuration(
+ client_side=True,
+ header_encoding='utf-8',
+ validate_outbound_headers=False,
+ validate_inbound_headers=False,
+
+ )
+ self._conn = _LockedObject(h2.connection.H2Connection(config=config1))
+
+ # Streams are stored in a dictionary keyed off their stream IDs. We
+ # also save the most recent one for easy access without having to walk
+ # the dictionary.
+ #
+ # We add a set of all streams that we or the remote party forcefully
+ # closed with RST_STREAM, to avoid encountering issues where frames
+ # were already in flight before the RST was processed.
+ #
+ # Finally, we add a set of streams that recently received data. When
+ # using multiple threads, this avoids reading on threads that have just
+ # acquired the I/O lock whose streams have already had their data read
+ # for them by prior threads.
+ self.streams = {}
+ self.recent_stream = None
+ self.next_stream_id = 1
+ self.reset_streams = set()
+ self.recent_recv_streams = set()
+
+ # The socket used to send data.
+ self._sock = None
+
+ # Instantiate a window manager.
+ #self.window_manager = self.__wm_class(65535)
+
+ return
+
+ def __init__(self,host,**kwargs):
+ HTTP20Connection.__init__(self,host,**kwargs)
+ self.__init_state()
+
+ def connect(self):
+ """
+ Connect to the server specified when the object was created. This is a
+ no-op if we're already connected.
+
+ Concurrency
+ -----------
+
+ This method is thread-safe. It may be called from multiple threads, and
+ is a noop for all threads apart from the first.
+
+ :returns: Nothing.
+
+ """
+ #print("connecting to ATS")
+ with self._lock:
+ if self._sock is not None:
+ return
+ sni = self.host
+ if not self.proxy_host:
+ host = self.host
+ port = self.port
+ else:
+ host = self.proxy_host
+ port = self.proxy_port
+
+ sock = socket.create_connection((host, port))
+
+ if self.secure:
+ #assert not self.proxy_host, "Proxy with HTTPS not supported."
+ sock, proto = wrap_socket(sock, sni, self.ssl_context,
+ force_proto=self.force_proto)
+ else:
+ proto = H2C_PROTOCOL
+
+ log.debug("Selected NPN protocol: %s", proto)
+ assert proto in H2_NPN_PROTOCOLS or proto == H2C_PROTOCOL
+
+ self._sock = BufferedSocket(sock, self.network_buffer_size)
+
+ self._send_preamble()
+
+def createDummyBodywithLength(numberOfbytes):
+ if numberOfbytes==0:
+ return None
+ body= 'a'
+ while numberOfbytes!=1:
+ body += 'b'
+ numberOfbytes -= 1
+ return body
+
+def handleResponse(response,*args, **kwargs):
+ print(response.status_code)
+ #resp=args[0]
+ #expected_output_split = resp.getHeaders().split('\r\n')[ 0].split(' ', 2)
+ #expected_output = (int(expected_output_split[1]), str( expected_output_split[2]))
+ #r = result.Result(session_filename, expected_output[0], response.status_code)
+ #print(r.getResultString(colorize=True))
+# make sure len of the message body is greater than length
+def gen():
+ yield 'pforpersia,champaignurbana'.encode('utf-8')
+ yield 'there'.encode('utf-8')
+
+def txn_replay(session_filename, txn, proxy, result_queue, h2conn,request_IDs):
+ """ Replays a single transaction
+ :param request_session: has to be a valid requests session"""
+ req = txn.getRequest()
+ resp = txn.getResponse()
+ # Construct HTTP request & fire it off
+ txn_req_headers = req.getHeaders()
+ txn_req_headers_dict = extractHeader.header_to_dict(txn_req_headers)
+ txn_req_headers_dict['Content-MD5'] = txn._uuid # used as unique identifier
+ if 'body' in txn_req_headers_dict:
+ del txn_req_headers_dict['body']
+ responseID = -1
+ #print("Replaying session")
+ try:
+ #response = request_session.request(extractHeader.extract_txn_req_method(txn_req_headers),
+ # 'http://' + extractHeader.extract_host(txn_req_headers) + extractHeader.extract_GET_path(txn_req_headers),
+ # headers=txn_req_headers_dict,stream=False) # making stream=False raises contentdecoding exception? kill me
+ method = extractHeader.extract_txn_req_method(txn_req_headers)
+ response = None
+ mbody=None
+ #txn_req_headers_dict['Host'] = "localhost"
+ if 'Transfer-Encoding' in txn_req_headers_dict:
+ # deleting the host key, since the STUPID post/get functions are going to add host field anyway, so there will be multiple host fields in the header
+ # This confuses the ATS and it returns 400 "Invalid HTTP request". I don't believe this
+ # BUT, this is not a problem if the data is not chunked encoded.. Strange, huh?
+ #del txn_req_headers_dict['Host']
+ if 'Content-Length' in txn_req_headers_dict:
+ #print("ewww !")
+ del txn_req_headers_dict['Content-Length']
+ mbody = gen()
+ if 'Content-Length' in txn_req_headers_dict:
+ nBytes=int(txn_req_headers_dict['Content-Length'])
+ mbody = createDummyBodywithLength(nBytes)
+ if 'Connection' in txn_req_headers_dict:
+ del txn_req_headers_dict['Connection']
+ #str2 = extractHeader.extract_host(txn_req_headers)+ extractHeader.extract_GET_path(txn_req_headers)
+ #print(str2)
+ if method == 'GET':
+ responseID = h2conn.request('GET',url=extractHeader.extract_GET_path(txn_req_headers),headers=txn_req_headers_dict, body=mbody)
+ #print("get response", responseID)
+ return responseID
+ #request_IDs.append(responseID)
+ #response = h2conn.get_response(id)
+ #print(response.headers)
+ #if 'Content-Length' in response.headers:
+ # content = response.read()
+ #print("len: {0} received {1}".format(response.headers['Content-Length'],content))
+
+ elif method == 'POST':
+ responseID = h2conn.request('POST',url=extractHeader.extract_GET_path(txn_req_headers),headers=txn_req_headers_dict, body=mbody)
+ print("get response", responseID)
+ return responseID
+
+ elif method == 'HEAD':
+ responseID = h2conn.request('HEAD',url=extractHeader.extract_GET_path(txn_req_headers),headers=txn_req_headers_dict)
+ print("get response", responseID)
+ return responseID
+
+ #print(response.headers)
+ #print("logged respose")
+ expected=extractHeader.responseHeader_to_dict(resp.getHeaders())
+ #print(expected)
+ if mainProcess.verbose:
+ expected_output_split = resp.getHeaders().split('\r\n')[ 0].split(' ', 2)
+ expected_output = (int(expected_output_split[1]), str( expected_output_split[2]))
+ r = result.Result(session_filename, expected_output[0], response.status_code)
+ print(r.getResultString(response.headers,expected,colorize=True))
+
+ #return responseID
+
+ except UnicodeEncodeError as e:
+ # these unicode errors are due to the interaction between Requests and our wiretrace data.
+ # TODO fix
+ print("UnicodeEncodeError exception")
+
+ except:
+ e=sys.exc_info()
+ print("ERROR in requests: ",e,response, session_filename)
+
+def session_replay(input, proxy, result_queue):
+ global bSTOP
+ ''' Replay all transactions in session
+
+ This entire session will be replayed in one requests.Session (so one socket / TCP connection)'''
+ #if timing_control:
+ # time.sleep(float(session._timestamp)) # allow other threads to run
+ while bSTOP == False:
+ for session in iter(input.get, 'STOP'):
+ print(bSTOP)
+ if session == 'STOP':
+ print("Queue is empty")
+ bSTOP = True
+ break
+ txn = session.returnFirstTransaction()
+ req = txn.getRequest()
+ # Construct HTTP request & fire it off
+ txn_req_headers = req.getHeaders()
+ txn_req_headers_dict = extractHeader.header_to_dict(txn_req_headers)
+ with h2ATS(txn_req_headers_dict['Host'], secure=True, proxy_host=Config.proxy_host,proxy_port=Config.proxy_ssl_port) as h2conn:
+ request_IDs = []
+ respList = []
+ for txn in session.getTransactionIter():
+ try:
+ ret = txn_replay(session._filename, txn, proxy, result_queue, h2conn,request_IDs)
+ respList.append(txn.getResponse())
+ request_IDs.append(ret)
+ #print("txn return value is ",ret)
+ except:
+ e=sys.exc_info()
+ print("ERROR in replaying: ",e,txn.getRequest().getHeaders())
+ for id in request_IDs:
+ expectedH = respList.pop(0);
+ #print("extracting",id)
+ response = h2conn.get_response(id)
+ #print("code {0}:{1}".format(response.status,response.headers))
+ response_dict = {}
+ if mainProcess.verbose:
+ for field,value in response.headers.items():
+ response_dict[field.decode('utf-8')] = value.decode('utf-8')
+
+ expected_output_split = expectedH.getHeaders().split('\r\n')[ 0].split(' ', 2)
+ expected_output = (int(expected_output_split[1]), str( expected_output_split[2]))
+ r = result.Result("", expected_output[0], response.status)
+ expected_Dict = extractHeader.responseHeader_to_dict(expectedH.getHeaders())
+ print(r.getResultString(response_dict,expected_Dict,colorize=True))
+ #r.Compare(response_dict,expected_Dict)
+
+ bSTOP = True
+ print("Queue is empty")
+ input.put('STOP')
+ break
+
+def client_replay(input, proxy, result_queue, nThread):
+ Threads = []
+ for i in range(nThread):
+ t = Thread(target=session_replay, args=[input, proxy, result_queue])
+ t.start()
+ Threads.append(t)
+
+ for t1 in Threads:
+ t1.join()
diff --git a/tests/tools/traffic-replay/mainProcess.py b/tests/tools/traffic-replay/mainProcess.py
new file mode 100644
index 0000000..a2f38f0
--- /dev/null
+++ b/tests/tools/traffic-replay/mainProcess.py
@@ -0,0 +1,63 @@
+#!/bin/env python3
+'''
+'''
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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 sys
+import json
+import socket
+import os
+import threading
+import time
+import argparse
+import subprocess
+import shlex
+from multiprocessing import Pool, Process
+from collections import deque
+from progress.bar import Bar
+import sessionvalidation.sessionvalidation as sv
+import lib.result as result
+import WorkerTask
+import Scheduler
+import Config
+verbose = False
+def check_for_ats(hostname, port):
+ ''' Checks to see if ATS is running on `hostname` and `port`
+ If not running, this function will terminate the script
+ '''
+ sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ result = sock.connect_ex((hostname, port))
+ if result != 0:
+ # hostname:port is not being listened to
+ print('==========')
+ print('Error: Apache Traffic Server is not running on {0}:{1}'.format(hostname, port))
+ print('Aborting')
+ print('==========')
+ sys.exit()
+# Note: this function can't handle multi-line (ie wrapped line) headers
+# Hopefully this isn't an issue because multi-line headers are deprecated now
+
+def main(path, replay_type, Bverbose):
+ global verbose
+ verbose = Bverbose
+ check_for_ats(Config.proxy_host, Config.proxy_nonssl_port)
+ proxy = {"http": "http://{0}:{1}".format(Config.proxy_host, Config.proxy_nonssl_port)}
+ Scheduler.LaunchWorkers(path,Config.nProcess,proxy,replay_type, Config.nThread)
+
+
+
--
To stop receiving notification emails like this one, please contact
['"commits@trafficserver.apache.org" <co...@trafficserver.apache.org>'].