Merge branch 'DamianZaremba-url-rewrite2'

Rollback Elasticsearch

.gitignore

@@ -0,0 +1,2 @@
+*.swp
+*.gor

.travis.yml

@@ -0,0 +1,3 @@
+language: go 
+go: 1.3.3
+script: sudo -E bash -c "source /etc/profile && gvm use go1.3.3 && export GOPATH=$HOME/gopath:$GOPATH && go get && go test -v"

CHANGELOG.md

@@ -0,0 +1,25 @@
+v0.7.0 - 31 Oct 2013
+* New modular architecture. Listener and Replay functionality merged.
+* Added option to equally split traffic between multiple outputs: --split-output true
+* Saving requests to file and replaying from it
+* Injecting custom headers to http requests
+* Advanced stats using ElasticSearch
+
+v0.3.5 - 15 Sep 2013
+* Significantly improved test coverage
+* Fixed bug with redirect replay https://github.com/buger/gor/pull/15
+* Added limit on listener side
+* Improved stability (catch and log panic, instead of exiting)
+* Added License file
+
+v0.3.3 - 22 Jun 2013
+* Using TCP instead of UDP for communication between Listener and Replay
+* Significantly improved performance
+* Fixed bugs causing locking and message dropping (concurrency issues)
+* Rewrote concurrency model to use more channels
+
+v0.3 - 10 Jun 2013
+* Use RAW_SOCKETS instead of tcpdump
+* Own TCP stack
+* All HTTP request types support
+* Simplified request parsing

Dockerfile

@@ -0,0 +1,7 @@
+FROM google/golang
+
+WORKDIR /gopath/src/gor
+
+ADD . /gopath/src/gor
+
+RUN go get

ELASTICSEARCH.md

