Yay! http.Client replaced by own simple http client

2024-04-21 12:32:02 +00:00 · 2015-07-06 21:08:34 +05:00
parent bc48e1ad49
commit 9a0d9166c3
8 changed files with 102 additions and 203 deletions
@@ -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
@@ -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,
 	}
@@ -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))
 	}
@@ -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()
 }
@@ -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()
@@ -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
 }
@@ -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)
 }
@@ -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