diff --git a/ELASTICSEARCH.md b/ELASTICSEARCH.md new file mode 100644 index 0000000..d15b6ec --- /dev/null +++ b/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) diff --git a/README.md b/README.md index 293eddd..60965aa 100644 --- a/README.md +++ b/README.md @@ -129,6 +129,15 @@ gor --input-raw :80 --output-http "http://user:pass@staging .com" Note: This will overwrite any Authorization headers in the original request. +## 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. @@ -165,6 +174,9 @@ https://github.com/buger/gor/releases # 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' diff --git a/elasticsearch/elasticsearch.go b/elasticsearch/elasticsearch.go new file mode 100644 index 0000000..8e60c99 --- /dev/null +++ b/elasticsearch/elasticsearch.go @@ -0,0 +1,153 @@ +package elasticsearch + +import ( + "encoding/json" + "github.com/mattbaird/elastigo/api" + "github.com/mattbaird/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 +} diff --git a/output_http.go b/output_http.go index fc7aa75..38e4fc8 100644 --- a/output_http.go +++ b/output_http.go @@ -3,12 +3,14 @@ package main import ( "bufio" "bytes" + es "github.com/buger/gor/elasticsearch" "io" "log" "net/http" "net/url" "strconv" "strings" + "time" ) type RedirectNotAllowed struct{} @@ -48,10 +50,12 @@ type HTTPOutput struct { headers HTTPHeaders methods HTTPMethods + elasticSearch *es.ESPlugin + bufStats *GorStat } -func NewHTTPOutput(options string, headers HTTPHeaders, methods HTTPMethods, urlRegexp HTTPUrlRegexp, headerFilters HTTPHeaderFilters, headerHashFilters HTTPHeaderHashFilters) io.Writer { +func NewHTTPOutput(options string, headers HTTPHeaders, methods HTTPMethods, urlRegexp HTTPUrlRegexp, headerFilters HTTPHeaderFilters, headerHashFilters HTTPHeaderHashFilters, elasticSearchAddr string) io.Writer { o := new(HTTPOutput) optionsArr := strings.Split(options, "|") @@ -72,6 +76,11 @@ func NewHTTPOutput(options string, headers HTTPHeaders, methods HTTPMethods, url o.buf = make(chan []byte, 100) o.bufStats = NewGorStat("output_http") + if elasticSearchAddr != "" { + o.elasticSearch = new(es.ESPlugin) + o.elasticSearch.Init(elasticSearchAddr) + } + if len(optionsArr) > 1 { o.limit, _ = strconv.Atoi(optionsArr[1]) } @@ -134,7 +143,9 @@ func (o *HTTPOutput) sendRequest(client *http.Client, data []byte) { 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 { @@ -149,6 +160,9 @@ func (o *HTTPOutput) sendRequest(client *http.Client, data []byte) { 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) { diff --git a/output_http_test.go b/output_http_test.go index 42c8ed4..a9c1dff 100644 --- a/output_http_test.go +++ b/output_http_test.go @@ -63,7 +63,7 @@ func TestHTTPOutput(t *testing.T) { wg.Done() }) - output := NewHTTPOutput(listener.Addr().String(), headers, methods, HTTPUrlRegexp{}, HTTPHeaderFilters{}, HTTPHeaderHashFilters{}) + output := NewHTTPOutput(listener.Addr().String(), headers, methods, HTTPUrlRegexp{}, HTTPHeaderFilters{}, HTTPHeaderHashFilters{}, "") Plugins.Inputs = []io.Reader{input} Plugins.Outputs = []io.Writer{output} @@ -96,7 +96,7 @@ func BenchmarkHTTPOutput(b *testing.B) { wg.Done() }) - output := NewHTTPOutput(listener.Addr().String(), headers, methods, HTTPUrlRegexp{}, HTTPHeaderFilters{}, HTTPHeaderHashFilters{}) + output := NewHTTPOutput(listener.Addr().String(), headers, methods, HTTPUrlRegexp{}, HTTPHeaderFilters{}, HTTPHeaderHashFilters{}, "") Plugins.Inputs = []io.Reader{input} Plugins.Outputs = []io.Writer{output} diff --git a/plugins.go b/plugins.go index 62791af..805faaf 100644 --- a/plugins.go +++ b/plugins.go @@ -41,6 +41,6 @@ func InitPlugins() { } for _, options := range Settings.outputHTTP { - Plugins.Outputs = append(Plugins.Outputs, NewHTTPOutput(options, Settings.outputHTTPHeaders, Settings.outputHTTPMethods, Settings.outputHTTPUrlRegexp, Settings.outputHTTPHeaderFilters, Settings.outputHTTPHeaderHashFilters)) + Plugins.Outputs = append(Plugins.Outputs, NewHTTPOutput(options, Settings.outputHTTPHeaders, Settings.outputHTTPMethods, Settings.outputHTTPUrlRegexp, Settings.outputHTTPHeaderFilters, Settings.outputHTTPHeaderHashFilters, Settings.outputHTTPElasticSearch)) } }