github/goreplay
4
0
Fork 0
mirror of https://github.com/buger/goreplay.git synced 2024-04-21 12:32:02 +00:00
Code Issues Releases Wiki Activity

Remove elastic search functionality

Browse Source
This commit is contained in:
Leonid Bugaev
2014-07-22 22:50:04 +04:00
parent bae1a84f40
commit cd27667246
6 changed files with 4 additions and 248 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ELASTICSEARCH.md
-65
View File
@@ -1,65 +0,0 @@
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)
README.md
-12
View File
@@ -130,15 +130,6 @@ 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.
@@ -175,9 +166,6 @@ 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'
elasticsearch/elasticsearch.go
-153
View File
@@ -1,153 +0,0 @@
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
}
output_http.go
+1 -15
View File
@@ -3,14 +3,12 @@ package main
import (
"bufio"
"bytes"
es "github.com/buger/gor/elasticsearch"
"io"
"log"
"net/http"
"net/url"
"strconv"
"strings"
"time"
)
type RedirectNotAllowed struct{}
@@ -50,12 +48,10 @@ 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, elasticSearchAddr string) io.Writer {
func NewHTTPOutput(options string, headers HTTPHeaders, methods HTTPMethods, urlRegexp HTTPUrlRegexp, headerFilters HTTPHeaderFilters, headerHashFilters HTTPHeaderHashFilters) io.Writer {
o := new(HTTPOutput)
optionsArr := strings.Split(options, "|")
@@ -76,11 +72,6 @@ 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])
}
@@ -143,9 +134,7 @@ func (o *HTTPOutput) sendRequest(client *http.Client, data []byte) {
request.Header.Set(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 {
@@ -160,9 +149,6 @@ 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 (o *HTTPOutput) String() string {
output_http_test.go
+2 -2
View File
@@ -42,7 +42,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}
@@ -75,7 +75,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}
plugins.go
+1 -1
View File
@@ -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, Settings.outputHTTPElasticSearch))
Plugins.Outputs = append(Plugins.Outputs, NewHTTPOutput(options, Settings.outputHTTPHeaders, Settings.outputHTTPMethods, Settings.outputHTTPUrlRegexp, Settings.outputHTTPHeaderFilters, Settings.outputHTTPHeaderHashFilters))
}
}