From 9a0d9166c310717d7b86e6c58800e63c197b75fb Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Mon, 6 Jul 2015 21:08:34 +0500 Subject: [PATCH] Yay! http.Client replaced by own simple http client --- Makefile | 2 +- elasticsearch.go | 86 +++++++++++------------ http_client.go | 2 + http_client_test.go | 52 ++++++++------ input_raw_test.go | 14 +--- output_http.go | 108 +++-------------------------- proto/proto.go | 11 +++ raw_socket_listener/tcp_message.go | 30 +------- 8 files changed, 102 insertions(+), 203 deletions(-) diff --git a/Makefile b/Makefile index 07d3f7c..c75b5f2 100644 --- a/Makefile +++ b/Makefile @@ -14,7 +14,7 @@ dbuild: docker build -t gor . dtest: - docker run -v `pwd`:$(SOURCE_PATH) -t -i --env GORACE="halt_on_error=1" gor go test ./... $(ARGS) -race -v + docker run -v `pwd`:$(SOURCE_PATH) -t -i --env GORACE="halt_on_error=1" gor go test ./... $(ARGS) -race -v -timeout 15s dfmt: docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go fmt diff --git a/elasticsearch.go b/elasticsearch.go index 2ef8ea4..1819c9d 100644 --- a/elasticsearch.go +++ b/elasticsearch.go @@ -4,8 +4,8 @@ import ( "encoding/json" "github.com/buger/elastigo/api" "github.com/buger/elastigo/core" + "github.com/buger/gor/proto" "log" - "net/http" "regexp" "time" ) @@ -26,26 +26,26 @@ type ESPlugin struct { } 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"` + ReqUrl []byte `json:"Req_URL"` + ReqMethod []byte `json:"Req_Method"` + ReqUserAgent []byte `json:"Req_User-Agent"` + ReqAcceptLanguage []byte `json:"Req_Accept-Language,omitempty"` + ReqAccept []byte `json:"Req_Accept,omitempty"` + ReqAcceptEncoding []byte `json:"Req_Accept-Encoding,omitempty"` + ReqIfModifiedSince []byte `json:"Req_If-Modified-Since,omitempty"` + ReqConnection []byte `json:"Req_Connection,omitempty"` + ReqCookies []byte `json:"Req_Cookies,omitempty"` + RespStatus []byte `json:"Resp_Status"` + RespStatusCode []byte `json:"Resp_Status-Code"` + RespProto []byte `json:"Resp_Proto,omitempty"` + RespContentLength []byte `json:"Resp_Content-Length,omitempty"` + RespContentType []byte `json:"Resp_Content-Type,omitempty"` + RespTransferEncoding []byte `json:"Resp_Transfer-Encoding,omitempty"` + RespContentEncoding []byte `json:"Resp_Content-Encoding,omitempty"` + RespExpires []byte `json:"Resp_Expires,omitempty"` + RespCacheControl []byte `json:"Resp_Cache-Control,omitempty"` + RespVary []byte `json:"Resp_Vary,omitempty"` + RespSetCookie []byte `json:"Resp_Set-Cookie,omitempty"` Rtt int64 `json:"RTT"` Timestamp time.Time } @@ -111,8 +111,8 @@ func (p *ESPlugin) RttDurationToMs(d time.Duration) int64 { return int64(fl) } -func (p *ESPlugin) ResponseAnalyze(req *http.Request, resp *http.Response, start, stop time.Time) { - if resp == nil { +func (p *ESPlugin) ResponseAnalyze(req, resp []byte, start, stop time.Time) { + if len(resp) == 0 { // nil http response - skipped elasticsearch export for this request return } @@ -120,26 +120,26 @@ func (p *ESPlugin) ResponseAnalyze(req *http.Request, resp *http.Response, start 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"), + ReqUrl: proto.Path(req), + ReqMethod: proto.Method(req), + ReqUserAgent: proto.GetHeader(req, "User-Agent"), + ReqAcceptLanguage: proto.GetHeader(req, "Accept-Language"), + ReqAccept: proto.GetHeader(req, "Accept"), + ReqAcceptEncoding: proto.GetHeader(req, "Accept-Encoding"), + ReqIfModifiedSince: proto.GetHeader(req, "If-Modified-Since"), + ReqConnection: proto.GetHeader(req, "Connection"), + ReqCookies: proto.GetHeader(req, "Cookie"), + RespStatus: proto.Status(resp), + RespStatusCode: proto.Status(resp), + RespProto: proto.Method(resp), + RespContentLength: proto.GetHeader(resp, "Content-Length"), + RespContentType: proto.GetHeader(resp, "Content-Type"), + RespTransferEncoding: proto.GetHeader(resp, "Transfer-Encoding"), + RespContentEncoding: proto.GetHeader(resp, "Content-Encoding"), + RespExpires: proto.GetHeader(resp, "Expires"), + RespCacheControl: proto.GetHeader(resp, "Cache-Control"), + RespVary: proto.GetHeader(resp, "Vary"), + RespSetCookie: proto.GetHeader(resp, "Set-Cookie"), Rtt: rtt, Timestamp: t, } diff --git a/http_client.go b/http_client.go index 4f78e9d..e5f859d 100644 --- a/http_client.go +++ b/http_client.go @@ -93,6 +93,8 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) { c.conn.SetWriteDeadline(timeout) + data = proto.SetHeader(data, []byte("Host"), []byte(c.baseURL.Host)) + if c.config.Debug { Debug("Sending:", string(data)) } diff --git a/http_client_test.go b/http_client_test.go index 25f8b4c..45a3ba9 100644 --- a/http_client_test.go +++ b/http_client_test.go @@ -36,13 +36,18 @@ func TestHTTPClientURLPort(t *testing.T) { func TestHTTPClientSend(t *testing.T) { wg := new(sync.WaitGroup) - GET_payload := []byte("GET / HTTP/1.1\r\n\r\n") + payload := func(reqType string) []byte { + switch reqType { + case "GET": + return []byte("GET / HTTP/1.1\r\n\r\n") + case "POST": + return []byte("POST /post HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") + case "POST_CHUNKED": + return []byte("POST / 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") + } - // Post request terminates by reading Content-Length without double CRLF - POST_payload := []byte("POST /post HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - - // Chunked requests terminated with double CRLF - POST_CHUNKED_payload := []byte("POST / 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") + return []byte("") + } server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -65,13 +70,13 @@ func TestHTTPClientSend(t *testing.T) { wg.Done() })) - client := NewHTTPClient(server.URL, &HTTPClientConfig{}) + client := NewHTTPClient(server.URL, &HTTPClientConfig{Debug: false}) wg.Add(4) - client.Send(POST_payload) - client.Send(GET_payload) - client.Send(POST_CHUNKED_payload) - client.Send(POST_payload) + client.Send(payload("POST")) + client.Send(payload("GET")) + client.Send(payload("POST_CHUNKED")) + client.Send(payload("POST")) wg.Wait() } @@ -79,13 +84,18 @@ func TestHTTPClientSend(t *testing.T) { func TestHTTPClientHTTPSSend(t *testing.T) { wg := new(sync.WaitGroup) - GET_payload := []byte("GET / HTTP/1.1\r\n\r\n") + payload := func(reqType string) []byte { + switch reqType { + case "GET": + return []byte("GET / HTTP/1.1\r\n\r\n") + case "POST": + return []byte("POST /post HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") + case "POST_CHUNKED": + return []byte("POST / 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") + } - // Post request terminates by reading Content-Length without double CRLF - POST_payload := []byte("POST /post HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2") - - // Chunked requests terminated with double CRLF - POST_CHUNKED_payload := []byte("POST / 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") + return []byte("") + } server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -111,10 +121,10 @@ func TestHTTPClientHTTPSSend(t *testing.T) { client := NewHTTPClient(server.URL, &HTTPClientConfig{}) wg.Add(4) - client.Send(GET_payload) - client.Send(POST_payload) - client.Send(POST_CHUNKED_payload) - client.Send(POST_payload) + client.Send(payload("POST")) + client.Send(payload("GET")) + client.Send(payload("POST_CHUNKED")) + client.Send(payload("POST")) wg.Wait() } diff --git a/input_raw_test.go b/input_raw_test.go index 60bcf49..477eff8 100644 --- a/input_raw_test.go +++ b/input_raw_test.go @@ -118,14 +118,6 @@ func TestInputRAWChunkedEncoding(t *testing.T) { input := NewRAWInput(origin_address) - // We will use it to get content of raw HTTP request - test_output := NewTestOutput(func(data []byte) { - if strings.Contains(string(data), "Transfer-Encoding: chunked") { - t.Error("Should not contain chunked header") - } - wg.Done() - }) - listener := startHTTP(func(req *http.Request) { defer req.Body.Close() body, _ := ioutil.ReadAll(req.Body) @@ -139,14 +131,14 @@ func TestInputRAWChunkedEncoding(t *testing.T) { }) replay_address := listener.Addr().String() - http_output := NewHTTPOutput(replay_address, &HTTPOutputConfig{}) + http_output := NewHTTPOutput(replay_address, &HTTPOutputConfig{Debug: true}) Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{test_output, http_output} + Plugins.Outputs = []io.Writer{http_output} go Start(quit) - wg.Add(3) + wg.Add(2) curl := exec.Command("curl", "http://"+origin_address, "--header", "Transfer-Encoding: chunked", "--data-binary", "@README.md") err := curl.Run() diff --git a/output_http.go b/output_http.go index c1b42f9..9b77e37 100644 --- a/output_http.go +++ b/output_http.go @@ -1,59 +1,12 @@ package main import ( - "bufio" - "bytes" - "crypto/tls" "io" - "io/ioutil" "log" - "net/http" - "net/url" - "strings" "sync/atomic" "time" ) -type RedirectNotAllowed struct{} - -func (e *RedirectNotAllowed) Error() string { - return "Redirects not allowed" -} - -// customCheckRedirect disables redirects https://github.com/buger/gor/pull/15 -func (o *HTTPOutput) customCheckRedirect(req *http.Request, via []*http.Request) error { - if len(via) >= o.config.redirectLimit { - return new(RedirectNotAllowed) - } - return nil -} - -// ParseRequest in []byte returns a http request or an error -func ParseRequest(data []byte) (request *http.Request, err error) { - var body []byte - - buf := bytes.NewBuffer(data) - reader := bufio.NewReader(buf) - - // ReadRequest does not read POST bodies, we have to do it by ourseves - request, err = http.ReadRequest(reader) - - if err != nil { - return - } - - if request.Method == "POST" { - body, _ = ioutil.ReadAll(reader) - - bodyBuf := bytes.NewBuffer(body) - - request.Body = ioutil.NopCloser(bodyBuf) - request.ContentLength = int64(bodyBuf.Len()) - } - - return -} - const InitialDynamicWorkers = 10 type HTTPOutputConfig struct { @@ -63,6 +16,8 @@ type HTTPOutputConfig struct { workers int elasticSearch string + + Debug bool } type HTTPOutput struct { @@ -88,10 +43,6 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) io.Writer { o := new(HTTPOutput) - if !strings.HasPrefix(address, "http") { - address = "http://" + address - } - o.address = address o.config = config @@ -134,14 +85,10 @@ func (o *HTTPOutput) WorkerMaster() { } func (o *HTTPOutput) Worker() { - tr := &http.Transport{ - TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, - } - - client := &http.Client{ - Transport: tr, - CheckRedirect: o.customCheckRedirect, - } + client := NewHTTPClient(o.address, &HTTPClientConfig{ + FollowRedirects: o.config.redirectLimit, + Debug: o.config.Debug, + }) death_count := 0 @@ -194,55 +141,20 @@ func (o *HTTPOutput) Write(data []byte) (n int, err error) { 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 - } - - // Change HOST of original request - URL := o.address + request.URL.Path + "?" + request.URL.RawQuery - - request.RequestURI = "" - request.URL, _ = url.ParseRequestURI(URL) - +func (o *HTTPOutput) sendRequest(client *HTTPClient, request []byte) { start := time.Now() - resp, err := client.Do(request) + resp, err := client.Send(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 { + if err != nil { log.Println("Request error:", err) } if o.elasticSearch != nil { - o.elasticSearch.ResponseAnalyze(request, resp, start, stop) + 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 } diff --git a/proto/proto.go b/proto/proto.go index f38645a..07622db 100644 --- a/proto/proto.go +++ b/proto/proto.go @@ -39,6 +39,12 @@ func Header(payload []byte, name []byte) (value []byte, headerStart, valueStart, return } +func GetHeader(payload []byte, name string) []byte { + val, _, _, _ := Header(payload, []byte(name)) + + return val +} + func SetHeader(payload, name, value []byte) []byte { _, hs, vs, he := Header(payload, name) @@ -84,4 +90,9 @@ func Method(payload []byte) []byte { end := bytes.IndexByte(payload, ' ') return payload[:end] +} + +// Status in response have same position as Path in request +func Status(payload []byte) []byte { + return Path(payload) } \ No newline at end of file diff --git a/raw_socket_listener/tcp_message.go b/raw_socket_listener/tcp_message.go index 9e45a85..69c4b93 100644 --- a/raw_socket_listener/tcp_message.go +++ b/raw_socket_listener/tcp_message.go @@ -4,10 +4,6 @@ import ( "log" "sort" "time" - "bytes" - "net/http/httputil" - "bufio" - "io/ioutil" ) const MSG_EXPIRE = 2000 * time.Millisecond @@ -75,30 +71,6 @@ func (t *TCPMessage) Timeout() { } } -var bTransferEncodingChunked = []byte("Transfer-Encoding: chunked\r\n") -var b2xCRLF = []byte("\r\n\r\n") - -// Norimalize requests with `Transfer-Encoding: chunked` header, because they have special body format -func fixChunkedEncoding(data []byte) []byte { - if bytes.Equal(data[0:4], bPOST) { - body_idx := bytes.Index(data, b2xCRLF) - chunked_header_idx := bytes.Index(data[:body_idx], bTransferEncodingChunked) - - if chunked_header_idx != -1 { - buf := bytes.NewBuffer(data[body_idx+4:]) - // Adding 4 bytes to skip 2xCLRF - bodyReader := bufio.NewReader(buf) - body, _ := ioutil.ReadAll(httputil.NewChunkedReader(bodyReader)) - - // Exclude Transfer-Encoding header and append new body - return append(append(append(data[:chunked_header_idx], - data[chunked_header_idx+len(bTransferEncodingChunked):body_idx]...), b2xCRLF...), body...) - } - } - - return data -} - // Bytes sorts packets in right orders and return message content func (t *TCPMessage) Bytes() (output []byte) { sort.Sort(BySeq(t.packets)) @@ -107,7 +79,7 @@ func (t *TCPMessage) Bytes() (output []byte) { output = append(output, v.Data...) } - return fixChunkedEncoding(output) + return output } // AddPacket to the message and ensure packet uniqueness