@@ -0,0 +1,65 @@
+gor & elasticsearch
+===================
+
+Prerequisites
+-------------
+
+- elasticsearch
+- kibana (Get it here: http://www.elasticsearch.org/overview/kibana/)
+- gor
+
+
+elasticsearch
+-------------
+
+The default elasticsearch configuration is just fine for most workloads. You won't need clustering, sharding or something like that.
+
+In this example we're installing it on our gor replay server which gives us the elasticsearch listener on _http://localhost:9200_
+
+
+kibana
+------
+
+Kibana (elasticsearch analytics web-ui) is just as simple. 
+Download it, extract it and serve it via a simple webserver.
+(Could be nginx or apache)
+
+You could also use a shell, ```cd``` into the kibana directory and start a little quick and dirty python webserver with:
+
+```
+python -m SimpleHTTPServer 8000
+```
+
+In this example we're also choosing the gor replay server as our kibana host. If you choose a different server you'll have to point kibana to your elasticsearch host.
+
+
+gor
+---
+
+Start your gor replay server with elasticsearch option:
+
+```
+./gor --input-raw :8000 --output-http http://staging.com  --output-http-elasticsearch localhost:9200/gor
+```
+
+
+(You don't have to create the index upfront. That will be done for you automatically)
+
+
+Now visit your kibana url, load the predefined dashboard from the gist https://gist.github.com/gottwald/b2c875037f24719a9616 and watch the data rush in.
+
+
+Troubleshooting
+---------------
+
+The replay process may complain about __too many open files__.
+That's because your typical linux shell has a small open files soft limit at 1024.
+You can easily raise that when you do this before starting your _gor replay_ process:
+
+```
+ulimit -n 64000
+```
+
+Please be aware, this is not a permanent setting. It's just valid for the following jobs you start from that shell.
+
+We reached the 1024 limit in our tests with a ubuntu box replaying about 9000 requests per minute. (We had very slow responses there, should be way more with fast responses)

LICENSE.txt

@@ -0,0 +1,191 @@
+Apache License
+Version 2.0, January 2004
+http://www.apache.org/licenses/
+
+TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+1. Definitions.
+
+"License" shall mean the terms and conditions for use, reproduction, and
+distribution as defined by Sections 1 through 9 of this document.
+
+"Licensor" shall mean the copyright owner or entity authorized by the copyright
+owner that is granting the License.
+
+"Legal Entity" shall mean the union of the acting entity and all other entities
+that control, are controlled by, or are under common control with that entity.
+For the purposes of this definition, "control" means (i) the power, direct or
+indirect, to cause the direction or management of such entity, whether by
+contract or otherwise, or (ii) ownership of fifty percent (50%) or more of the
+outstanding shares, or (iii) beneficial ownership of such entity.
+
+"You" (or "Your") shall mean an individual or Legal Entity exercising
+permissions granted by this License.
+
+"Source" form shall mean the preferred form for making modifications, including
+but not limited to software source code, documentation source, and configuration
+files.
+
+"Object" form shall mean any form resulting from mechanical transformation or
+translation of a Source form, including but not limited to compiled object code,
+generated documentation, and conversions to other media types.
+
+"Work" shall mean the work of authorship, whether in Source or Object form, made
+available under the License, as indicated by a copyright notice that is included
+in or attached to the work (an example is provided in the Appendix below).
+
+"Derivative Works" shall mean any work, whether in Source or Object form, that
+is based on (or derived from) the Work and for which the editorial revisions,
+annotations, elaborations, or other modifications represent, as a whole, an
+original work of authorship. For the purposes of this License, Derivative Works
+shall not include works that remain separable from, or merely link (or bind by
+name) to the interfaces of, the Work and Derivative Works thereof.
+
+"Contribution" shall mean any work of authorship, including the original version
+of the Work and any modifications or additions to that Work or Derivative Works
+thereof, that is intentionally submitted to Licensor for inclusion in the Work
+by the copyright owner or by an individual or Legal Entity authorized to submit
+on behalf of the copyright owner. For the purposes of this definition,
+"submitted" means any form of electronic, verbal, or written communication sent
+to the Licensor or its representatives, including but not limited to
+communication on electronic mailing lists, source code control systems, and
+issue tracking systems that are managed by, or on behalf of, the Licensor for
+the purpose of discussing and improving the Work, but excluding communication
+that is conspicuously marked or otherwise designated in writing by the copyright
+owner as "Not a Contribution."
+
+"Contributor" shall mean Licensor and any individual or Legal Entity on behalf
+of whom a Contribution has been received by Licensor and subsequently
+incorporated within the Work.
+
+2. Grant of Copyright License.
+
+Subject to the terms and conditions of this License, each Contributor hereby
+grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free,
+irrevocable copyright license to reproduce, prepare Derivative Works of,
+publicly display, publicly perform, sublicense, and distribute the Work and such
+Derivative Works in Source or Object form.
+
+3. Grant of Patent License.
+
+Subject to the terms and conditions of this License, each Contributor hereby
+grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free,
+irrevocable (except as stated in this section) patent license to make, have
+made, use, offer to sell, sell, import, and otherwise transfer the Work, where
+such license applies only to those patent claims licensable by such Contributor
+that are necessarily infringed by their Contribution(s) alone or by combination
+of their Contribution(s) with the Work to which such Contribution(s) was
+submitted. If You institute patent litigation against any entity (including a
+cross-claim or counterclaim in a lawsuit) alleging that the Work or a
+Contribution incorporated within the Work constitutes direct or contributory
+patent infringement, then any patent licenses granted to You under this License
+for that Work shall terminate as of the date such litigation is filed.
+
+4. Redistribution.
+
+You may reproduce and distribute copies of the Work or Derivative Works thereof
+in any medium, with or without modifications, and in Source or Object form,
+provided that You meet the following conditions:
+
+You must give any other recipients of the Work or Derivative Works a copy of
+this License; and
+You must cause any modified files to carry prominent notices stating that You
+changed the files; and
+You must retain, in the Source form of any Derivative Works that You distribute,
+all copyright, patent, trademark, and attribution notices from the Source form
+of the Work, excluding those notices that do not pertain to any part of the
+Derivative Works; and
+If the Work includes a "NOTICE" text file as part of its distribution, then any
+Derivative Works that You distribute must include a readable copy of the
+attribution notices contained within such NOTICE file, excluding those notices
+that do not pertain to any part of the Derivative Works, in at least one of the
+following places: within a NOTICE text file distributed as part of the
+Derivative Works; within the Source form or documentation, if provided along
+with the Derivative Works; or, within a display generated by the Derivative
+Works, if and wherever such third-party notices normally appear. The contents of
+the NOTICE file are for informational purposes only and do not modify the
+License. You may add Your own attribution notices within Derivative Works that
+You distribute, alongside or as an addendum to the NOTICE text from the Work,
+provided that such additional attribution notices cannot be construed as
+modifying the License.
+You may add Your own copyright statement to Your modifications and may provide
+additional or different license terms and conditions for use, reproduction, or
+distribution of Your modifications, or for any such Derivative Works as a whole,
+provided Your use, reproduction, and distribution of the Work otherwise complies
+with the conditions stated in this License.
+
+5. Submission of Contributions.
+
+Unless You explicitly state otherwise, any Contribution intentionally submitted
+for inclusion in the Work by You to the Licensor shall be under the terms and
+conditions of this License, without any additional terms or conditions.
+Notwithstanding the above, nothing herein shall supersede or modify the terms of
+any separate license agreement you may have executed with Licensor regarding
+such Contributions.
+
+6. Trademarks.
+
+This License does not grant permission to use the trade names, trademarks,
+service marks, or product names of the Licensor, except as required for
+reasonable and customary use in describing the origin of the Work and
+reproducing the content of the NOTICE file.
+
+7. Disclaimer of Warranty.
+
+Unless required by applicable law or agreed to in writing, Licensor provides the
+Work (and each Contributor provides its Contributions) on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied,
+including, without limitation, any warranties or conditions of TITLE,
+NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A PARTICULAR PURPOSE. You are
+solely responsible for determining the appropriateness of using or
+redistributing the Work and assume any risks associated with Your exercise of
+permissions under this License.
+
+8. Limitation of Liability.
+
+In no event and under no legal theory, whether in tort (including negligence),
+contract, or otherwise, unless required by applicable law (such as deliberate
+and grossly negligent acts) or agreed to in writing, shall any Contributor be
+liable to You for damages, including any direct, indirect, special, incidental,
+or consequential damages of any character arising as a result of this License or
+out of the use or inability to use the Work (including but not limited to
+damages for loss of goodwill, work stoppage, computer failure or malfunction, or
+any and all other commercial damages or losses), even if such Contributor has
+been advised of the possibility of such damages.
+
+9. Accepting Warranty or Additional Liability.
+
+While redistributing the Work or Derivative Works thereof, You may choose to
+offer, and charge a fee for, acceptance of support, warranty, indemnity, or
+other liability obligations and/or rights consistent with this License. However,
+in accepting such obligations, You may act only on Your own behalf and on Your
+sole responsibility, not on behalf of any other Contributor, and only if You
+agree to indemnify, defend, and hold each Contributor harmless for any liability
+incurred by, or claims asserted against, such Contributor by reason of your
+accepting any such warranty or additional liability.
+
+END OF TERMS AND CONDITIONS
+
+APPENDIX: How to apply the Apache License to your work
+
+To apply the Apache License to your work, attach the following boilerplate
+notice, with the fields enclosed by brackets "[]" replaced with your own
+identifying information. (Don't include the brackets!) The text should be
+enclosed in the appropriate comment syntax for the file format. We also
+recommend that a file or class name and description of purpose be included on
+the same "printed page" as the copyright notice for easier identification within
+third-party archives.
+
+   Copyright [yyyy] [name of copyright owner]
+
+   Licensed 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.

Makefile

@@ -0,0 +1,25 @@
+SOURCE = emitter.go gor.go gor_stat.go input_dummy.go input_file.go input_raw.go input_tcp.go limiter.go output_dummy.go output_file.go output_http.go output_tcp.go plugins.go settings.go settings_header_filters.go settings_header_hash_filters.go settings_headers.go settings_methods.go settings_option.go settings_url_regexp.go test_input.go elasticsearch.go settings_url_map.go
+
+all: build-x86 build-x64
+
+build-x64:
+	GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build
+	tar -czf gor_x64.tar.gz gor
+	rm gor
+
+build-x86:
+	GOOS=linux GOARCH=386 CGO_ENABLED=0 go build
+	tar -czf gor_x86.tar.gz gor
+	rm gor
+
+dbuild:
+	docker build -t gor .
+
+dtest:
+	docker run -v `pwd`:/gopath/src/gor -t -i gor go test -v
+
+drun:
+	docker run -v `pwd`:/gopath/src/gor -t -i gor go run $(SOURCE) --input-dummy=0 --output-dummy=0 --verbose
+
+dbash: 
+	docker run -v `pwd`:/gopath/src/gor -t -i gor /bin/bash

Procfile

@@ -0,0 +1,4 @@
+web: python -m SimpleHTTPServer 8000
+replayed_web: python -m SimpleHTTPServer 8001
+listener: sudo -E go run ./bin/gor.go --input-raw :8000 --output-tcp :8002 --verbose
+replay: go run ./bin/gor.go --input-tcp :8002 --output-http localhost:8001 --verbose

README.md

@@ -1,75 +1,357 @@
+[![Stories in Ready](https://badge.waffle.io/buger/gor.png?label=ready)](https://waffle.io/buger/gor)
+[![Build Status](https://travis-ci.org/buger/gor.png?branch=master)](https://travis-ci.org/buger/gor)
+
 ## About
 
-Gor is simple http traffic replication tool written in Go. 
-Its main goal to replay traffic from production servers to staging and dev environments.
+Gor is a simple http traffic replication tool written in Go.
+Its main goal is to replay traffic from production servers to staging and dev environments.
 
 
-Now you can test your code on real user sessions in an automated and repeatable fashion.  
+Now you can test your code on real user sessions in an automated and repeatable fashion.
 **No more falling down in production!**
 
-Gor consists of 2 parts: listener and replay servers.
+Here is basic workflow: The listener server catches http traffic and sends it to the replay server or saves to file.The replay server forwards traffic to a given address.
 
-Listener catch http traffic from given port in real-time and send it to replay server via UDP. 
-Replay server forwards traffic to given address.
 
-## Basic example
+![Diagram](http://i.imgur.com/9mqj2SK.png)
 
+
+## Examples
+
+### Capture traffic from port
 ```bash
 # Run on servers where you want to catch traffic. You can run it on each `web` machine.
-sudo gor listen -p 80 -r replay.server.local:28020 
+sudo gor --input-raw :80 --output-tcp replay.local:28020
 
-# Replay server (replay.server.local). 
-gor replay -f http://staging.server
+# Replay server (replay.local).
+gor --input-tcp replay.local:28020 --output-http http://staging.com
+```
+
+### Using 1 Gor instance for both listening and replaying
+It's recommended to use separate server for replaying traffic, but if you have enough CPU resources you can use single Gor instance.
+
+```
+sudo gor --input-raw :80 --output-http "http://staging.com"
 ```
 
 ## Advanced use
 
 ### Rate limiting
-Replay server support rate limiting. It can be useful if you want forward only part of production traffic, not to overload staging environment. You can specify desired request per second using "|" operator after server address:
+Both replay and listener support rate limiting. It can be useful if you want
+forward only part of production traffic and not overload your staging
+environment. 
+Current throttling works like this: If for current second it reached specified requests limit - disregard the rest, on next second counter reseted.
 
+You can specify your desired requests per second using the
+"|" operator after the server address:
+
+#### Limiting replay
 ```
-# staging.server not get more than 10 requests per second
-gor replay -f "http://staging.server|10"
+# staging.server will not get more than 10 requests per second
+gor --input-tcp :28020 --output-http "http://staging.com|10"
+```
+
+#### Limiting listener
+```
+# replay server will not get more than 10 requests per second
+# useful for high-load environments
+gor --input-raw :80 --output-tcp "replay.local:28020|10"
+```
+
+#### Match on regexp of url
+```
+# only forward requests being sent to the api... domains
+gor --input-raw :8080 --output-http staging.com --output-http-url-regexp ^www.
+```
+
+#### Filter based on regexp of header
+```
+# only forward requests with an api version of 1.0x
+gor --input-raw :8080 --output-http staging.com --output-http-header-filter api-version:^1\.0\d
+```
+
+#### Filter based on hash of header
+```
+# send 1/32 of all users consistently to staging
+gor --input-raw :8080 --output-http staging.com --output-http-header-hash-filter user-id:1/32
 ```
 
 ### Forward to multiple addresses
 
-You can forward traffic to multiple endpoints. Just separate addresses by coma.
+You can forward traffic to multiple endpoints. Just add multiple --output-* arguments.
 ```
-gor replay -f "http://staging.server|10,http://dev.server|5"
+gor --input-tcp :28020 --output-http "http://staging.com"  --output-http "http://dev.com"
+```
+
+#### Splitting traffic
+By default it will send same traffic to all outputs, but you have options to equally split it:
+
+```
+gor --input-tcp :28020 --output-http "http://staging.com"  --output-http "http://dev.com" --split-output true
+```
+
+### Saving requests to file
+You can save requests to file, and replay them later:
+```
+# write to file
+gor --input-raw :80 --output-file requests.gor
+
+# read from file
+gor --input-file requests.gor --output-http "http://staging.com"
+```
+
+**Note:** Replay will preserve the original time differences between requests.
+
+### Injecting headers
+
+Additional headers can be injected/overwritten into requests during replay. This may be useful if you need to identify requests generated by Gor or enable feature flagged functionality in an application:
+
+```
+gor --input-raw :80 --output-http "http://staging.server" \
+    --output-http-header "User-Agent: Replayed by Gor" \
+    --output-http-header "Enable-Feature-X: true"
+```
+
+## Filtering HTTP methods
+
+Requests not matching a specified whitelist can be filtered out. For example to strip non-nullipotent requests:
+
+```
+gor --input-raw :80 --output-http "http://staging.server" \
+    --output-http-method GET \
+    --output-http-method OPTIONS
+```
+
+### Basic Auth
+
+If your development or staging environment is protected by Basic Authentication then those credentials can be injected in during the replay:
+
+```
+gor --input-raw :80 --output-http "http://user:pass@staging .com"
+```
+
+Note: This will overwrite any Authorization headers in the original request.
+
+#### Rewrite the target urls based on a mapping
+```
+# rewrite url to match the following
+gor --input-raw :8080 --output-http staging.com --output-http-rewrite-url /xml_test/interface.php:/api/service.do
+```
+
+
+## Stats 
+### ElasticSearch 
+For deep response analyze based on url, cookie, user-agent and etc. you can export response metadata to ElasticSearch. See [ELASTICSEARCH.md](ELASTICSEARCH.md) for more details.
+
+```
+gor --input-tcp :80 --output-http "http://staging.com" --output-http-elasticsearch "es_host:api_port/index_name"
 ```
 
 ## Additional help
+
+Feel free to ask question directly by email or by creating github issue.
+
+## Latest releases (including binaries)
+
+https://github.com/buger/gor/releases
+
+## Command line reference
+`gor -h` output:
 ```
-$ gor listen -h
-Usage of ./bin/gor-linux:
-  -i="any": By default it try to listen on all network interfaces.To get list of interfaces run `ifconfig`
-  -p=80: Specify the http server port whose traffic you want to capture
-  -r="localhost:28020": Address of replay server.
+  -cpuprofile="": write cpu profile to file
+  -memprofile="": write memory profile to this file
+  
+  -input-dummy=[]: Used for testing outputs. Emits 'Get /' request every 1s
+
+  -input-file=[]: Read requests from file: 
+    gor --input-file ./requests.gor --output-http staging.com
+
+  -input-raw=[]: Capture traffic from given port (use RAW sockets and require *sudo* access):
+    # Capture traffic from 8080 port
+    gor --input-raw :8080 --output-http staging.com
+
+  -input-tcp=[]: Used for internal communication between Gor instances. Example: 
+    # Receive requests from other Gor instances on 28020 port, and redirect output to staging
+    gor --input-tcp :28020 --output-http staging.com
+
+  -output-dummy=[]: Used for testing inputs. Just prints data coming from inputs.
+
+  -output-file=[]: Write incoming requests to file: 
+    gor --input-raw :80 --output-file ./requests.gor
+
+  -output-http=[]: Forwards incoming requests to given http address.
+    # Redirect all incoming requests to staging.com address 
+    gor --input-raw :80 --output-http http://staging.com
+
+  -output-http-elasticsearch="": Send request and response stats to ElasticSearch:
+    gor --input-raw :8080 --output-http staging.com --output-http-elasticsearch 'es_host:api_port/index_name'
+
+  -output-http-header=[]: Inject additional headers to http reqest:
+    gor --input-raw :8080 --output-http staging.com --output-http-header 'User-Agent: Gor'
+
+  -output-http-header-filter=[]: A regexp to match a specific header against. Requests with non-matching headers will be dropped:
+    gor --input-raw :8080 --output-http staging.com --output-http-header-filter api-version:^v1
+
+  -output-http-header-hash-filter=[]: Takes a fraction of requests, consistently taking or rejecting a request based on the FNV32-1A hash of a specific header. The fraction must have a denominator that is a power of two:
+    gor --input-raw :8080 --output-http staging.com --output-http-header-hash-filter user-id:1/4
+
+  -output-http-url-regexp=: A regexp to match requests against. Anything else will be dropped:
+    gor --input-raw :8080 --output-http staging.com --output-http-url-regexp ^www.
+    
+  -output-http-workers=-1: Number of http output workers desired. Use default -1 for dynamic worker scaling.  Gor will add http workers if its work queue starts getting too full and kill them .
+    
+  -output-http-stats=false: If set to `true` it gives out queuing stats for the HTTP output every 5 seconds in the form latest,mean,max,count,count/second.
+    
+  -output-tcp=[]: Used for internal communication between Gor instances. Example: 
+    # Listen for requests on 80 port and forward them to other Gor instance on 28020 port
+    gor --input-raw :80 --output-tcp replay.local:28020
+    
+  -output-tcp-stats=false: If set to `true` it gives out queuing stats for the TCP output every 5 seconds in the form latest,mean,max,count,count/second.
+    
+  -split-output=false: By default each output gets same traffic. If set to `true` it splits traffic equally among all outputs.
+
+  -output-http-rewrite-url=[]: Rewrites the url in the request based on a mapping
+    gor --input-raw :8080 --output-http staging.com --output-http-rewrite-url /xml_test/interface.php:/api/service.do
 ```
 
-```
-$ gor replay -h
-Usage of ./bin/gor-linux:
-  -f="http://localhost:8080": http address to forward traffic.
-	You can limit requests per second by adding `|#{num}` after address.
-	If you have multiple addresses with different limits. For example: http://staging.example.com|100,http://dev.example.com|10
-  -ip="0.0.0.0": ip addresses to listen on
-  -p=28020: specify port number
-```
+## Building from source
+
+1. Setup standard Go environment http://golang.org/doc/code.html and ensure that $GOPATH environment variable properly set.
+2. `go get github.com/buger/gor`.
+3. `cd $GOPATH/src/github.com/buger/gor`
+4. `go build` to get binary, or `go test` to run tests
+
+## Development
+Project contains Docker environment.
+
+1. Build container: `make dbuild`
+2. Run tests: `make dtest`
+3. Bash access to container: `make dbash`. Inside container you have python to run simple web server `python -m SimpleHTTPServer 8080` and `curl` to make http requests. 
+
+
+## Questions and support 
+
+All bug-reports and suggestions should go though Github Issues or our [Google Group](https://groups.google.com/forum/#!forum/gor-users). Or you can just send email to gor-users@googlegroups.com
+
+If you have some private questions you can send direct mail to leonsbox@gmail.com
 
 ## FAQ
 
-### Why `gor listener` requires sudo or root access?
-Listener works by sniffing traffic from given port. Its accessible only using sudo or root access.
+### What OS are supported?
+For now only Linux based. *BSD (including MacOS is not supported yet, check https://github.com/buger/gor/issues/22 for details)
 
-### Do you support all http request types?
-Right now it support only "GET" requests.
+### Why does the `--input-raw` requires sudo or root access?
+Listener works by sniffing traffic from a given port. It's accessible
+only by using sudo or root access.
 
-## TODO
+### I'm getting 'too many open files' error
+Typical linux shell has a small open files soft limit at 1024. You can easily raise that when you do this before starting your gor replay process:
+  
+  ulimit -n 64000
 
-Use buffering for request throttling instead of simple rate limiting. 
+More about ulimit: http://blog.thecodingmachine.com/content/solving-too-many-open-files-exception-red5-or-any-other-application
 
-Better stats
+### What do the stats commands do?
+Gor can report stats on the output-tcp and output-http request queues. Stats are reported to the console every 5 seconds in the form `latest,mean,max,count,count/second` by using the `-output-http-stats` and `-output-tcp-stats` options.
 
-Optimize for load testing cases
+Examples:
+
+```
+2014/04/23 21:17:50 output_tcp:latest,mean,max,count,count/second
+2014/04/23 21:17:50 output_tcp:0,0,0,0,0
+2014/04/23 21:17:55 output_tcp:1,1,2,68,13
+2014/04/23 21:18:00 output_tcp:1,1,2,92,18
+2014/04/23 21:18:05 output_tcp:1,1,2,119,23
+2014/04/23 21:18:10 output_tcp:1,0,1,95,19
+2014/04/23 21:18:15 output_tcp:1,1,2,92,18
+2014/04/23 21:18:20 output_tcp:1,1,2,108,21
+2014/04/23 21:18:25 output_tcp:1,1,2,117,23
+2014/04/23 21:18:30 output_tcp:1,1,2,113,22
+2014/04/23 21:18:35 output_tcp:21,20,21,132,26
+2014/04/23 21:18:40 output_tcp:100,99,100,99,19
+```
+
+```
+Version: 0.8
+2014/04/23 21:19:46 output_http:latest,mean,max,count,count/second
+2014/04/23 21:19:46 output_http:0,0,0,0,0
+2014/04/23 21:19:51 output_http:0,0,0,0,0
+2014/04/23 21:19:56 output_http:0,0,0,0,0
+2014/04/23 21:20:01 output_http:1,0,1,50,10
+2014/04/23 21:20:06 output_http:1,1,4,72,14
+2014/04/23 21:20:11 output_http:1,0,1,179,35
+2014/04/23 21:20:16 output_http:1,0,1,148,29
+2014/04/23 21:20:21 output_http:1,1,2,91,18
+2014/04/23 21:20:26 output_http:1,1,2,150,30
+2014/04/23 21:18:15 output_http:100,99,100,70,14
+2014/04/23 21:18:21 output_http:100,99,100,55,11
+2014/04/23 21:18:28 output_http:100,99,100,55,11
+2014/04/23 21:18:34 output_http:100,99,100,57,11
+2014/04/23 21:18:41 output_http:100,99,100,61,12
+2014/04/23 21:18:48 output_http:100,99,100,56,11
+2014/04/23 21:18:56 output_http:100,99,100,58,11
+2014/04/23 21:19:01 output_http:100,99,100,31,6
+2014/04/23 21:19:08 output_http:100,99,100,61,12
+2014/04/23 21:19:15 output_http:100,99,100,64,12
+2014/04/23 21:19:21 output_http:100,99,100,70,14
+2014/04/23 21:19:28 output_http:100,99,100,61,12
+2014/04/23 21:19:35 output_http:100,99,100,56,11
+```
+
+### How can I tell if I have bottlenecks?
+Key areas that sometimes experience bottlenecks are the output-tcp and output-http functions which have internal queues for requests. Each queue has an upper limit of 100. Enable stats reporting to see if any queues are experiencing bottleneck behavior.
+ 
+#### output-http bottlenecks
+When running a Gor replay the output-http feature may bottleneck if:
+
+  * the replay has inadequate bandwidth. If the replay is receiving or sending more messages than its network adapter can handle the output-http-stats  may report that the output-http queue is filling up. See if there is a way to upgrade the replay's bandwidth.
+  * with `--output-http-workers` set to anything other than `-1` the `-output-http` target is unable to respond to messages in a timely manner. The http output workers which take messages off the output-http queue, process the request, and ensure that the request did not result in an error may not be able to keep up with the number of incoming requests. If the replay is not using dynamic worker scaling (`--output-http-workers=-1`)  The optimal number of output-http-workers can be determined with the formula `output-workers = (Average number of requests per second)/(Average target response time per second)`.
+
+#### output-tcp bottlenecks
+When using the Gor listener the output-tcp feature may bottleneck if:
+
+  * the replay is unable to accept and process more requests than the listener is able generate. Prior to troubleshooting the output-tcp bottleneck, ensure that the replay target is not experiencing any bottlenecks. 
+  * the replay target has inadequate bandwidth to handle all its incoming requests.  If a replay target's incoming bandwidth is maxed out the output-tcp-stats may report that the output-tcp queue is filling up. See if there is a way to upgrade the replay's bandwidth.
+
+### The CPU average across my load-balanced targets is higher than the source
+If you are replaying traffic from multiple listeners to a load-balanced target and you use sticky sessions, you may observe that the target servers have a higher CPU load than the listener servers. This may be because the sticky session cookie of the original load balancer is not honored by the target load balancer thus resulting in requests that would normally hit the same target server hitting different servers on the backend thus reducing some caching benefits gained via the load balancing.  Try running just one listener against one replay target and see if the CPU utilization comparison is more accurate.
+ 
+### How does dynamic http worker scaling work?
+By using the Gor setting `--output-http-workers=-1` Gor will create more http output workers when the http output queue length is greater than 10.  The number of workers created (N) is equal to the queue length at the time which it is checked and found to have a length greater than 10. The queue length is checked every time a message is written to the http output queue.  No more workers will be spawned until that request to spawn N workers is satisfied.  If a dynamic worker cannot process a message at that time, it will sleep for 100 milliseconds. If a dynamic worker cannot process a message for 2 seconds it dies.  
+
+## Tuning
+
+To achieve the top most performance you should tune the source server system limits:
+
+    net.ipv4.tcp_max_tw_buckets = 65536
+    net.ipv4.tcp_tw_recycle = 1
+    net.ipv4.tcp_tw_reuse = 0
+    net.ipv4.tcp_max_syn_backlog = 131072
+    net.ipv4.tcp_syn_retries = 3
+    net.ipv4.tcp_synack_retries = 3
+    net.ipv4.tcp_retries1 = 3
+    net.ipv4.tcp_retries2 = 8
+    net.ipv4.tcp_rmem = 16384 174760 349520
+    net.ipv4.tcp_wmem = 16384 131072 262144
+    net.ipv4.tcp_mem = 262144 524288 1048576
+    net.ipv4.tcp_max_orphans = 65536
+    net.ipv4.tcp_fin_timeout = 10
+    net.ipv4.tcp_low_latency = 1
+    net.ipv4.tcp_syncookies = 0
+
+
+## Contributing
+
+1. Fork it
+2. Create your feature branch (git checkout -b my-new-feature)
+3. Commit your changes (git commit -am 'Added some feature')
+4. Push to the branch (git push origin my-new-feature)
+5. Create new Pull Request
+
+## Companies using Gor
+
+* [Granify](http://granify.com)
+* [GOV.UK](https://www.gov.uk) ([Government Digital Service](http://digital.cabinetoffice.gov.uk/))
+* [theguardian.com](http://theguardian.com)
+* [TomTom](http://www.tomtom.com/)
+* To add your company drop me a line to github.com/buger or leonsbox@gmail.com

elasticsearch.go

@@ -0,0 +1,153 @@
+package main
+
+import (
+	"encoding/json"
+	"github.com/buger/elastigo/api"
+	"github.com/buger/elastigo/core"
+	"log"
+	"net/http"
+	"regexp"
+	"time"
+)
+
+type ESUriErorr struct{}
+
+func (e *ESUriErorr) Error() string {
+	return "Wrong ElasticSearch URL format. Expected to be: host:port/index_name"
+}
+
+type ESPlugin struct {
+	Active  bool
+	ApiPort string
+	Host    string
+	Index   string
+	indexor *core.BulkIndexer
+	done    chan bool
+}
+
+type ESRequestResponse struct {
+	ReqUrl               string         `json:"Req_URL"`
+	ReqMethod            string         `json:"Req_Method"`
+	ReqUserAgent         string         `json:"Req_User-Agent"`
+	ReqAcceptLanguage    string         `json:"Req_Accept-Language,omitempty"`
+	ReqAccept            string         `json:"Req_Accept,omitempty"`
+	ReqAcceptEncoding    string         `json:"Req_Accept-Encoding,omitempty"`
+	ReqIfModifiedSince   string         `json:"Req_If-Modified-Since,omitempty"`
+	ReqConnection        string         `json:"Req_Connection,omitempty"`
+	ReqCookies           []*http.Cookie `json:"Req_Cookies,omitempty"`
+	RespStatus           string         `json:"Resp_Status"`
+	RespStatusCode       int            `json:"Resp_Status-Code"`
+	RespProto            string         `json:"Resp_Proto,omitempty"`
+	RespContentLength    int64          `json:"Resp_Content-Length,omitempty"`
+	RespContentType      string         `json:"Resp_Content-Type,omitempty"`
+	RespTransferEncoding []string       `json:"Resp_Transfer-Encoding,omitempty"`
+	RespContentEncoding  string         `json:"Resp_Content-Encoding,omitempty"`
+	RespExpires          string         `json:"Resp_Expires,omitempty"`
+	RespCacheControl     string         `json:"Resp_Cache-Control,omitempty"`
+	RespVary             string         `json:"Resp_Vary,omitempty"`
+	RespSetCookie        string         `json:"Resp_Set-Cookie,omitempty"`
+	Rtt                  int64          `json:"RTT"`
+	Timestamp            time.Time
+}
+
+// Parse ElasticSearch URI
+//
+// Proper format is: host:port/index_name
+func parseURI(URI string) (err error, host string, port string, index string) {
+	rURI := regexp.MustCompile("(.+):([0-9]+)/(.+)")
+	match := rURI.FindAllStringSubmatch(URI, -1)
+
+	if len(match) == 0 {
+		err = new(ESUriErorr)
+	} else {
+		host = match[0][1]
+		port = match[0][2]
+		index = match[0][3]
+	}
+
+	return
+}
+
+func (p *ESPlugin) Init(URI string) {
+	var err error
+
+	err, p.Host, p.ApiPort, p.Index = parseURI(URI)
+
+	if err != nil {
+		log.Fatal("Can't initialize ElasticSearch plugin.", err)
+	}
+
+	api.Domain = p.Host
+	api.Port = p.ApiPort
+
+	p.indexor = core.NewBulkIndexerErrors(50, 60)
+	p.done = make(chan bool)
+	p.indexor.Run(p.done)
+
+	// Only start the ErrorHandler goroutine when in verbose mode
+	// no need to burn ressources otherwise
+	// go p.ErrorHandler()
+
+	log.Println("Initialized Elasticsearch Plugin")
+	return
+}
+
+func (p *ESPlugin) IndexerShutdown() {
+	p.done <- true
+	return
+}
+
+func (p *ESPlugin) ErrorHandler() {
+	for {
+		errBuf := <-p.indexor.ErrorChannel
+		log.Println(errBuf.Err)
+	}
+}
+
+func (p *ESPlugin) RttDurationToMs(d time.Duration) int64 {
+	sec := d / time.Second
+	nsec := d % time.Second
+	fl := float64(sec) + float64(nsec)*1e-6
+	return int64(fl)
+}
+
+func (p *ESPlugin) ResponseAnalyze(req *http.Request, resp *http.Response, start, stop time.Time) {
+	if resp == nil {
+		// nil http response - skipped elasticsearch export for this request
+		return
+	}
+	t := time.Now()
+	rtt := p.RttDurationToMs(stop.Sub(start))
+
+	esResp := ESRequestResponse{
+		ReqUrl:               req.URL.String(),
+		ReqMethod:            req.Method,
+		ReqUserAgent:         req.UserAgent(),
+		ReqAcceptLanguage:    req.Header.Get("Accept-Language"),
+		ReqAccept:            req.Header.Get("Accept"),
+		ReqAcceptEncoding:    req.Header.Get("Accept-Encoding"),
+		ReqIfModifiedSince:   req.Header.Get("If-Modified-Since"),
+		ReqConnection:        req.Header.Get("Connection"),
+		ReqCookies:           req.Cookies(),
+		RespStatus:           resp.Status,
+		RespStatusCode:       resp.StatusCode,
+		RespProto:            resp.Proto,
+		RespContentLength:    resp.ContentLength,
+		RespContentType:      resp.Header.Get("Content-Type"),
+		RespTransferEncoding: resp.TransferEncoding,
+		RespContentEncoding:  resp.Header.Get("Content-Encoding"),
+		RespExpires:          resp.Header.Get("Expires"),
+		RespCacheControl:     resp.Header.Get("Cache-Control"),
+		RespVary:             resp.Header.Get("Vary"),
+		RespSetCookie:        resp.Header.Get("Set-Cookie"),
+		Rtt:                  rtt,
+		Timestamp:            t,
+	}
+	j, err := json.Marshal(&esResp)
+	if err != nil {
+		log.Println(err)
+	} else {
+		p.indexor.Index(p.Index, "RequestResponse", "", "", &t, j, true)
+	}
+	return
+}

emitter.go

@@ -0,0 +1,53 @@
+package main
+
+import (
+	"io"
+)
+
+func Start(stop chan int) {
+	for _, in := range Plugins.Inputs {
+		go CopyMulty(in, Plugins.Outputs...)
+	}
+
+	select {
+	case <-stop:
+		return
+	}
+}
+
+// Copy from 1 reader to multiple writers
+func CopyMulty(src io.Reader, writers ...io.Writer) (err error) {
+	buf := make([]byte, 32*1024)
+	wIndex := 0
+
+	for {
+		nr, er := src.Read(buf)
+		if nr > 0 && len(buf) > nr{
+			Debug("Sending", src, ": ", string(buf[0:nr]))
+
+			if Settings.splitOutput {
+				// Simple round robin
+				writers[wIndex].Write(buf[0:nr])
+
+				wIndex++
+
+				if wIndex >= len(writers) {
+					wIndex = 0
+				}
+			} else {
+				for _, dst := range writers {
+					dst.Write(buf[0:nr])
+				}
+			}
+
+		}
+		if er == io.EOF {
+			break
+		}
+		if er != nil {
+			err = er
+			break
+		}
+	}
+	return err
+}

emitter_test.go

@@ -0,0 +1,99 @@
+package main
+
+import (
+	"io"
+	"sync"
+	"sync/atomic"
+	"testing"
+)
+
+func TestEmitter(t *testing.T) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	input := NewTestInput()
+	output := NewTestOutput(func(data []byte) {
+		wg.Done()
+	})
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	go Start(quit)
+
+	for i := 0; i < 1000; i++ {
+		wg.Add(1)
+		input.EmitGET()
+	}
+
+	wg.Wait()
+
+	close(quit)
+}
+
+func TestEmitterRoundRobin(t *testing.T) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	input := NewTestInput()
+
+	var counter1, counter2 int32
+
+	output1 := NewTestOutput(func(data []byte) {
+		atomic.AddInt32(&counter1, 1)
+		wg.Done()
+	})
+
+	output2 := NewTestOutput(func(data []byte) {
+		atomic.AddInt32(&counter2, 1)
+		wg.Done()
+	})
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output1, output2}
+
+	Settings.splitOutput = true
+
+	go Start(quit)
+
+	for i := 0; i < 1000; i++ {
+		wg.Add(1)
+		input.EmitGET()
+	}
+
+	wg.Wait()
+
+	close(quit)
+
+	if counter1 == 0 || counter2 == 0 {
+		t.Errorf("Round robin should split traffic equally: %d vs %d", counter1, counter2)
+	}
+
+	Settings.splitOutput = false
+}
+
+func BenchmarkEmitter(b *testing.B) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	input := NewTestInput()
+
+	output := NewTestOutput(func(data []byte) {
+		wg.Done()
+	})
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	go Start(quit)
+
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		wg.Add(1)
+		input.EmitGET()
+	}
+
+	wg.Wait()
+	close(quit)
+}

gor.go

@@ -1,46 +1,78 @@
 // Gor is simple http traffic replication tool written in Go. Its main goal to replay traffic from production servers to staging and dev environments.
 // Now you can test your code on real user sessions in an automated and repeatable fashion.
-//
-// Gor consists of 2 parts: listener and replay servers.
-// Listener catch http traffic from given port in real-time and send it to replay server via UDP. Replay server forwards traffic to given address.
 package main
 
 import (
 	"flag"
 	"fmt"
-	"github.com/buger/gor/listener"
-	"github.com/buger/gor/replay"
+	"log"
 	"os"
+	"runtime/debug"
+	"runtime/pprof"
+	"time"
 )
 
-const (
-	VERSION = "0.1.1"
+var (
+	mode       string
+	cpuprofile = flag.String("cpuprofile", "", "write cpu profile to file")
+	memprofile = flag.String("memprofile", "", "write memory profile to this file")
 )
 
 func main() {
+	// Don't exit on panic
+	defer func() {
+		if r := recover(); r != nil {
+			if _, ok := r.(error); !ok {
+				fmt.Printf("PANIC: pkg: %v %s \n", r, debug.Stack())
+			}
+		}
+	}()
+
 	fmt.Println("Version:", VERSION)
 
-	mode := "unknown"
-
-	if len(os.Args) > 1 {
-		mode = os.Args[1]
-	}
-
-	if mode != "listen" && mode != "replay" {
-		fmt.Println("Usage: \n\tgor listen -h\n\tgor replay -h")
-		return
-	}
-
-	// Remove mode attr
-	os.Args = append(os.Args[:1], os.Args[2:]...)
-
 	flag.Parse()
+	InitPlugins()
 
-	switch mode {
-	case "listen":
-		listener.Run()
-	case "replay":
-		replay.Run()
+	if len(Plugins.Inputs) == 0 || len(Plugins.Outputs) == 0 {
+		log.Fatal("Required at least 1 input and 1 output")
 	}
 
+	if *memprofile != "" {
+		profileMEM(*memprofile)
+	}
+
+	if *cpuprofile != "" {
+		profileCPU(*cpuprofile)
+	}
+
+	Start(nil)
+}
+
+func profileCPU(cpuprofile string) {
+	if cpuprofile != "" {
+		f, err := os.Create(cpuprofile)
+		if err != nil {
+			log.Fatal(err)
+		}
+		pprof.StartCPUProfile(f)
+
+		time.AfterFunc(60*time.Second, func() {
+			pprof.StopCPUProfile()
+			f.Close()
+			log.Println("Stop profiling after 60 seconds")
+		})
+	}
+}
+
+func profileMEM(memprofile string) {
+	if memprofile != "" {
+		f, err := os.Create(memprofile)
+		if err != nil {
+			log.Fatal(err)
+		}
+		time.AfterFunc(60*time.Second, func() {
+			pprof.WriteHeapProfile(f)
+			f.Close()
+		})
+	}
 }

gor_stat.go

@@ -0,0 +1,67 @@
+package main
+
+import (
+	"time"
+	"log"
+	"strconv"
+)
+
+const (
+	rate = 5
+)
+
+type GorStat struct {
+	statName	string
+	latest		int
+	mean	 	int
+	max			int
+	count		int
+}
+
+func NewGorStat(statName string) (s *GorStat) {
+	s = new(GorStat)
+	s.statName = statName
+	s.latest = 0
+	s.mean = 0
+	s.max = 0
+	s.count = 0
+
+	if Settings.stats {
+		log.Println(s.statName + ":latest,mean,max,count,count/second")
+		go s.reportStats()
+	}
+	return
+}
+
+func (s *GorStat) Write(latest int) {
+	if Settings.stats {
+		if latest > s.max {
+			s.max = latest
+		}
+		if latest != 0 {
+			s.mean = (s.mean + latest) / 2
+		}
+		s.latest = latest
+		s.count = s.count + 1
+	}
+}
+
+func (s *GorStat) Reset() {
+	s.latest = 0
+	s.max = 0
+	s.mean = 0
+	s.count = 0
+}
+
+func (s *GorStat) String() string {
+	return s.statName + ":" + strconv.Itoa(s.latest) + "," + strconv.Itoa(s.mean) + "," + strconv.Itoa(s.max) + "," + strconv.Itoa(s.count) + "," + strconv.Itoa(s.count / rate)
+}
+
+func (s *GorStat) reportStats() {
+	for {
+			log.Println(s)
+			s.Reset()
+			time.Sleep(rate * time.Second)
+	}
+}
+

input_dummy.go

@@ -0,0 +1,40 @@
+package main
+
+import (
+	"time"
+)
+
+type DummyInput struct {
+	data chan []byte
+}
+
+func NewDummyInput(options string) (di *DummyInput) {
+	di = new(DummyInput)
+	di.data = make(chan []byte)
+
+	go di.emit()
+
+	return
+}
+
+func (i *DummyInput) Read(data []byte) (int, error) {
+	buf := <-i.data
+	copy(data, buf)
+
+	return len(buf), nil
+}
+
+func (i *DummyInput) emit() {
+	ticker := time.NewTicker(time.Second)
+
+	for {
+		select {
+		case <-ticker.C:
+			i.data <- []byte("GET / HTTP/1.1\r\n\r\n")
+		}
+	}
+}
+
+func (i *DummyInput) String() string {
+	return "Dummy Input"
+}

input_file.go

@@ -0,0 +1,66 @@
+package main
+
+import (
+	"encoding/gob"
+	"log"
+	"os"
+	"time"
+)
+
+type FileInput struct {
+	data    chan []byte
+	path    string
+	decoder *gob.Decoder
+}
+
+func NewFileInput(path string) (i *FileInput) {
+	i = new(FileInput)
+	i.data = make(chan []byte)
+	i.path = path
+	i.Init(path)
+
+	go i.emit()
+
+	return
+}
+
+func (i *FileInput) Init(path string) {
+	file, err := os.Open(path)
+
+	if err != nil {
+		log.Fatal(i, "Cannot open file %q. Error: %s", path, err)
+	}
+
+	i.decoder = gob.NewDecoder(file)
+}
+
+func (i *FileInput) Read(data []byte) (int, error) {
+	buf := <-i.data
+	copy(data, buf)
+
+	return len(buf), nil
+}
+
+func (i *FileInput) String() string {
+	return "File input: " + i.path
+}
+
+func (i *FileInput) emit() {
+	var lastTime int64
+
+	for {
+		raw := new(RawRequest)
+		err := i.decoder.Decode(raw)
+
+		if err != nil {
+			return
+		}
+
+		if lastTime != 0 {
+			time.Sleep(time.Duration(raw.Timestamp - lastTime))
+		}
+		lastTime = raw.Timestamp
+
+		i.data <- raw.Request
+	}
+}

input_raw.go

@@ -0,0 +1,53 @@
+package main
+
+import (
+	raw "github.com/buger/gor/raw_socket_listener"
+	"log"
+	"net"
+	"strings"
+)
+
+type RAWInput struct {
+	data    chan []byte
+	address string
+}
+
+func NewRAWInput(address string) (i *RAWInput) {
+	i = new(RAWInput)
+	i.data = make(chan []byte)
+	i.address = address
+
+	go i.listen(address)
+
+	return
+}
+
+func (i *RAWInput) Read(data []byte) (int, error) {
+	buf := <-i.data
+	copy(data, buf)
+
+	return len(buf), nil
+}
+
+func (i *RAWInput) listen(address string) {
+	address = strings.Replace(address, "[::]", "127.0.0.1", -1)
+
+	host, port, err := net.SplitHostPort(address)
+
+	if err != nil {
+		log.Fatal("input-raw: error while parsing address", err)
+	}
+
+	listener := raw.NewListener(host, port)
+
+	for {
+		// Receiving TCPMessage object
+		m := listener.Receive()
+
+		i.data <- m.Bytes()
+	}
+}
+
+func (i *RAWInput) String() string {
+	return "RAW Socket input: " + i.address
+}

input_raw_test.go

@@ -0,0 +1,39 @@
+package main
+
+import (
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+	"testing"
+)
+
+func TestRAWInput(t *testing.T) {
+
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	listener := startHTTP(func(req *http.Request) {})
+
+	input := NewRAWInput(listener.Addr().String())
+	output := NewTestOutput(func(data []byte) {
+		wg.Done()
+	})
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	address := strings.Replace(listener.Addr().String(), "[::]", "127.0.0.1", -1)
+
+	go Start(quit)
+
+	for i := 0; i < 100; i++ {
+		wg.Add(1)
+		res, _ := http.Get("http://" + address)
+		res.Body.Close()
+	}
+
+	wg.Wait()
+
+	close(quit)
+}

input_tcp.go

@@ -0,0 +1,84 @@
+package main
+
+import (
+	"bufio"
+	"io"
+	"log"
+	"net"
+)
+
+// Can be tested using nc tool:
+//    echo "asdad" | nc 127.0.0.1 27017
+//
+type TCPInput struct {
+	data     chan []byte
+	address  string
+	listener net.Listener
+}
+
+func NewTCPInput(address string) (i *TCPInput) {
+	i = new(TCPInput)
+	i.data = make(chan []byte)
+	i.address = address
+
+	i.listen(address)
+
+	return
+}
+
+func (i *TCPInput) Read(data []byte) (int, error) {
+	buf := <-i.data
+	copy(data, buf)
+
+	return len(buf), nil
+}
+
+func (i *TCPInput) listen(address string) {
+	listener, err := net.Listen("tcp", address)
+	i.listener = listener
+
+	if err != nil {
+		log.Fatal("Can't start:", err)
+	}
+
+	go func() {
+		for {
+			conn, err := listener.Accept()
+
+			if err != nil {
+				log.Println("Error while Accept()", err)
+				continue
+			}
+
+			go i.handleConnection(conn)
+		}
+	}()
+}
+
+func (i *TCPInput) handleConnection(conn net.Conn) {
+	defer conn.Close()
+
+	reader := bufio.NewReader(conn)
+
+	for {
+		buf,err := reader.ReadBytes('¶')
+		buf_len := len(buf)
+		if buf_len > 0 {
+			new_buf_len := len(buf) - 2
+			if new_buf_len > 0 {
+				new_buf := make([]byte, new_buf_len)
+				copy(new_buf, buf[:new_buf_len])
+				i.data <- new_buf
+				if err != nil {
+					if err != io.EOF {
+						log.Printf("error: %s\n", err)
+					}
+				}
+			}
+		}
+	}
+}
+
+func (i *TCPInput) String() string {
+	return "TCP input: " + i.address
+}

input_tcp_test.go

@@ -0,0 +1,112 @@
+package main
+
+import (
+	"io"
+	"log"
+	"net"
+	"sync"
+	"testing"
+)
+
+func TestTCPInput(t *testing.T) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	input := NewTCPInput(":0")
+	output := NewTestOutput(func(data []byte) {
+		wg.Done()
+	})
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	go Start(quit)
+
+	tcpAddr, err := net.ResolveTCPAddr("tcp", input.listener.Addr().String())
+
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	conn, err := net.DialTCP("tcp", nil, tcpAddr)
+
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	msg := []byte("GET / HTTP/1.1\r\n\r\n")
+
+	for i := 0; i < 100; i++ {
+		wg.Add(1)
+		new_buf := make([]byte, len(msg) + 2)
+		msg = append(msg,[]byte("¶")...)
+		copy(new_buf, msg)
+		conn.Write(new_buf)
+	}
+
+	wg.Wait()
+
+	close(quit)
+}
+
+func BenchmarkTCPInput(b *testing.B) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	input := NewTCPInput(":0")
+	output := NewTestOutput(func(data []byte) {
+		wg.Done()
+	})
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	go Start(quit)
+
+	tcpAddr, err := net.ResolveTCPAddr("tcp", input.listener.Addr().String())
+
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	var connections []net.Conn
+
+	// Creating simple pool of workers, same as output_tcp have
+	dataChan := make(chan []byte, 1000)
+
+	for i := 0; i < 10; i++ {
+		conn, _ := net.DialTCP("tcp", nil, tcpAddr)
+		connections = append(connections, conn)
+
+		go func(conn net.Conn) {
+			for {
+				data := <-dataChan
+
+				new_buf := make([]byte, len(data) + 2)
+				data = append(data,[]byte("¶")...)
+				copy(new_buf, data)
+				conn.Write(new_buf)
+			}
+		}(conn)
+	}
+
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	msg := []byte("GET / HTTP/1.1\r\n\r\n")
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		wg.Add(1)
+		dataChan <- msg
+	}
+
+	wg.Wait()
+
+	for _, conn := range connections {
+		conn.Close()
+	}
+
+	close(quit)
+}

limiter.go

@@ -0,0 +1,45 @@
+package main
+
+import (
+	"fmt"
+	"io"
+	"time"
+)
+
+type Limiter struct {
+	writer io.Writer
+	limit  int
+
+	currentRPS  int
+	currentTime int64
+}
+
+func NewLimiter(writer io.Writer, limit int) (l *Limiter) {
+	l = new(Limiter)
+	l.limit = limit
+	l.writer = writer
+	l.currentTime = time.Now().UnixNano()
+
+	return
+}
+
+func (l *Limiter) Write(data []byte) (n int, err error) {
+	if (time.Now().UnixNano() - l.currentTime) > time.Second.Nanoseconds() {
+		l.currentTime = time.Now().UnixNano()
+		l.currentRPS = 0
+	}
+
+	if l.currentRPS >= l.limit {
+		return 0, nil
+	}
+
+	n, err = l.writer.Write(data)
+
+	l.currentRPS++
+
+	return
+}
+
+func (l *Limiter) String() string {
+	return fmt.Sprintf("Limiting %s to: %d", l.writer, l.limit)
+}

limiter_test.go

@@ -0,0 +1,31 @@
+package main
+
+import (
+	"io"
+	"sync"
+	"testing"
+)
+
+func TestLimiter(t *testing.T) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	input := NewTestInput()
+	output := NewLimiter(NewTestOutput(func(data []byte) {
+		wg.Done()
+	}), 10)
+	wg.Add(10)
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	go Start(quit)
+
+	for i := 0; i < 100; i++ {
+		input.EmitGET()
+	}
+
+	wg.Wait()
+
+	close(quit)
+}

listener/listener.go

@@ -1,159 +0,0 @@
-// Listener capture TCP traffic right from given port using `tcpdump` utility.
-// Note: it requires sudo or root access.
-//
-// Rigt now it suport only HTTP, and only GET requests.
-package listener
-
-import (
-	"bufio"
-	"bytes"
-	"encoding/gob"
-	"flag"
-	"fmt"
-	"io"
-	"log"
-	"net"
-	"os"
-	"os/exec"
-	"regexp"
-	"strings"
-)
-
-type HttpRequest struct {
-	Tag     string            // Not used yet
-	Method  string            // Right now only 'GET'
-	Url     string            // Request URL
-	Headers map[string]string // Request Headers
-}
-
-// Enable debug logging only if "--verbose" flag passed
-func Debug(v ...interface{}) {
-	if Settings.verbose { log.Println(v...) }
-}
-
-// Parse `tcpdump` output to find HTTP GET requests
-// When HttpRequest found it get send to `requests` channel
-func parseRequest(pipe io.ReadCloser, requests chan *HttpRequest) {
-	request_re := regexp.MustCompile("(GET) (/.*) HTTP/1.1")
-	headers_re := regexp.MustCompile("([^ ]*): (.*)")
-
-	reader := bufio.NewScanner(pipe)
-
-	var request *HttpRequest
-
-	var requestStarted = false
-
-	for reader.Scan() {
-		line := reader.Text()
-
-		// HTTP/1.1 match finds both requests and response
-		// Index is used instead of Regexp just for speed
-		if strings.Index(line, "HTTP/1.1") != -1 {
-			// Allow only requests
-			match := request_re.FindAllString(line, -1)
-
-			if len(match) > 0 {
-				info := strings.Split(match[0], " ")
-
-				request = &HttpRequest{
-					Method:  info[0],
-					Url:     info[1],
-					Headers: make(map[string]string),
-				}
-
-				requestStarted = true
-			}
-		}
-
-		if requestStarted {
-			// We assume that empty line is end of request info
-			// This is true only for GET requests
-			if line == "" {
-				requests <- request
-				requestStarted = false
-			} else {
-				// All headers comes in this format:
-				//
-				//     User-Agent: Mozilla
-				//     Content-Type: text/html
-				//
-				match := headers_re.FindAllString(line, -1)
-
-				if len(match) > 0 {
-					header := strings.Split(match[0], ": ")
-
-					request.Headers[header[0]] = header[1]
-				}
-			}
-		}
-	}
-}
-
-// Sends request to replay server via UDP
-// Before sending it encode request object using standard gob encoder
-func forwardRequest(requests chan *HttpRequest) {
-	serverAddr, err := net.ResolveUDPAddr("udp4", Settings.ReplayServer())
-	conn, err := net.DialUDP("udp", nil, serverAddr)
-
-	defer conn.Close()
-
-	if err != nil {
-		log.Fatal("Connection error", err)
-	}
-
-	for {
-		select {
-		case request := <-requests:
-			Debug("Forwarding:", request.Url, "to", Settings.ReplayServer())
-
-			msg := bytes.Buffer{}
-
-			enc := gob.NewEncoder(&msg)
-			err := enc.Encode(request)
-
-			conn.Write(msg.Bytes())
-
-			if err != nil {
-				log.Println("encode error:", err)
-			}
-		}
-	}
-}
-
-
-func greeting() {	
-	fmt.Println("Listening for HTTP traffic on", Settings.port, "port")
-	fmt.Println("Running: tcpdump "+strings.Join(Settings.TCPDumpConfig()," "))
-	fmt.Println("Forwarding requests to replay server:", Settings.ReplayServer())
-}
-
-
-// Because its sub-program, Run acts as `main`
-func Run() {
-	if os.Getuid() != 0 {
-		fmt.Println("Please start the listener as root or sudo!")
-		fmt.Println("This is required since listener sniff traffic on given port.")
-		os.Exit(1)
-	}
-	
-	// TODO: use RAW_SOCKETS instead of tcpdump
-	cmd := exec.Command("tcpdump", Settings.TCPDumpConfig()...)
-
-	greeting()
-
-	stdout, _ := cmd.StdoutPipe()
-	cmd.Stderr = os.Stderr
-
-	if err := cmd.Start(); err != nil {
-		log.Fatal(err)
-	}
-
-	requests := make(chan *HttpRequest)
-
-	go parseRequest(stdout, requests)
-	go forwardRequest(requests)
-
-	if err := cmd.Wait(); err != nil {
-		flag.Usage()
-	}
-}

listener/settings.go

@@ -1,54 +0,0 @@
-package listener
-
-import (
-    "strconv"
-    "strings"
-    "flag"
-    "os"
-)
-
-const (
-    defaultPort             = 80
-    defaultNetworkInterface = "any"
-
-    defaultReplayAddress = "localhost:28020"
-)
-
-type ListenerSettings struct {
-    networkInterface string
-    port             int
-
-    replayAddress string
-
-    verbose bool
-}
-
-var Settings ListenerSettings = ListenerSettings{}
-
-func (s *ListenerSettings) ReplayServer() string {
-    if !strings.Contains(s.replayAddress, ":") {
-        return s.replayAddress + ":28020"
-    }
-
-    return s.replayAddress
-}
-
-// tcpdump -vv -A -i all port 8080
-func (s *ListenerSettings) TCPDumpConfig() []string {
-    return []string{"-vv", "-A", "-i", Settings.networkInterface, "port "+strconv.Itoa(Settings.port)}
-}
-
-
-func init() {
-    if len(os.Args) < 2 || os.Args[1] != "listen" {
-        return
-    }
-
-    flag.IntVar(&Settings.port, "p", defaultPort, "Specify the http server port whose traffic you want to capture")
-
-    flag.StringVar(&Settings.networkInterface, "i", defaultNetworkInterface, "By default it try to listen on all network interfaces.To get list of interfaces run `ifconfig`")
-
-    flag.StringVar(&Settings.replayAddress, "r", defaultReplayAddress, "Address of replay server.")
-
-    flag.BoolVar(&Settings.verbose, "verbose", false, "Log requests")
-}

output_dummy.go

@@ -0,0 +1,24 @@
+package main
+
+import (
+	"fmt"
+)
+
+type DummyOutput struct {
+}
+
+func NewDummyOutput(options string) (di *DummyOutput) {
+	di = new(DummyOutput)
+
+	return
+}
+
+func (i *DummyOutput) Write(data []byte) (int, error) {
+	fmt.Println("Writing message: ", data)
+
+	return len(data), nil
+}
+
+func (i *DummyOutput) String() string {
+	return "Dummy Output"
+}

output_file.go

@@ -0,0 +1,51 @@
+package main
+
+import (
+	"encoding/gob"
+	"log"
+	"os"
+	"time"
+)
+
+type RawRequest struct {
+	Timestamp int64
+	Request   []byte
+}
+
+type FileOutput struct {
+	path    string
+	encoder *gob.Encoder
+	file    *os.File
+}
+
+func NewFileOutput(path string) (o *FileOutput) {
+	o = new(FileOutput)
+	o.path = path
+	o.Init(path)
+
+	return
+}
+
+func (o *FileOutput) Init(path string) {
+	var err error
+
+	o.file, err = os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0660)
+
+	if err != nil {
+		log.Fatal(o, "Cannot open file %q. Error: %s", path, err)
+	}
+
+	o.encoder = gob.NewEncoder(o.file)
+}
+
+func (o *FileOutput) Write(data []byte) (n int, err error) {
+	raw := RawRequest{time.Now().UnixNano(), data}
+
+	o.encoder.Encode(raw)
+
+	return len(data), nil
+}
+
+func (o *FileOutput) String() string {
+	return "File output: " + o.path
+}

output_file_test.go

@@ -0,0 +1,42 @@
+package main
+
+import (
+	"io"
+	"sync"
+	"testing"
+)
+
+func TestFileOutput(t *testing.T) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	input := NewTestInput()
+	output := NewFileOutput("/tmp/test_requests.gor")
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	go Start(quit)
+
+	for i := 0; i < 100; i++ {
+		wg.Add(2)
+		input.EmitGET()
+		input.EmitPOST()
+	}
+	close(quit)
+
+	quit = make(chan int)
+
+	input2 := NewFileInput("/tmp/test_requests.gor")
+	output2 := NewTestOutput(func(data []byte) {
+		wg.Done()
+	})
+
+	Plugins.Inputs = []io.Reader{input2}
+	Plugins.Outputs = []io.Writer{output2}
+
+	go Start(quit)
+
+	wg.Wait()
+	close(quit)
+}

output_http.go

@@ -0,0 +1,231 @@
+package main
+
+import (
+	"bufio"
+	"bytes"	
+	"io"
+	"log"
+	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
+	"time"
+)
+
+type RedirectNotAllowed struct{}
+
+func (e *RedirectNotAllowed) Error() string {
+	return "Redirects not allowed"
+}
+
+// customCheckRedirect disables redirects https://github.com/buger/gor/pull/15
+func customCheckRedirect(req *http.Request, via []*http.Request) error {
+	if len(via) >= 0 {
+		return new(RedirectNotAllowed)
+	}
+	return nil
+}
+
+// ParseRequest in []byte returns a http request or an error
+func ParseRequest(data []byte) (request *http.Request, err error) {
+	buf := bytes.NewBuffer(data)
+	reader := bufio.NewReader(buf)
+
+	request, err = http.ReadRequest(reader)
+
+	return
+}
+
+type HTTPOutput struct {
+	address string
+	limit   int
+	buf     chan []byte
+	needWorker chan int
+
+	urlRegexp             HTTPUrlRegexp
+	headerFilters         HTTPHeaderFilters
+	headerHashFilters     HTTPHeaderHashFilters
+    outputHTTPUrlRewrite  UrlRewriteMap
+
+	headers HTTPHeaders
+	methods HTTPMethods
+
+	elasticSearch *ESPlugin
+
+	bufStats *GorStat
+}
+
+func NewHTTPOutput(options string, headers HTTPHeaders, methods HTTPMethods, urlRegexp HTTPUrlRegexp, headerFilters HTTPHeaderFilters, headerHashFilters HTTPHeaderHashFilters, elasticSearchAddr string, outputHTTPUrlRewrite UrlRewriteMap) io.Writer {
+
+	o := new(HTTPOutput)
+
+	optionsArr := strings.Split(options, "|")
+	address := optionsArr[0]
+
+	if !strings.HasPrefix(address, "http") {
+		address = "http://" + address
+	}
+
+	o.address = address
+	o.headers = headers
+	o.methods = methods
+
+	o.urlRegexp = urlRegexp
+	o.headerFilters = headerFilters
+	o.headerHashFilters = headerHashFilters
+	o.outputHTTPUrlRewrite = outputHTTPUrlRewrite
+
+	o.buf = make(chan []byte, 100)
+	if Settings.outputHTTPStats {
+		o.bufStats = NewGorStat("output_http")
+	}
+	if Settings.outputHTTPWorkers == -1 {
+		o.needWorker = make(chan int)
+	}
+
+	if elasticSearchAddr != "" {
+		o.elasticSearch = new(ESPlugin)
+		o.elasticSearch.Init(elasticSearchAddr)
+	}
+
+	if len(optionsArr) > 1 {
+		o.limit, _ = strconv.Atoi(optionsArr[1])
+	}
+
+	go o.WorkerMaster(Settings.outputHTTPWorkers)
+
+	if o.limit > 0 {
+		return NewLimiter(o, o.limit)
+	} else {
+		return o
+	}
+}
+
+func (o *HTTPOutput) WorkerMaster(n int) {
+	for i := 0; i < n; i++ {
+		go o.Worker()
+	}
+
+	if Settings.outputHTTPWorkers == -1 {
+		for {
+			new_workers := <-o.needWorker
+			for i := 0; i < new_workers; i++ {
+				go o.Worker()
+			}
+		}
+	}
+}
+
+func (o *HTTPOutput) Worker() {
+	client := &http.Client{
+		CheckRedirect: customCheckRedirect,
+	}
+	death_count := 0
+	Loop:
+		for {
+			select {
+				case data := <-o.buf:
+				o.sendRequest(client, data)
+				death_count = 0
+			default:
+				if Settings.outputHTTPWorkers == -1 {
+					death_count += 1
+				}
+				if death_count > 20 {
+					break Loop
+				} else {
+					time.Sleep(time.Millisecond * 100)
+				}
+
+			}
+		}
+}
+
+func (o *HTTPOutput) Write(data []byte) (n int, err error) {
+	buf := make([]byte, len(data))
+	copy(buf, data)
+
+	o.buf <- buf
+	buf_len := len(o.buf)
+	if Settings.outputHTTPStats {
+		o.bufStats.Write(len(o.buf))
+	}
+
+	if Settings.outputHTTPWorkers == -1 {
+		if buf_len > 10 {
+			if len(o.needWorker) == 0 {
+				o.needWorker <- buf_len
+			}
+		}
+	}
+	return len(data), nil
+}
+
+func (o *HTTPOutput) sendRequest(client *http.Client, data []byte) {
+	request, err := ParseRequest(data)
+
+	if err != nil {
+		log.Println("Cannot parse request", string(data), err)
+		return
+	}
+
+	if len(o.methods) > 0 && !o.methods.Contains(request.Method) {
+		return
+	}
+
+	if !(o.urlRegexp.Good(request) && o.headerFilters.Good(request) && o.headerHashFilters.Good(request)) {
+		return
+	}
+
+    // Rewrite the path as necessary
+    request.URL.Path = o.outputHTTPUrlRewrite.Rewrite(request.URL.Path)
+
+	// Change HOST of original request
+	URL := o.address + request.URL.Path + "?" + request.URL.RawQuery
+
+	request.RequestURI = ""
+	request.URL, _ = url.ParseRequestURI(URL)
+
+	for _, header := range o.headers {
+		SetHeader(request, header.Name, header.Value)
+	}
+
+	start := time.Now()
+	resp, err := client.Do(request)
+	stop := time.Now()
+
+	// We should not count Redirect as errors
+	if urlErr, ok := err.(*url.Error); ok {
+		if _, ok := urlErr.Err.(*RedirectNotAllowed); ok {
+			err = nil
+		}
+	}
+
+	if err == nil {
+		defer resp.Body.Close()
+	} else {
+		log.Println("Request error:", err)
+	}
+
+	if o.elasticSearch != nil {
+		o.elasticSearch.ResponseAnalyze(request, resp, start, stop)
+	}
+}
+
+func SetHeader(request *http.Request, name string, value string) {
+
+	// Need to check here for the Host header as it needs to be set on the request and not as a separate header
+	// http.ReadRequest sets it by default to the URL Host of the request being read
+	if name == "Host" {
+		request.Host = value
+	} else {
+		request.Header.Set(name, value)
+	}
+
+	return
+
+}
+
+func (o *HTTPOutput) String() string {
+	return "HTTP output: " + o.address
+}

output_http_test.go

@@ -0,0 +1,114 @@
+package main
+
+import (
+	"io"
+	"net"
+	"net/http"
+	"sync"
+	"testing"
+	"time"
+)
+
+func startHTTP(cb func(*http.Request)) net.Listener {
+	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		go cb(r)
+	})
+
+	listener, _ := net.Listen("tcp", ":0")
+
+	go http.Serve(listener, handler)
+
+	return listener
+}
+
+func TestSetHeader(t *testing.T) {
+
+	req := &http.Request{
+		Header: make(map[string][]string),
+	}
+	req.Host = "test.com"
+
+	SetHeader(req, "Host", "test2.com")
+
+	if req.Host != "test2.com" {
+		t.Error("Expected test2.com - got ", req.Host)
+	}
+
+	SetHeader(req, "test_header", "test_value")
+
+	if req.Header.Get("test_header") != "test_value" {
+		t.Error("Wrong header value found")
+	}
+
+}
+
+func TestHTTPOutput(t *testing.T) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	input := NewTestInput()
+
+	headers := HTTPHeaders{HTTPHeader{"User-Agent", "Gor"}}
+	methods := HTTPMethods{"GET", "PUT", "POST"}
+
+	listener := startHTTP(func(req *http.Request) {
+		if req.Header.Get("User-Agent") != "Gor" {
+			t.Error("Wrong header")
+		}
+
+		if req.Method == "OPTIONS" {
+			t.Error("Wrong method")
+		}
+
+		wg.Done()
+	})
+
+	output := NewHTTPOutput(listener.Addr().String(), headers, methods, HTTPUrlRegexp{}, HTTPHeaderFilters{}, HTTPHeaderHashFilters{}, "", UrlRewriteMap{})
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	go Start(quit)
+
+	for i := 0; i < 100; i++ {
+		wg.Add(2)
+		input.EmitPOST()
+		input.EmitOPTIONS()
+		input.EmitGET()
+	}
+
+	wg.Wait()
+
+	close(quit)
+}
+
+func BenchmarkHTTPOutput(b *testing.B) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	input := NewTestInput()
+
+	headers := HTTPHeaders{HTTPHeader{"User-Agent", "Gor"}}
+	methods := HTTPMethods{"GET", "PUT", "POST"}
+
+	listener := startHTTP(func(req *http.Request) {
+		time.Sleep(50 * time.Millisecond)
+		wg.Done()
+	})
+
+	output := NewHTTPOutput(listener.Addr().String(), headers, methods, HTTPUrlRegexp{}, HTTPHeaderFilters{}, HTTPHeaderHashFilters{}, "",UrlRewriteMap{})
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	go Start(quit)
+
+	for i := 0; i < b.N; i++ {
+		wg.Add(1)
+		input.EmitPOST()
+	}
+
+	wg.Wait()
+
+	close(quit)
+}

output_tcp.go

@@ -0,0 +1,78 @@
+package main
+
+import (
+	"fmt"
+	"io"
+	"log"
+	"net"
+	"strconv"
+	"strings"
+)
+
+type TCPOutput struct {
+	address string
+	limit   int
+	buf     chan []byte
+	bufStats *GorStat
+}
+
+func NewTCPOutput(options string) io.Writer {
+	o := new(TCPOutput)
+
+	optionsArr := strings.Split(options, "|")
+	o.address = optionsArr[0]
+
+	o.buf = make(chan []byte, 100)
+	if Settings.outputTCPStats {
+		o.bufStats = NewGorStat("output_tcp")
+	}
+
+	if len(optionsArr) > 1 {
+		o.limit, _ = strconv.Atoi(optionsArr[1])
+	}
+
+	for i := 0; i < 10; i++ {
+		go o.worker()
+	}
+
+	if o.limit > 0 {
+		return NewLimiter(o, o.limit)
+	} else {
+		return o
+	}
+}
+
+func (o *TCPOutput) worker() {
+	conn, _ := o.connect(o.address)
+	defer conn.Close()
+
+	for {
+		conn.Write(<-o.buf)
+	}
+}
+
+func (o *TCPOutput) Write(data []byte) (n int, err error) {
+	new_buf := make([]byte, len(data) + 2)
+	data = append(data,[]byte("¶")...)
+	copy(new_buf, data)
+	o.buf <- new_buf
+	if Settings.outputTCPStats {
+		o.bufStats.Write(len(o.buf))
+	}
+
+	return len(data), nil
+}
+
+func (o *TCPOutput) connect(address string) (conn net.Conn, err error) {
+	conn, err = net.Dial("tcp", address)
+
+	if err != nil {
+		log.Println("Connection error ", err, o.address)
+	}
+
+	return
+}
+
+func (o *TCPOutput) String() string {
+	return fmt.Sprintf("TCP output %s, limit: %d", o.address, o.limit)
+}

output_tcp_test.go

@@ -0,0 +1,94 @@
+package main
+
+import (
+	"bufio"
+	"io"
+	"log"
+	"net"
+	"sync"
+	"testing"
+)
+
+func TestTCPOutput(t *testing.T) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	listener := startTCP(func(data []byte) {
+		wg.Done()
+	})
+	input := NewTestInput()
+	output := NewTCPOutput(listener.Addr().String())
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	go Start(quit)
+
+	for i := 0; i < 100; i++ {
+		wg.Add(1)
+		input.EmitGET()
+	}
+
+	wg.Wait()
+
+	close(quit)
+}
+
+func startTCP(cb func([]byte)) net.Listener {
+	listener, err := net.Listen("tcp", ":0")
+
+	if err != nil {
+		log.Fatal("Can't start:", err)
+	}
+
+	go func() {
+		for {
+			conn, _ := listener.Accept()
+			
+			go func() {
+				reader := bufio.NewReader(conn)
+				for {
+					buf,err := reader.ReadBytes('¶')
+					new_buf_len := len(buf) - 2
+					new_buf := make([]byte, new_buf_len)
+					copy(new_buf, buf[:new_buf_len])
+					if err != nil {
+						if err != io.EOF {
+							log.Printf("error: %s\n", err)
+						}
+					}
+					cb(new_buf)
+				}
+				conn.Close()
+			}()
+		}
+	}()
+
+	return listener
+}
+
+func BenchmarkTCPOutput(b *testing.B) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	listener := startTCP(func(data []byte) {
+		wg.Done()
+	})
+	input := NewTestInput()
+	output := NewTCPOutput(listener.Addr().String())
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	go Start(quit)
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		wg.Add(1)
+		input.EmitGET()
+	}
+
+	wg.Wait()
+
+	close(quit)
+}

plugins.go

@@ -0,0 +1,46 @@
+package main
+
+import (
+	"io"
+)
+
+type InOutPlugins struct {
+	Inputs  []io.Reader
+	Outputs []io.Writer
+}
+
+var Plugins *InOutPlugins = new(InOutPlugins)
+
+func InitPlugins() {
+	for _, options := range Settings.inputDummy {
+		Plugins.Inputs = append(Plugins.Inputs, NewDummyInput(options))
+	}
+
+	for _, options := range Settings.outputDummy {
+		Plugins.Outputs = append(Plugins.Outputs, NewDummyOutput(options))
+	}
+
+	for _, options := range Settings.inputRAW {
+		Plugins.Inputs = append(Plugins.Inputs, NewRAWInput(options))
+	}
+
+	for _, options := range Settings.inputTCP {
+		Plugins.Inputs = append(Plugins.Inputs, NewTCPInput(options))
+	}
+
+	for _, options := range Settings.outputTCP {
+		Plugins.Outputs = append(Plugins.Outputs, NewTCPOutput(options))
+	}
+
+	for _, options := range Settings.inputFile {
+		Plugins.Inputs = append(Plugins.Inputs, NewFileInput(options))
+	}
+
+	for _, options := range Settings.outputFile {
+		Plugins.Outputs = append(Plugins.Outputs, NewFileOutput(options))
+	}
+
+	for _, options := range Settings.outputHTTP {
+		Plugins.Outputs = append(Plugins.Outputs, NewHTTPOutput(options, Settings.outputHTTPHeaders, Settings.outputHTTPMethods, Settings.outputHTTPUrlRegexp, Settings.outputHTTPHeaderFilters, Settings.outputHTTPHeaderHashFilters, Settings.outputHTTPElasticSearch, Settings.outputHTTPUrlRewrite))
+	}
+}

raw_socket_listener/listener.go

@@ -0,0 +1,141 @@
+package raw_socket
+
+import (
+	"encoding/binary"
+	"log"
+	"net"
+	"strconv"
+)
+
+// Capture traffic from socket using RAW_SOCKET's
+// http://en.wikipedia.org/wiki/Raw_socket
+//
+// RAW_SOCKET allow you listen for traffic on any port (e.g. sniffing) because they operate on IP level.
+// Ports is TCP feature, same as flow control, reliable transmission and etc.
+// Since we can't use default TCP libraries RAWTCPLitener implements own TCP layer
+// TCP packets is parsed using tcp_packet.go, and flow control is managed by tcp_message.go
+type Listener struct {
+	messages map[string]*TCPMessage // buffer of TCPMessages waiting to be send
+
+	c_packets  chan *TCPPacket
+	c_messages chan *TCPMessage // Messages ready to be send to client
+
+	c_del_message chan *TCPMessage // Used for notifications about completed or expired messages
+
+	addr string // IP to listen
+	port int    // Port to listen
+}
+
+// RAWTCPListen creates a listener to capture traffic from RAW_SOCKET
+func NewListener(addr string, port string) (rawListener *Listener) {
+	rawListener = &Listener{}
+
+	rawListener.c_packets = make(chan *TCPPacket, 100)
+	rawListener.c_messages = make(chan *TCPMessage, 100)
+	rawListener.c_del_message = make(chan *TCPMessage, 100)
+	rawListener.messages = make(map[string]*TCPMessage)
+
+	rawListener.addr = addr
+	rawListener.port, _ = strconv.Atoi(port)
+
+	go rawListener.listen()
+	go rawListener.readRAWSocket()
+
+	return
+}
+
+func (t *Listener) listen() {
+	for {
+		select {
+		// If message ready for deletion it means that its also complete or expired by timeout
+		case message := <-t.c_del_message:
+			t.c_messages <- message
+			delete(t.messages, message.ID)
+
+		// We need to use channels to process each packet to avoid data races
+		case packet := <-t.c_packets:
+			t.processTCPPacket(packet)
+		}
+	}
+}
+
+func (t *Listener) readRAWSocket() {
+	conn, e := net.ListenPacket("ip4:tcp", t.addr)
+	defer conn.Close()
+
+	if e != nil {
+		log.Fatal(e)
+	}
+
+	buf := make([]byte, 4096*2)
+
+	for {
+		// Note: ReadFrom receive messages without IP header
+		n, addr, err := conn.ReadFrom(buf)
+
+		if err != nil {
+			log.Println("Error:", err)
+			continue
+		}
+
+		if n > 0 {
+			t.parsePacket(addr, buf[:n])
+		}
+	}
+}
+
+func (t *Listener) parsePacket(addr net.Addr, buf []byte) {
+	if t.isIncomingDataPacket(buf) {
+		new_buf := make([]byte, len(buf))
+		copy(new_buf, buf)
+
+		t.c_packets <- ParseTCPPacket(addr, new_buf)
+	}
+}
+
+func (t *Listener) isIncomingDataPacket(buf []byte) bool {
+	// To avoid full packet parsing every time, we manually parsing values needed for packet filtering
+	// http://en.wikipedia.org/wiki/Transmission_Control_Protocol
+	dest_port := binary.BigEndian.Uint16(buf[2:4])
+
+	// Because RAW_SOCKET can't be bound to port, we have to control it by ourself
+	if int(dest_port) == t.port {
+		// Get the 'data offset' (size of the TCP header in 32-bit words)
+		dataOffset := (buf[12] & 0xF0) >> 4
+
+		// We need only packets with data inside
+		// Check that the buffer is larger than the size of the TCP header
+		if len(buf) > int(dataOffset*4) {
+			// We should create new buffer because go slices is pointers. So buffer data shoud be immutable.
+			return true
+		}
+	}
+
+	return false
+}
+
+// Trying to add packet to existing message or creating new message
+//
+// For TCP message unique id is Acknowledgment number (see tcp_packet.go)
+func (t *Listener) processTCPPacket(packet *TCPPacket) {
+	defer func() { recover() }()
+
+	var message *TCPMessage
+	m_id := packet.Addr.String() + strconv.Itoa(int(packet.Ack))
+
+	message, ok := t.messages[m_id]
+
+	if !ok {
+		// We sending c_del_message channel, so message object can communicate with Listener and notify it if message completed
+		message = NewTCPMessage(m_id, t.c_del_message)
+		t.messages[m_id] = message
+	}
+
+	// Adding packet to message
+	message.c_packets <- packet
+}
+
+// Receive TCP messages from the listener channel
+func (t *Listener) Receive() *TCPMessage {
+	return <-t.c_messages
+}

raw_socket_listener/tcp_message.go

@@ -0,0 +1,94 @@
+package raw_socket
+
+import (
+	"log"
+	"sort"
+	"time"
+)
+
+const MSG_EXPIRE = 2000 * time.Millisecond
+
+// TCPMessage ensure that all TCP packets for given request is received, and processed in right sequence
+// Its needed because all TCP message can be fragmented or re-transmitted
+//
+// Each TCP Packet have 2 ids: acknowledgment - message_id, and sequence - packet_id
+// Message can be compiled from unique packets with same message_id which sorted by sequence
+// Message is received if we didn't receive any packets for 2000ms
+type TCPMessage struct {
+	ID      string // Message ID
+	packets []*TCPPacket
+
+	timer *time.Timer // Used for expire check
+
+	c_packets chan *TCPPacket
+
+	c_del_message chan *TCPMessage
+}
+
+// NewTCPMessage pointer created from a Acknowledgment number and a channel of messages readuy to be deleted
+func NewTCPMessage(ID string, c_del chan *TCPMessage) (msg *TCPMessage) {
+	msg = &TCPMessage{ID: ID}
+
+	msg.c_packets = make(chan *TCPPacket)
+	msg.c_del_message = c_del // used for notifying that message completed or expired
+
+	// Every time we receive packet we reset this timer
+	msg.timer = time.AfterFunc(MSG_EXPIRE, msg.Timeout)
+
+	go msg.listen()
+
+	return
+}
+
+func (t *TCPMessage) listen() {
+	for {
+		select {
+		case packet, more := <-t.c_packets:
+			if more {
+				t.AddPacket(packet)
+			} else {
+				// Stop loop if channel closed
+				return
+			}
+		}
+	}
+}
+
+// Timeout notifies message to stop listening, close channel and message ready to be sent
+func (t *TCPMessage) Timeout() {
+	close(t.c_packets)   // Notify to stop listen loop and close channel
+	t.c_del_message <- t // Notify RAWListener that message is ready to be send to replay server
+}
+
+// Bytes sorts packets in right orders and return message content
+func (t *TCPMessage) Bytes() (output []byte) {
+	sort.Sort(BySeq(t.packets))
+
+	for _, v := range t.packets {
+		output = append(output, v.Data...)
+	}
+
+	return
+}
+
+// AddPacket to the message and ensure packet uniqueness
+// TCP allows that packet can be re-send multiple times
+func (t *TCPMessage) AddPacket(packet *TCPPacket) {
+	packetFound := false
+
+	for _, pkt := range t.packets {
+		if packet.Seq == pkt.Seq {
+			packetFound = true
+			break
+		}
+	}
+
+	if packetFound {
+		log.Println("Received packet with same sequence")
+	} else {
+		t.packets = append(t.packets, packet)
+	}
+
+	// Reset message timeout timer
+	t.timer.Reset(MSG_EXPIRE)
+}

raw_socket_listener/tcp_packet.go

@@ -0,0 +1,100 @@
+package raw_socket
+
+import (
+	"encoding/binary"
+	"net"
+	"strconv"
+	"strings"
+)
+
+// TCP Flags
+const (
+	TCP_FIN = 1 << iota
+	TCP_SYN
+	TCP_RST
+	TCP_PSH
+	TCP_ACK
+	TCP_URG
+	TCP_ECE
+	TCP_CWR
+	TCP_NS
+)
+
+// Simple TCP packet parser
+//
+// Packet structure: http://en.wikipedia.org/wiki/Transmission_Control_Protocol
+type TCPPacket struct {
+	SrcPort    uint16
+	DestPort   uint16
+	Seq        uint32
+	Ack        uint32
+	DataOffset uint8
+	Flags      uint16
+	Window     uint16
+	Checksum   uint16
+	Urgent     uint16
+
+	Data []byte
+
+	Addr net.Addr
+}
+
+func ParseTCPPacket(addr net.Addr, b []byte) (p *TCPPacket) {
+	p = &TCPPacket{Data: b}
+	p.ParseBasic()
+	p.Addr = addr
+
+	return p
+}
+
+// Parse TCP Packet, inspired by: https://github.com/miekg/pcap/blob/master/packet.go
+func (t *TCPPacket) Parse() {
+	t.ParseBasic()
+	t.SrcPort = binary.BigEndian.Uint16(t.Data[0:2])
+	t.DestPort = binary.BigEndian.Uint16(t.Data[2:4])
+	t.Flags = binary.BigEndian.Uint16(t.Data[12:14]) & 0x1FF
+	t.Window = binary.BigEndian.Uint16(t.Data[14:16])
+	t.Checksum = binary.BigEndian.Uint16(t.Data[16:18])
+	t.Urgent = binary.BigEndian.Uint16(t.Data[18:20])
+}
+
+// ParseBasic set of fields
+func (t *TCPPacket) ParseBasic() {
+	t.Seq = binary.BigEndian.Uint32(t.Data[4:8])
+	t.Ack = binary.BigEndian.Uint32(t.Data[8:12])
+	t.DataOffset = (t.Data[12] & 0xF0) >> 4
+
+	t.Data = t.Data[t.DataOffset*4:]
+}
+
+// String output for a TCP Packet
+func (t *TCPPacket) String() string {
+	return strings.Join([]string{
+		"Source port: " + strconv.Itoa(int(t.SrcPort)),
+		"Dest port:" + strconv.Itoa(int(t.DestPort)),
+		"Sequence:" + strconv.Itoa(int(t.Seq)),
+		"Acknowledgment:" + strconv.Itoa(int(t.Ack)),
+		"Header len:" + strconv.Itoa(int(t.DataOffset)),
+
+		"Flag ns:" + strconv.FormatBool(t.Flags&TCP_NS != 0),
+		"Flag crw:" + strconv.FormatBool(t.Flags&TCP_CWR != 0),
+		"Flag ece:" + strconv.FormatBool(t.Flags&TCP_ECE != 0),
+		"Flag urg:" + strconv.FormatBool(t.Flags&TCP_URG != 0),
+		"Flag ack:" + strconv.FormatBool(t.Flags&TCP_ACK != 0),
+		"Flag psh:" + strconv.FormatBool(t.Flags&TCP_PSH != 0),
+		"Flag rst:" + strconv.FormatBool(t.Flags&TCP_RST != 0),
+		"Flag syn:" + strconv.FormatBool(t.Flags&TCP_SYN != 0),
+		"Flag fin:" + strconv.FormatBool(t.Flags&TCP_FIN != 0),
+
+		"Window size:" + strconv.Itoa(int(t.Window)),
+		"Checksum:" + strconv.Itoa(int(t.Checksum)),
+
+		"Data:" + string(t.Data),
+	}, "\n")
+}
+
+type BySeq []*TCPPacket
+
+func (a BySeq) Len() int			{ return len(a) }
+func (a BySeq) Swap(i, j int)		{ a[i], a[j] = a[j], a[i] }
+func (a BySeq) Less(i, j int) bool	{ return a[i].Seq < a[j].Seq }

replay/replay.go

@@ -1,102 +0,0 @@
-// Replay server receive requests objects from Listeners and forward it to given address.
-// Basic usage:
-//
-//     gor replay -f http://staging.server
-//
-//
-// Rate limiting
-//
-// It can be useful if you want forward only part of production traffic, not to overload staging environment. You can specify desired request per second using "|" operator after server address:
-//
-//     # staging.server not get more than 10 requests per second
-//     gor replay -f "http://staging.server|10"
-//
-//
-// Forward to multiple addresses
-//
-// Just separate addresses by coma:
-//    gor replay -f "http://staging.server|10,http://dev.server|20"
-//
-//
-//  For more help run:
-//
-//     gor replay -h
-//
-package replay
-
-import (
-	"bytes"
-	"encoding/gob"
-	"log"
-	"net"
-)
-
-const bufSize = 1024 * 10
-
-// Enable debug logging only if "--verbose" flag passed
-func Debug(v ...interface{}) {
-	if Settings.verbose { log.Println(v...) }
-}
-
-// Decode HttpRequest object using standard gob decoder
-func DecodeRequest(enc []byte) (request *HttpRequest, err error) {
-	var buf bytes.Buffer
-	buf.Write(enc)
-
-	request = &HttpRequest{}
-
-	encoder := gob.NewDecoder(&buf)
-	err = encoder.Decode(request)
-
-	return
-}
-
-// Because its sub-program, Run acts as `main`
-// Replay server listen to UDP traffic from Listeners
-// Each request processed by RequestFactory
-func Run() {
-	var buf [bufSize]byte
-
-	addr, err := net.ResolveUDPAddr("udp", Settings.Address())
-	if err != nil {
-		log.Fatal("Can't start:", err)
-	}
-
-	conn, err := net.ListenUDP("udp", addr)
-	log.Println("Starting replay server at:", Settings.Address())
-
-	if err != nil {
-		log.Fatal("Can't start:", err)
-	}
-
-	defer conn.Close()
-
-	for _, host := range Settings.ForwardedHosts() {
-		log.Println("Forwarding requests to:", host.Url, "limit:", host.Limit)
-	}
-
-	requestFactory := NewRequestFactory()
-
-	for {
-		rlen, _, err := conn.ReadFromUDP(buf[0:])
-
-		if err != nil {
-			continue
-		}
-
-		if rlen > 0 {
-			if rlen > bufSize {
-				log.Fatal("Too large udp packet", bufSize)
-			}
-
-			request, err := DecodeRequest(buf[0:rlen])
-
-			if err != nil {
-				log.Println("Decode error:", err)
-			} else {
-				requestFactory.Add(request)
-			}
-		}
-	}
-
-}

replay/request_factory.go

@@ -1,99 +0,0 @@
-package replay
-
-import (
-	"net/http"
-)
-
-// Userd for transfering Request info between Listener and Replay server
-type HttpRequest struct {
-	Tag     string            // Not used yet
-	Method  string            // Right now only 'GET'
-	Url     string            // Request URL
-	Headers map[string]string // Request Headers
-}
-
-type HttpResponse struct {
-	host *ForwardHost
-	req  *HttpRequest
-	resp *http.Response
-	err  error
-}
-
-// Class for processing requests
-//
-// Basic workflow:
-//
-// 1. When request added via Add() it get pushed to `responses` chan
-// 2. handleRequest() listen for `responses` chan and decide where request should be forwarded, and apply rate-limit if needed
-// 3. sendRequest() forwards request and returns response info to `responses` chan
-// 4. handleRequest() listen for `response` channel and updates stats
-type RequestFactory struct {
-	responses chan *HttpResponse
-	requests  chan *HttpRequest
-}
-
-// RequestFactory contstuctor
-// One created, it starts listening for incoming requests: requests channel
-func NewRequestFactory() (factory *RequestFactory) {
-	factory = &RequestFactory{}
-	factory.responses = make(chan *HttpResponse)
-	factory.requests = make(chan *HttpRequest)
-
-	go factory.handleRequests()
-
-	return
-}
-
-// Forward http request to given host
-func (f *RequestFactory) sendRequest(host *ForwardHost, request *HttpRequest) {
-	var req *http.Request
-
-	client := &http.Client{}
-
-	req, err := http.NewRequest("GET", host.Url+request.Url, nil)
-
-	// Forwarded request should have same headers
-	for key, value := range request.Headers {
-		req.Header.Add(key, value)
-	}
-
-	resp, err := client.Do(req)
-
-	if err == nil {
-		defer resp.Body.Close()
-	}
-
-	f.responses <- &HttpResponse{host, request, resp, err}
-}
-
-// Handle incoming requests, and they responses
-func (f *RequestFactory) handleRequests() {
-	hosts := Settings.ForwardedHosts()
-
-	for {
-		select {
-		case req := <-f.requests:
-			for _, host := range hosts {
-				// Ensure that we have actual stats for given timestamp
-				host.Stat.Touch()
-
-				if host.Limit == 0 || host.Stat.Count < host.Limit {
-					// Increment Stat.Count
-					host.Stat.IncReq()
-
-                    Debug("GET ",host.Url + req.Url)
-
-					go f.sendRequest(host, req)
-				}
-			}
-		case resp := <-f.responses:
-			// Increment returned http code stats, and elapsed time
-			resp.host.Stat.IncResp(resp)
-		}
-	}
-}
-
-// Add request to channel for further processing
-func (f *RequestFactory) Add(request *HttpRequest) {
-	f.requests <- request
-}

replay/request_stats.go

@@ -1,63 +0,0 @@
-package replay
-
-import (
-	"time"
-)
-
-// Stats stores in context of current timestamp
-type RequestStat struct {
-	timestamp int64
-
-	Codes map[int]int // { 200: 10, 404:2, 500:1 }
-
-	Count  int // All requests including errors
-	Errors int // Rquests with errors (timeout or host not reachable). Not include 50x errors.
-
-	host *ForwardHost
-}
-
-// Ensure that current stats is actual (for current timestamp)
-func (s *RequestStat) Touch() {
-	if s.timestamp != time.Now().Unix() {
-		s.reset()
-	}
-}
-
-// Called on request start
-func (s *RequestStat) IncReq() {
-	s.Count++
-}
-
-// Called after response
-func (s *RequestStat) IncResp(resp *HttpResponse) {
-	s.Touch()
-
-	if resp.err != nil {
-		s.Errors++
-		return
-	}
-
-	s.Codes[resp.resp.StatusCode]++
-}
-
-// Updated stats timestamp to current time and reset to zero all stats values
-// TODO: Further on reset it should write stats to file
-func (s *RequestStat) reset() {
-	if s.timestamp != 0 {        
-	   Debug("Host:", s.host.Url, "Requests:", s.Count, "Errors:", s.Errors, "Status codes:", s.Codes)
-	}
-
-	s.timestamp = time.Now().Unix()
-
-	s.Codes = make(map[int]int)
-	s.Count = 0
-	s.Errors = 0
-}
-
-// RequestStat constructor
-func NewRequestStats(host *ForwardHost) (stat *RequestStat) {
-	stat = &RequestStat{host: host}
-	stat.reset()
-
-	return
-}

replay/settings.go

@@ -1,80 +0,0 @@
-package replay
-
-import (
-	"strconv"
-	"strings"
-	"os"
-	"flag"
-)
-
-type ForwardHost struct {
-	Url   string
-	Limit int
-
-	Stat *RequestStat
-}
-
-type ReplaySettings struct {
-	port int
-	host string
-
-	forwardAddress string
-
-	verbose bool
-}
-
-// ForwardedHosts implements forwardAddress syntax support for multiple hosts (coma separated), and rate limiting by specifing "|maxRps" after host name.
-//
-//    -f "host1,http://host2|10,host3"
-//
-func (r *ReplaySettings) ForwardedHosts() (hosts []*ForwardHost) {
-	hosts = make([]*ForwardHost, 0, 10)
-
-	for _, address := range strings.Split(r.forwardAddress, ",") {
-		host_info := strings.Split(address, "|")
-
-		if strings.Index(host_info[0], "http") == -1 {
-			host_info[0] = "http://" + host_info[0]
-		}
-
-		host := &ForwardHost{Url: host_info[0]}
-		host.Stat = NewRequestStats(host)
-
-		if len(host_info) > 1 {
-			host.Limit, _ = strconv.Atoi(host_info[1])
-		}
-
-		hosts = append(hosts, host)
-	}
-
-	return
-}
-
-// Helper to return address with port, e.g.: 127.0.0.1:28020
-func (r *ReplaySettings) Address() string {
-	return r.host + ":" + strconv.Itoa(r.port)
-}
-
-var Settings ReplaySettings = ReplaySettings{}
-
-
-func init() {
-	if len(os.Args) < 2 || os.Args[1] != "replay" {
-		return
-	}
-
-	const (
-		defaultPort = 28020
-		defaultHost = "0.0.0.0"
-
-		defaultAddress = "http://localhost:8080"
-	)
-
-	flag.IntVar(&Settings.port, "p", defaultPort, "specify port number")
-
-	flag.StringVar(&Settings.host, "ip", defaultHost, "ip addresses to listen on")
-
-	flag.StringVar(&Settings.forwardAddress, "f", defaultAddress, "http address to forward traffic.\n\tYou can limit requests per second by adding `|num` after address.\n\tIf you have multiple addresses with different limits. For example: http://staging.example.com|100,http://dev.example.com|10")
-
-	flag.BoolVar(&Settings.verbose, "verbose", false, "Log requests")
-}

settings.go

@@ -0,0 +1,89 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"log"
+	"os"
+)
+
+const (
+	VERSION = "0.9.0"
+)
+
+type AppSettings struct {
+	verbose bool
+	stats   bool
+
+	splitOutput bool
+
+	inputDummy  MultiOption
+	outputDummy MultiOption
+
+	inputTCP  MultiOption
+	outputTCP MultiOption
+	outputTCPStats bool
+
+	inputFile  MultiOption
+	outputFile MultiOption
+
+	inputRAW MultiOption
+
+	outputHTTP                  MultiOption
+	outputHTTPHeaders           HTTPHeaders
+	outputHTTPMethods           HTTPMethods
+	outputHTTPUrlRegexp         HTTPUrlRegexp
+	outputHTTPUrlRewrite        UrlRewriteMap
+	outputHTTPHeaderFilters     HTTPHeaderFilters
+	outputHTTPHeaderHashFilters HTTPHeaderHashFilters
+	outputHTTPElasticSearch     string
+	outputHTTPWorkers           int
+	outputHTTPStats             bool
+}
+
+var Settings AppSettings = AppSettings{}
+
+func usage() {
+	fmt.Printf("Gor is a simple http traffic replication tool written in Go. Its main goal is to replay traffic from production servers to staging and dev environments.\nProject page: https://github.com/buger/gor\nAuthor: <Leonid Bugaev> leonsbox@gmail.com\nCurrent Version: %s\n\n", VERSION)
+	flag.PrintDefaults()
+	os.Exit(2)
+}
+
+func init() {
+	flag.Usage = usage
+
+	flag.BoolVar(&Settings.verbose, "verbose", false, "Turn on verbose/debug output")
+	flag.BoolVar(&Settings.stats, "stats", false, "Turn on queue stats output")
+
+	flag.BoolVar(&Settings.splitOutput, "split-output", false, "By default each output gets same traffic. If set to `true` it splits traffic equally among all outputs.")
+
+	flag.Var(&Settings.inputDummy, "input-dummy", "Used for testing outputs. Emits 'Get /' request every 1s")
+	flag.Var(&Settings.outputDummy, "output-dummy", "Used for testing inputs. Just prints data coming from inputs.")
+
+	flag.Var(&Settings.inputTCP, "input-tcp", "Used for internal communication between Gor instances. Example: \n\t# Receive requests from other Gor instances on 28020 port, and redirect output to staging\n\tgor --input-tcp :28020 --output-http staging.com")
+	flag.Var(&Settings.outputTCP, "output-tcp", "Used for internal communication between Gor instances. Example: \n\t# Listen for requests on 80 port and forward them to other Gor instance on 28020 port\n\tgor --input-raw :80 --output-tcp replay.local:28020")
+	flag.BoolVar(&Settings.outputTCPStats, "output-tcp-stats", false, "Report TCP output queue stats to console every 5 seconds.")
+
+	flag.Var(&Settings.inputFile, "input-file", "Read requests from file: \n\tgor --input-file ./requests.gor --output-http staging.com")
+	flag.Var(&Settings.outputFile, "output-file", "Write incoming requests to file: \n\tgor --input-raw :80 --output-file ./requests.gor")
+
+	flag.Var(&Settings.inputRAW, "input-raw", "Capture traffic from given port (use RAW sockets and require *sudo* access):\n\t# Capture traffic from 8080 port\n\tgor --input-raw :8080 --output-http staging.com")
+
+	flag.Var(&Settings.outputHTTP, "output-http", "Forwards incoming requests to given http address.\n\t# Redirect all incoming requests to staging.com address \n\tgor --input-raw :80 --output-http http://staging.com")
+	flag.Var(&Settings.outputHTTPHeaders, "output-http-header", "Inject additional headers to http reqest:\n\tgor --input-raw :8080 --output-http staging.com --output-http-header 'User-Agent: Gor'")
+	flag.Var(&Settings.outputHTTPMethods, "output-http-method", "Whitelist of HTTP methods to replay. Anything else will be dropped:\n\tgor --input-raw :8080 --output-http staging.com --output-http-method GET --output-http-method OPTIONS")
+	flag.Var(&Settings.outputHTTPUrlRegexp, "output-http-url-regexp", "A regexp to match requests against. Anything else will be dropped:\n\t gor --input-raw :8080 --output-http staging.com --output-http-url-regexp ^www.")
+	flag.Var(&Settings.outputHTTPHeaderFilters, "output-http-header-filter", "A regexp to match a specific header against. Requests with non-matching headers will be dropped:\n\t gor --input-raw :8080 --output-http staging.com --output-http-header-filter api-version:^v1")
+	flag.Var(&Settings.outputHTTPHeaderHashFilters, "output-http-header-hash-filter", "Takes a fraction of requests, consistently taking or rejecting a request based on the FNV32-1A hash of a specific header. The fraction must have a denominator that is a power of two:\n\t gor --input-raw :8080 --output-http staging.com --output-http-header-hash-filter user-id:1/4")
+	flag.IntVar(&Settings.outputHTTPWorkers, "output-http-workers", -1, "Gor uses dynamic worker scaling by default.  Enter a number to run a set number of workers.")
+	flag.BoolVar(&Settings.outputHTTPStats, "output-http-stats", false, "Report http output queue stats to console every 5 seconds.")
+
+	flag.StringVar(&Settings.outputHTTPElasticSearch, "output-http-elasticsearch", "", "Send request and response stats to ElasticSearch:\n\tgor --input-raw :8080 --output-http staging.com --output-http-elasticsearch 'es_host:api_port/index_name'")
+	flag.Var(&Settings.outputHTTPUrlRewrite, "output-http-rewrite-url", "Rewrite the requst url based on a mapping:\n\tgor --input-raw :8080 --output-http staging.com --output-http-rewrite-url /xml_test/interface.php:/api/service.do")
+}
+
+func Debug(args ...interface{}) {
+	if Settings.verbose {
+		log.Println(args...)
+	}
+}

settings_header_filters.go

@@ -0,0 +1,44 @@
+package main
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+	"regexp"
+	"strings"
+)
+
+type headerFilter struct {
+	name   string
+	regexp *regexp.Regexp
+}
+
+type HTTPHeaderFilters []headerFilter
+
+func (h *HTTPHeaderFilters) String() string {
+	return fmt.Sprint(*h)
+}
+
+func (h *HTTPHeaderFilters) Set(value string) error {
+	valArr := strings.SplitN(value, ":", 2)
+	if len(valArr) < 2 {
+		return errors.New("need both header and value, colon-delimited (ex. user_id:^169$).")
+	}
+	r, err := regexp.Compile(valArr[1])
+	if err != nil {
+		return err
+	}
+
+	*h = append(*h, headerFilter{name: valArr[0], regexp: r})
+
+	return nil
+}
+
+func (h *HTTPHeaderFilters) Good(req *http.Request) bool {
+	for _, f := range *h {
+		if !f.regexp.MatchString(req.Header.Get(f.name)) {
+			return false
+		}
+	}
+	return true
+}

settings_header_filters_test.go

@@ -0,0 +1,35 @@
+package main
+
+import (
+	"testing"
+	"net/http"
+)
+
+func TestHTTPHeaderFilters(t *testing.T) {
+	filters := HTTPHeaderFilters{}
+
+	err := filters.Set("Header1:^$")
+	if err != nil {
+		t.Error("Should not error on Header1:^$")
+	}
+
+	err = filters.Set("Header2:^:$")
+	if err != nil {
+		t.Error("Should not error on Header2:^:$")
+	}
+
+	err = filters.Set("Header3-^$")
+	if err == nil {
+		t.Error("Should error on Header2:^:$")
+	}
+
+	req := http.Request{}
+	req.Header = make(map[string][]string)
+	req.Header.Add("Header1", "")
+	req.Header.Add("Header2", ":")
+	req.Header.Add("Header3", "Irrelevant")
+
+	if(!filters.Good(&req)) {
+		t.Error("Request should pass filters")
+	}
+}

settings_header_hash_filters.go

@@ -0,0 +1,68 @@
+package main
+
+import (
+	"errors"
+	"fmt"
+	"hash/fnv"
+	"net/http"
+	"strconv"
+	"strings"
+)
+
+type headerHashFilter struct {
+	name    string
+	maxHash uint32
+}
+
+type HTTPHeaderHashFilters []headerHashFilter
+
+func (h *HTTPHeaderHashFilters) String() string {
+	return fmt.Sprint(*h)
+}
+
+func (h *HTTPHeaderHashFilters) Set(value string) error {
+	valArr := strings.SplitN(value, ":", 2)
+	if len(valArr) < 2 {
+		return errors.New("need both header and value, colon-delimited (ex. user_id:1/2).")
+	}
+
+	fracArr := strings.Split(valArr[1], "/")
+	if len(fracArr) < 2 {
+		return errors.New("need both a numerator and denominator specified, slash-delimited (ex. user_id:1/4).")
+	}
+
+	var num, den uint64
+	num, _ = strconv.ParseUint(fracArr[0], 10, 64)
+	den, _ = strconv.ParseUint(fracArr[1], 10, 64)
+
+	if num < 1 || den < 1 || num > den {
+		panic("need positive numerators and denominators, with the former less than the latter.")
+	}
+
+	for test := den; test != 1; test /= 2 {
+		if test%2 == 1 {
+			return errors.New("must have a denominator which is a power of two.")
+		}
+	}
+
+	var f headerHashFilter
+	f.name = valArr[0]
+	f.maxHash = (uint32)(num * (((uint64)(2 << 31)) / den))
+	*h = append(*h, f)
+
+	return nil
+}
+
+func (h *HTTPHeaderHashFilters) Good(req *http.Request) bool {
+	for _, f := range *h {
+		if req.Header.Get(f.name) == "" {
+			return false
+		}
+		hasher := fnv.New32a()
+		hasher.Write([]byte(req.Header.Get(f.name)))
+		if hasher.Sum32() > f.maxHash {
+			return false
+		}
+	}
+	return true
+}

settings_header_hash_filters_test.go

@@ -0,0 +1,43 @@
+package main
+
+import (
+	"testing"
+	"net/http"
+)
+
+func TestHTTPHeaderHashFilters(t *testing.T) {
+	filters := HTTPHeaderHashFilters{}
+
+	err := filters.Set("Header1:1/2")
+	if err != nil {
+		t.Error("Should not error on Header1:^$")
+	}
+
+	err = filters.Set("Header2:1/2")
+	if err != nil {
+		t.Error("Should not error on Header2:^:$")
+	}
+
+	err = filters.Set("HeaderIrrelevant:1/3")
+	if err == nil {
+		t.Error("Should error on HeaderIrrelevant:1/3")
+	}
+
+	req := http.Request{}
+	req.Header = make(map[string][]string)
+	req.Header.Add("Header1", "test3414")
+
+	if(filters.Good(&req)) {
+		t.Error("Request should not pass filters, Header2 does not exist")
+	}
+
+	req.Header.Add("Header2", "test2")
+	if(filters.Good(&req)) {
+		t.Error("Request should not pass filters, Header2 hash too high")
+	}
+
+	req.Header.Set("Header2", "test3414")
+	if(!filters.Good(&req)) {
+		t.Error("Request should pass filters")
+	}
+}

settings_headers.go

@@ -0,0 +1,32 @@
+package main
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+)
+
+type HTTPHeaders []HTTPHeader
+type HTTPHeader struct {
+	Name  string
+	Value string
+}
+
+func (h *HTTPHeaders) String() string {
+	return fmt.Sprint(*h)
+}
+
+func (h *HTTPHeaders) Set(value string) error {
+	v := strings.SplitN(value, ":", 2)
+	if len(v) != 2 {
+		return errors.New("Expected `Key: Value`")
+	}
+
+	header := HTTPHeader{
+		strings.TrimSpace(v[0]),
+		strings.TrimSpace(v[1]),
+	}
+
+	*h = append(*h, header)
+	return nil
+}

settings_methods.go

@@ -0,0 +1,26 @@
+package main
+
+import (
+	"fmt"
+	"strings"
+)
+
+type HTTPMethods []string
+
+func (h *HTTPMethods) String() string {
+	return fmt.Sprint(*h)
+}
+
+func (h *HTTPMethods) Set(value string) error {
+	*h = append(*h, strings.ToUpper(value))
+	return nil
+}
+
+func (h *HTTPMethods) Contains(value string) bool {
+	for _, method := range *h {
+		if value == method {
+			return true
+		}
+	}
+	return false
+}

settings_methods_test.go

@@ -0,0 +1,24 @@
+package main
+
+import (
+	"testing"
+)
+
+func TestHTTPMethods(t *testing.T) {
+	methods := HTTPMethods{}
+
+	methods.Set("lower")
+	methods.Set("UPPER")
+
+	if !methods.Contains("LOWER") {
+		t.Error("Does not contain LOWER")
+	}
+
+	if !methods.Contains("UPPER") {
+		t.Error("Does not contain UPPER")
+	}
+
+	if methods.Contains("ABSENT") {
+		t.Error("Does contain ABSENT")
+	}
+}

settings_option.go

@@ -0,0 +1,16 @@
+package main
+
+import (
+	"fmt"
+)
+
+type MultiOption []string
+
+func (h *MultiOption) String() string {
+	return fmt.Sprint(*h)
+}
+
+func (h *MultiOption) Set(value string) error {
+	*h = append(*h, value)
+	return nil
+}

settings_url_map.go

@@ -0,0 +1,36 @@
+package main
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+)
+
+type urlRewrite struct {
+	src    string
+	target string
+}
+
+type UrlRewriteMap []urlRewrite
+
+func (r *UrlRewriteMap) String() string {
+	return fmt.Sprint(*r)
+}
+
+func (r *UrlRewriteMap) Set(value string) error {
+	valArr := strings.SplitN(value, ":", 2)
+	if len(valArr) < 2 {
+		return errors.New("need both src and target, colon-delimited (ex. /a:/b).")
+	}
+	*r = append(*r, urlRewrite{src: valArr[0], target: valArr[1]})
+	return nil
+}
+
+func (r *UrlRewriteMap) Rewrite(path string) string {
+        for _, f := range *r {
+                if f.src == path {
+                        return f.target
+                }
+        }
+        return path
+}

settings_url_map_test.go

@@ -0,0 +1,27 @@
+package main
+
+import (
+	"testing"
+)
+
+func TestUrlRewriteMap(t *testing.T) {
+    var url string;
+
+	rewrites := UrlRewriteMap{}
+
+	err := rewrites.Set("/abc:/123")
+	if err != nil {
+		t.Error("Should not error on /abc:/123")
+	}
+
+	url = "/abc"
+
+    if(rewrites.Rewrite(url) == url) {
+        t.Error("Request url should have been rewritten, wasn't")
+    }
+
+    url = "/wibble"
+    if(rewrites.Rewrite(url) != url) {
+        t.Error("Request url should not have been rewritten, was")
+    }
+}

settings_url_regexp.go

@@ -0,0 +1,30 @@
+package main
+
+import (
+	"net/http"
+	"regexp"
+)
+
+type HTTPUrlRegexp struct {
+	regexp *regexp.Regexp
+}
+
+func (r *HTTPUrlRegexp) String() string {
+	if r.regexp == nil {
+		return ""
+	}
+	return r.regexp.String()
+}
+
+func (r *HTTPUrlRegexp) Set(value string) error {
+	regexp, err := regexp.Compile(value)
+	r.regexp = regexp
+	return err
+}
+
+func (r *HTTPUrlRegexp) Good(req *http.Request) bool {
+	if r.regexp == nil {
+		return true
+	}
+	return r.regexp.MatchString(req.Host + req.URL.String())
+}

settings_url_regexp_test.go

@@ -0,0 +1,26 @@
+package main
+
+import (
+	"testing"
+	"net/http"
+	"net/url"
+)
+
+func TestHTTPUrlRegexp(t *testing.T) {
+	filter := HTTPUrlRegexp{}
+
+	filter.Set("^www.google.com/admin/")
+
+	req := http.Request{}
+	req.Host = "www.google.com"
+	var err error
+	req.URL, err = url.Parse("/admin/testpage1")
+	if(!filter.Good(&req) || err != nil) {
+		t.Error("Request should pass filters")
+	}
+
+	req.URL, err = url.Parse("/user/testpage2")
+	if(filter.Good(&req) || err != nil) {
+		t.Error("Request should not pass filters")
+	}
+}

test_input.go

@@ -0,0 +1,35 @@
+package main
+
+type TestInput struct {
+	data chan []byte
+}
+
+func NewTestInput() (i *TestInput) {
+	i = new(TestInput)
+	i.data = make(chan []byte, 100)
+
+	return
+}
+
+func (i *TestInput) Read(data []byte) (int, error) {
+	buf := <-i.data
+	copy(data, buf)
+
+	return len(buf), nil
+}
+
+func (i *TestInput) EmitGET() {
+	i.data <- []byte("GET / HTTP/1.1\r\n\r\n")
+}
+
+func (i *TestInput) EmitPOST() {
+	i.data <- []byte("POST /pub/WWW/ HTTP/1.1\nHost: www.w3.org\r\n\r\na=1&b=2\r\n\r\n")
+}
+
+func (i *TestInput) EmitOPTIONS() {
+	i.data <- []byte("OPTIONS / HTTP/1.1\nHost: www.w3.org\r\n\r\n")
+}
+
+func (i *TestInput) String() string {
+	return "Test Input"
+}

test_output.go

@@ -0,0 +1,24 @@
+package main
+
+type writeCallback func(data []byte)
+
+type TestOutput struct {
+	cb writeCallback
+}
+
+func NewTestOutput(cb writeCallback) (i *TestOutput) {
+	i = new(TestOutput)
+	i.cb = cb
+
+	return
+}
+
+func (i *TestOutput) Write(data []byte) (int, error) {
+	i.cb(data)
+
+	return len(data), nil
+}
+
+func (i *TestOutput) String() string {
+	return "Test Input"
+}