Revert "Remove elastic search functionality"

This reverts commit cd27667246.
2024-04-21 12:32:02 +00:00 · 2014-09-04 22:48:08 +04:00
parent 6f6c6aba95
commit 674d6036d1
6 changed files with 248 additions and 4 deletions
@@ -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)
@@ -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'

@@ -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
+}
@@ -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) {
@@ -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}
@@ -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))
 	}
 }