diff --git a/output_http.go b/output_http.go index e2a6c96..eb3e925 100644 --- a/output_http.go +++ b/output_http.go @@ -7,8 +7,9 @@ import ( "time" ) -const InitialDynamicWorkers = 10 +const initialDynamicWorkers = 10 +// HTTPOutputConfig struct for holding http output configuration type HTTPOutputConfig struct { redirectLimit int @@ -20,6 +21,9 @@ type HTTPOutputConfig struct { Debug bool } +// HTTPOutput plugin manage pool of workers which send request to replayed server +// By default workers pool is dynamic and starts with 10 workers +// You can specify fixed number of workers using `--output-http-workers` type HTTPOutput struct { // Keep this as first element of struct because it guarantees 64bit // alignment. atomic.* functions crash on 32bit machines if operand is not @@ -39,6 +43,8 @@ type HTTPOutput struct { elasticSearch *ESPlugin } +// NewHTTPOutput constructor for HTTPOutput +// Initialize workers func NewHTTPOutput(address string, config *HTTPOutputConfig) io.Writer { o := new(HTTPOutput) @@ -55,7 +61,7 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) io.Writer { // Initial workers count if o.config.workers == 0 { - o.needWorker <- InitialDynamicWorkers + o.needWorker <- initialDynamicWorkers } else { o.needWorker <- o.config.workers } @@ -65,16 +71,16 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) io.Writer { o.elasticSearch.Init(o.config.elasticSearch) } - go o.WorkerMaster() + go o.workerMaster() return o } -func (o *HTTPOutput) WorkerMaster() { +func (o *HTTPOutput) workerMaster() { for { - new_workers := <-o.needWorker - for i := 0; i < new_workers; i++ { - go o.Worker() + newWorkers := <-o.needWorker + for i := 0; i < newWorkers; i++ { + go o.startWorker() } // Disable dynamic scaling if workers poll fixed size @@ -84,13 +90,13 @@ func (o *HTTPOutput) WorkerMaster() { } } -func (o *HTTPOutput) Worker() { +func (o *HTTPOutput) startWorker() { client := NewHTTPClient(o.address, &HTTPClientConfig{ FollowRedirects: o.config.redirectLimit, Debug: o.config.Debug, }) - death_count := 0 + deathCount := 0 atomic.AddInt64(&o.activeWorkers, 1) @@ -98,19 +104,19 @@ func (o *HTTPOutput) Worker() { select { case data := <-o.queue: o.sendRequest(client, data) - death_count = 0 + deathCount = 0 case <-time.After(time.Millisecond * 100): // When dynamic scaling enabled workers die after 2s of inactivity if o.config.workers == 0 { - death_count += 1 + deathCount++ } else { continue } - if death_count > 20 { + if deathCount > 20 { workersCount := atomic.LoadInt64(&o.activeWorkers) - // At least 1 worker should be alive + // At least 1 startWorker should be alive if workersCount != 1 { atomic.AddInt64(&o.activeWorkers, -1) return diff --git a/output_http_test.go b/output_http_test.go index 7cf8c24..16f8422 100644 --- a/output_http_test.go +++ b/output_http_test.go @@ -86,10 +86,10 @@ func TestOutputHTTPSSL(t *testing.T) { })) input := NewTestInput() - http_output := NewHTTPOutput(server.URL, &HTTPOutputConfig{}) + output := NewHTTPOutput(server.URL, &HTTPOutputConfig{}) Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{http_output} + Plugins.Outputs = []io.Writer{output} go Start(quit) diff --git a/output_tcp.go b/output_tcp.go index f6f0e9d..dd7cdc4 100644 --- a/output_tcp.go +++ b/output_tcp.go @@ -9,6 +9,9 @@ import ( "time" ) +// TCPOutput used for sending raw tcp payloads +// Currently used for internal communication between listener and replay server +// Can be used for transfering binary payloads like protocol buffers type TCPOutput struct { address string limit int @@ -16,6 +19,8 @@ type TCPOutput struct { bufStats *GorStat } +// NewTCPOutput constructor for TCPOutput +// Initialize 10 workers which hold keep-alive connection func NewTCPOutput(address string) io.Writer { o := new(TCPOutput) diff --git a/plugins.go b/plugins.go index 88134ef..cb681ae 100644 --- a/plugins.go +++ b/plugins.go @@ -6,23 +6,25 @@ import ( "strings" ) +// InOutPlugins struct for holding references to plugins type InOutPlugins struct { Inputs []io.Reader Outputs []io.Writer } -type ReaderOrWriter interface{} - -var Plugins *InOutPlugins = new(InOutPlugins) +// Plugins holds all the plugin objects +var Plugins *InOutPlugins +// extractLimitOptions detects if plugin get called with limiter support +// Returns address and limit func extractLimitOptions(options string) (string, string) { split := strings.Split(options, "|") if len(split) > 1 { return split[0], split[1] - } else { - return split[0], "" } + + return split[0], "" } // Automatically detects type of plugin and initialize it @@ -45,23 +47,24 @@ func registerPlugin(constructor interface{}, options ...interface{}) { // Calling our constructor with list of given options plugin := vc.Call(vo)[0].Interface() - plugin_wrapper := plugin + pluginWrapper := plugin if limit != "" { - plugin_wrapper = NewLimiter(plugin, limit) + pluginWrapper = NewLimiter(plugin, limit) } else { - plugin_wrapper = plugin + pluginWrapper = plugin } if _, ok := plugin.(io.Reader); ok { - Plugins.Inputs = append(Plugins.Inputs, plugin_wrapper.(io.Reader)) + Plugins.Inputs = append(Plugins.Inputs, pluginWrapper.(io.Reader)) } if _, ok := plugin.(io.Writer); ok { - Plugins.Outputs = append(Plugins.Outputs, plugin_wrapper.(io.Writer)) + Plugins.Outputs = append(Plugins.Outputs, pluginWrapper.(io.Writer)) } } +// InitPlugins specify and initialize all available plugins func InitPlugins() { for _, options := range Settings.inputDummy { registerPlugin(NewDummyInput, options) diff --git a/settings.go b/settings.go index 8c345e7..a08a9c1 100644 --- a/settings.go +++ b/settings.go @@ -8,21 +8,24 @@ import ( ) const ( + // VERSION specifies Gor current version VERSION = "0.9.8" ) -// Allows to specify multiple flags with same name and collects all values to array +// MultiOption allows to specify multiple flags with same name and collects all values into array type MultiOption []string func (h *MultiOption) String() string { return fmt.Sprint(*h) } +// Set gets called multiple times for each flag with same name func (h *MultiOption) Set(value string) error { *h = append(*h, value) return nil } +// AppSettings is the struct of main configuration type AppSettings struct { verbose bool debug bool @@ -49,7 +52,8 @@ type AppSettings struct { modifierConfig HTTPModifierConfig } -var Settings AppSettings = AppSettings{} +// Settings holds Gor configuration +var Settings 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: leonsbox@gmail.com\nCurrent Version: %s\n\n", VERSION) @@ -113,6 +117,7 @@ func init() { flag.Var(&Settings.modifierConfig.paramHashFilters, "http-param-limiter", "Takes a fraction of requests, consistently taking or rejecting a request based on the FNV32-1A hash of a specific GET param:\n\t gor --input-raw :8080 --output-http staging.com --http-param-limiter user_id:25%") } +// Debug gets called only if --verbose flag specified func Debug(args ...interface{}) { if Settings.verbose { fmt.Print("[DEBUG] ") diff --git a/test_input.go b/test_input.go index 8f62344..a9a359a 100644 --- a/test_input.go +++ b/test_input.go @@ -5,10 +5,12 @@ import ( "encoding/base64" ) +// TestInput used for testing purpose, it allows emitting requests on demand type TestInput struct { data chan []byte } +// NewTestInput constructor for TestInput func NewTestInput() (i *TestInput) { i = new(TestInput) i.data = make(chan []byte, 100) @@ -23,18 +25,22 @@ func (i *TestInput) Read(data []byte) (int, error) { return len(buf), nil } +// EmitGET emits GET request without headers func (i *TestInput) EmitGET() { i.data <- []byte("GET / HTTP/1.1\r\n\r\n") } +// EmitPOST emits POST request with Content-Length func (i *TestInput) EmitPOST() { i.data <- []byte("POST /pub/WWW/ HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") } +// EmitChunkedPOST emits POST request with `Transfer-Encoding: chunked` and chunked body func (i *TestInput) EmitChunkedPOST() { i.data <- []byte("POST /pub/WWW/ HTTP/1.1\r\nHost: www.w3.org\r\nTransfer-Encoding: chunked\r\n\r\n4\r\nWiki\r\n5\r\npedia\r\ne\r\n in\r\n\r\nchunks.\r\n0\r\n\r\n") } +// EmitLargePOST emits POST request with large payload (5mb) func (i *TestInput) EmitLargePOST() { size := 5 * 1024 * 1024 // 5 MB rb := make([]byte, size) @@ -45,6 +51,7 @@ func (i *TestInput) EmitLargePOST() { i.data <- []byte("POST / HTTP/1.1\nHost: www.w3.org\nContent-Length:5242880\r\n\r\n" + rs) } +// EmitOPTIONS emits OPTIONS request, similar to GET func (i *TestInput) EmitOPTIONS() { i.data <- []byte("OPTIONS / HTTP/1.1\nHost: www.w3.org\r\n\r\n") } diff --git a/test_output.go b/test_output.go index 1d34da0..e76fbb6 100644 --- a/test_output.go +++ b/test_output.go @@ -2,10 +2,12 @@ package main type writeCallback func(data []byte) +// TestOutput used in testing to intercept any output into callback type TestOutput struct { cb writeCallback } +// NewTestOutput constructor for TestOutput, accepts callback which get called on each incoming Write func NewTestOutput(cb writeCallback) (i *TestOutput) { i = new(TestOutput) i.cb = cb